]> git.sesse.net Git - casparcg/blob - core/consumer/output.cpp
2.1.0: -graph: Refactored: Safer and faster.
[casparcg] / core / consumer / output.cpp
1 /*\r
2 * Copyright (c) 2011 Sveriges Television AB <info@casparcg.com>\r
3 *\r
4 * This file is part of CasparCG (www.casparcg.com).\r
5 *\r
6 * CasparCG is free software: you can redistribute it and/or modify\r
7 * it under the terms of the GNU General Public License as published by\r
8 * the Free Software Foundation, either version 3 of the License, or\r
9 * (at your option) any later version.\r
10 *\r
11 * CasparCG is distributed in the hope that it will be useful,\r
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of\r
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the\r
14 * GNU General Public License for more details.\r
15 *\r
16 * You should have received a copy of the GNU General Public License\r
17 * along with CasparCG. If not, see <http://www.gnu.org/licenses/>.\r
18 *\r
19 * Author: Robert Nagy, ronag89@gmail.com\r
20 */\r
21 \r
22 #include "../StdAfx.h"\r
23 \r
24 #ifdef _MSC_VER\r
25 #pragma warning (disable : 4244)\r
26 #endif\r
27 \r
28 #include "output.h"\r
29 \r
30 #include "../video_format.h"\r
31 #include "../mixer/gpu/ogl_device.h"\r
32 #include "../mixer/read_frame.h"\r
33 \r
34 #include <common/concurrency/executor.h>\r
35 #include <common/utility/assert.h>\r
36 #include <common/utility/timer.h>\r
37 #include <common/memory/memshfl.h>\r
38 #include <common/env.h>\r
39 \r
40 #include <boost/circular_buffer.hpp>\r
41 #include <boost/timer.hpp>\r
42 #include <boost/range/algorithm.hpp>\r
43 #include <boost/range/adaptors.hpp>\r
44 #include <boost/property_tree/ptree.hpp>\r
45 \r
46 namespace caspar { namespace core {\r
47         \r
48 struct output::implementation\r
49 {               \r
50         const int                                                                               channel_index_;\r
51         const safe_ptr<diagnostics::graph>                              graph_;\r
52         boost::timer                                                                    consume_timer_;\r
53 \r
54         video_format_desc                                                               format_desc_;\r
55 \r
56         std::map<int, safe_ptr<frame_consumer>>                 consumers_;\r
57         \r
58         high_prec_timer                                                                 sync_timer_;\r
59 \r
60         boost::circular_buffer<safe_ptr<read_frame>>    frames_;\r
61 \r
62         executor                                                                                executor_;\r
63                 \r
64 public:\r
65         implementation(const safe_ptr<diagnostics::graph>& graph, const video_format_desc& format_desc, int channel_index) \r
66                 : channel_index_(channel_index)\r
67                 , graph_(graph)\r
68                 , format_desc_(format_desc)\r
69                 , executor_(L"output")\r
70         {\r
71                 graph_->set_color("consume-time", diagnostics::color(1.0f, 0.4f, 0.0f));\r
72         }       \r
73         \r
74         void add(int index, safe_ptr<frame_consumer> consumer)\r
75         {               \r
76                 remove(index);\r
77 \r
78                 consumer = create_consumer_cadence_guard(consumer);\r
79                 consumer->initialize(format_desc_, channel_index_);\r
80 \r
81                 executor_.invoke([&]\r
82                 {\r
83                         consumers_.insert(std::make_pair(index, consumer));\r
84                         CASPAR_LOG(info) << print() << L" " << consumer->print() << L" Added.";\r
85                 }, high_priority);\r
86         }\r
87 \r
88         void add(const safe_ptr<frame_consumer>& consumer)\r
89         {\r
90                 add(consumer->index(), consumer);\r
91         }\r
92 \r
93         void remove(int index)\r
94         {               \r
95                 // Destroy  consumer on calling thread:\r
96                 std::shared_ptr<frame_consumer> old_consumer;\r
97 \r
98                 executor_.invoke([&]\r
99                 {\r
100                         auto it = consumers_.find(index);\r
101                         if(it != consumers_.end())\r
102                         {\r
103                                 old_consumer = it->second;\r
104                                 consumers_.erase(it);\r
105                         }\r
106                 }, high_priority);\r
107 \r
108                 if(old_consumer)\r
109                 {\r
110                         auto str = old_consumer->print();\r
111                         old_consumer.reset();\r
112                         CASPAR_LOG(info) << print() << L" " << str << L" Removed.";\r
113                 }\r
114         }\r
115 \r
116         void remove(const safe_ptr<frame_consumer>& consumer)\r
117         {\r
118                 remove(consumer->index());\r
119         }\r
120         \r
121         void set_video_format_desc(const video_format_desc& format_desc)\r
122         {\r
123                 executor_.invoke([&]\r
124                 {\r
125                         auto it = consumers_.begin();\r
126                         while(it != consumers_.end())\r
127                         {                                               \r
128                                 try\r
129                                 {\r
130                                         it->second->initialize(format_desc_, channel_index_);\r
131                                         ++it;\r
132                                 }\r
133                                 catch(...)\r
134                                 {\r
135                                         CASPAR_LOG_CURRENT_EXCEPTION();\r
136                                         CASPAR_LOG(info) << print() << L" " << it->second->print() << L" Removed.";\r
137                                         consumers_.erase(it++);\r
138                                 }\r
139                         }\r
140                         \r
141                         format_desc_ = format_desc;\r
142                         frames_.clear();\r
143                 });\r
144         }\r
145 \r
146         std::pair<int, int> minmax_buffer_depth() const\r
147         {               \r
148                 if(consumers_.empty())\r
149                         return std::make_pair(0, 0);\r
150                 \r
151                 auto buffer_depths = consumers_ | \r
152                                                          boost::adaptors::map_values | // std::function is MSVC workaround\r
153                                                          boost::adaptors::transformed(std::function<int(const safe_ptr<frame_consumer>&)>([](const safe_ptr<frame_consumer>& x){return x->buffer_depth();})); \r
154                 \r
155 \r
156                 return std::make_pair(*boost::range::min_element(buffer_depths), *boost::range::max_element(buffer_depths));\r
157         }\r
158 \r
159         bool has_synchronization_clock() const\r
160         {\r
161                 return boost::range::count_if(consumers_ | boost::adaptors::map_values, [](const safe_ptr<frame_consumer>& x){return x->has_synchronization_clock();}) > 0;\r
162         }\r
163 \r
164         void send(const std::pair<safe_ptr<read_frame>, std::shared_ptr<void>>& packet)\r
165         {\r
166                 executor_.begin_invoke([=]\r
167                 {\r
168                         try\r
169                         {\r
170                                 consume_timer_.restart();\r
171 \r
172                                 auto input_frame = packet.first;\r
173 \r
174                                 if(!has_synchronization_clock())\r
175                                         sync_timer_.tick(1.0/format_desc_.fps);\r
176 \r
177                                 if(input_frame->image_data().size() != format_desc_.size)\r
178                                 {\r
179                                         sync_timer_.tick(1.0/format_desc_.fps);\r
180                                         return;\r
181                                 }\r
182                                         \r
183                                 auto minmax = minmax_buffer_depth();\r
184 \r
185                                 frames_.set_capacity(minmax.second - minmax.first + 1);\r
186                                 frames_.push_back(input_frame);\r
187 \r
188                                 if(!frames_.full())\r
189                                         return;\r
190 \r
191                                 auto it = consumers_.begin();\r
192                                 while(it != consumers_.end())\r
193                                 {\r
194                                         auto consumer   = it->second;\r
195                                         auto frame              = frames_.at(consumer->buffer_depth()-minmax.first);\r
196                                                 \r
197                                         try\r
198                                         {\r
199                                                 if(consumer->send(frame))\r
200                                                         ++it;\r
201                                                 else\r
202                                                 {\r
203                                                         CASPAR_LOG(info) << print() << L" " << it->second->print() << L" Removed.";\r
204                                                         consumers_.erase(it++);\r
205                                                 }\r
206                                         }\r
207                                         catch(...)\r
208                                         {\r
209                                                 CASPAR_LOG_CURRENT_EXCEPTION();\r
210                                                 try\r
211                                                 {\r
212                                                         consumer->initialize(format_desc_, channel_index_);\r
213                                                         if(consumer->send(frame))\r
214                                                                 ++it;\r
215                                                         else\r
216                                                         {\r
217                                                                 CASPAR_LOG(info) << print() << L" " << it->second->print() << L" Removed.";\r
218                                                                 consumers_.erase(it++);\r
219                                                         }\r
220                                                 }\r
221                                                 catch(...)\r
222                                                 {\r
223                                                         CASPAR_LOG_CURRENT_EXCEPTION();\r
224                                                         CASPAR_LOG(error) << "Failed to recover consumer: " << consumer->print() << L". Removing it.";\r
225                                                         consumers_.erase(it++);\r
226                                                 }\r
227                                         }\r
228                                 }\r
229                                                 \r
230                                 graph_->set_value("consume-time", consume_timer_.elapsed()*format_desc_.fps*0.5);\r
231                         }\r
232                         catch(...)\r
233                         {\r
234                                 CASPAR_LOG_CURRENT_EXCEPTION();\r
235                         }\r
236                 });\r
237         }\r
238 \r
239         std::wstring print() const\r
240         {\r
241                 return L"output[" + boost::lexical_cast<std::wstring>(channel_index_) + L"]";\r
242         }\r
243 \r
244         boost::unique_future<boost::property_tree::wptree> info()\r
245         {\r
246                 return std::move(executor_.begin_invoke([&]() -> boost::property_tree::wptree\r
247                 {                       \r
248                         boost::property_tree::wptree info;\r
249                         BOOST_FOREACH(auto& consumer, consumers_)\r
250                         {\r
251                                 info.add_child(L"consumers.consumer", consumer.second->info())\r
252                                         .add(L"index", consumer.first); \r
253                         }\r
254                         return info;\r
255                 }, high_priority));\r
256         }\r
257 };\r
258 \r
259 output::output(const safe_ptr<diagnostics::graph>& graph, const video_format_desc& format_desc, int channel_index) : impl_(new implementation(graph, format_desc, channel_index)){}\r
260 void output::add(int index, const safe_ptr<frame_consumer>& consumer){impl_->add(index, consumer);}\r
261 void output::add(const safe_ptr<frame_consumer>& consumer){impl_->add(consumer);}\r
262 void output::remove(int index){impl_->remove(index);}\r
263 void output::remove(const safe_ptr<frame_consumer>& consumer){impl_->remove(consumer);}\r
264 void output::send(const std::pair<safe_ptr<read_frame>, std::shared_ptr<void>>& frame) {impl_->send(frame); }\r
265 void output::set_video_format_desc(const video_format_desc& format_desc){impl_->set_video_format_desc(format_desc);}\r
266 boost::unique_future<boost::property_tree::wptree> output::info() const{return impl_->info();}\r
267 }}