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 #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() << msg_info("params cannot be empty"));
 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(user_error()
 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 }}