git.sesse.net Git - casparcg/blob - core/consumer/output.cpp

   1 /*\r
   2 * Copyright (c) 2011 Sveriges Television AB <info@casparcg.com>\r
   3 *\r
   4 * This file is part of CasparCG (www.casparcg.com).\r
   5 *\r
   6 * CasparCG is free software: you can redistribute it and/or modify\r
   7 * it under the terms of the GNU General Public License as published by\r
   8 * the Free Software Foundation, either version 3 of the License, or\r
   9 * (at your option) any later version.\r
  10 *\r
  11 * CasparCG is distributed in the hope that it will be useful,\r
  12 * but WITHOUT ANY WARRANTY; without even the implied warranty of\r
  13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the\r
  14 * GNU General Public License for more details.\r
  15 *\r
  16 * You should have received a copy of the GNU General Public License\r
  17 * along with CasparCG. If not, see <http://www.gnu.org/licenses/>.\r
  18 *\r
  19 * Author: Robert Nagy, ronag89@gmail.com\r
  20 */\r
  21 \r
  22 #include "../StdAfx.h"\r
  23 \r
  24 #ifdef _MSC_VER\r
  25 #pragma warning (disable : 4244)\r
  26 #endif\r
  27 \r
  28 #include "output.h"\r
  29 \r
  30 #include "frame_consumer.h"\r
  31 \r
  32 #include "../video_format.h"\r
  33 #include "../frame/frame.h"\r
  34 \r
  35 #include <common/assert.h>\r
  36 #include <common/future.h>\r
  37 #include <common/executor.h>\r
  38 #include <common/diagnostics/graph.h>\r
  39 #include <common/prec_timer.h>\r
  40 #include <common/memshfl.h>\r
  41 #include <common/env.h>\r
  42 \r
  43 #include <boost/circular_buffer.hpp>\r
  44 #include <boost/lexical_cast.hpp>\r
  45 #include <boost/property_tree/ptree.hpp>\r
  46 #include <boost/range/algorithm.hpp>\r
  47 #include <boost/range/adaptors.hpp>\r
  48 #include <boost/timer.hpp>\r
  49 \r
  50 namespace caspar { namespace core {\r
  51         \r
  52 class port : public monitor::observable\r
  53 {\r
  54         port(const port&);\r
  55         port& operator=(const port&);\r
  56 \r
  57         monitor::basic_subject                          event_subject_;\r
  58         std::shared_ptr<frame_consumer>         consumer_;\r
  59         int                                                                     index_;\r
  60         int                                                                     channel_index_;\r
  61 public:\r
  62         port(int index, int channel_index, spl::shared_ptr<frame_consumer> consumer)\r
  63                 : event_subject_(monitor::path("port") % index)\r
  64                 , consumer_(std::move(consumer))\r
  65                 , index_(index)\r
  66                 , channel_index_(channel_index)\r
  67         {\r
  68                 consumer_->subscribe(event_subject_);\r
  69         }\r
  70 \r
  71         port(port&& other)\r
  72                 : event_subject_(std::move(other.event_subject_))\r
  73                 , consumer_(std::move(other.consumer_))\r
  74                 , index_(other.index_)\r
  75                 , channel_index_(other.channel_index_)\r
  76         {\r
  77         }\r
  78 \r
  79         port& operator=(port&& other)\r
  80         {\r
  81                 event_subject_  = std::move(other.event_subject_);\r
  82                 consumer_               = std::move(other.consumer_);\r
  83                 index_                  = std::move(other.index_);\r
  84                 channel_index_  = std::move(other.channel_index_);\r
  85         }\r
  86 \r
  87         void video_format_desc(const struct video_format_desc& format_desc)\r
  88         {\r
  89                 consumer_->initialize(format_desc, channel_index_);\r
  90         }\r
  91                 \r
  92         bool send(class const_frame frame)\r
  93         {\r
  94                 event_subject_ << monitor::event("type") % consumer_->name();\r
  95                 return consumer_->send(frame);\r
  96         }\r
  97         \r
  98         int index() const\r
  99         {\r
 100                 return index_;\r
 101         }\r
 102 \r
 103         int buffer_depth() const\r
 104         {\r
 105                 return consumer_->buffer_depth();\r
 106         }\r
 107 \r
 108         bool has_synchronization_clock() const\r
 109         {\r
 110                 return consumer_->has_synchronization_clock();\r
 111         }\r
 112 \r
 113         boost::property_tree::wptree info() const\r
 114         {\r
 115                 return consumer_->info();\r
 116         }\r
 117 \r
 118         void subscribe(const monitor::observable::observer_ptr& o) override\r
 119         {\r
 120                 event_subject_.subscribe(o);\r
 121         }\r
 122 \r
 123         void unsubscribe(const monitor::observable::observer_ptr& o) override\r
 124         {\r
 125                 event_subject_.unsubscribe(o);\r
 126         }       \r
 127 };\r
 128 \r
 129 struct output::impl\r
 130 {               \r
 131         spl::shared_ptr<diagnostics::graph>     graph_;\r
 132         monitor::basic_subject                          event_subject_;\r
 133         const int                                                       channel_index_;\r
 134         video_format_desc                                       format_desc_;\r
 135         std::map<int, port>                                     ports_; \r
 136         prec_timer                                                      sync_timer_;\r
 137         boost::circular_buffer<const_frame>     frames_;\r
 138         executor                                                        executor_;              \r
 139 public:\r
 140         impl(spl::shared_ptr<diagnostics::graph> graph, const video_format_desc& format_desc, int channel_index) \r
 141                 : graph_(std::move(graph))\r
 142                 , event_subject_("output")\r
 143                 , channel_index_(channel_index)\r
 144                 , format_desc_(format_desc)\r
 145                 , executor_(L"output")\r
 146         {\r
 147                 graph_->set_color("consume-time", diagnostics::color(1.0f, 0.4f, 0.0f, 0.8));\r
 148         }       \r
 149         \r
 150         void add(int index, spl::shared_ptr<frame_consumer> consumer)\r
 151         {               \r
 152                 remove(index);\r
 153 \r
 154                 consumer->initialize(format_desc_, channel_index_);\r
 155                 \r
 156                 executor_.begin_invoke([this, index, consumer]\r
 157                 {                       \r
 158                         port p(index, channel_index_, std::move(consumer));\r
 159                         p.subscribe(event_subject_);\r
 160                         ports_.insert(std::make_pair(index, std::move(p)));\r
 161                 }, task_priority::high_priority);\r
 162         }\r
 163 \r
 164         void add(const spl::shared_ptr<frame_consumer>& consumer)\r
 165         {\r
 166                 add(consumer->index(), consumer);\r
 167         }\r
 168 \r
 169         void remove(int index)\r
 170         {               \r
 171                 executor_.begin_invoke([=]\r
 172                 {\r
 173                         auto it = ports_.find(index);\r
 174                         if(it != ports_.end())\r
 175                                 ports_.erase(it);                                       \r
 176                 }, task_priority::high_priority);\r
 177         }\r
 178 \r
 179         void remove(const spl::shared_ptr<frame_consumer>& consumer)\r
 180         {\r
 181                 remove(consumer->index());\r
 182         }\r
 183         \r
 184         void video_format_desc(const core::video_format_desc& format_desc)\r
 185         {\r
 186                 executor_.invoke([&]\r
 187                 {\r
 188                         if(format_desc_ == format_desc)\r
 189                                 return;\r
 190 \r
 191                         auto it = ports_.begin();\r
 192                         while(it != ports_.end())\r
 193                         {                                               \r
 194                                 try\r
 195                                 {\r
 196                                         it->second.video_format_desc(format_desc);\r
 197                                         ++it;\r
 198                                 }\r
 199                                 catch(...)\r
 200                                 {\r
 201                                         CASPAR_LOG_CURRENT_EXCEPTION();\r
 202                                         ports_.erase(it++);\r
 203                                 }\r
 204                         }\r
 205                         \r
 206                         format_desc_ = format_desc;\r
 207                         frames_.clear();\r
 208                 });\r
 209         }\r
 210 \r
 211         std::pair<int, int> minmax_buffer_depth() const\r
 212         {               \r
 213                 if(ports_.empty())\r
 214                         return std::make_pair(0, 0);\r
 215                 \r
 216                 auto buffer_depths = ports_ | \r
 217                                                          boost::adaptors::map_values | // std::function is MSVC workaround\r
 218                                                          boost::adaptors::transformed(std::function<int(const port&)>([](const port& p){return p.buffer_depth();})); \r
 219                 \r
 220 \r
 221                 return std::make_pair(*boost::range::min_element(buffer_depths), *boost::range::max_element(buffer_depths));\r
 222         }\r
 223 \r
 224         bool has_synchronization_clock() const\r
 225         {\r
 226                 return boost::range::count_if(ports_ | boost::adaptors::map_values,\r
 227                                                                           [](const port& p){return p.has_synchronization_clock();}) > 0;\r
 228         }\r
 229                 \r
 230         void operator()(const_frame input_frame, const core::video_format_desc& format_desc)\r
 231         {\r
 232                 video_format_desc(format_desc);\r
 233 \r
 234                 executor_.invoke([=]\r
 235                 {                       \r
 236                         boost::timer frame_timer;\r
 237 \r
 238                         if(!has_synchronization_clock())\r
 239                                 sync_timer_.tick(1.0/format_desc_.fps);\r
 240                                 \r
 241                         if(input_frame.size() != format_desc_.size)\r
 242                         {\r
 243                                 sync_timer_.tick(1.0/format_desc_.fps);\r
 244                                 return;\r
 245                         }\r
 246                                         \r
 247                         auto minmax = minmax_buffer_depth();\r
 248 \r
 249                         frames_.set_capacity(std::max(2, minmax.second - minmax.first) + 1); // std::max(2, x) since we want to guarantee some pipeline depth for asycnhronous mixer read-back.\r
 250                         frames_.push_back(input_frame);\r
 251 \r
 252                         if(!frames_.full())\r
 253                                 return;\r
 254 \r
 255                         for(auto it = ports_.begin(); it != ports_.end();)\r
 256                         {\r
 257                                 auto& port              = it->second;\r
 258                                 auto& frame             = frames_.at(port.buffer_depth()-minmax.first);\r
 259                                         \r
 260                                 try\r
 261                                 {\r
 262                                         if(port.send(frame))\r
 263                                                 ++it;\r
 264                                         else\r
 265                                                 ports_.erase(it++);                                     \r
 266                                 }\r
 267                                 catch(...)\r
 268                                 {\r
 269                                         CASPAR_LOG_CURRENT_EXCEPTION();\r
 270                                         ports_.erase(it++);\r
 271                                 }\r
 272                         }\r
 273 \r
 274                         graph_->set_value("consume-time", frame_timer.elapsed()*format_desc.fps*0.5);\r
 275                 });\r
 276         }\r
 277 \r
 278         std::wstring print() const\r
 279         {\r
 280                 return L"output[" + boost::lexical_cast<std::wstring>(channel_index_) + L"]";\r
 281         }\r
 282 \r
 283         boost::unique_future<boost::property_tree::wptree> info()\r
 284         {\r
 285                 return std::move(executor_.begin_invoke([&]() -> boost::property_tree::wptree\r
 286                 {                       \r
 287                         boost::property_tree::wptree info;\r
 288                         BOOST_FOREACH(auto& port, ports_ | boost::adaptors::map_values)\r
 289                         {\r
 290                                 info.add_child(L"consumers.consumer", port.info())\r
 291                                         .add(L"index", port.index()); \r
 292                         }\r
 293                         return info;\r
 294                 }, task_priority::high_priority));\r
 295         }\r
 296 };\r
 297 \r
 298 output::output(spl::shared_ptr<diagnostics::graph> graph, const video_format_desc& format_desc, int channel_index) : impl_(new impl(std::move(graph), format_desc, channel_index)){}\r
 299 void output::add(int index, const spl::shared_ptr<frame_consumer>& consumer){impl_->add(index, consumer);}\r
 300 void output::add(const spl::shared_ptr<frame_consumer>& consumer){impl_->add(consumer);}\r
 301 void output::remove(int index){impl_->remove(index);}\r
 302 void output::remove(const spl::shared_ptr<frame_consumer>& consumer){impl_->remove(consumer);}\r
 303 boost::unique_future<boost::property_tree::wptree> output::info() const{return impl_->info();}\r
 304 void output::operator()(const_frame frame, const video_format_desc& format_desc){(*impl_)(std::move(frame), format_desc);}\r
 305 void output::subscribe(const monitor::observable::observer_ptr& o) {impl_->event_subject_.subscribe(o);}\r
 306 void output::unsubscribe(const monitor::observable::observer_ptr& o) {impl_->event_subject_.unsubscribe(o);}\r
 307 }}