rpcsx/rpcs3/Emu/Cell/lv2/sys_cond.cpp

#include "stdafx.h"

#include "util/serialization.hpp"
#include "Emu/IdManager.h"
#include "Emu/IPC.h"
#include "Emu/System.h"

#include "Emu/Cell/ErrorCodes.h"
#include "Emu/Cell/PPUThread.h"

#include "sys_cond.h"

#include "util/asm.hpp"

LOG_CHANNEL(sys_cond);

lv2_cond::lv2_cond(utils::serial& ar)
	: key(ar)
	, name(ar)
	, mtx_id(ar)
	, mutex(idm::get_unlocked<lv2_obj, lv2_mutex>(mtx_id)) // May be nullptr
{
}

CellError lv2_cond::on_id_create()
{
	exists++;

	static auto do_it = [](lv2_cond* _this) -> CellError
	{
		if (lv2_obj::check(_this->mutex))
		{
			_this->mutex->cond_count++;
			return {};
		}

		// Mutex has been destroyed, cannot create conditional variable
		return CELL_ESRCH;
	};

	if (mutex)
	{
		return do_it(this);
	}

	ensure(!!Emu.DeserialManager());

	Emu.DeferDeserialization([this]()
	{
		if (!mutex)
		{
			mutex = ensure(idm::get_unlocked<lv2_obj, lv2_mutex>(mtx_id));
		}

		// Defer function
		ensure(CellError{} == do_it(this));
	});

	return {};
}

std::shared_ptr<void> lv2_cond::load(utils::serial& ar)
{
	auto cond = std::make_shared<lv2_cond>(ar);
	return lv2_obj::load(cond->key, cond);
}

void lv2_cond::save(utils::serial& ar)
{
	ar(key, name, mtx_id);
}

error_code sys_cond_create(ppu_thread& ppu, vm::ptr<u32> cond_id, u32 mutex_id, vm::ptr<sys_cond_attribute_t> attr)
{
	ppu.state += cpu_flag::wait;

	sys_cond.warning("sys_cond_create(cond_id=*0x%x, mutex_id=0x%x, attr=*0x%x)", cond_id, mutex_id, attr);

	auto mutex = idm::get<lv2_obj, lv2_mutex>(mutex_id);

	if (!mutex)
	{
		return CELL_ESRCH;
	}

	const auto _attr = *attr;

	const u64 ipc_key = lv2_obj::get_key(_attr);

	if (const auto error = lv2_obj::create<lv2_cond>(_attr.pshared, ipc_key, _attr.flags, [&]
	{
		return std::make_shared<lv2_cond>(
			ipc_key,
			_attr.name_u64,
			mutex_id,
			std::move(mutex));
	}))
	{
		return error;
	}

	ppu.check_state();
	*cond_id = idm::last_id();
	return CELL_OK;
}

error_code sys_cond_destroy(ppu_thread& ppu, u32 cond_id)
{
	ppu.state += cpu_flag::wait;

	sys_cond.warning("sys_cond_destroy(cond_id=0x%x)", cond_id);

	const auto cond = idm::withdraw<lv2_obj, lv2_cond>(cond_id, [&](lv2_cond& cond) -> CellError
	{
		std::lock_guard lock(cond.mutex->mutex);

		if (atomic_storage<ppu_thread*>::load(cond.sq))
		{
			return CELL_EBUSY;
		}

		cond.mutex->cond_count--;
		lv2_obj::on_id_destroy(cond, cond.key);
		return {};
	});

	if (!cond)
	{
		return CELL_ESRCH;
	}

	if (cond.ret)
	{
		return cond.ret;
	}

	return CELL_OK;
}

error_code sys_cond_signal(ppu_thread& ppu, u32 cond_id)
{
	ppu.state += cpu_flag::wait;

	sys_cond.trace("sys_cond_signal(cond_id=0x%x)", cond_id);

	const auto cond = idm::check<lv2_obj, lv2_cond>(cond_id, [&, notify = lv2_obj::notify_all_t()](lv2_cond& cond)
	{
		if (atomic_storage<ppu_thread*>::load(cond.sq))
		{
			std::lock_guard lock(cond.mutex->mutex);

			if (const auto cpu = cond.schedule<ppu_thread>(cond.sq, cond.mutex->protocol))
			{
				if (static_cast<ppu_thread*>(cpu)->state & cpu_flag::again)
				{
					ppu.state += cpu_flag::again;
					return;
				}

				// TODO: Is EBUSY returned after reqeueing, on sys_cond_destroy?

				if (cond.mutex->try_own(*cpu))
				{
					cond.awake(cpu);
				}
			}
		}
	});

	if (!cond)
	{
		return CELL_ESRCH;
	}

	return CELL_OK;
}

error_code sys_cond_signal_all(ppu_thread& ppu, u32 cond_id)
{
	ppu.state += cpu_flag::wait;

	sys_cond.trace("sys_cond_signal_all(cond_id=0x%x)", cond_id);

	const auto cond = idm::check<lv2_obj, lv2_cond>(cond_id, [&, notify = lv2_obj::notify_all_t()](lv2_cond& cond)
	{
		if (atomic_storage<ppu_thread*>::load(cond.sq))
		{
			std::lock_guard lock(cond.mutex->mutex);

			for (auto cpu = +cond.sq; cpu; cpu = cpu->next_cpu)
			{
				if (cpu->state & cpu_flag::again)
				{
					ppu.state += cpu_flag::again;
					return;
				}
			}

			cpu_thread* result = nullptr;
			auto sq = cond.sq;
			atomic_storage<ppu_thread*>::release(cond.sq, nullptr);

			while (const auto cpu = cond.schedule<ppu_thread>(sq, SYS_SYNC_PRIORITY))
			{
				if (cond.mutex->try_own(*cpu))
				{
					ensure(!std::exchange(result, cpu));
				}
			}

			if (result)
			{
				cond.awake(result);
			}
		}
	});

	if (!cond)
	{
		return CELL_ESRCH;
	}

	return CELL_OK;
}

error_code sys_cond_signal_to(ppu_thread& ppu, u32 cond_id, u32 thread_id)
{
	ppu.state += cpu_flag::wait;

	sys_cond.trace("sys_cond_signal_to(cond_id=0x%x, thread_id=0x%x)", cond_id, thread_id);

	const auto cond = idm::check<lv2_obj, lv2_cond>(cond_id, [&, notify = lv2_obj::notify_all_t()](lv2_cond& cond) -> int
	{
		if (!idm::check_unlocked<named_thread<ppu_thread>>(thread_id))
		{
			return -1;
		}

		if (atomic_storage<ppu_thread*>::load(cond.sq))
		{
			std::lock_guard lock(cond.mutex->mutex);

			for (auto cpu = +cond.sq; cpu; cpu = cpu->next_cpu)
			{
				if (cpu->id == thread_id)
				{
					if (static_cast<ppu_thread*>(cpu)->state & cpu_flag::again)
					{
						ppu.state += cpu_flag::again;
						return 0;
					}

					ensure(cond.unqueue(cond.sq, cpu));

					if (cond.mutex->try_own(*cpu))
					{
						cond.awake(cpu);
					}

					return 1;
				}
			}
		}

		return 0;
	});

	if (!cond || cond.ret == -1)
	{
		return CELL_ESRCH;
	}

	if (!cond.ret)
	{
		return not_an_error(CELL_EPERM);
	}

	return CELL_OK;
}

error_code sys_cond_wait(ppu_thread& ppu, u32 cond_id, u64 timeout)
{
	ppu.state += cpu_flag::wait;

	sys_cond.trace("sys_cond_wait(cond_id=0x%x, timeout=%lld)", cond_id, timeout);

	// Further function result
	ppu.gpr[3] = CELL_OK;

	auto& sstate = *ppu.optional_savestate_state;

	const auto cond = idm::get<lv2_obj, lv2_cond>(cond_id, [&, notify = lv2_obj::notify_all_t()](lv2_cond& cond) -> s64
	{
		if (!ppu.loaded_from_savestate && atomic_storage<u32>::load(cond.mutex->control.raw().owner) != ppu.id)
		{
			return -1;
		}

		lv2_obj::prepare_for_sleep(ppu);

		std::lock_guard lock(cond.mutex->mutex);

		const u64 syscall_state = sstate.try_read<u64>().second;
		sstate.clear();

		if (ppu.loaded_from_savestate)
		{
			if (syscall_state & 1)
			{
				// Mutex sleep
				ensure(!cond.mutex->try_own(ppu));
			}
			else
			{
				lv2_obj::emplace(cond.sq, &ppu);
			}

			cond.sleep(ppu, timeout);
			return static_cast<u32>(syscall_state >> 32);
		}

		// Register waiter
		lv2_obj::emplace(cond.sq, &ppu);

		// Unlock the mutex
		const u32 count = cond.mutex->lock_count.exchange(0);

		if (const auto cpu = cond.mutex->reown<ppu_thread>())
		{
			if (cpu->state & cpu_flag::again)
			{
				ensure(cond.unqueue(cond.sq, &ppu));
				ppu.state += cpu_flag::again;
				return 0;
			}

			cond.mutex->append(cpu);
		}

		// Sleep current thread and schedule mutex waiter
		cond.sleep(ppu, timeout);

		// Save the recursive value
		return count;
	});

	if (!cond)
	{
		return CELL_ESRCH;
	}

	if (ppu.state & cpu_flag::again)
	{
		return {};
	}

	if (cond.ret < 0)
	{
		return CELL_EPERM;
	}

	while (auto state = +ppu.state)
	{
		if (state & cpu_flag::signal && ppu.state.test_and_reset(cpu_flag::signal))
		{
			break;
		}

		if (is_stopped(state))
		{
			std::lock_guard lock(cond->mutex->mutex);

			bool mutex_sleep = false;
			bool cond_sleep = false;

			for (auto cpu = atomic_storage<ppu_thread*>::load(cond->sq); cpu; cpu = cpu->next_cpu)
			{
				if (cpu == &ppu)
				{
					cond_sleep = true;
					break;
				}
			}

			for (auto cpu = atomic_storage<ppu_thread*>::load(cond->mutex->control.raw().sq); cpu; cpu = cpu->next_cpu)
			{
				if (cpu == &ppu)
				{
					mutex_sleep = true;
					break;
				}
			}

			if (!cond_sleep && !mutex_sleep)
			{
				break;
			}

			const u64 optional_syscall_state = u32{mutex_sleep} | (u64{static_cast<u32>(cond.ret)} << 32);
			sstate(optional_syscall_state);

			ppu.state += cpu_flag::again;
			return {};
		}

		for (usz i = 0; cpu_flag::signal - ppu.state && i < 50; i++)
		{
			busy_wait(500);
		}

		if (ppu.state & cpu_flag::signal)
 		{
			continue;
		}

		if (timeout)
		{
			if (lv2_obj::wait_timeout(timeout, &ppu))
			{
				const u64 start_time = ppu.start_time;

				// Wait for rescheduling
				if (ppu.check_state())
				{
					continue;
				}

				std::lock_guard lock(cond->mutex->mutex);

				// Try to cancel the waiting
				if (cond->unqueue(cond->sq, &ppu))
				{
					// TODO: Is EBUSY returned after reqeueing, on sys_cond_destroy?
					ppu.gpr[3] = CELL_ETIMEDOUT;

					// Own or requeue
					if (cond->mutex->try_own(ppu))
					{
						break;
					}
				}
				else if (atomic_storage<u32>::load(cond->mutex->control.raw().owner) == ppu.id)
				{
					break;
				}

				cond->mutex->sleep(ppu);
				ppu.start_time = start_time; // Restore start time because awake has been called
				timeout = 0;
				continue;
			}
		}
		else
		{
			thread_ctrl::wait_on(ppu.state, state);
		}
	}

	// Verify ownership
	ensure(atomic_storage<u32>::load(cond->mutex->control.raw().owner) == ppu.id);

	// Restore the recursive value
	cond->mutex->lock_count.release(static_cast<u32>(cond.ret));

	return not_an_error(ppu.gpr[3]);
}