git.sesse.net Git - casparcg/blob - 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
  33 #include <boost/thread.hpp>
  34
  35 #include <future>
  36 #include <vector>
  37 #include <map>
  38
  39 namespace caspar { namespace core {
  40
  41 struct frame_consumer_registry::impl
  42 {
  43         std::vector<consumer_factory_t>                                                         consumer_factories;
  44         std::map<std::wstring, preconfigured_consumer_factory_t>        preconfigured_consumer_factories;
  45         spl::shared_ptr<help_repository>                                                        help_repo;
  46
  47         impl(spl::shared_ptr<help_repository> help_repo)
  48                 : help_repo(std::move(help_repo))
  49         {
  50         }
  51 };
  52
  53 frame_consumer_registry::frame_consumer_registry(spl::shared_ptr<help_repository> help_repo)
  54         : impl_(new impl(std::move(help_repo)))
  55 {
  56 }
  57
  58 void frame_consumer_registry::register_consumer_factory(const std::wstring& name, const consumer_factory_t& factory, const help_item_describer& describer)
  59 {
  60         impl_->consumer_factories.push_back(factory);
  61         impl_->help_repo->register_item({ L"consumer" }, std::move(name), describer);
  62 }
  63
  64 void frame_consumer_registry::register_preconfigured_consumer_factory(
  65                 const std::wstring& element_name,
  66                 const preconfigured_consumer_factory_t& factory)
  67 {
  68         impl_->preconfigured_consumer_factories.insert(std::make_pair(element_name, factory));
  69 }
  70
  71 tbb::atomic<bool>& destroy_consumers_in_separate_thread()
  72 {
  73         static tbb::atomic<bool> state;
  74
  75         return state;
  76 }
  77
  78 void destroy_consumers_synchronously()
  79 {
  80         destroy_consumers_in_separate_thread() = false;
  81 }
  82
  83 class destroy_consumer_proxy : public frame_consumer
  84 {
  85         std::shared_ptr<frame_consumer> consumer_;
  86 public:
  87         destroy_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer)
  88                 : consumer_(std::move(consumer))
  89         {
  90                 destroy_consumers_in_separate_thread() = true;
  91         }
  92
  93         ~destroy_consumer_proxy()
  94         {
  95                 static tbb::atomic<int> counter;
  96                 static std::once_flag counter_init_once;
  97                 std::call_once(counter_init_once, []{ counter = 0; });
  98
  99                 if (!destroy_consumers_in_separate_thread())
 100                         return;
 101
 102                 ++counter;
 103                 CASPAR_VERIFY(counter < 8);
 104
 105                 auto consumer = new std::shared_ptr<frame_consumer>(std::move(consumer_));
 106                 boost::thread([=]
 107                 {
 108                         std::unique_ptr<std::shared_ptr<frame_consumer>> pointer_guard(consumer);
 109                         auto str = (*consumer)->print();
 110
 111                         try
 112                         {
 113                                 ensure_gpf_handler_installed_for_thread(u8(L"Destroyer: " + str).c_str());
 114
 115                                 if (!consumer->unique())
 116                                         CASPAR_LOG(trace) << str << L" Not destroyed on asynchronous destruction thread: " << consumer->use_count();
 117                                 else
 118                                         CASPAR_LOG(trace) << str << L" Destroying on asynchronous destruction thread.";
 119                         }
 120                         catch(...){}
 121
 122                         pointer_guard.reset();
 123
 124                 }).detach();
 125         }
 126
 127         std::future<bool> send(const_frame frame) override                                                                                                      {return consumer_->send(std::move(frame));}
 128         virtual void initialize(const video_format_desc& format_desc, int channel_index)        override        {return consumer_->initialize(format_desc, channel_index);}
 129         std::wstring print() const override                                                                                                                                     {return consumer_->print();}
 130         std::wstring name() const override                                                                                                                                      {return consumer_->name();}
 131         boost::property_tree::wptree info() const override                                                                                                      {return consumer_->info();}
 132         bool has_synchronization_clock() const override                                                                                                         {return consumer_->has_synchronization_clock();}
 133         int buffer_depth() const override                                                                                                                                       {return consumer_->buffer_depth();}
 134         int index() const override                                                                                                                                                      {return consumer_->index();}
 135         int64_t presentation_frame_age_millis() const override                                                                                          {return consumer_->presentation_frame_age_millis();}
 136         monitor::subject& monitor_output() override                                                                                                                     {return consumer_->monitor_output();}
 137 };
 138
 139 class print_consumer_proxy : public frame_consumer
 140 {
 141         std::shared_ptr<frame_consumer> consumer_;
 142 public:
 143         print_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer)
 144                 : consumer_(std::move(consumer))
 145         {
 146                 CASPAR_LOG(info) << consumer_->print() << L" Initialized.";
 147         }
 148
 149         ~print_consumer_proxy()
 150         {
 151                 auto str = consumer_->print();
 152                 CASPAR_LOG(trace) << 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         virtual void initialize(const video_format_desc& format_desc, int channel_index)        override        {return consumer_->initialize(format_desc, channel_index);}
 159         std::wstring print() const override                                                                                                                                     {return consumer_->print();}
 160         std::wstring name() const override                                                                                                                                      {return consumer_->name();}
 161         boost::property_tree::wptree info() const override                                                                                                      {return consumer_->info();}
 162         bool has_synchronization_clock() const override                                                                                                         {return consumer_->has_synchronization_clock();}
 163         int buffer_depth() const override                                                                                                                                       {return consumer_->buffer_depth();}
 164         int index() const override                                                                                                                                                      {return consumer_->index();}
 165         int64_t presentation_frame_age_millis() const override                                                                                          {return consumer_->presentation_frame_age_millis();}
 166         monitor::subject& monitor_output() override                                                                                                                     {return consumer_->monitor_output();}
 167 };
 168
 169 class recover_consumer_proxy : public frame_consumer
 170 {
 171         std::shared_ptr<frame_consumer> consumer_;
 172         int                                                             channel_index_  = -1;
 173         video_format_desc                               format_desc_;
 174 public:
 175         recover_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer)
 176                 : consumer_(std::move(consumer))
 177         {
 178         }
 179
 180         virtual std::future<bool> send(const_frame frame)
 181         {
 182                 try
 183                 {
 184                         return consumer_->send(frame);
 185                 }
 186                 catch(...)
 187                 {
 188                         CASPAR_LOG_CURRENT_EXCEPTION();
 189                         try
 190                         {
 191                                 consumer_->initialize(format_desc_, channel_index_);
 192                                 return consumer_->send(frame);
 193                         }
 194                         catch(...)
 195                         {
 196                                 CASPAR_LOG_CURRENT_EXCEPTION();
 197                                 CASPAR_LOG(error) << print() << " Failed to recover consumer.";
 198                                 return make_ready_future(false);
 199                         }
 200                 }
 201         }
 202
 203         virtual void initialize(const video_format_desc& format_desc, int channel_index)
 204         {
 205                 format_desc_    = format_desc;
 206                 channel_index_  = channel_index;
 207                 return consumer_->initialize(format_desc, channel_index);
 208         }
 209
 210         std::wstring print() const override                                                                             {return consumer_->print();}
 211         std::wstring name() const override                                                                              {return consumer_->name();}
 212         boost::property_tree::wptree info() const override                                              {return consumer_->info();}
 213         bool has_synchronization_clock() const override                                                 {return consumer_->has_synchronization_clock();}
 214         int buffer_depth() const override                                                                               {return consumer_->buffer_depth();}
 215         int index() const override                                                                                              {return consumer_->index();}
 216         int64_t presentation_frame_age_millis() const override                                  {return consumer_->presentation_frame_age_millis();}
 217         monitor::subject& monitor_output() override                                                             {return consumer_->monitor_output();}
 218 };
 219
 220 // This class is used to guarantee that audio cadence is correct. This is important for NTSC audio.
 221 class cadence_guard : public frame_consumer
 222 {
 223         spl::shared_ptr<frame_consumer>         consumer_;
 224         std::vector<int>                                        audio_cadence_;
 225         video_format_desc                                       format_desc_;
 226         boost::circular_buffer<std::size_t>     sync_buffer_;
 227 public:
 228         cadence_guard(const spl::shared_ptr<frame_consumer>& consumer)
 229                 : consumer_(consumer)
 230         {
 231         }
 232
 233         void initialize(const video_format_desc& format_desc, int channel_index) override
 234         {
 235                 audio_cadence_  = format_desc.audio_cadence;
 236                 sync_buffer_    = boost::circular_buffer<std::size_t>(format_desc.audio_cadence.size());
 237                 format_desc_    = format_desc;
 238                 consumer_->initialize(format_desc, channel_index);
 239         }
 240
 241         std::future<bool> send(const_frame frame) override
 242         {
 243                 if(audio_cadence_.size() == 1)
 244                         return consumer_->send(frame);
 245
 246                 std::future<bool> result = make_ready_future(true);
 247
 248                 if(boost::range::equal(sync_buffer_, audio_cadence_) && audio_cadence_.front() * format_desc_.audio_channels == static_cast<int>(frame.audio_data().size()))
 249                 {
 250                         // Audio sent so far is in sync, now we can send the next chunk.
 251                         result = consumer_->send(frame);
 252                         boost::range::rotate(audio_cadence_, std::begin(audio_cadence_)+1);
 253                 }
 254                 else
 255                         CASPAR_LOG(trace) << print() << L" Syncing audio.";
 256
 257                 sync_buffer_.push_back(static_cast<int>(frame.audio_data().size() / format_desc_.audio_channels));
 258
 259                 return std::move(result);
 260         }
 261
 262         std::wstring print() const override                                                                             {return consumer_->print();}
 263         std::wstring name() const override                                                                              {return consumer_->name();}
 264         boost::property_tree::wptree info() const override                                              {return consumer_->info();}
 265         bool has_synchronization_clock() const override                                                 {return consumer_->has_synchronization_clock();}
 266         int buffer_depth() const override                                                                               {return consumer_->buffer_depth();}
 267         int index() const override                                                                                              {return consumer_->index();}
 268         int64_t presentation_frame_age_millis() const override                                  {return consumer_->presentation_frame_age_millis();}
 269         monitor::subject& monitor_output() override                                                             {return consumer_->monitor_output();}
 270 };
 271
 272 spl::shared_ptr<core::frame_consumer> frame_consumer_registry::create_consumer(
 273                 const std::vector<std::wstring>& params, interaction_sink* sink) const
 274 {
 275         if(params.empty())
 276                 CASPAR_THROW_EXCEPTION(invalid_argument() << arg_name_info("params") << arg_value_info(""));
 277
 278         auto consumer = frame_consumer::empty();
 279         auto& consumer_factories = impl_->consumer_factories;
 280         std::any_of(consumer_factories.begin(), consumer_factories.end(), [&](const consumer_factory_t& factory) -> bool
 281                 {
 282                         try
 283                         {
 284                                 consumer = factory(params, sink);
 285                         }
 286                         catch(...)
 287                         {
 288                                 CASPAR_LOG_CURRENT_EXCEPTION();
 289                         }
 290                         return consumer != frame_consumer::empty();
 291                 });
 292
 293         if(consumer == frame_consumer::empty())
 294                 CASPAR_THROW_EXCEPTION(file_not_found() << msg_info("No match found for supplied commands. Check syntax."));
 295
 296         return spl::make_shared<destroy_consumer_proxy>(
 297                         spl::make_shared<print_consumer_proxy>(
 298                          spl::make_shared<recover_consumer_proxy>(
 299                           spl::make_shared<cadence_guard>(
 300                            std::move(consumer)))));
 301 }
 302
 303 spl::shared_ptr<frame_consumer> frame_consumer_registry::create_consumer(
 304                 const std::wstring& element_name,
 305                 const boost::property_tree::wptree& element,
 306                 interaction_sink* sink) const
 307 {
 308         auto& preconfigured_consumer_factories = impl_->preconfigured_consumer_factories;
 309         auto found = preconfigured_consumer_factories.find(element_name);
 310
 311         if (found == preconfigured_consumer_factories.end())
 312                 CASPAR_THROW_EXCEPTION(file_not_found()
 313                         << msg_info(L"No consumer factory registered for element name " + element_name));
 314
 315         return spl::make_shared<destroy_consumer_proxy>(
 316                         spl::make_shared<print_consumer_proxy>(
 317                                         spl::make_shared<recover_consumer_proxy>(
 318                                                         spl::make_shared<cadence_guard>(
 319                                                                         found->second(element, sink)))));
 320 }
 321
 322 const spl::shared_ptr<frame_consumer>& frame_consumer::empty()
 323 {
 324         class empty_frame_consumer : public frame_consumer
 325         {
 326         public:
 327                 std::future<bool> send(const_frame) override { return make_ready_future(false); }
 328                 void initialize(const video_format_desc&, int) override{}
 329                 std::wstring print() const override {return L"empty";}
 330                 std::wstring name() const override {return L"empty";}
 331                 bool has_synchronization_clock() const override {return false;}
 332                 int buffer_depth() const override {return 0;};
 333                 int index() const override {return -1;}
 334                 int64_t presentation_frame_age_millis() const override {return -1;}
 335                 monitor::subject& monitor_output() override {static monitor::subject monitor_subject(""); return monitor_subject;}
 336                 boost::property_tree::wptree info() const override
 337                 {
 338                         boost::property_tree::wptree info;
 339                         info.add(L"type", L"empty");
 340                         return info;
 341                 }
 342         };
 343         static spl::shared_ptr<frame_consumer> consumer = spl::make_shared<empty_frame_consumer>();
 344         return consumer;
 345 }
 346
 347 }}