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