X-Git-Url: https://git.sesse.net/?a=blobdiff_plain;ds=sidebyside;f=core%2Fproducer%2Fframe_producer.cpp;h=4379100ed0faa6794905a1dfda68257b71880850;hb=c65d48e9a2afc6c8bef32b1ce82dd093d76304fc;hp=b549007321c7c8d5a8285bf58b9447b68826dfa0;hpb=8d10ef21e321757c0e99cbfcef682016809ed252;p=casparcg

diff --git a/core/producer/frame_producer.cpp b/core/producer/frame_producer.cpp
index b54900732..4379100ed 100644
--- a/core/producer/frame_producer.cpp
+++ b/core/producer/frame_producer.cpp
@@ -22,104 +22,46 @@
 #include "../StdAfx.h"
 
 #include "frame_producer.h"
-#include "frame/basic_frame.h"
-#include "frame/frame_transform.h"
+
+#include "../frame/draw_frame.h"
+#include "../frame/frame_transform.h"
 
 #include "color/color_producer.h"
-#include "playlist/playlist_producer.h"
 #include "separated/separated_producer.h"
 
-#include <common/memory/safe_ptr.h>
+#include <common/assert.h>
+#include <common/except.h>
 #include <common/concurrency/executor.h>
-#include <common/exception/exceptions.h>
-#include <common/utility/move_on_copy.h>
+#include <common/concurrency/async.h>
+#include <common/spl/memory.h>
 
 namespace caspar { namespace core {
 	
 std::vector<const producer_factory_t> g_factories;
-	
-class destroy_producer_proxy : public frame_producer
-{	
-	std::unique_ptr<std::shared_ptr<frame_producer>> producer_;
-public:
-	destroy_producer_proxy(safe_ptr<frame_producer>&& producer) 
-		: producer_(new std::shared_ptr<frame_producer>(std::move(producer)))
-	{
-	}
-
-	~destroy_producer_proxy()
-	{		
-		static auto destroyers = std::make_shared<tbb::concurrent_bounded_queue<std::shared_ptr<executor>>>();
-		static tbb::atomic<int> destroyer_count;
-
-		try
-		{
-			std::shared_ptr<executor> destroyer;
-			if(!destroyers->try_pop(destroyer))
-			{
-				destroyer.reset(new executor(L"destroyer"));
-				destroyer->set_priority_class(below_normal_priority_class);
-				if(++destroyer_count > 16)
-					CASPAR_LOG(warning) << L"Potential destroyer dead-lock detected.";
-				CASPAR_LOG(trace) << "Created destroyer: " << destroyer_count;
-			}
-				
-			auto producer = producer_.release();
-			auto pool	  = destroyers;
-			destroyer->begin_invoke([=]
-			{
-				std::unique_ptr<std::shared_ptr<frame_producer>> producer2(producer);
-
-				auto str = (*producer2)->print();
-				try
-				{
-					if(!producer->unique())
-						CASPAR_LOG(trace) << str << L" Not destroyed on safe asynchronous destruction thread: " << producer->use_count();
-					else
-						CASPAR_LOG(trace) << str << L" Destroying on safe asynchronous destruction thread.";
-				}
-				catch(...){}
-								
-				producer2.reset();
-				CASPAR_LOG(debug) << str << L" Destroyed.";
-				pool->push(destroyer);
-			}); 
-		}
-		catch(...)
-		{
-			CASPAR_LOG_CURRENT_EXCEPTION();
-			try
-			{
-				auto str = (*producer_)->print();
-				producer_.reset();
-				CASPAR_LOG(debug) << str << L" Destroyed.";
-			}
-			catch(...){}
-		}
-	}
 
-	virtual safe_ptr<basic_frame>								receive(int flags) override												{return (*producer_)->receive(flags);}
-	virtual safe_ptr<basic_frame>								last_frame() const override		 										{return (*producer_)->last_frame();}
-	virtual std::wstring										print() const override													{return (*producer_)->print();}
-	virtual boost::property_tree::wptree 						info() const override													{return (*producer_)->info();}
-	virtual boost::unique_future<std::wstring>					call(const std::wstring& str) override									{return (*producer_)->call(str);}
-	virtual safe_ptr<frame_producer>							get_following_producer() const override									{return (*producer_)->get_following_producer();}
-	virtual void												set_leading_producer(const safe_ptr<frame_producer>& producer) override	{(*producer_)->set_leading_producer(producer);}
-	virtual uint32_t											nb_frames() const override												{return (*producer_)->nb_frames();}
-};
+void register_producer_factory(const producer_factory_t& factory)
+{
+	g_factories.push_back(factory);
+}
 
-safe_ptr<core::frame_producer> create_producer_destroy_proxy(safe_ptr<core::frame_producer>&& producer)
+boost::unique_future<std::wstring> frame_producer::call(const std::wstring&) 
 {
-	return make_safe<destroy_producer_proxy>(std::move(producer));
+	BOOST_THROW_EXCEPTION(not_supported());
 }
 
+const spl::shared_ptr<frame_producer>& frame_producer::empty() // nothrow
+{
+
 struct empty_frame_producer : public frame_producer
 {
-	virtual safe_ptr<basic_frame> receive(int){return basic_frame::empty();}
-	virtual safe_ptr<basic_frame> last_frame() const{return basic_frame::empty();}
-	virtual void set_frame_factory(const safe_ptr<frame_factory>&){}
+	virtual spl::shared_ptr<draw_frame> receive(int){return draw_frame::empty();}
+	virtual spl::shared_ptr<draw_frame> last_frame() const{return draw_frame::empty();}
+	virtual void set_frame_factory(const spl::shared_ptr<frame_factory>&){}
 	virtual uint32_t nb_frames() const {return 0;}
 	virtual std::wstring print() const { return L"empty";}
+	virtual void subscribe(const monitor::observable::observer_ptr& o){}
+	virtual void unsubscribe(const monitor::observable::observer_ptr& o){}	
+	virtual std::wstring name() const {return L"empty";}
 	
 	virtual boost::property_tree::wptree info() const override
 	{
@@ -129,32 +71,67 @@ struct empty_frame_producer : public frame_producer
 	}
 };
 
-const safe_ptr<frame_producer>& frame_producer::empty() // nothrow
-{
-	static safe_ptr<frame_producer> producer = make_safe<empty_frame_producer>();
+	static spl::shared_ptr<frame_producer> producer = spl::make_shared<empty_frame_producer>();
 	return producer;
 }	
 
-safe_ptr<basic_frame> receive_and_follow(safe_ptr<frame_producer>& producer, int flags)
+class producer_proxy : public frame_producer
 {	
-	auto frame = producer->receive(flags);
-	if(frame != basic_frame::eof())
-		return frame;
+	std::shared_ptr<frame_producer> producer_;
+public:
+	producer_proxy(spl::shared_ptr<frame_producer>&& producer) 
+		: producer_(std::move(producer))
+	{
+		CASPAR_LOG(info) << producer_->print() << L" Initialized.";
+	}
+
+	virtual ~producer_proxy()
+	{		
+		static tbb::atomic<int> counter = tbb::atomic<int>();
+		
+		++counter;
+		CASPAR_VERIFY(counter < 32);
 		
-	CASPAR_LOG(info) << producer->print() << " End Of File.";
-	auto following = producer->get_following_producer();
-	following->set_leading_producer(producer);
-	producer = std::move(following);
+		auto producer = new spl::shared_ptr<frame_producer>(std::move(producer_));
+		async([=]
+		{
+			std::unique_ptr<spl::shared_ptr<frame_producer>> pointer_guard(producer);
+			auto str = (*producer)->print();
+			try
+			{
+				if(!producer->unique())
+					CASPAR_LOG(trace) << str << L" Not destroyed on asynchronous destruction thread: " << producer->use_count();
+				else
+					CASPAR_LOG(trace) << str << L" Destroying on asynchronous destruction thread.";
+			}
+			catch(...){}
+			
+			CASPAR_LOG(trace) << str << L" Uninitializing.";
+			pointer_guard.reset();
+			CASPAR_LOG(info) << str << L" Uninitialized.";
 
-	return receive_and_follow(producer, flags);	
-}
+			--counter;
+		}); 
+	}
+	
+	virtual spl::shared_ptr<draw_frame>	receive(int flags) override																				{return producer_->receive(flags);}
+	virtual spl::shared_ptr<draw_frame>	last_frame() const override																				{return producer_->last_frame();}
+	virtual std::wstring										print() const override															{return producer_->print();}
+	virtual std::wstring										name() const override															{return producer_->name();}
+	virtual boost::property_tree::wptree 						info() const override															{return producer_->info();}
+	virtual boost::unique_future<std::wstring>					call(const std::wstring& str) override											{return producer_->call(str);}
+	virtual void												leading_producer(const spl::shared_ptr<frame_producer>& producer) override		{return producer_->leading_producer(producer);}
+	virtual uint32_t											nb_frames() const override														{return producer_->nb_frames();}
+	virtual void subscribe(const monitor::observable::observer_ptr& o)																			{return producer_->subscribe(o);}
+	virtual void unsubscribe(const monitor::observable::observer_ptr& o)																		{return producer_->unsubscribe(o);}
+};
 
-void register_producer_factory(const producer_factory_t& factory)
+spl::shared_ptr<core::frame_producer> wrap_producer(spl::shared_ptr<core::frame_producer> producer)
 {
-	g_factories.push_back(factory);
+	return spl::make_shared<producer_proxy>(std::move(producer));
 }
 
-safe_ptr<core::frame_producer> do_create_producer(const safe_ptr<frame_factory>& my_frame_factory, const std::vector<std::wstring>& params)
+spl::shared_ptr<core::frame_producer> do_create_producer(const spl::shared_ptr<frame_factory>& my_frame_factory, const std::vector<std::wstring>& params)
 {
 	if(params.empty())
 		BOOST_THROW_EXCEPTION(invalid_argument() << arg_name_info("params") << arg_value_info(""));
@@ -175,15 +152,14 @@ safe_ptr<core::frame_producer> do_create_producer(const safe_ptr<frame_factory>&
 
 	if(producer == frame_producer::empty())
 		producer = create_color_producer(my_frame_factory, params);
-	
-	if(producer == frame_producer::empty())
-		producer = create_playlist_producer(my_frame_factory, params);
 
+	if(producer == frame_producer::empty())
+		return producer;
+		
 	return producer;
 }
 
-
-safe_ptr<core::frame_producer> create_producer(const safe_ptr<frame_factory>& my_frame_factory, const std::vector<std::wstring>& params)
+spl::shared_ptr<core::frame_producer> create_producer(const spl::shared_ptr<frame_factory>& my_frame_factory, const std::vector<std::wstring>& params)
 {	
 	auto producer = do_create_producer(my_frame_factory, params);
 	auto key_producer = frame_producer::empty();
@@ -212,14 +188,14 @@ safe_ptr<core::frame_producer> create_producer(const safe_ptr<frame_factory>& my
 		std::wstring str;
 		BOOST_FOREACH(auto& param, params)
 			str += param + L" ";
-		BOOST_THROW_EXCEPTION(file_not_found() << msg_info("No match found for supplied commands. Check syntax.") << warg_value_info(str));
+		BOOST_THROW_EXCEPTION(file_not_found() << msg_info("No match found for supplied commands. Check syntax.") << arg_value_info(u8(str)));
 	}
 
 	return producer;
 }
 
 
-safe_ptr<core::frame_producer> create_producer(const safe_ptr<frame_factory>& factory, const std::wstring& params)
+spl::shared_ptr<core::frame_producer> create_producer(const spl::shared_ptr<frame_factory>& factory, const std::wstring& params)
 {
 	std::wstringstream iss(params);
 	std::vector<std::wstring> tokens;