1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254
|
// SPDX-FileCopyrightText: 2002-2025 PCSX2 Dev Team
// SPDX-License-Identifier: GPL-3.0+
#pragma once
#include "common/Pcsx2Defs.h"
#if defined(__APPLE__)
#include <mach/semaphore.h>
#elif !defined(_WIN32)
#include <semaphore.h>
#endif
#include <atomic>
#include <functional>
namespace Threading
{
// --------------------------------------------------------------------------------------
// Platform Specific External APIs
// --------------------------------------------------------------------------------------
// The following set of documented functions have Linux/Win32 specific implementations,
// which are found in WinThreads.cpp and LnxThreads.cpp
extern u64 GetThreadCpuTime();
extern u64 GetThreadTicksPerSecond();
/// Set the name of the current thread
extern void SetNameOfCurrentThread(const char* name);
// Releases a timeslice to other threads.
extern void Timeslice();
// For use in spin/wait loops.
extern void SpinWait();
// Optional implementation to enable hires thread/process scheduler for the operating system.
// Needed by Windows, but might not be relevant to other platforms.
extern void EnableHiresScheduler();
extern void DisableHiresScheduler();
// sleeps the current thread for the given number of milliseconds.
extern void Sleep(int ms);
// sleeps the current thread until the specified time point, or later.
extern void SleepUntil(u64 ticks);
// --------------------------------------------------------------------------------------
// ThreadHandle
// --------------------------------------------------------------------------------------
// Abstracts an OS's handle to a thread, closing the handle when necessary. Currently,
// only used for getting the CPU time for a thread.
//
class ThreadHandle
{
public:
ThreadHandle();
ThreadHandle(ThreadHandle&& handle);
ThreadHandle(const ThreadHandle& handle);
~ThreadHandle();
/// Returns a new handle for the calling thread.
static ThreadHandle GetForCallingThread();
ThreadHandle& operator=(ThreadHandle&& handle);
ThreadHandle& operator=(const ThreadHandle& handle);
operator void*() const { return m_native_handle; }
operator bool() const { return (m_native_handle != nullptr); }
/// Returns the amount of CPU time consumed by the thread, at the GetThreadTicksPerSecond() frequency.
u64 GetCPUTime() const;
/// Sets the affinity for a thread to the specified processors.
/// Obviously, only works up to 64 processors.
bool SetAffinity(u64 processor_mask) const;
protected:
void* m_native_handle = nullptr;
// We need the thread ID for affinity adjustments on Linux.
#if defined(__linux__)
unsigned int m_native_id = 0;
#endif
};
// --------------------------------------------------------------------------------------
// Thread
// --------------------------------------------------------------------------------------
// Abstracts a native thread in a lightweight manner. Provides more functionality than
// std::thread (allowing stack size adjustments).
//
class Thread : public ThreadHandle
{
public:
using EntryPoint = std::function<void()>;
Thread();
Thread(Thread&& thread);
Thread(const Thread&) = delete;
Thread(EntryPoint func);
~Thread();
ThreadHandle& operator=(Thread&& thread);
ThreadHandle& operator=(const Thread& handle) = delete;
__fi bool Joinable() const { return (m_native_handle != nullptr); }
__fi u32 GetStackSize() const { return m_stack_size; }
/// Sets the stack size for the thread. Do not call if the thread has already been started.
void SetStackSize(u32 size);
bool Start(EntryPoint func);
void Detach();
void Join();
protected:
#ifdef _WIN32
static unsigned __stdcall ThreadProc(void* param);
#else
static void* ThreadProc(void* param);
#endif
u32 m_stack_size = 0;
};
/// A semaphore that may not have a fast userspace path
/// (Used in other semaphore-based algorithms where the semaphore is just used for its thread sleep/wake ability)
class KernelSemaphore
{
#if defined(_WIN32)
void* m_sema;
#elif defined(__APPLE__)
semaphore_t m_sema;
#else
sem_t m_sema;
#endif
public:
KernelSemaphore();
~KernelSemaphore();
void Post();
void Wait();
bool TryWait();
};
/// A semaphore for notifying a work-processing thread of new work in a (separate) queue
///
/// Usage:
/// - Processing thread loops on `WaitForWork()` followed by processing all work in the queue
/// - Threads adding work first add their work to the queue, then call `NotifyOfWork()`
class WorkSema
{
/// Semaphore for sleeping the worker thread
KernelSemaphore m_sema;
/// Semaphore for sleeping thread waiting on worker queue empty
KernelSemaphore m_empty_sema;
/// Current state (see enum below)
std::atomic<s32> m_state{0};
// Expected call frequency is NotifyOfWork > WaitForWork > WaitForEmpty
// So optimize states for fast NotifyOfWork
enum
{
/* Any <-2 state: STATE_DEAD: Thread has crashed and is awaiting revival */
STATE_SPINNING = -2, ///< Worker thread is spinning waiting for work
STATE_SLEEPING = -1, ///< Worker thread is sleeping on m_sema
STATE_RUNNING_0 = 0, ///< Worker thread is processing work, but no work has been added since it last checked for new work
/* Any >0 state: STATE_RUNNING_N: Worker thread is processing work, and work has been added since it last checked for new work */
STATE_FLAG_WAITING_EMPTY = 1 << 30, ///< Flag to indicate that a thread is sleeping on m_empty_sema (can be applied to any STATE_RUNNING)
};
bool IsDead(s32 state)
{
return state < STATE_SPINNING;
}
bool IsReadyForSleep(s32 state)
{
s32 waiting_empty_cleared = state & (STATE_FLAG_WAITING_EMPTY - 1);
return waiting_empty_cleared == STATE_RUNNING_0;
}
s32 NextStateWaitForWork(s32 current)
{
s32 new_state = IsReadyForSleep(current) ? STATE_SLEEPING : STATE_RUNNING_0;
return new_state | (current & STATE_FLAG_WAITING_EMPTY); // Preserve waiting empty flag for RUNNING_N → RUNNING_0
}
public:
/// Notify the worker thread that you've added new work to its queue
void NotifyOfWork()
{
// State change:
// DEAD: Stay in DEAD (starting DEAD state is INT_MIN so we can assume we won't flip over to anything else)
// SPINNING: Change state to RUNNING. Thread will notice and process the new data
// SLEEPING: Change state to RUNNING and wake worker. Thread will wake up and process the new data.
// RUNNING_0: Change state to RUNNING_N.
// RUNNING_N: Stay in RUNNING_N
s32 old = m_state.fetch_add(2, std::memory_order_release);
if (old == STATE_SLEEPING)
m_sema.Post();
}
/// Checks if there's any work in the queue
bool CheckForWork();
/// Wait for work to be added to the queue
void WaitForWork();
/// Wait for work to be added to the queue, spinning for a bit before sleeping the thread
void WaitForWorkWithSpin();
/// Wait for the worker thread to finish processing all entries in the queue or die
/// Returns false if the thread is dead
bool WaitForEmpty();
/// Wait for the worker thread to finish processing all entries in the queue or die, spinning a bit before sleeping the thread
/// Returns false if the thread is dead
bool WaitForEmptyWithSpin();
/// Called by the worker thread to notify others of its death
/// Dead threads don't process work, and WaitForEmpty will return instantly even though there may be work in the queue
void Kill();
/// Reset the semaphore to the initial state
/// Should be called by the worker thread if it restarts after dying
void Reset();
};
/// A semaphore that definitely has a fast userspace path
class UserspaceSemaphore
{
KernelSemaphore m_sema;
std::atomic<int32_t> m_counter{0};
public:
UserspaceSemaphore() = default;
~UserspaceSemaphore() = default;
void Post()
{
if (m_counter.fetch_add(1, std::memory_order_release) < 0)
m_sema.Post();
}
void Wait()
{
if (m_counter.fetch_sub(1, std::memory_order_acquire) <= 0)
m_sema.Wait();
}
bool TryWait()
{
int32_t counter = m_counter.load(std::memory_order_relaxed);
while (counter > 0 && !m_counter.compare_exchange_weak(counter, counter - 1, std::memory_order_acquire, std::memory_order_relaxed))
;
return counter > 0;
}
};
} // namespace Threading
|