From: Ronag <Ronag@362d55ac-95cf-4e76-9f9a-cbaa9c17b72d>
Date: Thu, 25 Aug 2011 20:39:15 +0000 (+0000)
Subject: 2.0. executor: Removed dangerous code.
X-Git-Tag: 2.0.1~57
X-Git-Url: https://git.sesse.net/?p=casparcg;a=commitdiff_plain;h=f8c5ab340ceea14f419e77c408ff2bcf37a61604

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 ad86d4a37..31df71fcf 100644
--- a/common/concurrency/executor.h
+++ b/common/concurrency/executor.h
@@ -28,6 +28,7 @@
 #include <tbb/concurrent_queue.h>
 
 #include <boost/thread.hpp>
+#include <boost/optional.hpp>
 #include <boost/noncopyable.hpp>
 
 #include <functional>
@@ -177,28 +178,7 @@ public:
 					
 		return std::move(future);		
 	}
-
-	template<typename Func>
-	auto try_begin_invoke(Func&& func, task_priority priority = normal_priority) -> boost::unique_future<decltype(func())> // noexcept
-	{
-		// Create a move on copy adaptor to avoid copying the functor into the queue, tbb::concurrent_queue does not support move semantics.
-		auto task_adaptor = make_move_on_copy(create_task(func));
-		
-		auto future = task_adaptor.value.get_future();
-
-		if(priority == normal_priority || execution_queue_[normal_priority].try_push(nullptr))
-		{			
-			execution_queue_[priority].try_push([=]
-			{
-				try{task_adaptor.value();}
-				catch(boost::task_already_started&){}
-				catch(...){CASPAR_LOG_CURRENT_EXCEPTION();}
-			});
-		}
-		
-		return std::move(future);			
-	}
-
+	
 	template<typename Func>
 	auto invoke(Func&& func, task_priority prioriy = normal_priority) -> decltype(func()) // noexcept
 	{
@@ -207,16 +187,7 @@ public:
 		
 		return begin_invoke(std::forward<Func>(func), prioriy).get();
 	}
-
-	template<typename Func>
-	auto try_invoke(Func&& func, task_priority prioriy = normal_priority) -> decltype(func()) // noexcept
-	{
-		if(boost::this_thread::get_id() == thread_.get_id())  // Avoids potential deadlock.
-			return func();
-		
-		return try_begin_invoke(std::forward<Func>(func), prioriy).get();
-	}
-
+	
 	void yield() // noexcept
 	{
 		if(boost::this_thread::get_id() != thread_.get_id())  // Only yield when calling from execution thread.
diff --git a/modules/ogl/consumer/ogl_consumer.cpp b/modules/ogl/consumer/ogl_consumer.cpp
index 313991c5f..2b15a4975 100644
--- 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_;
 	size_t					square_height_;
 
-	boost::circular_buffer<safe_ptr<core::read_frame>> frame_buffer_;
+	boost::circular_buffer<safe_ptr<core::read_frame>>			input_buffer_;
+	tbb::concurrent_bounded_queue<safe_ptr<core::read_frame>>	frame_buffer_;
 
 	executor				executor_;
 public:
@@ -89,11 +90,14 @@ public:
 		, screen_index_(screen_index)
 		, format_desc_(format_desc_)
 		, graph_(diagnostics::create_graph(narrow(print())))
-		, frame_buffer_(core::consumer_buffer_depth())
+		, input_buffer_(core::consumer_buffer_depth()-1)
 		, executor_(print())
 	{		
+		frame_buffer_.set_capacity(2);
+
 		graph_->add_guide("frame-time", 0.5);
 		graph_->set_color("frame-time", diagnostics::color(1.0f, 0.0f, 0.0f));
+		graph_->set_color("dropped-frame", diagnostics::color(0.3f, 0.6f, 0.3f));
 		
 		format_desc_ = format_desc;
 
@@ -180,6 +184,14 @@ public:
 			glBindBufferARB(GL_PIXEL_UNPACK_BUFFER_ARB, 0);
 		});
 		CASPAR_LOG(info) << print() << " Sucessfully Initialized.";
+
+		executor_.begin_invoke([this]{tick();});
+	}
+
+	~ogl_consumer()
+	{
+		executor_.stop();
+		frame_buffer_.push(make_safe<core::read_frame>());
 	}
 	
 	const core::video_format_desc& get_video_format_desc() const
@@ -277,35 +289,49 @@ public:
 
 	void send(const safe_ptr<core::read_frame>& frame)
 	{
-		frame_buffer_.push_back(frame);
+		input_buffer_.push_back(frame);
 
-		if(frame_buffer_.full())
-			do_send(frame_buffer_.front());
+		if(input_buffer_.full())
+		{
+			if(!frame_buffer_.try_push(input_buffer_.front()))
+				graph_->add_tag("dropped-frame");
+		}
 	}
 		
-	void do_send(const safe_ptr<core::read_frame>& frame)
-	{		
-		executor_.try_begin_invoke([=]
+	std::wstring print() const
+	{	
+		return  L"ogl[" + boost::lexical_cast<std::wstring>(screen_index_) + L"|" + format_desc_.name + L"]";
+	}
+
+	void tick()
+	{
+		try
 		{
 			perf_timer_.restart();
 			sf::Event e;
+		
+			safe_ptr<core::read_frame> frame;
+
 			while(window_.GetEvent(e))
 			{
 				if(e.Type == sf::Event::Resized)
-					 calculate_aspect();
+					calculate_aspect();
 			}
-			render(frame);
-			window_.Display();
-			graph_->update_value("frame-time", static_cast<float>(perf_timer_.elapsed()*format_desc_.fps*0.5));
-		});
-	}
 
-	std::wstring print() const
-	{	
-		return  L"ogl[" + boost::lexical_cast<std::wstring>(screen_index_) + L"|" + format_desc_.name + L"]";
-	}
+			frame_buffer_.pop(frame);
+
+			if(!frame->image_data().empty())
+				render(frame);
 
-	size_t buffer_depth() const{return 2;}
+			window_.Display();
+			graph_->update_value("frame-time", static_cast<float>(perf_timer_.elapsed()*format_desc_.fps*0.5));	
+		}
+		catch(...)
+		{
+			CASPAR_LOG_CURRENT_EXCEPTION();
+		}
+		executor_.begin_invoke([=]{tick();});
+	}
 };