git.sesse.net Git - casparcg/blob - core/producer/frame_producer_device.cpp

   1 #include "../StdAfx.h"\r
   2 \r
   3 #include "frame_producer_device.h"\r
   4 \r
   5 #include "../mixer/frame/draw_frame.h"\r
   6 #include "../mixer/frame_factory.h"\r
   7 \r
   8 #include "layer.h"\r
   9 \r
  10 #include <common/concurrency/executor.h>\r
  11 \r
  12 #include <boost/range/algorithm_ext/erase.hpp>\r
  13 #include <boost/lexical_cast.hpp>\r
  14 \r
  15 #include <tbb/parallel_for.h>\r
  16 #include <tbb/spin_mutex.h>\r
  17 \r
  18 #include <array>\r
  19 #include <memory>\r
  20 \r
  21 namespace caspar { namespace core {\r
  22 \r
  23 struct frame_producer_device::implementation : boost::noncopyable\r
  24 {               \r
  25         std::array<layer, frame_producer_device::MAX_LAYER> layers_;            \r
  26 \r
  27         tbb::spin_mutex output_mutex_;\r
  28         output_func output_;\r
  29 \r
  30         const safe_ptr<frame_factory> factory_;\r
  31         \r
  32         mutable executor executor_;\r
  33 public:\r
  34         implementation(const safe_ptr<frame_factory>& factory, const output_func& output)  \r
  35                 : factory_(factory)\r
  36                 , output_(output)\r
  37         {\r
  38                 executor_.start();\r
  39                 executor_.begin_invoke([=]{tick();});\r
  40         }\r
  41 \r
  42         ~implementation()\r
  43         {\r
  44                 CASPAR_LOG(info) << "Shutting down producer-device.";\r
  45         }\r
  46                                         \r
  47         void tick()\r
  48         {               \r
  49                 auto frames = draw();\r
  50                 output_func output;\r
  51                 {\r
  52                         tbb::spin_mutex::scoped_lock lock(output_mutex_);\r
  53                         output = output_;\r
  54                 }\r
  55                 output(frames);\r
  56                 executor_.begin_invoke([=]{tick();});\r
  57         }\r
  58         \r
  59         std::vector<safe_ptr<draw_frame>> draw()\r
  60         {       \r
  61                 std::vector<safe_ptr<draw_frame>> frames(layers_.size(), draw_frame::empty());\r
  62                 tbb::parallel_for(tbb::blocked_range<size_t>(0, frames.size(), 1), // Use grain-size = 1.\r
  63                 [&](const tbb::blocked_range<size_t>& r)\r
  64                 {\r
  65                         for(size_t i = r.begin(); i != r.end(); ++i)\r
  66                         {\r
  67                                 frames[i] = layers_[i].receive();\r
  68                                 frames[i]->set_layer_index(i);\r
  69                         }\r
  70                 });             \r
  71                 boost::range::remove_erase(frames, draw_frame::empty());\r
  72                 return frames;\r
  73         }\r
  74 \r
  75         void load(size_t index, const safe_ptr<frame_producer>& producer, bool play_on_load)\r
  76         {\r
  77                 producer->initialize(factory_);\r
  78                 executor_.invoke([&]\r
  79                 {\r
  80                         layers_.at(index).load(producer, play_on_load);\r
  81                 });\r
  82         }\r
  83                         \r
  84         void preview(size_t index, const safe_ptr<frame_producer>& producer)\r
  85         {\r
  86                 producer->initialize(factory_);\r
  87                 executor_.invoke([&]\r
  88                 {                       \r
  89                         layers_.at(index).preview(producer);\r
  90                 });\r
  91         }\r
  92 \r
  93         void pause(size_t index)\r
  94         {               \r
  95                 executor_.invoke([&]\r
  96                 {\r
  97                         layers_.at(index).pause();\r
  98                 });\r
  99         }\r
 100 \r
 101         void play(size_t index)\r
 102         {               \r
 103                 executor_.invoke([&]\r
 104                 {\r
 105                         layers_.at(index).play();\r
 106                 });\r
 107         }\r
 108 \r
 109         void stop(size_t index)\r
 110         {               \r
 111                 executor_.invoke([&]\r
 112                 {\r
 113                         layers_.at(index).stop();\r
 114                 });\r
 115         }\r
 116 \r
 117         void clear(size_t index)\r
 118         {\r
 119                 executor_.invoke([&]\r
 120                 {\r
 121                         layers_.at(index) = std::move(layer());\r
 122                 });\r
 123         }\r
 124                 \r
 125         void clear()\r
 126         {\r
 127                 executor_.invoke([&]\r
 128                 {\r
 129                         for(auto it = layers_.begin(); it != layers_.end(); ++it)\r
 130                                 *it = std::move(layer());\r
 131                 });\r
 132         }       \r
 133         \r
 134         void swap_layer(size_t index, size_t other_index)\r
 135         {\r
 136                 executor_.invoke([&]\r
 137                 {\r
 138                         layers_.at(index).swap(layers_[other_index]);\r
 139                 });\r
 140         }\r
 141 \r
 142         void swap_layer(size_t index, size_t other_index, frame_producer_device& other)\r
 143         {\r
 144                 executor_.invoke([&]\r
 145                 {\r
 146                         layers_.at(index).swap(other.impl_->layers_.at(other_index));\r
 147                 });\r
 148         }\r
 149 \r
 150         void swap_output(frame_producer_device& other)\r
 151         {\r
 152                 if(other.impl_.get() == this) // Avoid deadlock.\r
 153                         return;\r
 154 \r
 155                 tbb::spin_mutex::scoped_lock lock1(output_mutex_);\r
 156                 tbb::spin_mutex::scoped_lock lock2(other.impl_->output_mutex_);\r
 157                 output_.swap(other.impl_->output_);\r
 158         }\r
 159         \r
 160         boost::unique_future<safe_ptr<frame_producer>> foreground(size_t index) const\r
 161         {\r
 162                 return executor_.begin_invoke([=]() -> safe_ptr<frame_producer>\r
 163                 {                       \r
 164                         return layers_.at(index).foreground();\r
 165                 });\r
 166         }\r
 167 };\r
 168 \r
 169 frame_producer_device::frame_producer_device(const safe_ptr<frame_factory>& factory, const output_func& output) : impl_(new implementation(factory, output)){}\r
 170 frame_producer_device::frame_producer_device(frame_producer_device&& other) : impl_(std::move(other.impl_)){}\r
 171 void frame_producer_device::load(size_t index, const safe_ptr<frame_producer>& producer, bool play_on_load){impl_->load(index, producer, play_on_load);}\r
 172 void frame_producer_device::preview(size_t index, const safe_ptr<frame_producer>& producer){impl_->preview(index, producer);}\r
 173 void frame_producer_device::pause(size_t index){impl_->pause(index);}\r
 174 void frame_producer_device::play(size_t index){impl_->play(index);}\r
 175 void frame_producer_device::stop(size_t index){impl_->stop(index);}\r
 176 void frame_producer_device::clear(size_t index){impl_->clear(index);}\r
 177 void frame_producer_device::clear(){impl_->clear();}\r
 178 void frame_producer_device::swap_layer(size_t index, size_t other_index){impl_->swap_layer(index, other_index);}\r
 179 void frame_producer_device::swap_layer(size_t index, size_t other_index, frame_producer_device& other){impl_->swap_layer(index, other_index, other);}\r
 180 void frame_producer_device::swap_output(frame_producer_device& other){impl_->swap_output(other);}\r
 181 boost::unique_future<safe_ptr<frame_producer>> frame_producer_device::foreground(size_t index) const{   return impl_->foreground(index);}\r
 182 }}