X-Git-Url: https://git.sesse.net/?a=blobdiff_plain;ds=sidebyside;f=modules%2Fdecklink%2Fproducer%2Fdecklink_producer.cpp;h=34d1bfcf31f60f5a0c46f93bcab33ba45f7750ac;hb=ed40e650cbbdf60295fdab81859eaac07aa4345b;hp=de51381ca2cd2fe940e0d4086154c8b8842c8a56;hpb=821db263adca238a650defe0da1970f23cc4803f;p=casparcg

diff --git a/modules/decklink/producer/decklink_producer.cpp b/modules/decklink/producer/decklink_producer.cpp
index de51381ca..34d1bfcf3 100644
--- a/modules/decklink/producer/decklink_producer.cpp
+++ b/modules/decklink/producer/decklink_producer.cpp
@@ -27,21 +27,24 @@
 
 #include "../../ffmpeg/producer/filter/filter.h"
 #include "../../ffmpeg/producer/util.h"
+#include "../../ffmpeg/producer/frame_muxer.h"
 
+#include <common/log/log.h>
 #include <common/diagnostics/graph.h>
 #include <common/concurrency/com_context.h>
 #include <common/exception/exceptions.h>
 #include <common/memory/memclr.h>
 
 #include <core/mixer/write_frame.h>
-#include <core/producer/frame/audio_transform.h>
+#include <core/producer/frame/frame_transform.h>
 #include <core/producer/frame/frame_factory.h>
-#include <core/producer/frame_muxer.h>
 
-#include <tbb/concurrent_queue.h>
-#include <tbb/atomic.h>
+#include <agents.h>
+#include <agents_extras.h>
+#include <ppl.h>
 
 #include <boost/algorithm/string.hpp>
+#include <boost/foreach.hpp>
 #include <boost/timer.hpp>
 
 #if defined(_MSC_VER)
@@ -70,55 +73,43 @@ extern "C"
 
 #include <functional>
 
-namespace caspar { 
+namespace caspar { namespace decklink {
 		
-class decklink_producer : public IDeckLinkInputCallback
-{	
-	CComPtr<IDeckLink>											decklink_;
-	CComQIPtr<IDeckLinkInput>									input_;
-	
-	const std::wstring											model_name_;
-	const core::video_format_desc								format_desc_;
-	const size_t												device_index_;
+typedef std::pair<CComPtr<IDeckLinkVideoInputFrame>, CComPtr<IDeckLinkAudioInputPacket>> frame_packet;
 
-	std::shared_ptr<diagnostics::graph>							graph_;
-	boost::timer												tick_timer_;
-	boost::timer												frame_timer_;
+class decklink_producer : boost::noncopyable, public IDeckLinkInputCallback
+{	
+	Concurrency::ITarget<frame_packet>&	target_;
 
-	std::vector<int16_t>										audio_samples_;
+	CComPtr<IDeckLink>					decklink_;
+	CComQIPtr<IDeckLinkInput>			input_;
 	
-	safe_ptr<core::frame_factory>								frame_factory_;
+	const std::wstring					model_name_;
+	const core::video_format_desc		format_desc_;
+	const size_t						device_index_;
 
-	tbb::concurrent_bounded_queue<safe_ptr<core::basic_frame>>	frame_buffer_;
-	safe_ptr<core::basic_frame>									tail_;
-
-	std::exception_ptr											exception_;
-	filter														filter_;
-		
-	core::frame_muxer											muxer_;
+	safe_ptr<diagnostics::graph>		graph_;
+	boost::timer						tick_timer_;
+	boost::timer						frame_timer_;
 
 public:
-	decklink_producer(const core::video_format_desc& format_desc, size_t device_index, const safe_ptr<core::frame_factory>& frame_factory, const std::wstring& filter)
-		: decklink_(get_device(device_index))
+	decklink_producer(Concurrency::ITarget<frame_packet>& target, const core::video_format_desc& format_desc, size_t device_index)
+		: target_(target)
+		, decklink_(get_device(device_index))
 		, input_(decklink_)
 		, model_name_(get_model_name(decklink_))
 		, format_desc_(format_desc)
 		, device_index_(device_index)
-		, frame_factory_(frame_factory)
-		, tail_(core::basic_frame::empty())
-		, filter_(filter)
-		, muxer_(double_rate(filter) ? format_desc.fps * 2.0 : format_desc.fps, frame_factory->get_video_format_desc().mode, frame_factory->get_video_format_desc().fps)
-	{
-		frame_buffer_.set_capacity(2);
-		
-		graph_ = diagnostics::create_graph(boost::bind(&decklink_producer::print, this));
+		, graph_ (diagnostics::create_graph("", false))
+	{		
 		graph_->add_guide("tick-time", 0.5);
-		graph_->set_color("tick-time", diagnostics::color(0.1f, 0.7f, 0.8f));
+		graph_->set_color("tick-time", diagnostics::color(0.0f, 0.6f, 0.9f));	
 		graph_->set_color("late-frame", diagnostics::color(0.6f, 0.3f, 0.3f));
 		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));
 		graph_->set_color("output-buffer", diagnostics::color(0.0f, 1.0f, 0.0f));
-		
+		graph_->update_text(narrow(print()));
+
 		auto display_mode = get_display_mode(input_, format_desc_.format, bmdFormat8BitYUV, bmdVideoInputFlagDefault);
 		
 		// NOTE: bmdFormat8BitARGB is currently not supported by any decklink card. (2011-05-08)
@@ -127,7 +118,7 @@ public:
 									<< msg_info(narrow(print()) + " Could not enable video input.")
 									<< boost::errinfo_api_function("EnableVideoInput"));
 
-		if(FAILED(input_->EnableAudioInput(bmdAudioSampleRate48kHz, bmdAudioSampleType16bitInteger, 2))) 
+		if(FAILED(input_->EnableAudioInput(bmdAudioSampleRate48kHz, bmdAudioSampleType32bitInteger, format_desc_.audio_channels))) 
 			BOOST_THROW_EXCEPTION(caspar_exception() 
 									<< msg_info(narrow(print()) + " Could not enable audio input.")
 									<< boost::errinfo_api_function("EnableAudioInput"));
@@ -143,10 +134,13 @@ public:
 									<< boost::errinfo_api_function("StartStreams"));
 
 		CASPAR_LOG(info) << print() << L" Successfully Initialized.";
+
+		graph_->start();
 	}
 
 	~decklink_producer()
 	{
+		Concurrency::scoped_oversubcription_token oversubscribe;
 		if(input_ != nullptr) 
 		{
 			input_->StopStreams();
@@ -165,138 +159,191 @@ public:
 
 	virtual HRESULT STDMETHODCALLTYPE VideoInputFrameArrived(IDeckLinkVideoInputFrame* video, IDeckLinkAudioInputPacket* audio)
 	{	
-		if(!video)
-			return S_OK;
+		if(!Concurrency::asend(target_, frame_packet(CComPtr<IDeckLinkVideoInputFrame>(video), CComPtr<IDeckLinkAudioInputPacket>(audio))))
+			graph_->add_tag("dropped-frame");
+		return S_OK;
+	}
+		
+	std::wstring print() const
+	{
+		return model_name_ + L" [" + boost::lexical_cast<std::wstring>(device_index_) + L"]";
+	}
+};
+	
+class decklink_producer_proxy : public Concurrency::agent, public core::frame_producer
+{		
+	Concurrency::bounded_buffer<ffmpeg::video_message_t>	video_frames_;
+	Concurrency::bounded_buffer<ffmpeg::audio_message_t>	audio_buffers_;
+	Concurrency::bounded_buffer<ffmpeg::frame_message_t>	muxed_frames_;
 
-		try
-		{
-			graph_->update_value("tick-time", tick_timer_.elapsed()*format_desc_.fps*0.5);
-			tick_timer_.restart();
+	const core::video_format_desc		format_desc_;
+	const size_t						device_index_;
 
-			frame_timer_.restart();
+	safe_ptr<core::basic_frame>			last_frame_;
+	const int64_t						length_;
 
-			void* bytes = nullptr;
-			if(FAILED(video->GetBytes(&bytes)) || !bytes)
-				return S_OK;
-			
-			safe_ptr<AVFrame> av_frame(avcodec_alloc_frame(), av_free);	
-			avcodec_get_frame_defaults(av_frame.get());
-						
-			av_frame->data[0]			= reinterpret_cast<uint8_t*>(bytes);
-			av_frame->linesize[0]		= video->GetRowBytes();			
-			av_frame->format			= PIX_FMT_UYVY422;
-			av_frame->width				= video->GetWidth();
-			av_frame->height			= video->GetHeight();
-			av_frame->interlaced_frame	= format_desc_.mode != core::video_mode::progressive;
-			av_frame->top_field_first	= format_desc_.mode == core::video_mode::upper ? 1 : 0;
-					
-			filter_.push(av_frame);
-			BOOST_FOREACH(auto& av_frame2, filter_.poll())
-				muxer_.push(make_write_frame(this, av_frame2, frame_factory_));		
-									
-			// It is assumed that audio is always equal or ahead of video.
-			if(audio && SUCCEEDED(audio->GetBytes(&bytes)))
-			{
-				auto sample_frame_count = audio->GetSampleFrameCount();
-				auto audio_data = reinterpret_cast<short*>(bytes);
-				audio_samples_.insert(audio_samples_.end(), audio_data, audio_data + sample_frame_count*2);
+	ffmpeg::filter						filter_;
+		
+	ffmpeg::frame_muxer2				muxer_;
 
-				if(audio_samples_.size() > frame_factory_->get_video_format_desc().audio_samples_per_frame)
-				{
-					const auto begin = audio_samples_.begin();
-					const auto end   = begin +  frame_factory_->get_video_format_desc().audio_samples_per_frame;
-					muxer_.push(std::vector<int16_t>(begin, end));
-					audio_samples_.erase(begin, end);
-				}
-			}
-			else
-				muxer_.push(std::vector<int16_t>(frame_factory_->get_video_format_desc().audio_samples_per_frame, 0));
-					
-			while(!muxer_.empty())
-			{
-				if(!frame_buffer_.try_push(muxer_.pop()))
-					graph_->add_tag("dropped-frame");
-			}
+	mutable Concurrency::single_assignment<std::wstring> print_;
 
-			graph_->update_value("frame-time", frame_timer_.elapsed()*format_desc_.fps*0.5);
+	safe_ptr<Concurrency::semaphore> semaphore_;
 
-			graph_->set_value("output-buffer", static_cast<float>(frame_buffer_.size())/static_cast<float>(frame_buffer_.capacity()));	
-		}
-		catch(...)
-		{
-			exception_ = std::current_exception();
-			return E_FAIL;
-		}
+	volatile bool is_running_;
+public:
 
-		return S_OK;
-	}
-	
-	safe_ptr<core::basic_frame> get_frame()
+	explicit decklink_producer_proxy(const safe_ptr<core::frame_factory>& frame_factory, const core::video_format_desc& format_desc, size_t device_index, const std::wstring& filter_str, int64_t length)
+		: video_frames_(1)
+		, audio_buffers_(1)
+		, muxed_frames_(1)
+		, format_desc_(format_desc)
+		, device_index_(device_index)
+		, last_frame_(core::basic_frame::empty())
+		, length_(length)
+		, filter_(filter_str)
+		, muxer_(&video_frames_, &audio_buffers_, muxed_frames_, ffmpeg::double_rate(filter_str) ? format_desc.fps * 2.0 : format_desc.fps, frame_factory)
+		, is_running_(true)
+		, semaphore_(make_safe<Concurrency::semaphore>(3))
 	{
-		if(exception_ != nullptr)
-			std::rethrow_exception(exception_);
+		agent::start();
+	}
 
-		if(!frame_buffer_.try_pop(tail_))
-			graph_->add_tag("late-frame");
-		graph_->set_value("output-buffer", static_cast<float>(frame_buffer_.size())/static_cast<float>(frame_buffer_.capacity()));	
-		return tail_;
+	~decklink_producer_proxy()
+	{
+		is_running_ = false;
+		agent::wait(this);
 	}
-	
-	std::wstring print() const
+				
+	virtual safe_ptr<core::basic_frame> receive(int)
 	{
-		return model_name_ + L" [" + boost::lexical_cast<std::wstring>(device_index_) + L"]";
+		auto frame = core::basic_frame::late();
+
+		try
+		{
+			last_frame_ = frame = Concurrency::receive(muxed_frames_)->payload;
+		}
+		catch(Concurrency::operation_timed_out&)
+		{		
+			//graph_->add_tag("underflow");	
+		}
+
+		return frame;
 	}
-};
-	
-class decklink_producer_proxy : public core::frame_producer
-{		
-	com_context<decklink_producer> context_;
-public:
 
-	explicit decklink_producer_proxy(const safe_ptr<core::frame_factory>& frame_factory, const core::video_format_desc& format_desc, size_t device_index, const std::wstring& filter_str = L"")
-		: context_(L"decklink_producer[" + boost::lexical_cast<std::wstring>(device_index) + L"]")
+	virtual safe_ptr<core::basic_frame> last_frame() const
 	{
-		context_.reset([&]{return new decklink_producer(format_desc, device_index, frame_factory, filter_str);}); 
+		return disable_audio(last_frame_);
 	}
-				
-	virtual safe_ptr<core::basic_frame> receive()
+	
+	virtual int64_t nb_frames() const 
 	{
-		return context_->get_frame();
+		return length_;
 	}
 	
 	std::wstring print() const
 	{
-		return context_->print();
+		return print_.value();
+	}
+
+	virtual void run()
+	{
+		try
+		{
+			struct co_init
+			{
+				co_init()  {CoInitialize(NULL);}
+				~co_init() {CoUninitialize();}
+			} init;
+			
+			Concurrency::bounded_buffer<frame_packet> input_buffer(2);
+
+			std::unique_ptr<decklink_producer> producer;
+			{				
+				Concurrency::scoped_oversubcription_token oversubscribe;
+				producer.reset(new decklink_producer(input_buffer, format_desc_, device_index_));
+			}
+
+			Concurrency::send(print_, producer->print());
+
+			while(is_running_)
+			{
+				auto packet = Concurrency::receive(input_buffer);
+				auto video  = packet.first;
+				auto audio  = packet.second;
+				
+				void* bytes = nullptr;
+				if(FAILED(video->GetBytes(&bytes)) || !bytes)
+					continue;
+			
+				safe_ptr<AVFrame> av_frame(avcodec_alloc_frame(), av_free);	
+				avcodec_get_frame_defaults(av_frame.get());
+						
+				av_frame->data[0]			= reinterpret_cast<uint8_t*>(bytes);
+				av_frame->linesize[0]		= video->GetRowBytes();			
+				av_frame->format			= PIX_FMT_UYVY422;
+				av_frame->width				= video->GetWidth();
+				av_frame->height			= video->GetHeight();
+				av_frame->interlaced_frame	= format_desc_.field_mode != core::field_mode::progressive;
+				av_frame->top_field_first	= format_desc_.field_mode == core::field_mode::upper ? 1 : 0;
+					
+				filter_.push(av_frame);
+
+				Concurrency::parallel_invoke(
+				[&]
+				{
+					while(true)
+					{
+						auto frame = filter_.poll();
+						if(!frame)
+							break;
+						Concurrency::send(video_frames_, ffmpeg::make_message(frame, std::make_shared<ffmpeg::token>(semaphore_)));
+					}
+				},
+				[&]
+				{													
+					// It is assumed that audio is always equal or ahead of video.
+					if(audio && SUCCEEDED(audio->GetBytes(&bytes)))
+					{
+						auto sample_frame_count = audio->GetSampleFrameCount();
+						auto audio_data = reinterpret_cast<int32_t*>(bytes);
+						Concurrency::send(audio_buffers_, ffmpeg::make_message(std::make_shared<core::audio_buffer>(audio_data, audio_data + sample_frame_count*format_desc_.audio_channels), std::make_shared<ffmpeg::token>(semaphore_)));
+					}
+					else
+						Concurrency::send(audio_buffers_, ffmpeg::make_message(ffmpeg::empty_audio(), std::make_shared<ffmpeg::token>(semaphore_)));	
+				});
+			}
+
+		}
+		catch(...)
+		{
+			CASPAR_LOG_CURRENT_EXCEPTION();
+		}
+		
+		CASPAR_LOG(info) << print() << L" Successfully Uninitialized.";	
+
+		done();
 	}
 };
 
-safe_ptr<core::frame_producer> create_decklink_producer(const safe_ptr<core::frame_factory>& frame_factory, const std::vector<std::wstring>& params)
+safe_ptr<core::frame_producer> create_producer(const safe_ptr<core::frame_factory>& frame_factory, const std::vector<std::wstring>& params)
 {
 	if(params.empty() || !boost::iequals(params[0], "decklink"))
 		return core::frame_producer::empty();
 
-	size_t device_index = 1;
-	if(params.size() > 1)
-		device_index = lexical_cast_or_default(params[1], 1);
-
-	core::video_format_desc format_desc = core::video_format_desc::get(L"PAL");
-	if(params.size() > 2)
-	{
-		auto desc = core::video_format_desc::get(params[2]);
-		if(desc.format != core::video_format::invalid)
-			format_desc = desc;
-	}
+	auto device_index	= core::get_param(L"DEVICE", params, 1);
+	auto filter_str		= core::get_param<std::wstring>(L"FILTER", params, L""); 	
+	auto length			= core::get_param(L"LENGTH", params, std::numeric_limits<int64_t>::max()); 	
 	
-	std::wstring filter_str = L"";
+	boost::replace_all(filter_str, L"DEINTERLACE", L"YADIF=0:-1");
+	boost::replace_all(filter_str, L"DEINTERLACE_BOB", L"YADIF=1:-1");
 
-	auto filter_it = std::find(params.begin(), params.end(), L"FILTER");
-	if(filter_it != params.end())
-	{
-		if(++filter_it != params.end())
-			filter_str = *filter_it;
-	}
+	auto format_desc	= core::video_format_desc::get(core::get_param<std::wstring>(L"FORMAT", params, L"INVALID"));
 
-	return make_safe<decklink_producer_proxy>(frame_factory, format_desc, device_index, filter_str);
+	if(format_desc.format == core::video_format::invalid)
+		format_desc = frame_factory->get_video_format_desc();
+			
+	return make_safe<decklink_producer_proxy>(frame_factory, format_desc, device_index, filter_str, length);
 }
 
-}
\ No newline at end of file
+}}
\ No newline at end of file