rpcs3/Utilities/Thread.h

446 lines
8.6 KiB
C
Raw Normal View History

#pragma once
2015-06-30 22:25:52 +00:00
const class thread_ctrl_t* get_current_thread_ctrl();
2015-06-30 22:25:52 +00:00
// named thread control class
class thread_ctrl_t final
{
friend class named_thread_t;
2015-08-21 11:07:31 +00:00
template<typename T> friend void current_thread_register_atexit(T);
2015-06-30 22:25:52 +00:00
// thread handler
std::thread m_thread;
2015-06-30 22:25:52 +00:00
// name getter
2015-08-21 11:07:31 +00:00
const std::function<std::string()> m_name;
// functions executed at thread exit (temporarily)
std::vector<std::function<void()>> m_atexit;
public:
2015-06-30 22:25:52 +00:00
thread_ctrl_t(std::function<std::string()> name)
2015-08-21 11:07:31 +00:00
: m_name(std::move(name))
2015-06-30 22:25:52 +00:00
{
}
2015-06-30 22:25:52 +00:00
// get thread name
std::string get_name() const;
};
2015-08-21 11:07:31 +00:00
// register function at thread exit (temporarily)
template<typename T> void current_thread_register_atexit(T func)
{
extern thread_local thread_ctrl_t* g_tls_this_thread;
g_tls_this_thread->m_atexit.emplace_back(func);
}
class named_thread_t
{
2015-06-30 22:25:52 +00:00
// pointer to managed resource (shared with actual thread)
std::shared_ptr<thread_ctrl_t> m_thread;
2015-01-16 14:36:53 +00:00
2015-06-30 22:25:52 +00:00
public:
// thread mutex for external use
std::mutex mutex;
// thread condition variable for external use
std::condition_variable cv;
public:
2015-06-30 22:25:52 +00:00
// initialize in empty state
named_thread_t() = default;
2015-06-30 22:25:52 +00:00
// create named thread
named_thread_t(std::function<std::string()> name, std::function<void()> func);
2015-06-30 22:25:52 +00:00
// destructor, will terminate if thread is neither joined nor detached
virtual ~named_thread_t();
2015-01-16 14:36:53 +00:00
named_thread_t(const named_thread_t&) = delete;
named_thread_t& operator =(const named_thread_t&) = delete;
2014-02-19 17:27:52 +00:00
public:
2015-06-30 22:25:52 +00:00
// get thread name
std::string get_name() const;
2015-06-30 22:25:52 +00:00
// create named thread (current state must be empty)
void start(std::function<std::string()> name, std::function<void()> func);
2014-10-17 20:13:25 +00:00
2015-06-30 22:25:52 +00:00
// detach thread -> empty state
void detach();
2014-10-17 20:13:25 +00:00
2015-06-30 22:25:52 +00:00
// join thread -> empty state
void join();
2014-10-17 20:13:25 +00:00
2015-06-30 22:25:52 +00:00
// check if not empty
bool joinable() const { return m_thread.operator bool(); }
2014-10-17 20:13:25 +00:00
2015-06-30 22:25:52 +00:00
// check whether it is the current running thread
bool is_current() const;
2015-07-15 18:11:32 +00:00
// get internal thread pointer
2015-07-19 01:56:33 +00:00
const thread_ctrl_t* get_thread_ctrl() const { return m_thread.get(); }
2014-10-17 20:13:25 +00:00
};
class autojoin_thread_t final : private named_thread_t
2015-07-03 23:22:24 +00:00
{
public:
using named_thread_t::mutex;
using named_thread_t::cv;
2015-07-03 23:22:24 +00:00
public:
autojoin_thread_t() = delete;
autojoin_thread_t(std::function<std::string()> name, std::function<void()> func)
{
start(std::move(name), std::move(func));
}
virtual ~autojoin_thread_t() override
{
join();
}
using named_thread_t::is_current;
2015-07-03 23:22:24 +00:00
};
extern const std::function<bool()> SQUEUE_ALWAYS_EXIT;
extern const std::function<bool()> SQUEUE_NEVER_EXIT;
2015-01-16 17:09:53 +00:00
bool squeue_test_exit();
2014-12-25 20:30:34 +00:00
2014-12-24 22:24:17 +00:00
template<typename T, u32 sq_size = 256>
class squeue_t
{
struct squeue_sync_var_t
{
struct
{
u32 position : 31;
2014-12-25 22:58:43 +00:00
u32 pop_lock : 1;
2014-12-24 22:24:17 +00:00
};
struct
{
u32 count : 31;
2014-12-25 22:58:43 +00:00
u32 push_lock : 1;
2014-12-24 22:24:17 +00:00
};
};
2015-07-08 15:01:59 +00:00
atomic_t<squeue_sync_var_t> m_sync;
2014-12-24 22:24:17 +00:00
2014-12-28 13:15:22 +00:00
mutable std::mutex m_rcv_mutex;
mutable std::mutex m_wcv_mutex;
mutable std::condition_variable m_rcv;
mutable std::condition_variable m_wcv;
2014-12-24 22:24:17 +00:00
T m_data[sq_size];
2014-12-25 22:49:55 +00:00
enum squeue_sync_var_result : u32
{
SQSVR_OK = 0,
SQSVR_LOCKED = 1,
SQSVR_FAILED = 2,
};
2014-12-24 22:24:17 +00:00
public:
squeue_t()
2015-03-13 01:09:53 +00:00
: m_sync({})
2014-12-24 22:24:17 +00:00
{
}
u32 get_max_size() const
{
return sq_size;
}
2015-03-13 01:09:53 +00:00
bool is_full() const volatile
2014-12-24 22:24:17 +00:00
{
2015-03-13 01:09:53 +00:00
return m_sync.data.count == sq_size;
2014-12-24 22:24:17 +00:00
}
2015-01-16 17:09:53 +00:00
bool push(const T& data, const std::function<bool()>& test_exit)
2014-12-24 22:24:17 +00:00
{
u32 pos = 0;
while (u32 res = m_sync.atomic_op([&pos](squeue_sync_var_t& sync) -> u32
2014-12-24 22:24:17 +00:00
{
assert(sync.count <= sq_size);
assert(sync.position < sq_size);
2014-12-25 22:58:43 +00:00
if (sync.push_lock)
2014-12-24 22:24:17 +00:00
{
2014-12-25 22:49:55 +00:00
return SQSVR_LOCKED;
}
if (sync.count == sq_size)
{
return SQSVR_FAILED;
2014-12-24 22:24:17 +00:00
}
2014-12-25 22:58:43 +00:00
sync.push_lock = 1;
2014-12-24 22:24:17 +00:00
pos = sync.position + sync.count;
2014-12-25 22:49:55 +00:00
return SQSVR_OK;
2014-12-24 22:24:17 +00:00
}))
{
2015-01-16 17:09:53 +00:00
if (res == SQSVR_FAILED && (test_exit() || squeue_test_exit()))
2014-12-24 22:24:17 +00:00
{
return false;
}
std::unique_lock<std::mutex> wcv_lock(m_wcv_mutex);
m_wcv.wait_for(wcv_lock, std::chrono::milliseconds(1));
}
m_data[pos >= sq_size ? pos - sq_size : pos] = data;
m_sync.atomic_op([](squeue_sync_var_t& sync)
{
assert(sync.count <= sq_size);
assert(sync.position < sq_size);
2014-12-25 22:58:43 +00:00
assert(sync.push_lock);
sync.push_lock = 0;
2014-12-24 22:24:17 +00:00
sync.count++;
});
m_rcv.notify_one();
m_wcv.notify_one();
return true;
}
2015-01-16 17:09:53 +00:00
bool push(const T& data, const volatile bool* do_exit)
2014-12-24 22:24:17 +00:00
{
2015-01-16 17:09:53 +00:00
return push(data, [do_exit](){ return do_exit && *do_exit; });
}
2014-12-24 22:24:17 +00:00
force_inline bool push(const T& data)
2015-01-16 17:09:53 +00:00
{
return push(data, SQUEUE_NEVER_EXIT);
2014-12-24 22:24:17 +00:00
}
force_inline bool try_push(const T& data)
2015-01-16 17:09:53 +00:00
{
return push(data, SQUEUE_ALWAYS_EXIT);
2015-01-16 17:09:53 +00:00
}
bool pop(T& data, const std::function<bool()>& test_exit)
2014-12-24 22:24:17 +00:00
{
u32 pos = 0;
while (u32 res = m_sync.atomic_op([&pos](squeue_sync_var_t& sync) -> u32
2014-12-24 22:24:17 +00:00
{
assert(sync.count <= sq_size);
assert(sync.position < sq_size);
2014-12-25 22:49:55 +00:00
if (!sync.count)
{
return SQSVR_FAILED;
2014-12-24 22:24:17 +00:00
}
2014-12-25 22:58:43 +00:00
if (sync.pop_lock)
{
return SQSVR_LOCKED;
}
2014-12-24 22:24:17 +00:00
2014-12-25 22:58:43 +00:00
sync.pop_lock = 1;
2014-12-24 22:24:17 +00:00
pos = sync.position;
2014-12-25 22:49:55 +00:00
return SQSVR_OK;
2014-12-24 22:24:17 +00:00
}))
{
2015-01-16 17:09:53 +00:00
if (res == SQSVR_FAILED && (test_exit() || squeue_test_exit()))
2014-12-24 22:24:17 +00:00
{
return false;
}
std::unique_lock<std::mutex> rcv_lock(m_rcv_mutex);
m_rcv.wait_for(rcv_lock, std::chrono::milliseconds(1));
}
data = m_data[pos];
m_sync.atomic_op([](squeue_sync_var_t& sync)
{
assert(sync.count <= sq_size);
assert(sync.position < sq_size);
2014-12-25 22:58:43 +00:00
assert(sync.pop_lock);
sync.pop_lock = 0;
2014-12-24 22:24:17 +00:00
sync.position++;
sync.count--;
if (sync.position == sq_size)
{
sync.position = 0;
}
});
m_rcv.notify_one();
m_wcv.notify_one();
return true;
}
2015-01-16 17:09:53 +00:00
bool pop(T& data, const volatile bool* do_exit)
2014-12-24 22:24:17 +00:00
{
2015-01-16 17:09:53 +00:00
return pop(data, [do_exit](){ return do_exit && *do_exit; });
}
2014-12-24 22:24:17 +00:00
force_inline bool pop(T& data)
2015-01-16 17:09:53 +00:00
{
return pop(data, SQUEUE_NEVER_EXIT);
2014-12-24 22:24:17 +00:00
}
force_inline bool try_pop(T& data)
2015-01-16 17:09:53 +00:00
{
return pop(data, SQUEUE_ALWAYS_EXIT);
2015-01-16 17:09:53 +00:00
}
bool peek(T& data, u32 start_pos, const std::function<bool()>& test_exit)
2014-12-24 22:24:17 +00:00
{
assert(start_pos < sq_size);
u32 pos = 0;
while (u32 res = m_sync.atomic_op([&pos, start_pos](squeue_sync_var_t& sync) -> u32
2014-12-24 22:24:17 +00:00
{
assert(sync.count <= sq_size);
assert(sync.position < sq_size);
2014-12-25 22:49:55 +00:00
if (sync.count <= start_pos)
{
return SQSVR_FAILED;
2014-12-24 22:24:17 +00:00
}
2014-12-25 22:58:43 +00:00
if (sync.pop_lock)
{
return SQSVR_LOCKED;
}
2014-12-25 22:58:43 +00:00
sync.pop_lock = 1;
2014-12-24 22:24:17 +00:00
pos = sync.position + start_pos;
2014-12-25 22:49:55 +00:00
return SQSVR_OK;
2014-12-24 22:24:17 +00:00
}))
{
2015-01-16 17:09:53 +00:00
if (res == SQSVR_FAILED && (test_exit() || squeue_test_exit()))
2014-12-24 22:24:17 +00:00
{
return false;
}
std::unique_lock<std::mutex> rcv_lock(m_rcv_mutex);
m_rcv.wait_for(rcv_lock, std::chrono::milliseconds(1));
}
data = m_data[pos >= sq_size ? pos - sq_size : pos];
m_sync.atomic_op([](squeue_sync_var_t& sync)
{
assert(sync.count <= sq_size);
assert(sync.position < sq_size);
2014-12-25 22:58:43 +00:00
assert(sync.pop_lock);
sync.pop_lock = 0;
2014-12-24 22:24:17 +00:00
});
m_rcv.notify_one();
return true;
}
2015-01-16 17:09:53 +00:00
bool peek(T& data, u32 start_pos, const volatile bool* do_exit)
{
return peek(data, start_pos, [do_exit](){ return do_exit && *do_exit; });
}
force_inline bool peek(T& data, u32 start_pos = 0)
2014-12-24 22:24:17 +00:00
{
return peek(data, start_pos, SQUEUE_NEVER_EXIT);
2015-01-16 17:09:53 +00:00
}
2014-12-24 22:24:17 +00:00
force_inline bool try_peek(T& data, u32 start_pos = 0)
2015-01-16 17:09:53 +00:00
{
return peek(data, start_pos, SQUEUE_ALWAYS_EXIT);
2014-12-24 22:24:17 +00:00
}
class squeue_data_t
{
T* const m_data;
const u32 m_pos;
const u32 m_count;
squeue_data_t(T* data, u32 pos, u32 count)
: m_data(data)
, m_pos(pos)
, m_count(count)
{
}
public:
T& operator [] (u32 index)
{
assert(index < m_count);
index += m_pos;
index = index < sq_size ? index : index - sq_size;
return m_data[index];
}
};
void process(void(*proc)(squeue_data_t data))
{
u32 pos, count;
while (m_sync.atomic_op([&pos, &count](squeue_sync_var_t& sync) -> u32
{
assert(sync.count <= sq_size);
assert(sync.position < sq_size);
if (sync.pop_lock || sync.push_lock)
{
return SQSVR_LOCKED;
}
pos = sync.position;
count = sync.count;
sync.pop_lock = 1;
sync.push_lock = 1;
return SQSVR_OK;
}))
{
std::unique_lock<std::mutex> rcv_lock(m_rcv_mutex);
m_rcv.wait_for(rcv_lock, std::chrono::milliseconds(1));
}
proc(squeue_data_t(m_data, pos, count));
m_sync.atomic_op([](squeue_sync_var_t& sync)
{
assert(sync.count <= sq_size);
assert(sync.position < sq_size);
assert(sync.pop_lock && sync.push_lock);
sync.pop_lock = 0;
sync.push_lock = 0;
});
m_wcv.notify_one();
m_rcv.notify_one();
}
void clear()
{
while (m_sync.atomic_op([](squeue_sync_var_t& sync) -> u32
{
assert(sync.count <= sq_size);
assert(sync.position < sq_size);
if (sync.pop_lock || sync.push_lock)
{
return SQSVR_LOCKED;
}
sync.pop_lock = 1;
sync.push_lock = 1;
return SQSVR_OK;
}))
{
std::unique_lock<std::mutex> rcv_lock(m_rcv_mutex);
m_rcv.wait_for(rcv_lock, std::chrono::milliseconds(1));
}
m_sync.exchange({});
m_wcv.notify_one();
m_rcv.notify_one();
}
2014-12-24 22:24:17 +00:00
};