2.0. executor: Removed dangerous code.

     ogl_consumer: Refactored.

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

diff --git a/common/concurrency/executor.h b/common/concurrency/executor.h
index ad86d4a379358f11083f56df523ba66b7f703e45..31df71fcfc4f50dab9093f314643ed83e53b2131 100644 (file)
--- a/common/concurrency/executor.h
+++ b/common/concurrency/executor.h
@@ -28,6 +28,7 @@
 #include <tbb/concurrent_queue.h>\r
 \r
 #include <boost/thread.hpp>\r
+#include <boost/optional.hpp>\r
 #include <boost/noncopyable.hpp>\r
 \r
 #include <functional>\r
@@ -177,28 +178,7 @@ public:
                                        \r
                return std::move(future);               \r
        }\r
-\r
-       template<typename Func>\r
-       auto try_begin_invoke(Func&& func, task_priority priority = normal_priority) -> boost::unique_future<decltype(func())> // noexcept\r
-       {\r
-               // Create a move on copy adaptor to avoid copying the functor into the queue, tbb::concurrent_queue does not support move semantics.\r
-               auto task_adaptor = make_move_on_copy(create_task(func));\r
-               \r
-               auto future = task_adaptor.value.get_future();\r
-\r
-               if(priority == normal_priority || execution_queue_[normal_priority].try_push(nullptr))\r
-               {                       \r
-                       execution_queue_[priority].try_push([=]\r
-                       {\r
-                               try{task_adaptor.value();}\r
-                               catch(boost::task_already_started&){}\r
-                               catch(...){CASPAR_LOG_CURRENT_EXCEPTION();}\r
-                       });\r
-               }\r
-               \r
-               return std::move(future);                       \r
-       }\r
-\r
+       \r
        template<typename Func>\r
        auto invoke(Func&& func, task_priority prioriy = normal_priority) -> decltype(func()) // noexcept\r
        {\r
@@ -207,16 +187,7 @@ public:
                \r
                return begin_invoke(std::forward<Func>(func), prioriy).get();\r
        }\r
-\r
-       template<typename Func>\r
-       auto try_invoke(Func&& func, task_priority prioriy = normal_priority) -> decltype(func()) // noexcept\r
-       {\r
-               if(boost::this_thread::get_id() == thread_.get_id())  // Avoids potential deadlock.\r
-                       return func();\r
-               \r
-               return try_begin_invoke(std::forward<Func>(func), prioriy).get();\r
-       }\r
-\r
+       \r
        void yield() // noexcept\r
        {\r
                if(boost::this_thread::get_id() != thread_.get_id())  // Only yield when calling from execution thread.\r

diff --git a/modules/ogl/consumer/ogl_consumer.cpp b/modules/ogl/consumer/ogl_consumer.cpp
index 313991c5f7e0d6219fc3d8f06d4d8955625bebf1..2b15a49757a6f26ee5aad4bc28b50550d884f40c 100644 (file)
--- a/modules/ogl/consumer/ogl_consumer.cpp
+++ b/modules/ogl/consumer/ogl_consumer.cpp
@@ -78,7 +78,8 @@ struct ogl_consumer : boost::noncopyable
        size_t                                  square_width_;\r
        size_t                                  square_height_;\r
 \r
-       boost::circular_buffer<safe_ptr<core::read_frame>> frame_buffer_;\r
+       boost::circular_buffer<safe_ptr<core::read_frame>>                      input_buffer_;\r
+       tbb::concurrent_bounded_queue<safe_ptr<core::read_frame>>       frame_buffer_;\r
 \r
        executor                                executor_;\r
 public:\r
@@ -89,11 +90,14 @@ public:
                , screen_index_(screen_index)\r
                , format_desc_(format_desc_)\r
                , graph_(diagnostics::create_graph(narrow(print())))\r
-               , frame_buffer_(core::consumer_buffer_depth())\r
+               , input_buffer_(core::consumer_buffer_depth()-1)\r
                , executor_(print())\r
        {               \r
+               frame_buffer_.set_capacity(2);\r
+\r
                graph_->add_guide("frame-time", 0.5);\r
                graph_->set_color("frame-time", diagnostics::color(1.0f, 0.0f, 0.0f));\r
+               graph_->set_color("dropped-frame", diagnostics::color(0.3f, 0.6f, 0.3f));\r
                \r
                format_desc_ = format_desc;\r
 \r
@@ -180,6 +184,14 @@ public:
                        glBindBufferARB(GL_PIXEL_UNPACK_BUFFER_ARB, 0);\r
                });\r
                CASPAR_LOG(info) << print() << " Sucessfully Initialized.";\r
+\r
+               executor_.begin_invoke([this]{tick();});\r
+       }\r
+\r
+       ~ogl_consumer()\r
+       {\r
+               executor_.stop();\r
+               frame_buffer_.push(make_safe<core::read_frame>());\r
        }\r
        \r
        const core::video_format_desc& get_video_format_desc() const\r
@@ -277,35 +289,49 @@ public:
 \r
        void send(const safe_ptr<core::read_frame>& frame)\r
        {\r
-               frame_buffer_.push_back(frame);\r
+               input_buffer_.push_back(frame);\r
 \r
-               if(frame_buffer_.full())\r
-                       do_send(frame_buffer_.front());\r
+               if(input_buffer_.full())\r
+               {\r
+                       if(!frame_buffer_.try_push(input_buffer_.front()))\r
+                               graph_->add_tag("dropped-frame");\r
+               }\r
        }\r
                \r
-       void do_send(const safe_ptr<core::read_frame>& frame)\r
-       {               \r
-               executor_.try_begin_invoke([=]\r
+       std::wstring print() const\r
+       {       \r
+               return  L"ogl[" + boost::lexical_cast<std::wstring>(screen_index_) + L"|" + format_desc_.name + L"]";\r
+       }\r
+\r
+       void tick()\r
+       {\r
+               try\r
                {\r
                        perf_timer_.restart();\r
                        sf::Event e;\r
+               \r
+                       safe_ptr<core::read_frame> frame;\r
+\r
                        while(window_.GetEvent(e))\r
                        {\r
                                if(e.Type == sf::Event::Resized)\r
-                                        calculate_aspect();\r
+                                       calculate_aspect();\r
                        }\r
-                       render(frame);\r
-                       window_.Display();\r
-                       graph_->update_value("frame-time", static_cast<float>(perf_timer_.elapsed()*format_desc_.fps*0.5));\r
-               });\r
-       }\r
 \r
-       std::wstring print() const\r
-       {       \r
-               return  L"ogl[" + boost::lexical_cast<std::wstring>(screen_index_) + L"|" + format_desc_.name + L"]";\r
-       }\r
+                       frame_buffer_.pop(frame);\r
+\r
+                       if(!frame->image_data().empty())\r
+                               render(frame);\r
 \r
-       size_t buffer_depth() const{return 2;}\r
+                       window_.Display();\r
+                       graph_->update_value("frame-time", static_cast<float>(perf_timer_.elapsed()*format_desc_.fps*0.5));     \r
+               }\r
+               catch(...)\r
+               {\r
+                       CASPAR_LOG_CURRENT_EXCEPTION();\r
+               }\r
+               executor_.begin_invoke([=]{tick();});\r
+       }\r
 };\r
 \r
 \r