Merge remote-tracking branch 'upstream/master' into ppu_llvm_recompiler

Conflicts:
	Utilities/BEType.h
	rpcs3/Emu/Cell/PPUInterpreter.h
	rpcs3/emucore.vcxproj.filters

Utilities/SQueue.h

@@ -1,6 +1,6 @@
 #pragma once
 
-#include "Utilities/SMutex.h"
+static const volatile bool sq_no_wait = true;
 
 template<typename T, u32 SQSize = 666>
 class SQueue
@@ -22,13 +22,13 @@ public:
 		return SQSize;
 	}
 
-	bool Push(const T& data)
+	bool Push(const T& data, const volatile bool* do_exit)
 	{
 		while (true)
 		{
 			if (m_count >= SQSize)
 			{
-				if (Emu.IsStopped())
+				if (Emu.IsStopped() || do_exit && *do_exit)
 				{
 					return false;
 				}
@@ -49,13 +49,13 @@ public:
 		}
 	}
 
-	bool Pop(T& data)
+	bool Pop(T& data, const volatile bool* do_exit)
 	{
 		while (true)
 		{
 			if (!m_count)
 			{
-				if (Emu.IsStopped())
+				if (Emu.IsStopped() || do_exit && *do_exit)
 				{
 					return false;
 				}
@@ -78,38 +78,21 @@ public:
 		}
 	}
 
-	u32 GetCount()
-	{
-		std::lock_guard<std::mutex> lock(m_mutex);
-		return m_count;
-	}
-
-	u32 GetCountUnsafe()
-	{
-		return m_count;
-	}
-
-	bool IsEmpty()
-	{
-		std::lock_guard<std::mutex> lock(m_mutex);
-		return !m_count;
-	}
-
 	void Clear()
 	{
 		std::lock_guard<std::mutex> lock(m_mutex);
 		m_count = 0;
 	}
 
-	T& Peek(u32 pos = 0)
+	bool Peek(T& data, const volatile bool* do_exit, u32 pos = 0)
 	{
 		while (true)
 		{
 			if (m_count <= pos)
 			{
-				if (Emu.IsStopped())
+				if (Emu.IsStopped() || do_exit && *do_exit)
 				{
-					break;
+					return false;
 				}
 				
 				std::this_thread::sleep_for(std::chrono::milliseconds(1));
@@ -124,21 +107,7 @@ public:
 				}
 			}
 		}
-		return m_data[(m_pos + pos) % SQSize];
+		data = m_data[(m_pos + pos) % SQSize];
-	}
+		return true;
-
-	T& PeekIfExist(u32 pos = 0)
-	{
-		static T def_value;
-
-		std::lock_guard<std::mutex> lock(m_mutex);
-		if (m_count <= pos)
-		{
-			return def_value;
-		}
-		else
-		{
-			return m_data[(m_pos + pos) % SQSize];
-		}
 	}
 };

Utilities/Thread.cpp

@@ -219,11 +219,12 @@ bool waiter_map_t::is_stopped(u64 signal_id)
 	return false;
 }
 
-waiter_map_t::waiter_reg_t::waiter_reg_t(waiter_map_t& map, u64 signal_id)
+void waiter_map_t::waiter_reg_t::init()
-	: signal_id(signal_id)
-	, thread(GetCurrentNamedThread())
-	, map(map)
 {
+	if (thread) return;
+
+	thread = GetCurrentNamedThread();
+
 	std::lock_guard<std::mutex> lock(map.m_mutex);
 
 	// add waiter
@@ -232,10 +233,12 @@ waiter_map_t::waiter_reg_t::waiter_reg_t(waiter_map_t& map, u64 signal_id)
 
 waiter_map_t::waiter_reg_t::~waiter_reg_t()
 {
+	if (!thread) return;
+
 	std::lock_guard<std::mutex> lock(map.m_mutex);
 
 	// remove waiter
-	for (size_t i = map.m_waiters.size() - 1; i >= 0; i--)
+	for (s64 i = map.m_waiters.size() - 1; i >= 0; i--)
 	{
 		if (map.m_waiters[i].signal_id == signal_id && map.m_waiters[i].thread == thread)
 		{

Utilities/Thread.h

@@ -71,6 +71,48 @@ public:
 	bool joinable() const;
 };
 
+class s_mutex_t
+{
+
+};
+
+class s_shared_mutex_t
+{
+
+};
+
+class s_cond_var_t
+{
+	
+//public:
+//	s_cond_var_t();
+//	~s_cond_var_t();
+//
+//	s_cond_var_t(s_cond_var_t& right) = delete;
+//	s_cond_var_t& operator = (s_cond_var_t& right) = delete;
+//
+//	void wait();
+//	void wait_for();
+//
+//	void notify();
+//	void notify_all();
+};
+
+class slw_mutex_t
+{
+
+};
+
+class slw_recursive_mutex_t
+{
+
+};
+
+class slw_shared_mutex_t
+{
+
+};
+
 class waiter_map_t
 {
 	// TODO: optimize (use custom lightweight readers-writer lock)
@@ -88,34 +130,43 @@ class waiter_map_t
 
 	struct waiter_reg_t
 	{
+		NamedThreadBase* thread;
 		const u64 signal_id;
-		NamedThreadBase* const thread;
 		waiter_map_t& map;
 
-		waiter_reg_t(waiter_map_t& map, u64 signal_id);
+		waiter_reg_t(waiter_map_t& map, u64 signal_id)
+			: thread(nullptr)
+			, signal_id(signal_id)
+			, map(map)
+		{
+		}
+
 		~waiter_reg_t();
+
+		void init();
 	};
 
 	bool is_stopped(u64 signal_id);
 
 public:
-	waiter_map_t(const char* name) : m_name(name) {}
+	waiter_map_t(const char* name)
+		: m_name(name)
+	{
+	}
 
 	// wait until waiter_func() returns true, signal_id is an arbitrary number
 	template<typename WT> __forceinline void wait_op(u64 signal_id, const WT waiter_func)
 	{
-		// check condition
-		if (waiter_func()) return;
-
 		// register waiter
 		waiter_reg_t waiter(*this, signal_id);
 
-		while (true)
+		// check condition or if emulator is stopped
+		while (!waiter_func() && !is_stopped(signal_id))
 		{
+			// initialize waiter (only first time)
+			waiter.init();
 			// wait for 1 ms or until signal arrived
 			waiter.thread->WaitForAnySignal(1);
-			if (is_stopped(signal_id)) break;
-			if (waiter_func()) break;
 		}
 	}
 

rpcs3/Emu/Cell/PPUInterpreter.h

@@ -2970,15 +2970,15 @@ private:
 			else
 			{
 				u32 buf = 0;
-				u32 i   = 0;
+				u32 i   = 3;
 				while (N > 0)
 				{
 					N = N - 1;
 					buf |= vm::read8(EA) << (i * 8);
 					EA = EA + 1;
-					i++;
+					i--;
 				}
-				CPU.GPR[reg] = re32(buf);
+				CPU.GPR[reg] = buf;
 			}
 			reg = (reg + 1) % 32;
 		}

rpcs3/Emu/SysCalls/Modules/cellAdec.cpp

@@ -28,6 +28,7 @@ AudioDecoder::AudioDecoder(AudioCodecType type, u32 addr, u32 size, vm::ptr<Cell
 	, cbArg(arg)
 	, adecCb(nullptr)
 	, is_running(false)
+	, is_closed(false)
 	, is_finished(false)
 	, just_started(false)
 	, just_finished(false)
@@ -64,14 +65,14 @@ AudioDecoder::AudioDecoder(AudioCodecType type, u32 addr, u32 size, vm::ptr<Cell
 AudioDecoder::~AudioDecoder()
 {
 	// TODO: check finalization
-	if (ctx)
+	AdecFrame af;
+	while (frames.Pop(af, &sq_no_wait))
 	{
-		for (u32 i = frames.GetCount() - 1; ~i; i--)
-		{
-			AdecFrame& af = frames.Peek(i);
 		av_frame_unref(af.data);
 		av_frame_free(&af.data);
 	}
+	if (ctx)
+	{
 		avcodec_close(ctx);
 		avformat_close_input(&fmt);
 	}
@@ -95,17 +96,14 @@ int adecRawRead(void* opaque, u8* buf, int buf_size)
 next:
 	if (adec.reader.size < (u32)buf_size /*&& !adec.just_started*/)
 	{
-		while (!adec.job.GetCountUnsafe())
+		AdecTask task;
+		if (!adec.job.Peek(task, &adec.is_closed))
 		{
-			if (Emu.IsStopped())
+			if (Emu.IsStopped()) cellAdec->Warning("adecRawRead() aborted");
-			{
-				cellAdec->Warning("adecRawRead(): aborted");
 			return 0;
 		}
-			std::this_thread::sleep_for(std::chrono::milliseconds(1));
-		}
 
-		switch (adec.job.Peek().type)
+		switch (task.type)
 		{
 		case adecEndSeq:
 		case adecClose:
@@ -113,6 +111,7 @@ next:
 			buf_size = adec.reader.size;
 		}
 		break;
+		
 		case adecDecodeAu:
 		{
 			memcpy(buf, vm::get_ptr<void>(adec.reader.addr), adec.reader.size);
@@ -123,17 +122,21 @@ next:
 
 			adec.cbFunc.call(*adec.adecCb, adec.id, CELL_ADEC_MSG_TYPE_AUDONE, adec.task.au.auInfo_addr, adec.cbArg);
 
-				adec.job.Pop(adec.task);
+			adec.job.Pop(adec.task, nullptr);
 
 			adec.reader.addr = adec.task.au.addr;
 			adec.reader.size = adec.task.au.size;
 			//LOG_NOTICE(HLE, "Audio AU: size = 0x%x, pts = 0x%llx", adec.task.au.size, adec.task.au.pts);
 		}
 		break;
+		
 		default:
-			cellAdec->Error("adecRawRead(): sequence error (task %d)", adec.job.Peek().type);
+		{
+			cellAdec->Error("adecRawRead(): unknown task (%d)", task.type);
+			Emu.Pause();
 			return -1;
 		}
+		}
 
 		goto next;
 	}
@@ -264,24 +267,18 @@ u32 adecOpen(AudioDecoder* data)
 
 		while (true)
 		{
-			if (Emu.IsStopped())
+			if (Emu.IsStopped() || adec.is_closed)
 			{
 				break;
 			}
 
-			if (!adec.job.GetCountUnsafe() && adec.is_running)
+			//if (!adec.job.GetCountUnsafe() && adec.is_running)
-			{
+			//{
-				std::this_thread::sleep_for(std::chrono::milliseconds(1));
+			//	std::this_thread::sleep_for(std::chrono::milliseconds(1));
-				continue;
+			//	continue;
-			}
+			//}
 
-			/*if (adec.frames.GetCount() >= 50)
+			if (!adec.job.Pop(task, &adec.is_closed))
-			{
-				std::this_thread::sleep_for(std::chrono::milliseconds(1));
-				continue;
-			}*/
-
-			if (!adec.job.Pop(task))
 			{
 				break;
 			}
@@ -428,10 +425,10 @@ u32 adecOpen(AudioDecoder* data)
 
 				while (true)
 				{
-					if (Emu.IsStopped())
+					if (Emu.IsStopped() || adec.is_closed)
 					{
-						cellAdec->Warning("adecDecodeAu: aborted");
+						if (Emu.IsStopped()) cellAdec->Warning("adecDecodeAu: aborted");
-						return;
+						break;
 					}
 
 					/*if (!adec.ctx) // fake
@@ -534,30 +531,32 @@ u32 adecOpen(AudioDecoder* data)
 							//frame.pts, frame.data->nb_samples, frame.data->channels, frame.data->sample_rate,
 							//av_get_bytes_per_sample((AVSampleFormat)frame.data->format));
 
-						adec.frames.Push(frame);
+						if (adec.frames.Push(frame, &adec.is_closed))
+						{
 							frame.data = nullptr; // to prevent destruction
-
 							adec.cbFunc.call(*adec.adecCb, adec.id, CELL_ADEC_MSG_TYPE_PCMOUT, CELL_OK, adec.cbArg);
 						}
 					}
+				}
 
 				adec.cbFunc.call(*adec.adecCb, adec.id, CELL_ADEC_MSG_TYPE_AUDONE, task.au.auInfo_addr, adec.cbArg);
 			}
 			break;
 
-			case adecClose:
+			case adecClose: break;
-			{
-				adec.is_finished = true;
-				cellAdec->Notice("Audio Decoder thread ended");
-				return;
-			}
 
 			default:
+			{
 				cellAdec->Error("Audio Decoder thread error: unknown task(%d)", task.type);
+				Emu.Pause();
+				return;
 			}
 			}
+		}
+
 		adec.is_finished = true;
-		cellAdec->Warning("Audio Decoder thread aborted");
+		if (adec.is_closed) cellAdec->Notice("Audio Decoder thread ended");
+		if (Emu.IsStopped()) cellAdec->Warning("Audio Decoder thread aborted");
 	});
 
 	t.detach();
@@ -639,7 +638,8 @@ int cellAdecClose(u32 handle)
 		return CELL_ADEC_ERROR_ARG;
 	}
 
-	adec->job.Push(AdecTask(adecClose));
+	adec->is_closed = true;
+	adec->job.Push(AdecTask(adecClose), &sq_no_wait);
 
 	while (!adec->is_finished)
 	{
@@ -658,7 +658,7 @@ int cellAdecClose(u32 handle)
 
 int cellAdecStartSeq(u32 handle, u32 param_addr)
 {
-	cellAdec->Log("cellAdecStartSeq(handle=%d, param_addr=0x%x)", handle, param_addr);
+	cellAdec->Todo("cellAdecStartSeq(handle=%d, param_addr=0x%x)", handle, param_addr);
 
 	AudioDecoder* adec;
 	if (!Emu.GetIdManager().GetIDData(handle, adec))
@@ -667,16 +667,10 @@ int cellAdecStartSeq(u32 handle, u32 param_addr)
 	}
 
 	AdecTask task(adecStartSeq);
-	/*if (adec->type == CELL_ADEC_TYPE_ATRACX_2CH)
-	{
 
-	}
+	// TODO: using parameters
-	else*/
-	{
-		cellAdec->Todo("cellAdecStartSeq(): initialization");
-	}
 
-	adec->job.Push(task);
+	adec->job.Push(task, &adec->is_closed);
 	return CELL_OK;
 }
 
@@ -690,7 +684,7 @@ int cellAdecEndSeq(u32 handle)
 		return CELL_ADEC_ERROR_ARG;
 	}
 
-	adec->job.Push(AdecTask(adecEndSeq));
+	adec->job.Push(AdecTask(adecEndSeq), &adec->is_closed);
 	return CELL_OK;
 }
 
@@ -711,7 +705,7 @@ int cellAdecDecodeAu(u32 handle, vm::ptr<CellAdecAuInfo> auInfo)
 	task.au.pts = ((u64)auInfo->pts.upper << 32) | (u64)auInfo->pts.lower;
 	task.au.userdata = auInfo->userData;
 
-	adec->job.Push(task);
+	adec->job.Push(task, &adec->is_closed);
 	return CELL_OK;
 }
 
@@ -725,20 +719,23 @@ int cellAdecGetPcm(u32 handle, vm::ptr<float> outBuffer)
 		return CELL_ADEC_ERROR_ARG;
 	}
 
-	if (adec->frames.IsEmpty())
+	AdecFrame af;
+	if (!adec->frames.Pop(af, &sq_no_wait))
 	{
+		std::this_thread::sleep_for(std::chrono::milliseconds(1)); // hack
 		return CELL_ADEC_ERROR_EMPTY;
 	}
 
-	AdecFrame af;
-	adec->frames.Pop(af);
 	AVFrame* frame = af.data;
 
-	if (!af.data) // fake: empty data
+	if (!af.data)
 	{
+		// hack
 		return CELL_OK;
 	}
 
+	if (outBuffer)
+	{
 		// reverse byte order, extract data:
 		float* in_f[2];
 		in_f[0] = (float*)frame->extended_data[0];
@@ -748,12 +745,10 @@ int cellAdecGetPcm(u32 handle, vm::ptr<float> outBuffer)
 			outBuffer[i * 2 + 0] = in_f[0][i];
 			outBuffer[i * 2 + 1] = in_f[1][i];
 		}
+	}
 
-	if (af.data)
-	{
 	av_frame_unref(af.data);
 	av_frame_free(&af.data);
-	}
 	return CELL_OK;
 }
 
@@ -767,14 +762,13 @@ int cellAdecGetPcmItem(u32 handle, vm::ptr<u32> pcmItem_ptr)
 		return CELL_ADEC_ERROR_ARG;
 	}
 
-	if (adec->frames.IsEmpty())
+	AdecFrame af;
+	if (!adec->frames.Peek(af, &sq_no_wait))
 	{
 		std::this_thread::sleep_for(std::chrono::milliseconds(1)); // hack
 		return CELL_ADEC_ERROR_EMPTY;
 	}
 
-	AdecFrame& af = adec->frames.Peek();
-
 	AVFrame* frame = af.data;
 
 	auto pcm = vm::ptr<CellAdecPcmItem>::make(adec->memAddr + adec->memBias);

rpcs3/Emu/SysCalls/Modules/cellAdec.h

@@ -1074,6 +1074,7 @@ public:
 	SQueue<AdecTask> job;
 	u32 id;
 	volatile bool is_running;
+	volatile bool is_closed;
 	volatile bool is_finished;
 	bool just_started;
 	bool just_finished;

rpcs3/Emu/SysCalls/Modules/cellAudio.cpp

@@ -105,9 +105,9 @@ int cellAudioInit()
 					float* oal_buffer_float = nullptr;
 
 					if (g_is_u16)
-						queue.Pop(oal_buffer);
+						queue.Pop(oal_buffer, nullptr);
 					else
-						queue_float.Pop(oal_buffer_float);
+						queue_float.Pop(oal_buffer_float, nullptr);
 
 					if (g_is_u16)
 					{
@@ -153,7 +153,6 @@ int cellAudioInit()
 				m_config.counter++;
 
 				const u32 oal_pos = m_config.counter % BUFFER_NUM;
-				const u32 oal_pos_float = m_config.counter % BUFFER_NUM;
 
 				if (Emu.IsPaused())
 				{
@@ -350,10 +349,10 @@ int cellAudioInit()
 								_mm_cvtps_epi32(_mm_mul_ps((__m128&)(buf2ch[i + 4]), float2u16)));
 						}
 					}
-
+					else
 					for (u32 i = 0; i < (sizeof(buf2ch) / sizeof(float)); i++)
 					{
-						oal_buffer_float[oal_pos_float][oal_buffer_offset + i] = buf2ch[i];
+						oal_buffer_float[oal_pos][oal_buffer_offset + i] = buf2ch[i];
 					}
 				}
 
@@ -361,8 +360,8 @@ int cellAudioInit()
 
 				if (first_mix)
 				{
-					memset(&oal_buffer[oal_pos][0], 0, oal_buffer_size * sizeof(s16));
+					if (g_is_u16) memset(&oal_buffer[oal_pos][0], 0, oal_buffer_size * sizeof(s16));
-					memset(&oal_buffer_float[oal_pos_float][0], 0, oal_buffer_size * sizeof(float));
+					else memset(&oal_buffer_float[oal_pos][0], 0, oal_buffer_size * sizeof(float));
 				}
 				oal_buffer_offset += sizeof(buf2ch) / sizeof(float);
 
@@ -371,9 +370,9 @@ int cellAudioInit()
 					if(m_audio_out)
 					{
 						if (g_is_u16)
-							queue.Push(&oal_buffer[oal_pos][0]);
+							queue.Push(&oal_buffer[oal_pos][0], nullptr);
 
-						queue_float.Push(&oal_buffer_float[oal_pos_float][0]);
+						queue_float.Push(&oal_buffer_float[oal_pos][0], nullptr);
 					}
 
 					oal_buffer_offset = 0;
@@ -439,8 +438,8 @@ int cellAudioInit()
 			}
 			cellAudio->Notice("Audio thread ended");
 abort:
-			queue.Push(nullptr);
+			queue.Push(nullptr, nullptr);
-			queue_float.Push(nullptr);
+			queue_float.Push(nullptr, nullptr);
 
 			if(do_dump)
 				m_dump.Finalize();

rpcs3/Emu/SysCalls/Modules/cellDmux.cpp

@@ -62,7 +62,11 @@ bool ElementaryStream::is_full()
 {
 	if (released < put_count)
 	{
-		u32 first = entries.Peek();
+		u32 first;
+		if (!entries.Peek(first, &dmux->is_closed))
+		{
+			return false;
+		}
 		if (first >= put)
 		{
 			return (first - put) < GetMaxAU();
@@ -136,9 +140,10 @@ void ElementaryStream::finish(DemuxerStream& stream) // not multithread-safe (or
 		put_count++;
 		//if (fidMajor != 0xbd) LOG_NOTICE(HLE, "<<< es::finish(): peek=0x%x, first=0x%x, put=0x%x, size=0x%x", peek, first, put, size);
 	}
-	if (!entries.Push(addr))
+	if (!entries.Push(addr, &sq_no_wait))
 	{
 		cellDmux->Error("es::finish() aborted (no space)");
+		Emu.Pause();
 	}
 }
 
@@ -196,10 +201,15 @@ bool ElementaryStream::release()
 	if (released >= put_count)
 	{
 		cellDmux->Error("es::release(): buffer is empty");
+		Emu.Pause();
 		return false;
 	}
 
-	u32 addr = entries.Peek();
+	u32 addr;
+	if (!entries.Peek(addr, &dmux->is_closed))
+	{
+		return false; // ???
+	}
 
 	auto info = vm::ptr<CellDmuxAuInfo>::make(addr);
 	//if (fidMajor != 0xbd) LOG_WARNING(HLE, "es::release(): (%s) size = 0x%x, info = 0x%x, pts = 0x%x",
@@ -208,13 +218,15 @@ bool ElementaryStream::release()
 	if (released >= peek_count)
 	{
 		cellDmux->Error("es::release(): buffer has not been seen yet");
+		Emu.Pause();
 		return false;
 	}
 
 	released++;
-	if (!entries.Pop(addr))
+	if (!entries.Pop(addr, &sq_no_wait))
 	{
 		cellDmux->Error("es::release(): entries.Pop() aborted (no entries found)");
+		Emu.Pause();
 		return false;
 	}
 	//if (fidMajor != 0xbd) LOG_NOTICE(HLE, "<<< es::release(): peek=0x%x, first=0x%x, put=0x%x, size=0x%x", peek, first, put, size);
@@ -235,7 +247,12 @@ bool ElementaryStream::peek(u32& out_data, bool no_ex, u32& out_spec, bool updat
 		return false;
 	}
 
-	u32 addr = entries.Peek(peek_count - released);
+	u32 addr;
+	if (!entries.Peek(addr, &dmux->is_closed, peek_count - released))
+	{
+		return false; // ???
+	}
+
 	auto info = vm::ptr<CellDmuxAuInfo>::make(addr);
 	//if (fidMajor != 0xbd) LOG_WARNING(HLE, "es::peek(%sAu(Ex)): (%s) size = 0x%x, info = 0x%x, pts = 0x%x",
 	//wxString(update_index ? "Get" : "Peek").wx_str(),
@@ -325,12 +342,12 @@ u32 dmuxOpen(Demuxer* data)
 
 		while (true)
 		{
-			if (Emu.IsStopped())
+			if (Emu.IsStopped() || dmux.is_closed)
 			{
 				break;
 			}
 			
-			if (!dmux.job.GetCountUnsafe() && dmux.is_running)
+			if (!dmux.job.Peek(task, &sq_no_wait) && dmux.is_running)
 			{
 				// default task (demuxing) (if there is no other work)
 				be_t<u32> code;
@@ -522,6 +539,7 @@ u32 dmuxOpen(Demuxer* data)
 				case USER_DATA_START_CODE:
 				{
 					cellDmux->Error("USER_DATA_START_CODE found");
+					Emu.Pause();
 					return;
 				}
 
@@ -530,14 +548,12 @@ u32 dmuxOpen(Demuxer* data)
 					// search
 					stream.skip(1);
 				}
-					break;
-
 				}
 				continue;
 			}
 
 			// wait for task with yielding (if no default work)
-			if (!dmux.job.Pop(task))
+			if (!dmux.job.Pop(task, &dmux.is_closed))
 			{
 				break; // Emu is stopped
 			}
@@ -563,6 +579,7 @@ u32 dmuxOpen(Demuxer* data)
 				if (updates_count != updates_signaled)
 				{
 					cellDmux->Error("dmuxSetStream: stream update inconsistency (input=%d, signaled=%d)", updates_count, updates_signaled);
+					Emu.Pause();
 					return;
 				}
 
@@ -572,7 +589,7 @@ u32 dmuxOpen(Demuxer* data)
 					//stream.addr, stream.size, stream.discontinuity, stream.userdata);
 
 				dmux.is_running = true;
-					dmux.fbSetStream.Push(task.stream.addr); // feedback
+				dmux.fbSetStream.Push(task.stream.addr, &dmux.is_closed); // feedback
 			}
 			break;
 
@@ -588,18 +605,11 @@ u32 dmuxOpen(Demuxer* data)
 				dmux.is_running = false;
 				if (task.type == dmuxResetStreamAndWaitDone)
 				{
-						dmux.fbSetStream.Push(0);
+					dmux.fbSetStream.Push(0, &dmux.is_closed);
 				}
 			}
 			break;
 
-			case dmuxClose:
-				{
-					dmux.is_finished = true;
-					cellDmux->Notice("Demuxer thread ended");
-					return;
-				}
-
 			case dmuxEnableEs:
 			{
 				ElementaryStream& es = *task.es.es_ptr;
@@ -646,12 +656,6 @@ u32 dmuxOpen(Demuxer* data)
 			}
 			break;
 
-			/*case dmuxReleaseAu:
-				{
-					task.es.es_ptr->release();
-				}
-				break;*/
-
 			case dmuxFlushEs:
 			{
 				ElementaryStream& es = *task.es.es_ptr;
@@ -680,12 +684,20 @@ u32 dmuxOpen(Demuxer* data)
 			}
 			break;
 
+			case dmuxClose: break;
+
 			default:
+			{
 				cellDmux->Error("Demuxer thread error: unknown task(%d)", task.type);
+				Emu.Pause();
 				return;
 			}	
 			}
-		cellDmux->Warning("Demuxer thread aborted");
+		}
+
+		dmux.is_finished = true;
+		if (Emu.IsStopped()) cellDmux->Warning("Demuxer thread aborted");
+		if (dmux.is_closed) cellDmux->Notice("Demuxer thread ended");
 	});
 
 	t.detach();
@@ -783,7 +795,8 @@ int cellDmuxClose(u32 demuxerHandle)
 		return CELL_DMUX_ERROR_ARG;
 	}
 
-	dmux->job.Push(DemuxerTask(dmuxClose));
+	dmux->is_closed = true;
+	dmux->job.Push(DemuxerTask(dmuxClose), &sq_no_wait);
 
 	while (!dmux->is_finished)
 	{
@@ -830,10 +843,10 @@ int cellDmuxSetStream(u32 demuxerHandle, const u32 streamAddress, u32 streamSize
 	info.discontinuity = discontinuity;
 	info.userdata = userData;
 
-	dmux->job.Push(task);
+	dmux->job.Push(task, &dmux->is_closed);
 
 	u32 addr;
-	if (!dmux->fbSetStream.Pop(addr))
+	if (!dmux->fbSetStream.Pop(addr, &dmux->is_closed))
 	{
 		cellDmux->Warning("cellDmuxSetStream(%d) aborted (fbSetStream.Pop())", demuxerHandle);
 		return CELL_OK;
@@ -856,7 +869,7 @@ int cellDmuxResetStream(u32 demuxerHandle)
 		return CELL_DMUX_ERROR_ARG;
 	}
 
-	dmux->job.Push(DemuxerTask(dmuxResetStream));
+	dmux->job.Push(DemuxerTask(dmuxResetStream), &dmux->is_closed);
 
 	return CELL_OK;
 }
@@ -871,10 +884,10 @@ int cellDmuxResetStreamAndWaitDone(u32 demuxerHandle)
 		return CELL_DMUX_ERROR_ARG;
 	}
 
-	dmux->job.Push(DemuxerTask(dmuxResetStreamAndWaitDone));
+	dmux->job.Push(DemuxerTask(dmuxResetStreamAndWaitDone), &dmux->is_closed);
 
 	u32 addr;
-	if (!dmux->fbSetStream.Pop(addr))
+	if (!dmux->fbSetStream.Pop(addr, &dmux->is_closed))
 	{
 		cellDmux->Warning("cellDmuxResetStreamAndWaitDone(%d) aborted (fbSetStream.Pop())", demuxerHandle);
 		return CELL_OK;
@@ -952,7 +965,7 @@ int cellDmuxEnableEs(u32 demuxerHandle, vm::ptr<const CellCodecEsFilterId> esFil
 	task.es.es = id;
 	task.es.es_ptr = es;
 
-	dmux->job.Push(task);
+	dmux->job.Push(task, &dmux->is_closed);
 	return CELL_OK;
 }
 
@@ -970,7 +983,7 @@ int cellDmuxDisableEs(u32 esHandle)
 	task.es.es = esHandle;
 	task.es.es_ptr = es;
 
-	es->dmux->job.Push(task);
+	es->dmux->job.Push(task, &es->dmux->is_closed);
 	return CELL_OK;
 }
 
@@ -988,7 +1001,7 @@ int cellDmuxResetEs(u32 esHandle)
 	task.es.es = esHandle;
 	task.es.es_ptr = es;
 
-	es->dmux->job.Push(task);
+	es->dmux->job.Push(task, &es->dmux->is_closed);
 	return CELL_OK;
 }
 
@@ -1115,7 +1128,7 @@ int cellDmuxFlushEs(u32 esHandle)
 	task.es.es = esHandle;
 	task.es.es_ptr = es;
 
-	es->dmux->job.Push(task);
+	es->dmux->job.Push(task, &es->dmux->is_closed);
 	return CELL_OK;
 }
 

rpcs3/Emu/SysCalls/Modules/cellDmux.h

@@ -415,12 +415,14 @@ public:
 	const u32 cbArg;
 	u32 id;
 	volatile bool is_finished;
+	volatile bool is_closed;
 	volatile bool is_running;
 
 	PPUThread* dmuxCb;
 
 	Demuxer(u32 addr, u32 size, vm::ptr<CellDmuxCbMsg> func, u32 arg)
 		: is_finished(false)
+		, is_closed(false)
 		, is_running(false)
 		, memAddr(addr)
 		, memSize(size)

rpcs3/Emu/SysCalls/Modules/cellSync.cpp

@@ -568,10 +568,11 @@ s32 cellSyncQueueTryPush(vm::ptr<CellSyncQueue> queue, vm::ptr<const void> buffe
 	assert(((u32)data.m_v1 & 0xffffff) <= depth && ((u32)data.m_v2 & 0xffffff) <= depth);
 
 	u32 position;
-	if (s32 res = queue->data.atomic_op(CELL_OK, [depth, &position](CellSyncQueue::data_t& queue) -> s32
+	s32 res = queue->data.atomic_op(CELL_OK, [depth, &position](CellSyncQueue::data_t& queue) -> s32
 	{
 		return syncQueueTryPushOp(queue, depth, position);
-	}))
+	});
+	if (res)
 	{
 		return res;
 	}
@@ -657,10 +658,11 @@ s32 cellSyncQueueTryPop(vm::ptr<CellSyncQueue> queue, vm::ptr<void> buffer)
 	assert(((u32)data.m_v1 & 0xffffff) <= depth && ((u32)data.m_v2 & 0xffffff) <= depth);
 
 	u32 position;
-	if (s32 res = queue->data.atomic_op(CELL_OK, [depth, &position](CellSyncQueue::data_t& queue) -> s32
+	s32 res = queue->data.atomic_op(CELL_OK, [depth, &position](CellSyncQueue::data_t& queue) -> s32
 	{
 		return syncQueueTryPopOp(queue, depth, position);
-	}))
+	});
+	if (res)
 	{
 		return res;
 	}
@@ -737,10 +739,11 @@ s32 cellSyncQueueTryPeek(vm::ptr<CellSyncQueue> queue, vm::ptr<void> buffer)
 	assert(((u32)data.m_v1 & 0xffffff) <= depth && ((u32)data.m_v2 & 0xffffff) <= depth);
 
 	u32 position;
-	if (s32 res = queue->data.atomic_op(CELL_OK, [depth, &position](CellSyncQueue::data_t& queue) -> s32
+	s32 res = queue->data.atomic_op(CELL_OK, [depth, &position](CellSyncQueue::data_t& queue) -> s32
 	{
 		return syncQueueTryPeekOp(queue, depth, position);
-	}))
+	});
+	if (res)
 	{
 		return res;
 	}

rpcs3/Emu/SysCalls/Modules/cellVdec.cpp

@@ -28,6 +28,7 @@ VideoDecoder::VideoDecoder(CellVdecCodecType type, u32 profile, u32 addr, u32 si
 	, cbFunc(func)
 	, cbArg(arg)
 	, is_finished(false)
+	, is_closed(false)
 	, is_running(false)
 	, just_started(false)
 	, just_finished(false)
@@ -64,14 +65,14 @@ VideoDecoder::VideoDecoder(CellVdecCodecType type, u32 profile, u32 addr, u32 si
 VideoDecoder::~VideoDecoder()
 {
 	// TODO: check finalization
-	if (ctx)
+	VdecFrame vf;
+	while (frames.Pop(vf, &sq_no_wait))
 	{
-		for (u32 i = frames.GetCount() - 1; ~i; i--)
-		{
-			VdecFrame& vf = frames.Peek(i);
 		av_frame_unref(vf.data);
 		av_frame_free(&vf.data);
 	}
+	if (ctx)
+	{
 		avcodec_close(ctx);
 		avformat_close_input(&fmt);
 	}
@@ -95,17 +96,14 @@ int vdecRead(void* opaque, u8* buf, int buf_size)
 next:
 	if (vdec.reader.size < (u32)buf_size /*&& !vdec.just_started*/)
 	{
-		while (!vdec.job.GetCountUnsafe())
+		VdecTask task;
+		if (!vdec.job.Peek(task, &vdec.is_closed))
 		{
-			if (Emu.IsStopped())
+			if (Emu.IsStopped()) cellVdec->Warning("vdecRead() aborted");
-			{
-				cellVdec->Warning("vdecRead(): aborted");
 			return 0;
 		}
-			std::this_thread::sleep_for(std::chrono::milliseconds(1));
-		}
 
-		switch (vdec.job.Peek().type)
+		switch (task.type)
 		{
 		case vdecEndSeq:
 		case vdecClose:
@@ -113,6 +111,7 @@ next:
 			buf_size = vdec.reader.size;
 		}
 		break;
+		
 		case vdecDecodeAu:
 		{
 			memcpy(buf, vm::get_ptr<void>(vdec.reader.addr), vdec.reader.size);
@@ -123,16 +122,20 @@ next:
 
 			vdec.cbFunc.call(*vdec.vdecCb, vdec.id, CELL_VDEC_MSG_TYPE_AUDONE, CELL_OK, vdec.cbArg);
 
-				vdec.job.Pop(vdec.task);
+			vdec.job.Pop(vdec.task, nullptr);
 
 			vdec.reader.addr = vdec.task.addr;
 			vdec.reader.size = vdec.task.size;
 			//LOG_NOTICE(HLE, "Video AU: size = 0x%x, pts = 0x%llx, dts = 0x%llx", vdec.task.size, vdec.task.pts, vdec.task.dts);
 		}
 		break;
+		
 		default:
-			cellVdec->Error("vdecRead(): sequence error (task %d)", vdec.job.Peek().type);
+		{
-			return 0;
+			cellVdec->Error("vdecRead(): unknown task (%d)", task.type);
+			Emu.Pause();
+			return -1;
+		}
 		}
 		
 		goto next;
@@ -199,24 +202,18 @@ u32 vdecOpen(VideoDecoder* data)
 
 		while (true)
 		{
-			if (Emu.IsStopped())
+			if (Emu.IsStopped() || vdec.is_closed)
 			{
 				break;
 			}
 
-			if (!vdec.job.GetCountUnsafe() && vdec.is_running)
+			//if (!vdec.job.GetCountUnsafe() && vdec.is_running)
-			{
+			//{
-				std::this_thread::sleep_for(std::chrono::milliseconds(1));
+			//	std::this_thread::sleep_for(std::chrono::milliseconds(1));
-				continue;
+			//	continue;
-			}
+			//}
 
-			if (vdec.frames.GetCount() >= 50)
+			if (!vdec.job.Pop(task, &vdec.is_closed))
-			{
-				std::this_thread::sleep_for(std::chrono::milliseconds(1));
-				continue;
-			}
-
-			if (!vdec.job.Pop(task))
 			{
 				break;
 			}
@@ -358,11 +355,10 @@ u32 vdecOpen(VideoDecoder* data)
 
 				while (true)
 				{
-					if (Emu.IsStopped() || vdec.job.PeekIfExist().type == vdecClose)
+					if (Emu.IsStopped() || vdec.is_closed)
 					{
-						vdec.is_finished = true;
+						if (Emu.IsStopped()) cellVdec->Warning("vdecDecodeAu: aborted");
-						cellVdec->Warning("vdecDecodeAu: aborted");
+						break;
-						return;
 					}
 
 					last_frame = av_read_frame(vdec.fmt, &au) < 0;
@@ -432,37 +428,39 @@ u32 vdecOpen(VideoDecoder* data)
 
 						//LOG_NOTICE(HLE, "got picture (pts=0x%llx, dts=0x%llx)", frame.pts, frame.dts);
 
-						vdec.frames.Push(frame); // !!!!!!!!
+						if (vdec.frames.Push(frame, &vdec.is_closed))
+						{
 							frame.data = nullptr; // to prevent destruction
-
 							vdec.cbFunc.call(*vdec.vdecCb, vdec.id, CELL_VDEC_MSG_TYPE_PICOUT, CELL_OK, vdec.cbArg);
 						}
 					}
+				}
 
 				vdec.cbFunc.call(*vdec.vdecCb, vdec.id, CELL_VDEC_MSG_TYPE_AUDONE, CELL_OK, vdec.cbArg);
 			}
 			break;
 
-			case vdecClose:
-			{
-				vdec.is_finished = true;
-				cellVdec->Notice("Video Decoder thread ended");
-				return;
-			}
-
 			case vdecSetFrameRate:
 			{
 				cellVdec->Error("TODO: vdecSetFrameRate(%d)", task.frc);
+				Emu.Pause();
 			}
 			break;
 
+			case vdecClose: break;
+
 			default:
+			{
 				cellVdec->Error("Video Decoder thread error: unknown task(%d)", task.type);
+				Emu.Pause();
+				return;
+			}
 			}
 		}
 
 		vdec.is_finished = true;
-		cellVdec->Warning("Video Decoder thread aborted");
+		if (Emu.IsStopped()) cellVdec->Warning("Video Decoder thread aborted");
+		if (vdec.is_closed) cellVdec->Notice("Video Decoder thread ended");
 	});
 
 	t.detach();
@@ -514,7 +512,8 @@ int cellVdecClose(u32 handle)
 		return CELL_VDEC_ERROR_ARG;
 	}
 
-	vdec->job.Push(VdecTask(vdecClose));
+	vdec->is_closed = true;
+	vdec->job.Push(VdecTask(vdecClose), &sq_no_wait);
 
 	while (!vdec->is_finished)
 	{
@@ -541,7 +540,7 @@ int cellVdecStartSeq(u32 handle)
 		return CELL_VDEC_ERROR_ARG;
 	}
 
-	vdec->job.Push(VdecTask(vdecStartSeq));
+	vdec->job.Push(VdecTask(vdecStartSeq), &vdec->is_closed);
 	return CELL_OK;
 }
 
@@ -555,29 +554,7 @@ int cellVdecEndSeq(u32 handle)
 		return CELL_VDEC_ERROR_ARG;
 	}
 
-	/*if (!vdec->job.IsEmpty())
+	vdec->job.Push(VdecTask(vdecEndSeq), &vdec->is_closed);
-	{
-		std::this_thread::sleep_for(std::chrono::milliseconds(1));
-		return CELL_VDEC_ERROR_BUSY; // ???
-	}
-
-	if (!vdec->frames.IsEmpty())
-	{
-		std::this_thread::sleep_for(std::chrono::milliseconds(1));
-		return CELL_VDEC_ERROR_BUSY; // ???
-	}*/
-
-	while (!vdec->job.IsEmpty() || !vdec->frames.IsEmpty())
-	{
-		if (Emu.IsStopped())
-		{
-			cellVdec->Warning("cellVdecEndSeq(%d) aborted", handle);
-			return CELL_OK;
-		}
-		std::this_thread::sleep_for(std::chrono::milliseconds(1));
-	}
-
-	vdec->job.Push(VdecTask(vdecEndSeq));
 	return CELL_OK;
 }
 
@@ -601,7 +578,7 @@ int cellVdecDecodeAu(u32 handle, CellVdecDecodeMode mode, vm::ptr<const CellVdec
 	task.userData = auInfo->userData;
 	task.specData = auInfo->codecSpecificData;
 
-	vdec->job.Push(task);
+	vdec->job.Push(task, &vdec->is_closed);
 	return CELL_OK;
 }
 
@@ -615,11 +592,19 @@ int cellVdecGetPicture(u32 handle, vm::ptr<const CellVdecPicFormat> format, vm::
 		return CELL_VDEC_ERROR_ARG;
 	}
 
-	if (vdec->frames.IsEmpty())
+	VdecFrame vf;
+	if (!vdec->frames.Pop(vf, &sq_no_wait))
 	{
+		std::this_thread::sleep_for(std::chrono::milliseconds(1)); // hack
 		return CELL_VDEC_ERROR_EMPTY;
 	}
 
+	if (!vf.data)
+	{
+		// hack
+		return CELL_OK;
+	}
+
 	if (outBuff)
 	{
 		u32 buf_size = a128(av_image_get_buffer_size(vdec->ctx->pix_fmt, vdec->ctx->width, vdec->ctx->height, 1));
@@ -636,10 +621,6 @@ int cellVdecGetPicture(u32 handle, vm::ptr<const CellVdecPicFormat> format, vm::
 			return CELL_OK;
 		}
 
-		VdecFrame vf;
-
-		vdec->frames.Pop(vf);
-
 		AVFrame& frame = *vf.data;
 
 		// TODO: zero padding bytes
@@ -650,11 +631,10 @@ int cellVdecGetPicture(u32 handle, vm::ptr<const CellVdecPicFormat> format, vm::
 			cellVdec->Error("cellVdecGetPicture: av_image_copy_to_buffer failed(%d)", err);
 			Emu.Pause();
 		}
+	}
 
 	av_frame_unref(vf.data);
 	av_frame_free(&vf.data);
-	}
-
 	return CELL_OK;
 }
 
@@ -668,14 +648,13 @@ int cellVdecGetPicItem(u32 handle, vm::ptr<u32> picItem_ptr)
 		return CELL_VDEC_ERROR_ARG;
 	}
 
-	if (vdec->frames.IsEmpty())
+	VdecFrame vf;
+	if (!vdec->frames.Peek(vf, &sq_no_wait))
 	{
 		std::this_thread::sleep_for(std::chrono::milliseconds(1)); // hack
 		return CELL_VDEC_ERROR_EMPTY;
 	}
 
-	VdecFrame& vf = vdec->frames.Peek();
-
 	AVFrame& frame = *vf.data;
 
 	auto info = vm::ptr<CellVdecPicItem>::make(vdec->memAddr + vdec->memBias);
@@ -782,7 +761,7 @@ int cellVdecSetFrameRate(u32 handle, CellVdecFrameRate frc)
 	VdecTask task(vdecSetFrameRate);
 	task.frc = frc;
 
-	vdec->job.Push(task);
+	vdec->job.Push(task, &vdec->is_closed);
 	return CELL_OK;
 }
 

rpcs3/Emu/SysCalls/Modules/cellVdec.h

@@ -647,7 +647,6 @@ struct CellVdecMpeg2Info
 
 enum VdecJobType : u32
 {
-	vdecInvalid,
 	vdecStartSeq,
 	vdecEndSeq,
 	vdecDecodeAu,
@@ -676,7 +675,6 @@ struct VdecTask
 	}
 
 	VdecTask()
-		: type(vdecInvalid)
 	{
 	}
 };
@@ -697,6 +695,7 @@ public:
 	SQueue<VdecTask> job;
 	u32 id;
 	volatile bool is_running;
+	volatile bool is_closed;
 	volatile bool is_finished;
 	bool just_started;
 	bool just_finished;