summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--src/common/CMakeLists.txt1
-rw-r--r--src/common/thread_worker.cpp66
-rw-r--r--src/common/thread_worker.h97
3 files changed, 86 insertions, 78 deletions
diff --git a/src/common/CMakeLists.txt b/src/common/CMakeLists.txt
index c05b78cd5..e03fffd8d 100644
--- a/src/common/CMakeLists.txt
+++ b/src/common/CMakeLists.txt
@@ -180,7 +180,6 @@ add_library(common STATIC
180 thread.cpp 180 thread.cpp
181 thread.h 181 thread.h
182 thread_queue_list.h 182 thread_queue_list.h
183 thread_worker.cpp
184 thread_worker.h 183 thread_worker.h
185 threadsafe_queue.h 184 threadsafe_queue.h
186 time_zone.cpp 185 time_zone.cpp
diff --git a/src/common/thread_worker.cpp b/src/common/thread_worker.cpp
deleted file mode 100644
index 32be49b15..000000000
--- a/src/common/thread_worker.cpp
+++ /dev/null
@@ -1,66 +0,0 @@
1// Copyright 2020 yuzu emulator team
2// Licensed under GPLv2 or any later version
3// Refer to the license.txt file included.
4
5#include "common/thread.h"
6#include "common/thread_worker.h"
7
8namespace Common {
9
10ThreadWorker::ThreadWorker(std::size_t num_workers, const std::string& name) {
11 workers_queued.store(static_cast<u64>(num_workers), std::memory_order_release);
12 const auto lambda = [this, thread_name{std::string{name}}] {
13 Common::SetCurrentThreadName(thread_name.c_str());
14
15 while (!stop) {
16 UniqueFunction<void> task;
17 {
18 std::unique_lock lock{queue_mutex};
19 if (requests.empty()) {
20 wait_condition.notify_all();
21 }
22 condition.wait(lock, [this] { return stop || !requests.empty(); });
23 if (stop) {
24 break;
25 }
26 task = std::move(requests.front());
27 requests.pop();
28 }
29 task();
30 work_done++;
31 }
32 workers_stopped++;
33 wait_condition.notify_all();
34 };
35 for (size_t i = 0; i < num_workers; ++i) {
36 threads.emplace_back(lambda);
37 }
38}
39
40ThreadWorker::~ThreadWorker() {
41 {
42 std::unique_lock lock{queue_mutex};
43 stop = true;
44 }
45 condition.notify_all();
46 for (std::thread& thread : threads) {
47 thread.join();
48 }
49}
50
51void ThreadWorker::QueueWork(UniqueFunction<void> work) {
52 {
53 std::unique_lock lock{queue_mutex};
54 requests.emplace(std::move(work));
55 work_scheduled++;
56 }
57 condition.notify_one();
58}
59
60void ThreadWorker::WaitForRequests() {
61 std::unique_lock lock{queue_mutex};
62 wait_condition.wait(
63 lock, [this] { return workers_stopped >= workers_queued || work_done >= work_scheduled; });
64}
65
66} // namespace Common
diff --git a/src/common/thread_worker.h b/src/common/thread_worker.h
index 12bbf5fef..16aa673bd 100644
--- a/src/common/thread_worker.h
+++ b/src/common/thread_worker.h
@@ -8,32 +8,107 @@
8#include <functional> 8#include <functional>
9#include <mutex> 9#include <mutex>
10#include <string> 10#include <string>
11#include <type_traits>
11#include <vector> 12#include <vector>
12#include <queue> 13#include <queue>
13 14
14#include "common/common_types.h" 15#include "common/thread.h"
15#include "common/unique_function.h" 16#include "common/unique_function.h"
16 17
17namespace Common { 18namespace Common {
18 19
19class ThreadWorker final { 20template <class StateType = void>
21class StatefulThreadWorker {
22 static constexpr bool with_state = !std::is_same_v<StateType, void>;
23
24 struct DummyCallable {
25 int operator()() const noexcept {
26 return 0;
27 }
28 };
29
30 using Task =
31 std::conditional_t<with_state, UniqueFunction<void, StateType*>, UniqueFunction<void>>;
32 using StateMaker = std::conditional_t<with_state, std::function<StateType()>, DummyCallable>;
33
20public: 34public:
21 explicit ThreadWorker(std::size_t num_workers, const std::string& name); 35 explicit StatefulThreadWorker(size_t num_workers, std::string name, StateMaker func = {})
22 ~ThreadWorker(); 36 : workers_queued{num_workers}, thread_name{std::move(name)} {
23 void QueueWork(UniqueFunction<void> work); 37 const auto lambda = [this, func] {
24 void WaitForRequests(); 38 Common::SetCurrentThreadName(thread_name.c_str());
39 {
40 std::conditional_t<with_state, StateType, int> state{func()};
41 while (!stop) {
42 Task task;
43 {
44 std::unique_lock lock{queue_mutex};
45 if (requests.empty()) {
46 wait_condition.notify_all();
47 }
48 condition.wait(lock, [this] { return stop || !requests.empty(); });
49 if (stop) {
50 break;
51 }
52 task = std::move(requests.front());
53 requests.pop();
54 }
55 if constexpr (with_state) {
56 task(&state);
57 } else {
58 task();
59 }
60 ++work_done;
61 }
62 }
63 ++workers_stopped;
64 wait_condition.notify_all();
65 };
66 for (size_t i = 0; i < num_workers; ++i) {
67 threads.emplace_back(lambda);
68 }
69 }
70
71 ~StatefulThreadWorker() {
72 {
73 std::unique_lock lock{queue_mutex};
74 stop = true;
75 }
76 condition.notify_all();
77 for (std::thread& thread : threads) {
78 thread.join();
79 }
80 }
81
82 void QueueWork(Task work) {
83 {
84 std::unique_lock lock{queue_mutex};
85 requests.emplace(std::move(work));
86 ++work_scheduled;
87 }
88 condition.notify_one();
89 }
90
91 void WaitForRequests() {
92 std::unique_lock lock{queue_mutex};
93 wait_condition.wait(lock, [this] {
94 return workers_stopped >= workers_queued || work_done >= work_scheduled;
95 });
96 }
25 97
26private: 98private:
27 std::vector<std::thread> threads; 99 std::vector<std::thread> threads;
28 std::queue<UniqueFunction<void>> requests; 100 std::queue<Task> requests;
29 std::mutex queue_mutex; 101 std::mutex queue_mutex;
30 std::condition_variable condition; 102 std::condition_variable condition;
31 std::condition_variable wait_condition; 103 std::condition_variable wait_condition;
32 std::atomic_bool stop{}; 104 std::atomic_bool stop{};
33 std::atomic<u64> work_scheduled{}; 105 std::atomic<size_t> work_scheduled{};
34 std::atomic<u64> work_done{}; 106 std::atomic<size_t> work_done{};
35 std::atomic<u64> workers_stopped{}; 107 std::atomic<size_t> workers_stopped{};
36 std::atomic<u64> workers_queued{}; 108 std::atomic<size_t> workers_queued{};
109 std::string thread_name;
37}; 110};
38 111
112using ThreadWorker = StatefulThreadWorker<>;
113
39} // namespace Common 114} // namespace Common