archive/rpcs3
1
0
Fork 0
mirror of https://github.com/RPCS3/rpcs3.git synced 2025-07-02 21:11:25 +12:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Rewrite condition variables

Browse source 
Implement helper functions balanced_wait_until and balanced_awaken
They include new path for Windows 8.1+ (WaitOnAddress)

shared_mutex, cond_variable, cond_one, cond_x16 modified to use it
Added helper function utils::popcnt16
Replace most semaphore<> with shared_mutex
This commit is contained in:
Nekotekina 2018-11-26 18:55:22 +03:00
parent f442a8a84c
commit 96cabeadff
23 changed files with 269 additions and 338 deletions
Download patch file Download diff file Expand all files Collapse all files

149
Utilities/sync.h
View file

@ -72,12 +72,12 @@ inline int futex(volatile void* uaddr, int futex_op, uint val, const timespec* t
switch (futex_op)
{
case FUTEX_WAIT:
case FUTEX_WAIT_PRIVATE:
{
mask = -1;
[[fallthrough]];
}
case FUTEX_WAIT_BITSET:
case FUTEX_WAIT_BITSET_PRIVATE:
{
if (*reinterpret_cast<volatile uint*>(uaddr) != val)
{
@ -115,12 +115,12 @@ inline int futex(volatile void* uaddr, int futex_op, uint val, const timespec* t
return res;
}
case FUTEX_WAKE:
case FUTEX_WAKE_PRIVATE:
{
mask = -1;
[[fallthrough]];
}
case FUTEX_WAKE_BITSET:
case FUTEX_WAKE_BITSET_PRIVATE:
{
int res = 0;
@ -149,3 +149,144 @@ inline int futex(volatile void* uaddr, int futex_op, uint val, const timespec* t
return g_futex(uaddr, futex_op, val, timeout, mask);
#endif
}
template <typename T, typename Pred>
bool balanced_wait_until(atomic_t<T>& var, u64 usec_timeout, Pred&& pred)
{
static_assert(sizeof(T) == 4);
const bool is_inf = usec_timeout > u64{UINT32_MAX / 1000} * 1000000;
// Optional second argument indicates that the predicate should try to retire
auto test_pred = [&](T& _new, auto... args)
{
T old = var.load();
while (true)
{
_new = old;
// Zero indicates failure without modifying the value
// Negative indicates failure but modifies the value
auto ret = std::invoke(std::forward<Pred>(pred), _new, args...);
if (LIKELY(!ret || var.compare_exchange(old, _new)))
{
return ret > 0;
}
}
};
T value;
#ifdef _WIN32
if (OptWaitOnAddress)
{
while (!test_pred(value))
{
if (OptWaitOnAddress(&var, &value, sizeof(u32), is_inf ? INFINITE : usec_timeout / 1000))
{
if (!test_pred(value) && !test_pred(value, nullptr))
{
return false;
}
break;
}
if (GetLastError() == ERROR_TIMEOUT)
{
// Retire
return test_pred(value, nullptr);
}
}
return true;
}
LARGE_INTEGER timeout;
timeout.QuadPart = usec_timeout * -10;
if (!usec_timeout || NtWaitForKeyedEvent(nullptr, &var, false, is_inf ? nullptr : &timeout))
{
// Timed out: retire
if (!test_pred(value, nullptr))
{
return false;
}
// Signaled in the last moment: restore balance
NtWaitForKeyedEvent(nullptr, &var, false, nullptr);
return true;
}
if (!test_pred(value) && !test_pred(value, nullptr))
{
// Stolen notification: restore balance
NtReleaseKeyedEvent(nullptr, &var, false, nullptr);
return false;
}
return true;
#else
struct timespec timeout;
timeout.tv_sec = usec_timeout / 1000000;
timeout.tv_nsec = (usec_timeout % 1000000) * 1000;
while (!test_pred(value))
{
if (futex(&var, FUTEX_WAIT_PRIVATE, static_cast<u32>(value), is_inf ? nullptr : &timeout) == 0)
{
if (!test_pred(value) && !test_pred(value, nullptr))
{
return false;
}
break;
}
switch (errno)
{
case EAGAIN: break;
case ETIMEDOUT: return test_pred(value, nullptr);
default: verify("Unknown futex error" HERE), 0;
}
}
return true;
#endif
}
template <typename T>
void balanced_awaken(atomic_t<T>& var, u32 weight)
{
static_assert(sizeof(T) == 4);
#ifdef _WIN32
if (OptWaitOnAddress)
{
if (weight > 1)
{
OptWakeByAddressAll(&var);
}
else if (weight == 1)
{
OptWakeByAddressSingle(&var);
}
return;
}
for (u32 i = 0; i < weight; i++)
{
NtReleaseKeyedEvent(nullptr, &var, false, nullptr);
}
#else
if (weight)
{
futex(&var, FUTEX_WAKE_PRIVATE, std::min<u32>(INT_MAX, weight));
}
return;
#endif
}