RPCS3/rpcs3
1
0
Fork 0
mirror of https://github.com/RPCS3/rpcs3.git synced 2026-04-04 14:08:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

New named_thread template (preview)

Browse source 
Old class named_thread renamed to old_thread
It's too hard to move in a single commit
This commit is contained in:
Nekotekina 2018-09-25 15:21:04 +03:00
parent 4ef384a161
commit 306f95a9ae
22 changed files with 392 additions and 204 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Utilities/GDBDebugServer.cpp
View file

@ -829,7 +829,7 @@ void GDBDebugServer::on_stop()
this->stop = true;
//just in case we are waiting for breakpoint
this->notify();
named_thread::on_stop();
old_thread::on_stop();
}
void GDBDebugServer::pause_from(cpu_thread* t) {

4
Utilities/GDBDebugServer.h
View file

@ -40,8 +40,8 @@ public:
const u64 ALL_THREADS = 0xffffffffffffffff;
const u64 ANY_THREAD = 0;
class GDBDebugServer : public named_thread {
class GDBDebugServer : public old_thread
{
socket_t server_socket;
socket_t client_socket;
std::weak_ptr<cpu_thread> selected_thread;

137
Utilities/Thread.cpp
View file

@ -35,10 +35,12 @@
#endif
#include "sync.h"
#include "Log.h"
thread_local u64 g_tls_fault_all = 0;
thread_local u64 g_tls_fault_rsx = 0;
thread_local u64 g_tls_fault_spu = 0;
extern thread_local std::string(*g_tls_log_prefix)();
[[noreturn]] void catch_all_exceptions()
{
@ -48,11 +50,11 @@ thread_local u64 g_tls_fault_spu = 0;
}
catch (const std::exception& e)
{
report_fatal_error("Unhandled exception of type '"s + typeid(e).name() + "': "s + e.what());
report_fatal_error("{" + g_tls_log_prefix() + "} Unhandled exception of type '"s + typeid(e).name() + "': "s + e.what());
}
catch (...)
{
report_fatal_error("Unhandled exception (unknown)");
report_fatal_error("{" + g_tls_log_prefix() + "} Unhandled exception (unknown)");
}
}
@ -1567,25 +1569,21 @@ extern atomic_t<u32> g_thread_count(0);
thread_local DECLARE(thread_ctrl::g_tls_this_thread) = nullptr;
extern thread_local std::string(*g_tls_log_prefix)();
DECLARE(thread_ctrl::g_native_core_layout) { native_core_arrangement::undefined };
void thread_ctrl::start(const std::shared_ptr<thread_ctrl>& ctrl, task_stack task)
void thread_base::start(const std::shared_ptr<thread_base>& ctrl, task_stack task)
{
#ifdef _WIN32
using thread_result = uint;
using thread_type = thread_result(__stdcall*)(void* arg);
#else
using thread_result = void*;
using thread_type = thread_result(*)(void* arg);
#endif
// Thread entry point
const thread_type entry = [](void* arg) -> thread_result
const native_entry entry = [](void* arg) -> thread_result
{
// Recover shared_ptr from short-circuited thread_ctrl object pointer
const std::shared_ptr<thread_ctrl> ctrl = static_cast<thread_ctrl*>(arg)->m_self;
// Recover shared_ptr from short-circuited thread_base object pointer
std::shared_ptr<thread_base> ctrl = static_cast<thread_base*>(arg)->m_self;
try
{
@ -1596,17 +1594,18 @@ void thread_ctrl::start(const std::shared_ptr<thread_ctrl>& ctrl, task_stack tas
{
// Capture exception
ctrl->finalize(std::current_exception());
finalize();
return 0;
}
ctrl->finalize(nullptr);
finalize();
return 0;
};
ctrl->m_self = ctrl;
ctrl->m_task = std::move(task);
// TODO: implement simple thread pool
#ifdef _WIN32
std::uintptr_t thread = _beginthreadex(nullptr, 0, entry, ctrl.get(), 0, nullptr);
verify("thread_ctrl::start" HERE), thread != 0;
@ -1619,14 +1618,24 @@ void thread_ctrl::start(const std::shared_ptr<thread_ctrl>& ctrl, task_stack tas
ctrl->m_thread = (uintptr_t)thread;
}
void thread_ctrl::initialize()
void thread_base::start(native_entry entry)
{
#ifdef _WIN32
m_thread = ::_beginthreadex(nullptr, 0, entry, this, CREATE_SUSPENDED, nullptr);
verify("thread_ctrl::start" HERE), m_thread, ::ResumeThread(reinterpret_cast<HANDLE>(+m_thread)) != -1;
#else
verify("thread_ctrl::start" HERE), pthread_create(reinterpret_cast<pthread_t*>(&m_thread.raw()), nullptr, entry, this) == 0;
#endif
}
void thread_base::initialize()
{
// Initialize TLS variable
g_tls_this_thread = this;
thread_ctrl::g_tls_this_thread = this;
g_tls_log_prefix = []
{
return g_tls_this_thread->m_name;
return thread_ctrl::g_tls_this_thread->m_name.get();
};
++g_thread_count;
@ -1645,7 +1654,7 @@ void thread_ctrl::initialize()
{
THREADNAME_INFO info;
info.dwType = 0x1000;
info.szName = m_name.c_str();
info.szName = m_name.get().c_str();
info.dwThreadID = -1;
info.dwFlags = 0;
@ -1660,17 +1669,17 @@ void thread_ctrl::initialize()
#endif
#if defined(__APPLE__)
pthread_setname_np(m_name.substr(0, 15).c_str());
pthread_setname_np(m_name.get().substr(0, 15).c_str());
#elif defined(__DragonFly__) || defined(__FreeBSD__) || defined(__OpenBSD__)
pthread_set_name_np(pthread_self(), m_name.c_str());
pthread_set_name_np(pthread_self(), m_name.get().c_str());
#elif defined(__NetBSD__)
pthread_setname_np(pthread_self(), "%s", (void*)m_name.c_str());
pthread_setname_np(pthread_self(), "%s", (void*)m_name.get().c_str());
#elif !defined(_WIN32)
pthread_setname_np(pthread_self(), m_name.substr(0, 15).c_str());
pthread_setname_np(pthread_self(), m_name.get().substr(0, 15).c_str());
#endif
}
void thread_ctrl::finalize(std::exception_ptr eptr) noexcept
std::shared_ptr<thread_base> thread_base::finalize(std::exception_ptr eptr) noexcept
{
// Report pending errors
error_code::error_report(0, 0, 0, 0);
@ -1693,7 +1702,7 @@ void thread_ctrl::finalize(std::exception_ptr eptr) noexcept
g_tls_log_prefix = []
{
return g_tls_this_thread->m_name;
return thread_ctrl::g_tls_this_thread->m_name.get();
};
LOG_NOTICE(GENERAL, "Thread time: %fs (%fGc); Faults: %u [rsx:%u, spu:%u];",
@ -1703,13 +1712,24 @@ void thread_ctrl::finalize(std::exception_ptr eptr) noexcept
g_tls_fault_rsx,
g_tls_fault_spu);
--g_thread_count;
// Untangle circular reference, set exception
std::lock_guard{m_mutex}, m_self.reset(), m_exception = eptr;
std::unique_lock lock(m_mutex);
// Signal joining waiters
m_jcv.notify_all();
// Possibly last reference to the thread object
std::shared_ptr<thread_base> self = std::move(m_self);
m_state = thread_state::finished;
m_exception = eptr;
// Signal waiting threads
lock.unlock(), m_jcv.notify_all();
return self;
}
void thread_base::finalize() noexcept
{
g_tls_log_prefix = []() -> std::string { return {}; };
thread_ctrl::g_tls_this_thread = nullptr;
--g_thread_count;
}
bool thread_ctrl::_wait_for(u64 usec)
@ -1718,7 +1738,7 @@ bool thread_ctrl::_wait_for(u64 usec)
struct half_lock
{
semaphore<>& ref;
shared_mutex& ref;
void lock()
{
@ -1777,7 +1797,7 @@ bool thread_ctrl::_wait_for(u64 usec)
return false;
}
[[noreturn]] void thread_ctrl::_throw()
[[noreturn]] void thread_base::_throw()
{
std::exception_ptr ex = std::exchange(m_exception, std::exception_ptr{});
m_signal &= ~3;
@ -1785,10 +1805,10 @@ bool thread_ctrl::_wait_for(u64 usec)
std::rethrow_exception(std::move(ex));
}
void thread_ctrl::_notify(cond_variable thread_ctrl::* ptr)
void thread_base::_notify(cond_variable thread_base::* ptr)
{
// Optimized lock + unlock
if (!m_mutex.get())
if (!m_mutex.is_free())
{
m_mutex.lock();
m_mutex.unlock();
@ -1797,12 +1817,12 @@ void thread_ctrl::_notify(cond_variable thread_ctrl::* ptr)
(this->*ptr).notify_one();
}
thread_ctrl::thread_ctrl(std::string&& name)
: m_name(std::move(name))
thread_base::thread_base(std::string_view name)
: m_name(name)
{
}
thread_ctrl::~thread_ctrl()
thread_base::~thread_base()
{
if (m_thread)
{
@ -1814,13 +1834,13 @@ thread_ctrl::~thread_ctrl()
}
}
std::exception_ptr thread_ctrl::get_exception() const
std::exception_ptr thread_base::get_exception() const
{
std::lock_guard lock(m_mutex);
return m_exception;
}
void thread_ctrl::set_exception(std::exception_ptr ptr)
void thread_base::set_exception(std::exception_ptr ptr)
{
std::lock_guard lock(m_mutex);
m_exception = ptr;
@ -1836,35 +1856,52 @@ void thread_ctrl::set_exception(std::exception_ptr ptr)
}
}
void thread_ctrl::join()
void thread_base::join() const
{
#ifdef _WIN32
//verify("thread_ctrl::join" HERE), WaitForSingleObjectEx((HANDLE)m_thread.load(), -1, false) == WAIT_OBJECT_0;
#endif
if (m_state == thread_state::finished)
{
return;
}
std::unique_lock lock(m_mutex);
while (m_self)
while (m_state != thread_state::finished)
{
m_jcv.wait(lock);
}
}
if (UNLIKELY(m_exception && !std::uncaught_exceptions()))
void thread_base::detach()
{
auto self = weak_from_this().lock();
if (!self)
{
std::rethrow_exception(m_exception);
LOG_FATAL(GENERAL, "Cannot detach thread '%s'", get_name());
return;
}
if (self->m_state.compare_and_swap_test(thread_state::created, thread_state::detached))
{
std::lock_guard lock(m_mutex);
if (m_state == thread_state::detached)
{
m_self = std::move(self);
}
}
}
void thread_ctrl::notify()
void thread_base::notify()
{
if (!(m_signal & 1))
{
m_signal |= 1;
_notify(&thread_ctrl::m_cond);
_notify(&thread_base::m_cond);
}
}
u64 thread_ctrl::get_cycles()
u64 thread_base::get_cycles()
{
u64 cycles;
@ -2059,23 +2096,23 @@ void thread_ctrl::set_thread_affinity_mask(u16 mask)
#endif
}
named_thread::named_thread()
old_thread::old_thread()
{
}
named_thread::~named_thread()
old_thread::~old_thread()
{
}
std::string named_thread::get_name() const
std::string old_thread::get_name() const
{
return fmt::format("('%s') Unnamed Thread", typeid(*this).name());
}
void named_thread::start_thread(const std::shared_ptr<void>& _this)
void old_thread::start_thread(const std::shared_ptr<void>& _this)
{
// Ensure it's not called from the constructor and the correct object is passed
verify("named_thread::start_thread" HERE), _this.get() == this;
verify("old_thread::start_thread" HERE), _this.get() == this;
// Run thread
thread_ctrl::spawn(m_thread, get_name(), [this, _this]()

327
Utilities/Thread.h
View file

@ -6,9 +6,11 @@
#include <exception>
#include <string>
#include <memory>
#include <string_view>
#include "sema.h"
#include "mutex.h"
#include "cond.h"
#include "lockless.h"
// Report error and call std::abort(), defined in main.cpp
[[noreturn]] void report_fatal_error(const std::string&);
@ -33,6 +35,55 @@ enum class thread_class : u32
ppu
};
enum class thread_state
{
created, // Initial state
detached, // Set if the thread has been detached successfully (only possible via shared_ptr)
aborting, // Set if the thread has been joined in destructor (mutually exclusive with detached)
finished // Final state, always set at the end of thread execution
};
template <class Context>
class named_thread;
template <typename T>
struct result_storage
{
alignas(T) std::byte data[sizeof(T)];
static constexpr bool empty = false;
using type = T;
T* get()
{
return reinterpret_cast<T*>(&data);
}
const T* get() const
{
return reinterpret_cast<const T*>(&data);
}
};
template <>
struct result_storage<void>
{
static constexpr bool empty = true;
using type = void;
};
template <class Context, typename... Args>
using result_storage_t = result_storage<std::invoke_result_t<Context, Args...>>;
// Detect on_stop() method (should return void)
template <typename T, typename = void>
struct thread_on_stop : std::bool_constant<false> {};
template <typename T>
struct thread_on_stop<T, decltype(std::declval<named_thread<T>&>().on_stop())> : std::bool_constant<true> {};
// Simple list of void() functors
class task_stack
{
@ -102,23 +153,24 @@ public:
}
};
// Thread control class
class thread_ctrl final
// Thread base class
class thread_base : public std::enable_shared_from_this<thread_base>
{
// Current thread
static thread_local thread_ctrl* g_tls_this_thread;
// Native thread entry point function type
#ifdef _WIN32
using native_entry = uint(__stdcall*)(void* arg);
#else
using native_entry = void*(*)(void* arg);
#endif
// Target cpu core layout
static atomic_t<native_core_arrangement> g_native_core_layout;
// Self pointer
std::shared_ptr<thread_ctrl> m_self;
// Self pointer for detached thread
std::shared_ptr<thread_base> m_self;
// Thread handle (platform-specific)
atomic_t<std::uintptr_t> m_thread{0};
// Thread mutex
mutable semaphore<> m_mutex;
mutable shared_mutex m_mutex;
// Thread condition variable
cond_variable m_cond;
@ -127,7 +179,10 @@ class thread_ctrl final
atomic_t<u32> m_signal{0};
// Thread joining condition variable
cond_variable m_jcv;
mutable cond_variable m_jcv;
// Thread state
atomic_t<thread_state> m_state = thread_state::created;
// Remotely set or caught exception
std::exception_ptr m_exception;
@ -135,38 +190,40 @@ class thread_ctrl final
// Thread initial task
task_stack m_task;
// Fixed name
std::string m_name;
// Thread name
lf_value<std::string> m_name;
// CPU cycles thread has run for
u64 m_cycles{0};
// Start thread
static void start(const std::shared_ptr<thread_ctrl>&, task_stack);
static void start(const std::shared_ptr<thread_base>&, task_stack);
void start(native_entry);
// Called at the thread start
void initialize();
// Called at the thread end
void finalize(std::exception_ptr) noexcept;
// Called at the thread end, returns moved m_self (may be null)
std::shared_ptr<thread_base> finalize(std::exception_ptr) noexcept;
// Internal waiting function, may throw. Infinite value is -1.
static bool _wait_for(u64 usec);
static void finalize() noexcept;
// Internal throwing function. Mutex must be locked and will be unlocked.
[[noreturn]] void _throw();
// Internal notification function
void _notify(cond_variable thread_ctrl::*);
void _notify(cond_variable thread_base::*);
friend class thread_ctrl;
template <class Context>
friend class named_thread;
public:
thread_ctrl(std::string&& name);
thread_base(std::string_view name);
thread_ctrl(const thread_ctrl&) = delete;
thread_ctrl& operator=(const thread_ctrl&) = delete;
~thread_ctrl();
~thread_base();
// Get thread name
const std::string& get_name() const
@ -174,6 +231,12 @@ public:
return m_name;
}
// Set thread name (not recommended)
void set_name(std::string_view name)
{
m_name.assign(name);
}
// Get CPU cycles since last time this function was called. First call returns 0.
u64 get_cycles();
@ -189,11 +252,36 @@ public:
// Set exception
void set_exception(std::exception_ptr ptr);
// Get thread result (may throw, simultaneous joining allowed)
void join();
// Wait for the thread (it does NOT change thread state, and can be called from multiple threads)
void join() const;
// Make thread to manage a shared_ptr of itself
void detach();
// Notify the thread
void notify();
};
// Collection of global function for current thread
class thread_ctrl final
{
// Current thread
static thread_local thread_base* g_tls_this_thread;
// Target cpu core layout
static atomic_t<native_core_arrangement> g_native_core_layout;
// Internal waiting function, may throw. Infinite value is -1.
static bool _wait_for(u64 usec);
friend class thread_base;
public:
// Read current state
static inline thread_state state()
{
return g_tls_this_thread->m_state;
}
// Wait once with timeout. Abortable, may throw. May spuriously return false.
static inline bool wait_for(u64 usec)
@ -208,7 +296,7 @@ public:
}
// Wait until pred(). Abortable, may throw.
template<typename F, typename RT = std::invoke_result_t<F>>
template <typename F, typename RT = std::invoke_result_t<F>>
static inline RT wait(F&& pred)
{
while (true)
@ -235,27 +323,27 @@ public:
static void test();
// Get current thread (may be nullptr)
static thread_ctrl* get_current()
static thread_base* get_current()
{
return g_tls_this_thread;
}
// Create detached named thread
template<typename N, typename F>
template <typename N, typename F>
static inline void spawn(N&& name, F&& func)
{
auto out = std::make_shared<thread_ctrl>(std::forward<N>(name));
auto out = std::make_shared<thread_base>(std::forward<N>(name));
thread_ctrl::start(out, std::forward<F>(func));
thread_base::start(out, std::forward<F>(func));
}
// Named thread factory
template<typename N, typename F>
static inline void spawn(std::shared_ptr<thread_ctrl>& out, N&& name, F&& func)
template <typename N, typename F>
static inline void spawn(std::shared_ptr<thread_base>& out, N&& name, F&& func)
{
out = std::make_shared<thread_ctrl>(std::forward<N>(name));
out = std::make_shared<thread_base>(std::forward<N>(name));
thread_ctrl::start(out, std::forward<F>(func));
thread_base::start(out, std::forward<F>(func));
}
// Detect layout
@ -269,21 +357,140 @@ public:
// Sets the preferred affinity mask for this thread
static void set_thread_affinity_mask(u16 mask);
template <typename F>
static inline std::shared_ptr<named_thread<F>> make_shared(std::string_view name, F&& lambda)
{
return std::make_shared<named_thread<F>>(name, std::forward<F>(lambda));
}
template <typename T, typename... Args>
static inline std::shared_ptr<named_thread<T>> make_shared(std::string_view name, Args&&... args)
{
return std::make_shared<named_thread<T>>(name, std::forward<Args>(args)...);
}
};
class named_thread
// Derived from the callable object Context, possibly a lambda
template <class Context>
class named_thread final : public Context, result_storage_t<Context>, public thread_base
{
// Pointer to managed resource (shared with actual thread)
std::shared_ptr<thread_ctrl> m_thread;
using result = result_storage_t<Context>;
using thread = thread_base;
// Type-erased thread entry point
#ifdef _WIN32
static inline uint __stdcall entry_point(void* arg) try
#else
static inline void* entry_point(void* arg) try
#endif
{
const auto maybe_last_ptr = static_cast<named_thread*>(static_cast<thread*>(arg))->entry_point();
thread::finalize();
return 0;
}
catch (...)
{
catch_all_exceptions();
}
std::shared_ptr<thread> entry_point()
{
thread::initialize();
if constexpr (result::empty)
{
// No result
Context::operator()();
}
else
{
// Construct the result using placement new (copy elision should happen)
new (result::get()) typename result::type(Context::operator()());
}
return thread::finalize(nullptr);
}
public:
named_thread();
virtual ~named_thread();
// Normal forwarding constructor
template <typename... Args, typename = std::enable_if_t<std::is_constructible_v<Context, Args&&...>>>
named_thread(std::string_view name, Args&&... args)
: Context(std::forward<Args>(args)...)
, thread(name)
{
thread::start(&named_thread::entry_point);
}
named_thread(const named_thread&) = delete;
// Lambda constructor, also the implicit deduction guide candidate
named_thread(std::string_view name, Context&& f)
: Context(std::forward<Context>(f))
, thread(name)
{
thread::start(&named_thread::entry_point);
}
named_thread& operator=(const named_thread&) = delete;
// Wait for the completion and access result (if not void)
[[nodiscard]] decltype(auto) operator()()
{
thread::join();
if constexpr (!result::empty)
{
return *result::get();
}
}
// Wait for the completion and access result (if not void)
[[nodiscard]] decltype(auto) operator()() const
{
thread::join();
if constexpr (!result::empty)
{
return *result::get();
}
}
// Access thread state
operator thread_state() const
{
return thread::m_state.load();
}
// Context type doesn't need virtual destructor
~named_thread()
{
// Notify thread if not detached or terminated
if (thread::m_state.compare_and_swap_test(thread_state::created, thread_state::aborting))
{
// Additional notification if on_stop() method exists
if constexpr (thread_on_stop<Context>())
{
Context::on_stop();
}
thread::notify();
thread::join();
}
}
};
// Old named_thread
class old_thread
{
// Pointer to managed resource (shared with actual thread)
std::shared_ptr<thread_base> m_thread;
public:
old_thread();
virtual ~old_thread();
old_thread(const old_thread&) = delete;
old_thread& operator=(const old_thread&) = delete;
// Get thread name
virtual std::string get_name() const;
@ -314,8 +521,7 @@ public:
m_thread->join();
}
// Access thread_ctrl
thread_ctrl* get() const
thread_base* get() const
{
return m_thread.get();
}
@ -330,32 +536,3 @@ public:
return m_thread->notify();
}
};
// Wrapper for named thread, joins automatically in the destructor, can only be used in function scope
class scope_thread final
{
std::shared_ptr<thread_ctrl> m_thread;
public:
template<typename N, typename F>
scope_thread(N&& name, F&& func)
{
thread_ctrl::spawn(m_thread, std::forward<N>(name), std::forward<F>(func));
}
scope_thread(const scope_thread&) = delete;
scope_thread& operator=(const scope_thread&) = delete;
// Destructor with exceptions allowed
~scope_thread() noexcept(false)
{
m_thread->join();
}
// Access thread_ctrl
thread_ctrl* get() const
{
return m_thread.get();
}
};