]> git.sesse.net Git - casparcg/blob - core/consumer/frame_consumer.cpp
Changed default log level to info and moved logging statements that we always want...
[casparcg] / core / consumer / frame_consumer.cpp
1 /*
2 * Copyright (c) 2011 Sveriges Television AB <info@casparcg.com>
3 *
4 * This file is part of CasparCG (www.casparcg.com).
5 *
6 * CasparCG is free software: you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License as published by
8 * the Free Software Foundation, either version 3 of the License, or
9 * (at your option) any later version.
10 *
11 * CasparCG is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14 * GNU General Public License for more details.
15 *
16 * You should have received a copy of the GNU General Public License
17 * along with CasparCG. If not, see <http://www.gnu.org/licenses/>.
18 *
19 * Author: Robert Nagy, ronag89@gmail.com
20 */
21
22 #include "../StdAfx.h"
23
24 #include "frame_consumer.h"
25
26 #include <common/except.h>
27 #include <common/future.h>
28 #include <common/os/general_protection_fault.h>
29
30 #include <core/video_format.h>
31 #include <core/frame/frame.h>
32 #include <core/frame/audio_channel_layout.h>
33
34 #include <boost/thread.hpp>
35
36 #include <future>
37 #include <vector>
38 #include <map>
39
40 namespace caspar { namespace core {
41
42 struct frame_consumer_registry::impl
43 {
44         std::vector<consumer_factory_t>                                                         consumer_factories;
45         std::map<std::wstring, preconfigured_consumer_factory_t>        preconfigured_consumer_factories;
46         spl::shared_ptr<help_repository>                                                        help_repo;
47
48         impl(spl::shared_ptr<help_repository> help_repo)
49                 : help_repo(std::move(help_repo))
50         {
51         }
52 };
53
54 frame_consumer_registry::frame_consumer_registry(spl::shared_ptr<help_repository> help_repo)
55         : impl_(new impl(std::move(help_repo)))
56 {
57 }
58
59 void frame_consumer_registry::register_consumer_factory(const std::wstring& name, const consumer_factory_t& factory, const help_item_describer& describer)
60 {
61         impl_->consumer_factories.push_back(factory);
62         impl_->help_repo->register_item({ L"consumer" }, std::move(name), describer);
63 }
64
65 void frame_consumer_registry::register_preconfigured_consumer_factory(
66                 const std::wstring& element_name,
67                 const preconfigured_consumer_factory_t& factory)
68 {
69         impl_->preconfigured_consumer_factories.insert(std::make_pair(element_name, factory));
70 }
71
72 tbb::atomic<bool>& destroy_consumers_in_separate_thread()
73 {
74         static tbb::atomic<bool> state;
75
76         return state;
77 }
78
79 void destroy_consumers_synchronously()
80 {
81         destroy_consumers_in_separate_thread() = false;
82 }
83
84 class destroy_consumer_proxy : public frame_consumer
85 {       
86         std::shared_ptr<frame_consumer> consumer_;
87 public:
88         destroy_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer) 
89                 : consumer_(std::move(consumer))
90         {
91                 destroy_consumers_in_separate_thread() = true;
92         }
93
94         ~destroy_consumer_proxy()
95         {               
96                 static tbb::atomic<int> counter;
97                 static std::once_flag counter_init_once;
98                 std::call_once(counter_init_once, []{ counter = 0; });
99
100                 if (!destroy_consumers_in_separate_thread())
101                         return;
102                         
103                 ++counter;
104                 CASPAR_VERIFY(counter < 8);
105                 
106                 auto consumer = new std::shared_ptr<frame_consumer>(std::move(consumer_));
107                 boost::thread([=]
108                 {
109                         std::unique_ptr<std::shared_ptr<frame_consumer>> pointer_guard(consumer);
110                         auto str = (*consumer)->print();
111
112                         try
113                         {
114                                 ensure_gpf_handler_installed_for_thread(u8(L"Destroyer: " + str).c_str());
115
116                                 if (!consumer->unique())
117                                         CASPAR_LOG(debug) << str << L" Not destroyed on asynchronous destruction thread: " << consumer->use_count();
118                                 else
119                                         CASPAR_LOG(debug) << str << L" Destroying on asynchronous destruction thread.";
120                         }
121                         catch(...){}
122
123                         pointer_guard.reset();
124
125                 }).detach(); 
126         }
127         
128         std::future<bool> send(const_frame frame) override                                                                                                                                                              {return consumer_->send(std::move(frame));}
129         void initialize(const video_format_desc& format_desc, const audio_channel_layout& channel_layout, int channel_index) override   {return consumer_->initialize(format_desc, channel_layout, channel_index);}
130         std::wstring print() const override                                                                                                                                                                                             {return consumer_->print();}    
131         std::wstring name() const override                                                                                                                                                                                              {return consumer_->name();}
132         boost::property_tree::wptree info() const override                                                                                                                                                              {return consumer_->info();}
133         bool has_synchronization_clock() const override                                                                                                                                                                 {return consumer_->has_synchronization_clock();}
134         int buffer_depth() const override                                                                                                                                                                                               {return consumer_->buffer_depth();}
135         int index() const override                                                                                                                                                                                                              {return consumer_->index();}
136         int64_t presentation_frame_age_millis() const override                                                                                                                                                  {return consumer_->presentation_frame_age_millis();}
137         monitor::subject& monitor_output() override                                                                                                                                                                             {return consumer_->monitor_output();}                                                                           
138 };
139
140 class print_consumer_proxy : public frame_consumer
141 {       
142         std::shared_ptr<frame_consumer> consumer_;
143 public:
144         print_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer) 
145                 : consumer_(std::move(consumer))
146         {
147         }
148
149         ~print_consumer_proxy()
150         {               
151                 auto str = consumer_->print();
152                 CASPAR_LOG(debug) << str << L" Uninitializing.";
153                 consumer_.reset();
154                 CASPAR_LOG(info) << str << L" Uninitialized.";
155         }
156         
157         std::future<bool> send(const_frame frame) override                                                                                                                                                              {return consumer_->send(std::move(frame));}
158         void initialize(const video_format_desc& format_desc, const audio_channel_layout& channel_layout, int channel_index) override
159         {
160                 consumer_->initialize(format_desc, channel_layout, channel_index);
161                 CASPAR_LOG(info) << consumer_->print() << L" Initialized.";
162         }
163         std::wstring print() const override                                                                                                                                                                                             {return consumer_->print();}
164         std::wstring name() const override                                                                                                                                                                                              {return consumer_->name();}
165         boost::property_tree::wptree info() const override                                                                                                                                                              {return consumer_->info();}
166         bool has_synchronization_clock() const override                                                                                                                                                                 {return consumer_->has_synchronization_clock();}
167         int buffer_depth() const override                                                                                                                                                                                               {return consumer_->buffer_depth();}
168         int index() const override                                                                                                                                                                                                              {return consumer_->index();}
169         int64_t presentation_frame_age_millis() const override                                                                                                                                                  {return consumer_->presentation_frame_age_millis();}
170         monitor::subject& monitor_output() override                                                                                                                                                                             {return consumer_->monitor_output();}                                                                           
171 };
172
173 class recover_consumer_proxy : public frame_consumer
174 {       
175         std::shared_ptr<frame_consumer> consumer_;
176         int                                                             channel_index_  = -1;
177         video_format_desc                               format_desc_;
178         audio_channel_layout                    channel_layout_ = audio_channel_layout::invalid();
179 public:
180         recover_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer) 
181                 : consumer_(std::move(consumer))
182         {
183         }
184         
185         std::future<bool> send(const_frame frame) override                              
186         {
187                 try
188                 {
189                         return consumer_->send(frame);
190                 }
191                 catch(...)
192                 {
193                         CASPAR_LOG_CURRENT_EXCEPTION();
194                         try
195                         {
196                                 consumer_->initialize(format_desc_, channel_layout_, channel_index_);
197                                 return consumer_->send(frame);
198                         }
199                         catch(...)
200                         {
201                                 CASPAR_LOG_CURRENT_EXCEPTION();
202                                 CASPAR_LOG(error) << print() << " Failed to recover consumer.";
203                                 return make_ready_future(false);
204                         }
205                 }
206         }
207
208         void initialize(const video_format_desc& format_desc, const audio_channel_layout& channel_layout, int channel_index) override
209         {
210                 format_desc_    = format_desc;
211                 channel_layout_ = channel_layout;
212                 channel_index_  = channel_index;
213                 return consumer_->initialize(format_desc, channel_layout, channel_index);
214         }
215
216         std::wstring print() const override                                                                             {return consumer_->print();}
217         std::wstring name() const override                                                                              {return consumer_->name();}
218         boost::property_tree::wptree info() const override                                              {return consumer_->info();}
219         bool has_synchronization_clock() const override                                                 {return consumer_->has_synchronization_clock();}
220         int buffer_depth() const override                                                                               {return consumer_->buffer_depth();}
221         int index() const override                                                                                              {return consumer_->index();}
222         int64_t presentation_frame_age_millis() const override                                  {return consumer_->presentation_frame_age_millis();}
223         monitor::subject& monitor_output() override                                                             {return consumer_->monitor_output();}                                                                           
224 };
225
226 // This class is used to guarantee that audio cadence is correct. This is important for NTSC audio.
227 class cadence_guard : public frame_consumer
228 {
229         spl::shared_ptr<frame_consumer>         consumer_;
230         std::vector<int>                                        audio_cadence_;
231         video_format_desc                                       format_desc_;
232         audio_channel_layout                            channel_layout_ = audio_channel_layout::invalid();
233         boost::circular_buffer<std::size_t>     sync_buffer_;
234 public:
235         cadence_guard(const spl::shared_ptr<frame_consumer>& consumer)
236                 : consumer_(consumer)
237         {
238         }
239         
240         void initialize(const video_format_desc& format_desc, const audio_channel_layout& channel_layout, int channel_index) override
241         {
242                 audio_cadence_  = format_desc.audio_cadence;
243                 sync_buffer_    = boost::circular_buffer<std::size_t>(format_desc.audio_cadence.size());
244                 format_desc_    = format_desc;
245                 channel_layout_ = channel_layout;
246                 consumer_->initialize(format_desc, channel_layout, channel_index);
247         }
248
249         std::future<bool> send(const_frame frame) override
250         {               
251                 if(audio_cadence_.size() == 1)
252                         return consumer_->send(frame);
253
254                 std::future<bool> result = make_ready_future(true);
255                 
256                 if(boost::range::equal(sync_buffer_, audio_cadence_) && audio_cadence_.front() * channel_layout_.num_channels == static_cast<int>(frame.audio_data().size()))
257                 {       
258                         // Audio sent so far is in sync, now we can send the next chunk.
259                         result = consumer_->send(frame);
260                         boost::range::rotate(audio_cadence_, std::begin(audio_cadence_)+1);
261                 }
262                 else
263                         CASPAR_LOG(trace) << print() << L" Syncing audio.";
264
265                 sync_buffer_.push_back(static_cast<int>(frame.audio_data().size() / channel_layout_.num_channels));
266                 
267                 return std::move(result);
268         }
269         
270         std::wstring print() const override                                                                             {return consumer_->print();}
271         std::wstring name() const override                                                                              {return consumer_->name();}
272         boost::property_tree::wptree info() const override                                              {return consumer_->info();}
273         bool has_synchronization_clock() const override                                                 {return consumer_->has_synchronization_clock();}
274         int buffer_depth() const override                                                                               {return consumer_->buffer_depth();}
275         int index() const override                                                                                              {return consumer_->index();}
276         int64_t presentation_frame_age_millis() const override                                  {return consumer_->presentation_frame_age_millis();}
277         monitor::subject& monitor_output() override                                                             {return consumer_->monitor_output();}                                                                           
278 };
279
280 spl::shared_ptr<core::frame_consumer> frame_consumer_registry::create_consumer(
281                 const std::vector<std::wstring>& params, interaction_sink* sink) const
282 {
283         if(params.empty())
284                 CASPAR_THROW_EXCEPTION(invalid_argument() << arg_name_info("params") << arg_value_info(""));
285         
286         auto consumer = frame_consumer::empty();
287         auto& consumer_factories = impl_->consumer_factories;
288         std::any_of(consumer_factories.begin(), consumer_factories.end(), [&](const consumer_factory_t& factory) -> bool
289                 {
290                         try
291                         {
292                                 consumer = factory(params, sink);
293                         }
294                         catch(...)
295                         {
296                                 CASPAR_LOG_CURRENT_EXCEPTION();
297                         }
298                         return consumer != frame_consumer::empty();
299                 });
300
301         if(consumer == frame_consumer::empty())
302                 CASPAR_THROW_EXCEPTION(file_not_found() << msg_info("No match found for supplied commands. Check syntax."));
303
304         return spl::make_shared<destroy_consumer_proxy>(
305                         spl::make_shared<print_consumer_proxy>(
306                          spl::make_shared<recover_consumer_proxy>(
307                           spl::make_shared<cadence_guard>(
308                            std::move(consumer)))));
309 }
310
311 spl::shared_ptr<frame_consumer> frame_consumer_registry::create_consumer(
312                 const std::wstring& element_name,
313                 const boost::property_tree::wptree& element,
314                 interaction_sink* sink) const
315 {
316         auto& preconfigured_consumer_factories = impl_->preconfigured_consumer_factories;
317         auto found = preconfigured_consumer_factories.find(element_name);
318
319         if (found == preconfigured_consumer_factories.end())
320                 CASPAR_THROW_EXCEPTION(file_not_found()
321                         << msg_info(L"No consumer factory registered for element name " + element_name));
322
323         return spl::make_shared<destroy_consumer_proxy>(
324                         spl::make_shared<print_consumer_proxy>(
325                                         spl::make_shared<recover_consumer_proxy>(
326                                                         spl::make_shared<cadence_guard>(
327                                                                         found->second(element, sink)))));
328 }
329
330 const spl::shared_ptr<frame_consumer>& frame_consumer::empty()
331 {
332         class empty_frame_consumer : public frame_consumer
333         {
334         public:
335                 std::future<bool> send(const_frame) override { return make_ready_future(false); }
336                 void initialize(const video_format_desc&, const audio_channel_layout&, int) override{}
337                 std::wstring print() const override {return L"empty";}
338                 std::wstring name() const override {return L"empty";}
339                 bool has_synchronization_clock() const override {return false;}
340                 int buffer_depth() const override {return 0;};
341                 int index() const override {return -1;}
342                 int64_t presentation_frame_age_millis() const override {return -1;}
343                 monitor::subject& monitor_output() override {static monitor::subject monitor_subject(""); return monitor_subject;}
344                 boost::property_tree::wptree info() const override
345                 {
346                         boost::property_tree::wptree info;
347                         info.add(L"type", L"empty");
348                         return info;
349                 }
350         };
351         static spl::shared_ptr<frame_consumer> consumer = spl::make_shared<empty_frame_consumer>();
352         return consumer;
353 }
354
355 }}