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