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         const frame_consumer* unwrapped() const override                                                                                                                                                                {return consumer_->unwrapped();}
 139 };
 140
 141 class print_consumer_proxy : public frame_consumer
 142 {
 143         std::shared_ptr<frame_consumer> consumer_;
 144 public:
 145         print_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer)
 146                 : consumer_(std::move(consumer))
 147         {
 148         }
 149
 150         ~print_consumer_proxy()
 151         {
 152                 auto str = consumer_->print();
 153                 CASPAR_LOG(debug) << str << L" Uninitializing.";
 154                 consumer_.reset();
 155                 CASPAR_LOG(info) << str << L" Uninitialized.";
 156         }
 157
 158         std::future<bool> send(const_frame frame) override                                                                                                                                                              {return consumer_->send(std::move(frame));}
 159         void initialize(const video_format_desc& format_desc, const audio_channel_layout& channel_layout, int channel_index) override
 160         {
 161                 consumer_->initialize(format_desc, channel_layout, channel_index);
 162                 CASPAR_LOG(info) << consumer_->print() << L" Initialized.";
 163         }
 164         std::wstring print() const override                                                                                                                                                                                             {return consumer_->print();}
 165         std::wstring name() const override                                                                                                                                                                                              {return consumer_->name();}
 166         boost::property_tree::wptree info() const override                                                                                                                                                              {return consumer_->info();}
 167         bool has_synchronization_clock() const override                                                                                                                                                                 {return consumer_->has_synchronization_clock();}
 168         int buffer_depth() const override                                                                                                                                                                                               {return consumer_->buffer_depth();}
 169         int index() const override                                                                                                                                                                                                              {return consumer_->index();}
 170         int64_t presentation_frame_age_millis() const override                                                                                                                                                  {return consumer_->presentation_frame_age_millis();}
 171         monitor::subject& monitor_output() override                                                                                                                                                                             {return consumer_->monitor_output();}
 172         const frame_consumer* unwrapped() const override                                                                                                                                                                {return consumer_->unwrapped();}
 173 };
 174
 175 class recover_consumer_proxy : public frame_consumer
 176 {
 177         std::shared_ptr<frame_consumer> consumer_;
 178         int                                                             channel_index_  = -1;
 179         video_format_desc                               format_desc_;
 180         audio_channel_layout                    channel_layout_ = audio_channel_layout::invalid();
 181 public:
 182         recover_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer)
 183                 : consumer_(std::move(consumer))
 184         {
 185         }
 186
 187         std::future<bool> send(const_frame frame) override
 188         {
 189                 try
 190                 {
 191                         return consumer_->send(frame);
 192                 }
 193                 catch(...)
 194                 {
 195                         CASPAR_LOG_CURRENT_EXCEPTION();
 196                         try
 197                         {
 198                                 consumer_->initialize(format_desc_, channel_layout_, channel_index_);
 199                                 return consumer_->send(frame);
 200                         }
 201                         catch(...)
 202                         {
 203                                 CASPAR_LOG_CURRENT_EXCEPTION();
 204                                 CASPAR_LOG(error) << print() << " Failed to recover consumer.";
 205                                 return make_ready_future(false);
 206                         }
 207                 }
 208         }
 209
 210         void initialize(const video_format_desc& format_desc, const audio_channel_layout& channel_layout, int channel_index) override
 211         {
 212                 format_desc_    = format_desc;
 213                 channel_layout_ = channel_layout;
 214                 channel_index_  = channel_index;
 215                 return consumer_->initialize(format_desc, channel_layout, channel_index);
 216         }
 217
 218         std::wstring print() const override                                                                             {return consumer_->print();}
 219         std::wstring name() const override                                                                              {return consumer_->name();}
 220         boost::property_tree::wptree info() const override                                              {return consumer_->info();}
 221         bool has_synchronization_clock() const override                                                 {return consumer_->has_synchronization_clock();}
 222         int buffer_depth() const override                                                                               {return consumer_->buffer_depth();}
 223         int index() const override                                                                                              {return consumer_->index();}
 224         int64_t presentation_frame_age_millis() const override                                  {return consumer_->presentation_frame_age_millis();}
 225         monitor::subject& monitor_output() override                                                             {return consumer_->monitor_output();}
 226         const frame_consumer* unwrapped() const override                                                {return consumer_->unwrapped();}
 227 };
 228
 229 // This class is used to guarantee that audio cadence is correct. This is important for NTSC audio.
 230 class cadence_guard : public frame_consumer
 231 {
 232         spl::shared_ptr<frame_consumer>         consumer_;
 233         std::vector<int>                                        audio_cadence_;
 234         video_format_desc                                       format_desc_;
 235         audio_channel_layout                            channel_layout_ = audio_channel_layout::invalid();
 236         boost::circular_buffer<std::size_t>     sync_buffer_;
 237 public:
 238         cadence_guard(const spl::shared_ptr<frame_consumer>& consumer)
 239                 : consumer_(consumer)
 240         {
 241         }
 242
 243         void initialize(const video_format_desc& format_desc, const audio_channel_layout& channel_layout, int channel_index) override
 244         {
 245                 audio_cadence_  = format_desc.audio_cadence;
 246                 sync_buffer_    = boost::circular_buffer<std::size_t>(format_desc.audio_cadence.size());
 247                 format_desc_    = format_desc;
 248                 channel_layout_ = channel_layout;
 249                 consumer_->initialize(format_desc, channel_layout, channel_index);
 250         }
 251
 252         std::future<bool> send(const_frame frame) override
 253         {
 254                 if(audio_cadence_.size() == 1)
 255                         return consumer_->send(frame);
 256
 257                 std::future<bool> result = make_ready_future(true);
 258
 259                 if(boost::range::equal(sync_buffer_, audio_cadence_) && audio_cadence_.front() * channel_layout_.num_channels == static_cast<int>(frame.audio_data().size()))
 260                 {
 261                         // Audio sent so far is in sync, now we can send the next chunk.
 262                         result = consumer_->send(frame);
 263                         boost::range::rotate(audio_cadence_, std::begin(audio_cadence_)+1);
 264                 }
 265                 else
 266                         CASPAR_LOG(trace) << print() << L" Syncing audio.";
 267
 268                 sync_buffer_.push_back(static_cast<int>(frame.audio_data().size() / channel_layout_.num_channels));
 269
 270                 return std::move(result);
 271         }
 272
 273         std::wstring print() const override                                                                             {return consumer_->print();}
 274         std::wstring name() const override                                                                              {return consumer_->name();}
 275         boost::property_tree::wptree info() const override                                              {return consumer_->info();}
 276         bool has_synchronization_clock() const override                                                 {return consumer_->has_synchronization_clock();}
 277         int buffer_depth() const override                                                                               {return consumer_->buffer_depth();}
 278         int index() const override                                                                                              {return consumer_->index();}
 279         int64_t presentation_frame_age_millis() const override                                  {return consumer_->presentation_frame_age_millis();}
 280         monitor::subject& monitor_output() override                                                             {return consumer_->monitor_output();}
 281         const frame_consumer* unwrapped() const override                                                {return consumer_->unwrapped();}
 282 };
 283
 284 spl::shared_ptr<core::frame_consumer> frame_consumer_registry::create_consumer(
 285                 const std::vector<std::wstring>& params, interaction_sink* sink, std::vector<spl::shared_ptr<video_channel>> channels) const
 286 {
 287         if(params.empty())
 288                 CASPAR_THROW_EXCEPTION(invalid_argument() << msg_info("params cannot be empty"));
 289
 290         auto consumer = frame_consumer::empty();
 291         auto& consumer_factories = impl_->consumer_factories;
 292         std::any_of(consumer_factories.begin(), consumer_factories.end(), [&](const consumer_factory_t& factory) -> bool
 293                 {
 294                         try
 295                         {
 296                                 consumer = factory(params, sink, channels);
 297                         }
 298                         catch(...)
 299                         {
 300                                 CASPAR_LOG_CURRENT_EXCEPTION();
 301                         }
 302                         return consumer != frame_consumer::empty();
 303                 });
 304
 305         if(consumer == frame_consumer::empty())
 306                 CASPAR_THROW_EXCEPTION(file_not_found() << msg_info("No match found for supplied commands. Check syntax."));
 307
 308         return spl::make_shared<destroy_consumer_proxy>(
 309                         spl::make_shared<print_consumer_proxy>(
 310                          spl::make_shared<recover_consumer_proxy>(
 311                           spl::make_shared<cadence_guard>(
 312                            std::move(consumer)))));
 313 }
 314
 315 spl::shared_ptr<frame_consumer> frame_consumer_registry::create_consumer(
 316                 const std::wstring& element_name,
 317                 const boost::property_tree::wptree& element,
 318                 interaction_sink* sink,
 319                 std::vector<spl::shared_ptr<video_channel>> channels) const
 320 {
 321         auto& preconfigured_consumer_factories = impl_->preconfigured_consumer_factories;
 322         auto found = preconfigured_consumer_factories.find(element_name);
 323
 324         if (found == preconfigured_consumer_factories.end())
 325                 CASPAR_THROW_EXCEPTION(user_error()
 326                         << msg_info(L"No consumer factory registered for element name " + element_name));
 327
 328         return spl::make_shared<destroy_consumer_proxy>(
 329                         spl::make_shared<print_consumer_proxy>(
 330                                         spl::make_shared<recover_consumer_proxy>(
 331                                                         spl::make_shared<cadence_guard>(
 332                                                                         found->second(element, sink, channels)))));
 333 }
 334
 335 const spl::shared_ptr<frame_consumer>& frame_consumer::empty()
 336 {
 337         class empty_frame_consumer : public frame_consumer
 338         {
 339         public:
 340                 std::future<bool> send(const_frame) override { return make_ready_future(false); }
 341                 void initialize(const video_format_desc&, const audio_channel_layout&, int) override{}
 342                 std::wstring print() const override {return L"empty";}
 343                 std::wstring name() const override {return L"empty";}
 344                 bool has_synchronization_clock() const override {return false;}
 345                 int buffer_depth() const override {return 0;};
 346                 int index() const override {return -1;}
 347                 int64_t presentation_frame_age_millis() const override {return -1;}
 348                 monitor::subject& monitor_output() override {static monitor::subject monitor_subject(""); return monitor_subject;}
 349                 boost::property_tree::wptree info() const override
 350                 {
 351                         boost::property_tree::wptree info;
 352                         info.add(L"type", L"empty");
 353                         return info;
 354                 }
 355         };
 356         static spl::shared_ptr<frame_consumer> consumer = spl::make_shared<empty_frame_consumer>();
 357         return consumer;
 358 }
 359
 360 }}