Threading: Add lightweight thread wrapper

This commit is contained in:
Connor McLaughlin 2022-05-07 21:07:22 +10:00 committed by refractionpcsx2
parent 433b88c0bf
commit 457ec7f6f5
4 changed files with 359 additions and 2 deletions

View File

@ -16,6 +16,7 @@
#if defined(__APPLE__)
#include <sched.h>
#include <pthread.h>
#include <unistd.h>
#include <mach/mach_init.h>
#include <mach/thread_act.h>
@ -23,6 +24,7 @@
#include "common/PrecompiledHeader.h"
#include "common/Threading.h"
#include "common/Assertions.h"
// Note: assuming multicore is safer because it forces the interlocked routines to use
// the LOCK prefix. The prefix works on single core CPUs fine (but is slow), but not
@ -146,6 +148,86 @@ bool Threading::ThreadHandle::SetAffinity(u64 processor_mask) const
return false;
}
Threading::Thread::Thread() = default;
Threading::Thread::Thread(Thread&& thread)
: ThreadHandle(thread)
, m_stack_size(thread.m_stack_size)
{
thread.m_stack_size = 0;
}
Threading::Thread::Thread(EntryPoint func)
: ThreadHandle()
{
if (!Start(std::move(func)))
pxFailRel("Failed to start implicitly started thread.");
}
Threading::Thread::~Thread()
{
pxAssertRel(!m_native_handle, "Thread should be detached or joined at destruction");
}
void Threading::Thread::SetStackSize(u32 size)
{
pxAssertRel(!m_native_handle, "Can't change the stack size on a started thread");
m_stack_size = size;
}
void* Threading::Thread::ThreadProc(void* param)
{
std::unique_ptr<EntryPoint> entry(static_cast<EntryPoint*>(param));
(*entry.get())();
return nullptr;
}
bool Threading::Thread::Start(EntryPoint func)
{
pxAssertRel(!m_native_handle, "Can't start an already-started thread");
std::unique_ptr<EntryPoint> func_clone(std::make_unique<EntryPoint>(std::move(func)));
pthread_attr_t attrs;
bool has_attributes = false;
if (m_stack_size != 0)
{
has_attributes = true;
pthread_attr_init(&attrs);
}
if (m_stack_size != 0)
pthread_attr_setstacksize(&attrs, m_stack_size);
pthread_t handle;
const int res = pthread_create(&handle, has_attributes ? &attrs : nullptr, ThreadProc, func_clone.get());
if (res != 0)
return false;
// thread started, it'll release the memory
m_native_handle = (void*)handle;
func_clone.release();
return true;
}
void Threading::Thread::Detach()
{
pxAssertRel(m_native_handle, "Can't detach without a thread");
pthread_detach((pthread_t)m_native_handle);
m_native_handle = nullptr;
}
void Threading::Thread::Join()
{
pxAssertRel(m_native_handle, "Can't join without a thread");
void* retval;
const int res = pthread_join((pthread_t)m_native_handle, &retval);
if (res != 0)
pxFailRel("pthread_join() for thread join failed");
m_native_handle = nullptr;
}
// name can be up to 16 bytes
void Threading::SetNameOfCurrentThread(const char* name)
{

View File

@ -18,6 +18,9 @@
#define _GNU_SOURCE
#endif
#include <memory>
#include <pthread.h>
#include <unistd.h>
#if defined(__linux__)
#include <sys/prctl.h>
@ -35,6 +38,7 @@
#endif
#include "common/Threading.h"
#include "common/Assertions.h"
// We wont need this until we actually have this more then just stubbed out, so I'm commenting this out
// to remove an unneeded dependency.
@ -204,6 +208,159 @@ bool Threading::ThreadHandle::SetAffinity(u64 processor_mask) const
#endif
}
Threading::Thread::Thread() = default;
Threading::Thread::Thread(Thread&& thread)
: ThreadHandle(thread)
, m_stack_size(thread.m_stack_size)
{
thread.m_stack_size = 0;
}
Threading::Thread::Thread(EntryPoint func)
: ThreadHandle()
{
if (!Start(std::move(func)))
pxFailRel("Failed to start implicitly started thread.");
}
Threading::Thread::~Thread()
{
pxAssertRel(!m_native_handle, "Thread should be detached or joined at destruction");
}
void Threading::Thread::SetStackSize(u32 size)
{
pxAssertRel(!m_native_handle, "Can't change the stack size on a started thread");
m_stack_size = size;
}
#ifdef __linux__
// For Linux, we have to do a bit of trickery here to get the thread's ID back from
// the thread itself, because it's not part of pthreads. We use a semaphore to signal
// when the thread has started, and filled in thread_id_ptr.
struct ThreadProcParameters
{
Threading::Thread::EntryPoint func;
Threading::KernelSemaphore* start_semaphore;
unsigned int* thread_id_ptr;
};
void* Threading::Thread::ThreadProc(void* param)
{
std::unique_ptr<ThreadProcParameters> entry(static_cast<ThreadProcParameters*>(param));
*entry->thread_id_ptr = gettid();
entry->start_semaphore->Post();
entry->func();
return nullptr;
}
bool Threading::Thread::Start(EntryPoint func)
{
pxAssertRel(!m_native_handle, "Can't start an already-started thread");
KernelSemaphore start_semaphore;
std::unique_ptr<ThreadProcParameters> params(std::make_unique<ThreadProcParameters>());
params->func = std::move(func);
params->start_semaphore = &start_semaphore;
params->thread_id_ptr = &m_native_id;
pthread_attr_t attrs;
bool has_attributes = false;
if (m_stack_size != 0)
{
has_attributes = true;
pthread_attr_init(&attrs);
}
if (m_stack_size != 0)
pthread_attr_setstacksize(&attrs, m_stack_size);
pthread_t handle;
const int res = pthread_create(&handle, has_attributes ? &attrs : nullptr, ThreadProc, params.get());
if (res != 0)
return false;
// wait until it sets our native id
start_semaphore.Wait();
// thread started, it'll release the memory
m_native_handle = (void*)handle;
params.release();
return true;
}
#else
void* Threading::Thread::ThreadProc(void* param)
{
std::unique_ptr<EntryPoint> entry(static_cast<EntryPoint*>(param));
(*entry.get())();
return nullptr;
}
bool Threading::Thread::Start(EntryPoint func)
{
pxAssertRel(!m_native_handle, "Can't start an already-started thread");
std::unique_ptr<EntryPoint> func_clone(std::make_unique<EntryPoint>(std::move(func)));
pthread_attr_t attrs;
bool has_attributes = false;
if (m_stack_size != 0)
{
has_attributes = true;
pthread_attr_init(&attrs);
}
if (m_stack_size != 0)
pthread_attr_setstacksize(&attrs, m_stack_size);
pthread_t handle;
const int res = pthread_create(&handle, has_attributes ? &attrs : nullptr, ThreadProc, func_clone.get());
if (res != 0)
return false;
// thread started, it'll release the memory
m_native_handle = (void*)handle;
func_clone.release();
return true;
}
#endif
void Threading::Thread::Detach()
{
pxAssertRel(m_native_handle, "Can't detach without a thread");
pthread_detach((pthread_t)m_native_handle);
m_native_handle = nullptr;
#ifdef __linux__
m_native_id = 0;
#endif
}
void Threading::Thread::Join()
{
pxAssertRel(m_native_handle, "Can't join without a thread");
void* retval;
const int res = pthread_join((pthread_t)m_native_handle, &retval);
if (res != 0)
pxFailRel("pthread_join() for thread join failed");
m_native_handle = nullptr;
#ifdef __linux__
m_native_id = 0;
#endif
}
Threading::ThreadHandle& Threading::Thread::operator=(Thread&& thread)
{
ThreadHandle::operator=(thread);
m_stack_size = thread.m_stack_size;
thread.m_stack_size = 0;
return *this;
}
void Threading::SetNameOfCurrentThread(const char* name)
{
#if defined(__linux__)

View File

@ -25,6 +25,7 @@
#endif
#include <atomic>
#include <functional>
namespace Threading
{
@ -84,7 +85,7 @@ namespace Threading
/// Obviously, only works up to 64 processors.
bool SetAffinity(u64 processor_mask) const;
private:
protected:
void* m_native_handle = nullptr;
// We need the thread ID for affinity adjustments on Linux.
@ -93,6 +94,46 @@ namespace Threading
#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

View File

@ -15,9 +15,11 @@
#if defined(_WIN32)
#include "common/RedtapeWindows.h"
#include "common/Threading.h"
#include "common/Assertions.h"
#include "common/emitter/tools.h"
#include "common/RedtapeWindows.h"
#include <process.h>
__fi void Threading::Sleep(int ms)
{
@ -130,6 +132,81 @@ bool Threading::ThreadHandle::SetAffinity(u64 processor_mask) const
return (SetThreadAffinityMask(GetCurrentThread(), (DWORD_PTR)processor_mask) != 0 || GetLastError() != ERROR_SUCCESS);
}
Threading::Thread::Thread() = default;
Threading::Thread::Thread(Thread&& thread)
: ThreadHandle(thread)
, m_stack_size(thread.m_stack_size)
{
thread.m_stack_size = 0;
}
Threading::Thread::Thread(EntryPoint func)
: ThreadHandle()
{
if (!Start(std::move(func)))
pxFailRel("Failed to start implicitly started thread.");
}
Threading::Thread::~Thread()
{
pxAssertRel(!m_native_handle, "Thread should be detached or joined at destruction");
}
void Threading::Thread::SetStackSize(u32 size)
{
pxAssertRel(!m_native_handle, "Can't change the stack size on a started thread");
m_stack_size = size;
}
unsigned Threading::Thread::ThreadProc(void* param)
{
std::unique_ptr<EntryPoint> entry(static_cast<EntryPoint*>(param));
(*entry.get())();
return 0;
}
bool Threading::Thread::Start(EntryPoint func)
{
pxAssertRel(!m_native_handle, "Can't start an already-started thread");
std::unique_ptr<EntryPoint> func_clone(std::make_unique<EntryPoint>(std::move(func)));
unsigned thread_id;
m_native_handle = reinterpret_cast<void*>(_beginthreadex(nullptr, m_stack_size, ThreadProc, func_clone.get(), 0, &thread_id));
if (!m_native_handle)
return false;
// thread started, it'll release the memory
func_clone.release();
return true;
}
void Threading::Thread::Detach()
{
pxAssertRel(m_native_handle, "Can't detach without a thread");
CloseHandle((HANDLE)m_native_handle);
m_native_handle = nullptr;
}
void Threading::Thread::Join()
{
pxAssertRel(m_native_handle, "Can't join without a thread");
const DWORD res = WaitForSingleObject((HANDLE)m_native_handle, INFINITE);
if (res != WAIT_OBJECT_0)
pxFailRel("WaitForSingleObject() for thread join failed");
CloseHandle((HANDLE)m_native_handle);
m_native_handle = nullptr;
}
Threading::ThreadHandle& Threading::Thread::operator=(Thread&& thread)
{
ThreadHandle::operator=(thread);
m_stack_size = thread.m_stack_size;
thread.m_stack_size = 0;
return *this;
}
u64 Threading::GetThreadCpuTime()
{
u64 ret = 0;