]> git.sesse.net Git - casparcg/blobdiff - core/producer/frame_producer_device.cpp
2.0.0.2: All pipeline stages share thread context. This is a better and simpler desig...
[casparcg] / core / producer / frame_producer_device.cpp
index cb6b08cf55eebdd2800fe48c1aa21fb2cb187919..a788a30b2a7ae132d394ab6674c5238ecbeb068a 100644 (file)
 #include "../StdAfx.h"\r
 \r
 #include "frame_producer_device.h"\r
-#include "destroy_producer_proxy.h"\r
+\r
+#include "layer.h"\r
 \r
 #include <core/producer/frame/basic_frame.h>\r
 #include <core/producer/frame/frame_factory.h>\r
 \r
 #include <common/diagnostics/graph.h>\r
-\r
-#include "layer.h"\r
-\r
 #include <common/concurrency/executor.h>\r
 \r
-#include <boost/range/algorithm_ext/erase.hpp>\r
-#include <boost/lexical_cast.hpp>\r
 #include <boost/timer.hpp>\r
 \r
 #include <tbb/parallel_for.h>\r
 \r
-#include <array>\r
-#include <memory>\r
 #include <map>\r
 \r
 namespace caspar { namespace core {\r
+               \r
+void destroy_producer(safe_ptr<frame_producer>& producer)\r
+{\r
+       if(!producer.unique())\r
+               CASPAR_LOG(warning) << producer->print() << L" Not destroyed on safe asynchronous destruction thread.";\r
+               \r
+       producer = frame_producer::empty();\r
+}\r
+\r
+class destroy_producer_proxy : public frame_producer\r
+{\r
+       safe_ptr<frame_producer> producer_;\r
+       executor& destroy_context_;\r
+public:\r
+       destroy_producer_proxy(executor& destroy_context, const safe_ptr<frame_producer>& producer) \r
+               : producer_(producer)\r
+               , destroy_context_(destroy_context){}\r
+\r
+       ~destroy_producer_proxy()\r
+       {               \r
+               if(destroy_context_.size() > 4)\r
+                       CASPAR_LOG(error) << L" Potential destroyer deadlock.";\r
+\r
+               destroy_context_.begin_invoke(std::bind(&destroy_producer, std::move(producer_)));\r
+       }\r
+\r
+       virtual safe_ptr<basic_frame>           receive()                                                                                                               {return core::receive(producer_);}\r
+       virtual std::wstring                            print() const                                                                                                   {return producer_->print();}\r
+       virtual void                                            param(const std::wstring& str)                                                                  {producer_->param(str);}\r
+       virtual safe_ptr<frame_producer>        get_following_producer() const                                                                  {return producer_->get_following_producer();}\r
+       virtual void                                            set_leading_producer(const safe_ptr<frame_producer>& producer)  {producer_->set_leading_producer(producer);}\r
+};\r
 \r
 struct frame_producer_device::implementation : boost::noncopyable\r
 {              \r
        std::map<int, layer>             layers_;               \r
        const video_format_desc          format_desc_;  \r
-       safe_ptr<diagnostics::graph> diag_;\r
        const output_t                           output_;\r
-\r
+       \r
+       safe_ptr<diagnostics::graph> diag_;\r
        boost::timer                             frame_timer_;\r
        boost::timer                             tick_timer_;\r
        boost::timer                             output_timer_;\r
        \r
-       mutable executor                         executor_;\r
+       executor&                                        context_;\r
+       executor&                                        destroy_context_;\r
 public:\r
-       implementation(const video_format_desc& format_desc, const output_t& output)  \r
+       implementation(executor& context, executor& destroy_context, const video_format_desc& format_desc, const output_t& output)  \r
                : format_desc_(format_desc)\r
                , diag_(diagnostics::create_graph(std::string("frame_producer_device")))\r
-               , executor_(L"frame_producer_device")\r
+               , context_(context)\r
+               , destroy_context_(destroy_context)\r
                , output_(output)\r
        {\r
                diag_->add_guide("frame-time", 0.5f);   \r
                diag_->set_color("frame-time", diagnostics::color(1.0f, 0.0f, 0.0f));\r
                diag_->set_color("tick-time", diagnostics::color(0.1f, 0.7f, 0.8f));\r
                diag_->set_color("output-time", diagnostics::color(0.5f, 1.0f, 0.2f));\r
-               executor_.begin_invoke([]\r
-               {\r
-                       SetThreadPriority(GetCurrentThread(), ABOVE_NORMAL_PRIORITY_CLASS);\r
-               });\r
-               executor_.begin_invoke([=]{tick();});           \r
+\r
+               context_.begin_invoke([=]{tick();});            \r
        }\r
                        \r
        void tick()\r
@@ -88,7 +113,7 @@ public:
                        CASPAR_LOG_CURRENT_EXCEPTION();\r
                }\r
 \r
-               executor_.begin_invoke([=]{tick();});\r
+               context_.begin_invoke([=]{tick();});\r
        }\r
                        \r
        std::map<int, safe_ptr<basic_frame>> render()\r
@@ -104,9 +129,9 @@ public:
                        frames[pair.first] = pair.second.receive();\r
                });\r
                \r
-               diag_->update_value("frame-time", static_cast<float>(frame_timer_.elapsed()*format_desc_.fps*0.5));\r
+               diag_->update_value("frame-time", frame_timer_.elapsed()*format_desc_.fps*0.5);\r
 \r
-               diag_->update_value("tick-time", static_cast<float>(tick_timer_.elapsed()*format_desc_.fps*0.5));\r
+               diag_->update_value("tick-time", tick_timer_.elapsed()*format_desc_.fps*0.5);\r
                tick_timer_.restart();\r
 \r
                return frames;\r
@@ -114,37 +139,37 @@ public:
 \r
        void load(int index, const safe_ptr<frame_producer>& producer, bool preview)\r
        {\r
-               executor_.invoke([&]{layers_[index].load(make_safe<destroy_producer_proxy>(producer), preview);});\r
+               context_.invoke([&]{layers_[index].load(make_safe<destroy_producer_proxy>(destroy_context_, producer), preview);});\r
        }\r
 \r
        void pause(int index)\r
        {               \r
-               executor_.invoke([&]{layers_[index].pause();});\r
+               context_.invoke([&]{layers_[index].pause();});\r
        }\r
 \r
        void play(int index)\r
        {               \r
-               executor_.invoke([&]{layers_[index].play();});\r
+               context_.invoke([&]{layers_[index].play();});\r
        }\r
 \r
        void stop(int index)\r
        {               \r
-               executor_.invoke([&]{layers_[index].stop();});\r
+               context_.invoke([&]{layers_[index].stop();});\r
        }\r
 \r
        void clear(int index)\r
        {\r
-               executor_.invoke([&]{layers_.erase(index);});\r
+               context_.invoke([&]{layers_.erase(index);});\r
        }\r
                \r
        void clear()\r
        {\r
-               executor_.invoke([&]{layers_.clear();});\r
+               context_.invoke([&]{layers_.clear();});\r
        }       \r
        \r
        void swap_layer(int index, size_t other_index)\r
        {\r
-               executor_.invoke([&]{layers_[index].swap(layers_[other_index]);});\r
+               context_.invoke([&]{layers_[index].swap(layers_[other_index]);});\r
        }\r
 \r
        void swap_layer(int index, size_t other_index, frame_producer_device& other)\r
@@ -158,7 +183,7 @@ public:
 \r
                        auto func = [&]{layers_[index].swap(other.impl_->layers_[other_index]);};\r
                \r
-                       executor_.invoke([&]{other.impl_->executor_.invoke(func);});\r
+                       context_.invoke([&]{other.impl_->context_.invoke(func);});\r
                }\r
        }\r
 \r
@@ -185,21 +210,22 @@ public:
                        });                                     \r
                };\r
                \r
-               executor_.invoke([&]{other.impl_->executor_.invoke(func);});\r
+               context_.invoke([&]{other.impl_->context_.invoke(func);});\r
        }\r
        \r
        boost::unique_future<safe_ptr<frame_producer>> foreground(int index)\r
        {\r
-               return executor_.begin_invoke([=]{return layers_[index].foreground();});\r
+               return context_.begin_invoke([=]{return layers_[index].foreground();});\r
        }\r
        \r
        boost::unique_future<safe_ptr<frame_producer>> background(int index)\r
        {\r
-               return executor_.begin_invoke([=]{return layers_[index].background();});\r
+               return context_.begin_invoke([=]{return layers_[index].background();});\r
        }\r
 };\r
 \r
-frame_producer_device::frame_producer_device(const video_format_desc& format_desc, const output_t& output) : impl_(new implementation(format_desc, output)){}\r
+frame_producer_device::frame_producer_device(executor& context, executor& destroy_context, const video_format_desc& format_desc, const output_t& output)\r
+       : impl_(new implementation(context, destroy_context, format_desc, output)){}\r
 frame_producer_device::frame_producer_device(frame_producer_device&& other) : impl_(std::move(other.impl_)){}\r
 void frame_producer_device::swap(frame_producer_device& other){impl_->swap(other);}\r
 void frame_producer_device::load(int index, const safe_ptr<frame_producer>& producer, bool preview){impl_->load(index, producer, preview);}\r