git.sesse.net Git - casparcg/blob - modules/reroute/producer/channel_producer.cpp

   1 /*
   2 * Copyright 2013 Sveriges Television AB http://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 "channel_producer.h"
  25
  26 #include <core/monitor/monitor.h>
  27 #include <core/consumer/frame_consumer.h>
  28 #include <core/consumer/output.h>
  29 #include <core/producer/frame_producer.h>
  30 #include <core/producer/framerate/framerate_producer.h>
  31 #include <core/video_channel.h>
  32
  33 #include <core/frame/frame.h>
  34 #include <core/frame/pixel_format.h>
  35 #include <core/frame/audio_channel_layout.h>
  36 #include <core/frame/draw_frame.h>
  37 #include <core/frame/frame_factory.h>
  38 #include <core/video_format.h>
  39
  40 #include <boost/thread/once.hpp>
  41 #include <boost/lexical_cast.hpp>
  42 #include <boost/property_tree/ptree.hpp>
  43 #include <boost/range/algorithm/copy.hpp>
  44
  45 #include <common/except.h>
  46 #include <common/memory.h>
  47 #include <common/memcpy.h>
  48 #include <common/semaphore.h>
  49 #include <common/future.h>
  50
  51 #include <tbb/concurrent_queue.h>
  52
  53 #include <asmlib.h>
  54
  55 #include <queue>
  56
  57 namespace caspar { namespace reroute {
  58
  59 class channel_consumer : public core::frame_consumer
  60 {
  61         core::monitor::subject                                                          monitor_subject_;
  62         tbb::concurrent_bounded_queue<core::const_frame>        frame_buffer_;
  63         core::video_format_desc                                                         format_desc_;
  64         core::audio_channel_layout                                                      channel_layout_                 = core::audio_channel_layout::invalid();
  65         int                                                                                                     channel_index_;
  66         int                                                                                                     consumer_index_;
  67         tbb::atomic<bool>                                                                       is_running_;
  68         tbb::atomic<int64_t>                                                            current_age_;
  69         semaphore                                                                                       frames_available_ { 0 };
  70         int                                                                                                     frames_delay_;
  71
  72 public:
  73         channel_consumer(int frames_delay)
  74                 : consumer_index_(next_consumer_index())
  75                 , frames_delay_(frames_delay)
  76         {
  77                 is_running_ = true;
  78                 current_age_ = 0;
  79                 frame_buffer_.set_capacity(3 + frames_delay);
  80         }
  81
  82         static int next_consumer_index()
  83         {
  84                 static tbb::atomic<int> consumer_index_counter;
  85                 static boost::once_flag consumer_index_counter_initialized;
  86
  87                 boost::call_once(consumer_index_counter_initialized, [&]()
  88                 {
  89                         consumer_index_counter = 0;
  90                 });
  91
  92                 return ++consumer_index_counter;
  93         }
  94
  95         ~channel_consumer()
  96         {
  97         }
  98
  99         // frame_consumer
 100
 101         std::future<bool> send(core::const_frame frame) override
 102         {
 103                 bool pushed = frame_buffer_.try_push(frame);
 104
 105                 if (pushed)
 106                         frames_available_.release();
 107
 108                 return make_ready_future(is_running_.load());
 109         }
 110
 111         void initialize(
 112                         const core::video_format_desc& format_desc,
 113                         const core::audio_channel_layout& channel_layout,
 114                         int channel_index) override
 115         {
 116                 format_desc_    = format_desc;
 117                 channel_layout_ = channel_layout;
 118                 channel_index_  = channel_index;
 119         }
 120
 121         std::wstring name() const override
 122         {
 123                 return L"channel-consumer";
 124         }
 125
 126         int64_t presentation_frame_age_millis() const override
 127         {
 128                 return current_age_;
 129         }
 130
 131         std::wstring print() const override
 132         {
 133                 return L"[channel-consumer|" + boost::lexical_cast<std::wstring>(channel_index_) + L"]";
 134         }
 135
 136         boost::property_tree::wptree info() const override
 137         {
 138                 boost::property_tree::wptree info;
 139                 info.add(L"type", L"channel-consumer");
 140                 info.add(L"channel-index", channel_index_);
 141                 return info;
 142         }
 143
 144         bool has_synchronization_clock() const override
 145         {
 146                 return false;
 147         }
 148
 149         int buffer_depth() const override
 150         {
 151                 return -1;
 152         }
 153
 154         int index() const override
 155         {
 156                 return 78500 + consumer_index_;
 157         }
 158
 159         core::monitor::subject& monitor_output() override
 160         {
 161                 return monitor_subject_;
 162         }
 163
 164         // channel_consumer
 165
 166         const core::video_format_desc& get_video_format_desc()
 167         {
 168                 return format_desc_;
 169         }
 170
 171         const core::audio_channel_layout& get_audio_channel_layout()
 172         {
 173                 return channel_layout_;
 174         }
 175
 176         void block_until_first_frame_available()
 177         {
 178                 if (!frames_available_.try_acquire(1 + frames_delay_, boost::chrono::seconds(2)))
 179                         CASPAR_LOG(warning)
 180                                         << print() << L" Timed out while waiting for first frame";
 181         }
 182
 183         core::const_frame receive()
 184         {
 185                 core::const_frame frame = core::const_frame::empty();
 186
 187                 if (!is_running_)
 188                         return frame;
 189
 190                 if (frame_buffer_.try_pop(frame))
 191                         current_age_ = frame.get_age_millis();
 192
 193                 return frame;
 194         }
 195
 196         void stop()
 197         {
 198                 is_running_ = false;
 199         }
 200 };
 201
 202 class channel_producer : public core::frame_producer_base
 203 {
 204         core::monitor::subject                                          monitor_subject_;
 205
 206         const spl::shared_ptr<core::frame_factory>      frame_factory_;
 207         const core::video_format_desc                           output_format_desc_;
 208         const spl::shared_ptr<channel_consumer>         consumer_;
 209         core::constraints                                                       pixel_constraints_;
 210
 211         std::queue<core::draw_frame>                            frame_buffer_;
 212
 213 public:
 214         explicit channel_producer(const core::frame_producer_dependencies& dependecies, const spl::shared_ptr<core::video_channel>& channel, int frames_delay)
 215                 : frame_factory_(dependecies.frame_factory)
 216                 , output_format_desc_(dependecies.format_desc)
 217                 , consumer_(spl::make_shared<channel_consumer>(frames_delay))
 218         {
 219                 pixel_constraints_.width.set(output_format_desc_.width);
 220                 pixel_constraints_.height.set(output_format_desc_.height);
 221                 channel->output().add(consumer_);
 222                 consumer_->block_until_first_frame_available();
 223                 CASPAR_LOG(info) << print() << L" Initialized";
 224         }
 225
 226         ~channel_producer()
 227         {
 228                 consumer_->stop();
 229                 CASPAR_LOG(info) << print() << L" Uninitialized";
 230         }
 231
 232         // frame_producer
 233
 234         core::draw_frame receive_impl() override
 235         {
 236                 auto format_desc = consumer_->get_video_format_desc();
 237
 238                 if(frame_buffer_.size() > 0)
 239                 {
 240                         auto frame = frame_buffer_.front();
 241                         frame_buffer_.pop();
 242                         return frame;
 243                 }
 244
 245                 auto read_frame = consumer_->receive();
 246                 if(read_frame == core::const_frame::empty() || read_frame.image_data().empty())
 247                         return core::draw_frame::late();
 248
 249                 core::pixel_format_desc desc;
 250                 desc.format = core::pixel_format::bgra;
 251                 desc.planes.push_back(core::pixel_format_desc::plane(format_desc.width, format_desc.height, 4));
 252                 auto frame = frame_factory_->create_frame(this, desc, consumer_->get_audio_channel_layout());
 253
 254                 frame.audio_data().reserve(read_frame.audio_data().size());
 255                 boost::copy(read_frame.audio_data(), std::back_inserter(frame.audio_data()));
 256
 257                 fast_memcpy(frame.image_data().begin(), read_frame.image_data().begin(), read_frame.image_data().size());
 258
 259                 frame_buffer_.push(core::draw_frame(std::move(frame)));
 260
 261                 return receive_impl();
 262         }
 263
 264         std::wstring name() const override
 265         {
 266                 return L"channel-producer";
 267         }
 268
 269         std::wstring print() const override
 270         {
 271                 return L"channel-producer[]";
 272         }
 273
 274         core::constraints& pixel_constraints() override
 275         {
 276                 return pixel_constraints_;
 277         }
 278
 279         boost::property_tree::wptree info() const override
 280         {
 281                 boost::property_tree::wptree info;
 282                 info.add(L"type", L"channel-producer");
 283                 return info;
 284         }
 285
 286         core::monitor::subject& monitor_output() override
 287         {
 288                 return monitor_subject_;
 289         }
 290 };
 291
 292 spl::shared_ptr<core::frame_producer> create_channel_producer(
 293                 const core::frame_producer_dependencies& dependencies,
 294                 const spl::shared_ptr<core::video_channel>& channel,
 295                 int frames_delay)
 296 {
 297         return core::create_framerate_producer(
 298                         spl::make_shared<channel_producer>(dependencies, channel, frames_delay),
 299                         [channel] { return channel->video_format_desc().framerate; }    ,
 300                         dependencies.format_desc.framerate,
 301                         dependencies.format_desc.field_mode,
 302                         dependencies.format_desc.audio_cadence);
 303 }
 304
 305 }}