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