2.0. Async opt 2.

git-svn-id: https://casparcg.svn.sourceforge.net/svnroot/casparcg/server/branches/2.0.0.2@1093 362d55ac-95cf-4e76-9f9a-cbaa9c17b72d

diff --git a/core/mixer/mixer.cpp b/core/mixer/mixer.cpp
index fc05764580bfc17b558b0d35b4f9f916e6883f35..a9ab74701be5dceb65db532cdf9038eed76bc19d 100644 (file)
--- a/core/mixer/mixer.cpp
+++ b/core/mixer/mixer.cpp
@@ -118,7 +118,7 @@ public:
                                        [&]{image = mix_image(frames);}, \r
                                        [&]{audio = mix_audio(frames);});\r
                        \r
-                       return make_safe<read_frame>(channel_.ogl(), std::move(image.get()), std::move(audio));\r
+                       return make_safe<read_frame>(channel_.ogl(), channel_.get_format_desc().size, std::move(image), std::move(audio));\r
                }\r
                catch(...)\r
                {\r

diff --git a/core/mixer/read_frame.cpp b/core/mixer/read_frame.cpp
index 81f708efa70ace883c679091860f7ebb3701be7b..a9844a26a955da2fbbc58b6efb37107ec4c94008 100644 (file)
--- a/core/mixer/read_frame.cpp
+++ b/core/mixer/read_frame.cpp
@@ -25,26 +25,37 @@
 #include "gpu/host_buffer.h"   \r
 #include "gpu/ogl_device.h"\r
 \r
+#include <tbb/mutex.h>\r
+\r
 namespace caspar { namespace core {\r
                                                                                                                                                                                                                                                                                                                        \r
 struct read_frame::implementation : boost::noncopyable\r
 {\r
-       ogl_device&                                             ogl_;\r
-       safe_ptr<host_buffer>                   image_data_;\r
-       std::vector<int16_t>                    audio_data_;\r
+       ogl_device&                                                                             ogl_;\r
+       size_t                                                                                  size_;\r
+       boost::unique_future<safe_ptr<host_buffer>>             future_image_data_;\r
+       std::shared_ptr<host_buffer>                                    image_data_;\r
+       tbb::mutex                                                                              mutex_;\r
+       std::vector<int16_t>                                                    audio_data_;\r
 \r
 public:\r
-       implementation(ogl_device& ogl, safe_ptr<host_buffer>&& image_data, std::vector<int16_t>&& audio_data) \r
+       implementation(ogl_device& ogl, size_t size, boost::unique_future<safe_ptr<host_buffer>>&& image_data, std::vector<int16_t>&& audio_data) \r
                : ogl_(ogl)\r
-               , image_data_(std::move(image_data))\r
+               , size_(size)\r
+               , future_image_data_(std::move(image_data))\r
                , audio_data_(std::move(audio_data)){}  \r
        \r
        const boost::iterator_range<const uint8_t*> image_data()\r
        {\r
-               if(!image_data_->data())\r
                {\r
-                       image_data_->wait(ogl_);\r
-                       ogl_.invoke([=]{image_data_->map();}, high_priority);\r
+                       tbb::mutex::scoped_lock lock(mutex_);\r
+\r
+                       if(!image_data_)\r
+                       {\r
+                               image_data_ = future_image_data_.get();\r
+                               image_data_.get()->wait(ogl_);\r
+                               ogl_.invoke([=]{image_data_.get()->map();}, high_priority);\r
+                       }\r
                }\r
 \r
                auto ptr = static_cast<const uint8_t*>(image_data_->data());\r
@@ -56,8 +67,8 @@ public:
        }\r
 };\r
 \r
-read_frame::read_frame(ogl_device& ogl, safe_ptr<host_buffer>&& image_data, std::vector<int16_t>&& audio_data) \r
-       : impl_(new implementation(ogl, std::move(image_data), std::move(audio_data))){}\r
+read_frame::read_frame(ogl_device& ogl, size_t size, boost::unique_future<safe_ptr<host_buffer>>&& image_data, std::vector<int16_t>&& audio_data) \r
+       : impl_(new implementation(ogl, size, std::move(image_data), std::move(audio_data))){}\r
 read_frame::read_frame(){}\r
 const boost::iterator_range<const uint8_t*> read_frame::image_data()\r
 {\r
@@ -69,7 +80,7 @@ const boost::iterator_range<const int16_t*> read_frame::audio_data()
        return impl_ ? impl_->audio_data() : boost::iterator_range<const int16_t*>();\r
 }\r
 \r
-size_t read_frame::image_size() const{return impl_ ? impl_->image_data_->size() : 0;}\r
+size_t read_frame::image_size() const{return impl_ ? impl_->size_ : 0;}\r
 \r
 //#include <tbb/scalable_allocator.h>\r
 //#include <tbb/parallel_for.h>\r

diff --git a/core/mixer/read_frame.h b/core/mixer/read_frame.h
index 1b2999952e8a4a6d7b936fa718d2b327f9c1ad5c..9575f1fd26364120d3471af873cead8a01f95b37 100644 (file)
--- a/core/mixer/read_frame.h
+++ b/core/mixer/read_frame.h
@@ -23,6 +23,7 @@
 \r
 #include <boost/noncopyable.hpp>\r
 #include <boost/range/iterator_range.hpp>\r
+#include <boost/thread/future.hpp>\r
 \r
 #include <cstdint>\r
 #include <memory>\r
@@ -37,7 +38,7 @@ class read_frame : boost::noncopyable
 {\r
 public:\r
        read_frame();\r
-       read_frame(ogl_device& ogl, safe_ptr<host_buffer>&& image_data, std::vector<int16_t>&& audio_data);\r
+       read_frame(ogl_device& ogl, size_t size, boost::unique_future<safe_ptr<host_buffer>>&& image_data, std::vector<int16_t>&& audio_data);\r
 \r
        virtual const boost::iterator_range<const uint8_t*> image_data();\r
        virtual const boost::iterator_range<const int16_t*> audio_data();\r

diff --git a/core/producer/separated/separated_producer.cpp b/core/producer/separated/separated_producer.cpp
index 42b465d7057c00eace94ccf99a21fcc5b37c0cff..95bf3840666183911a8fe1aa5200c57752e1db59 100644 (file)
--- a/core/producer/separated/separated_producer.cpp
+++ b/core/producer/separated/separated_producer.cpp
@@ -23,6 +23,8 @@
 \r
 #include <core/producer/frame/basic_frame.h>\r
 \r
+#include <tbb/parallel_invoke.h>\r
+\r
 namespace caspar { namespace core {    \r
 \r
 struct separated_producer : public frame_producer\r
@@ -44,11 +46,19 @@ struct separated_producer : public frame_producer
        \r
        virtual safe_ptr<basic_frame> receive(int hints)\r
        {\r
-               if(fill_ == core::basic_frame::late())\r
-                       fill_ = receive_and_follow(fill_producer_, hints);\r
-       \r
-               if(key_ == core::basic_frame::late())\r
-                       key_ = receive_and_follow(key_producer_, hints | ALPHA_HINT);\r
+               tbb::parallel_invoke\r
+               (\r
+                       [&]\r
+                       {\r
+                               if(fill_ == core::basic_frame::late())\r
+                                       fill_ = receive_and_follow(fill_producer_, hints);\r
+                       },\r
+                       [&]\r
+                       {\r
+                               if(key_ == core::basic_frame::late())\r
+                                       key_ = receive_and_follow(key_producer_, hints | ALPHA_HINT);\r
+                       }\r
+               );\r
 \r
                if(fill_ == basic_frame::eof())\r
                        return basic_frame::eof();\r

diff --git a/core/producer/stage.cpp b/core/producer/stage.cpp
index 404ccd645a626953dff37db51241495563ad9fa8..d8e1d262ca7c394c8c068cec131f51e1e6abc41e 100644 (file)
--- a/core/producer/stage.cpp
+++ b/core/producer/stage.cpp
@@ -33,6 +33,8 @@
 \r
 #include <boost/foreach.hpp>\r
 \r
+#include <tbb/parallel_for_each.h>\r
+\r
 #include <map>\r
 #include <set>\r
 \r
@@ -74,7 +76,7 @@ public:
 \r
 struct stage::implementation : boost::noncopyable\r
 {              \r
-       std::map<int, layer>                                            layers_;                \r
+       std::map<int, layer>                                            layers_;        \r
        video_channel_context&                                          channel_;\r
 public:\r
        implementation(video_channel_context& video_channel)  \r
@@ -88,8 +90,15 @@ public:
 \r
                try\r
                {\r
-                       BOOST_FOREACH(auto& layer, layers_)\r
+                       // Allocate placeholders.\r
+                       BOOST_FOREACH(auto layer, layers_)\r
+                               frames[layer.first] = basic_frame::empty();\r
+\r
+                       // Render layers\r
+                       tbb::parallel_for_each(layers_.begin(), layers_.end(), [&](std::map<int, layer>::value_type& layer)\r
+                       {\r
                                frames[layer.first] = layer.second.receive();\r
+                       });\r
                }\r
                catch(...)\r
                {\r

diff --git a/core/producer/transition/transition_producer.cpp b/core/producer/transition/transition_producer.cpp
index e25bb0c95de5cb81a96568611344ef7d2263c608..073e6b5176a78d3169f11b52d8efe6dda705d77c 100644 (file)
--- a/core/producer/transition/transition_producer.cpp
+++ b/core/producer/transition/transition_producer.cpp
@@ -27,6 +27,8 @@
 #include <core/producer/frame/image_transform.h>\r
 #include <core/producer/frame/audio_transform.h>\r
 \r
+#include <tbb/parallel_invoke.h>\r
+\r
 namespace caspar { namespace core {    \r
 \r
 struct transition_producer : public frame_producer\r
@@ -66,13 +68,24 @@ struct transition_producer : public frame_producer
                if(current_frame_++ >= info_.duration)\r
                        return basic_frame::eof();\r
                \r
-               auto dest = receive_and_follow(dest_producer_, hints);\r
-               if(dest == core::basic_frame::late())\r
-                       dest = dest_producer_->last_frame();\r
-               \r
-               auto source = receive_and_follow(source_producer_, hints);\r
-               if(source == core::basic_frame::late())\r
-                       source = source_producer_->last_frame();\r
+               auto dest       = core::basic_frame::empty();\r
+               auto source     = core::basic_frame::empty();\r
+\r
+               tbb::parallel_invoke\r
+               (\r
+                       [&]\r
+                       {\r
+                               dest = receive_and_follow(dest_producer_, hints);\r
+                               if(dest == core::basic_frame::late())\r
+                                       dest = dest_producer_->last_frame();\r
+                       },\r
+                       [&]\r
+                       {\r
+                               source = receive_and_follow(source_producer_, hints);\r
+                               if(source == core::basic_frame::late())\r
+                                       source = source_producer_->last_frame();\r
+                       }\r
+               );\r
 \r
                return last_frame_ = compose(dest, source);\r
        }\r

diff --git a/modules/ffmpeg/producer/ffmpeg_producer.cpp b/modules/ffmpeg/producer/ffmpeg_producer.cpp
index bae7396cce132b97ef08ac36a1d1dca433b33824..a533d056e635fb158d96bee966731a3f75df7781 100644 (file)
--- a/modules/ffmpeg/producer/ffmpeg_producer.cpp
+++ b/modules/ffmpeg/producer/ffmpeg_producer.cpp
@@ -70,8 +70,6 @@ struct ffmpeg_producer : public core::frame_producer
 \r
        safe_ptr<core::basic_frame>                                             last_frame_;\r
        \r
-       tbb::task_group                                                                 tasks_;\r
-\r
 public:\r
        explicit ffmpeg_producer(const safe_ptr<core::frame_factory>& frame_factory, const std::wstring& filename, const std::wstring& filter, bool loop, int start, int length) \r
                : filename_(filename)\r
@@ -94,19 +92,16 @@ public:
                for(int n = 0; n < 128 && muxer_.size() < 2; ++n)\r
                        decode_frame(0);\r
        }\r
-\r
-       ~ffmpeg_producer()\r
-       {\r
-               tasks_.cancel();\r
-               tasks_.wait();\r
-       }\r
-       \r
+               \r
        virtual safe_ptr<core::basic_frame> receive(int hints)\r
        {\r
-               tasks_.wait();\r
-\r
                auto frame = core::basic_frame::late();\r
                \r
+               frame_timer_.restart();\r
+\r
+               for(int n = 0; n < 64 && muxer_.size() < 2; ++n)\r
+                       decode_frame(hints);\r
+               \r
                if(!muxer_.empty())\r
                        frame = last_frame_ = muxer_.pop();     \r
                else\r
@@ -119,16 +114,8 @@ public:
                                ++nb_frames_;           \r
                        }\r
                }\r
-\r
-               tasks_.run([=]\r
-               {\r
-                       frame_timer_.restart();\r
-\r
-                       for(int n = 0; n < 64 && muxer_.size() < 2; ++n)\r
-                               decode_frame(hints);\r
-\r
-                       graph_->update_value("frame-time", static_cast<float>(frame_timer_.elapsed()*format_desc_.fps*0.5));\r
-               });\r
+               \r
+               graph_->update_value("frame-time", static_cast<float>(frame_timer_.elapsed()*format_desc_.fps*0.5));\r
                \r
                return frame;\r
        }\r