summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/common/thread_worker.cpp14
-rw-r--r--src/common/thread_worker.h5
2 files changed, 18 insertions, 1 deletions
diff --git a/src/common/thread_worker.cpp b/src/common/thread_worker.cpp
index f4d8bb0f0..fd130dfb4 100644
--- a/src/common/thread_worker.cpp
+++ b/src/common/thread_worker.cpp
@@ -8,9 +8,17 @@
8namespace Common { 8namespace Common {
9 9
10ThreadWorker::ThreadWorker(std::size_t num_workers, const std::string& name) { 10ThreadWorker::ThreadWorker(std::size_t num_workers, const std::string& name) {
11 workers_queued.store(static_cast<u64>(num_workers), std::memory_order_release);
11 const auto lambda = [this, thread_name{std::string{name}}] { 12 const auto lambda = [this, thread_name{std::string{name}}] {
12 Common::SetCurrentThreadName(thread_name.c_str()); 13 Common::SetCurrentThreadName(thread_name.c_str());
13 14
15 // TODO(Blinkhawk): Change the design, this is very prone to data races
16 // Wait for first request
17 {
18 std::unique_lock lock{queue_mutex};
19 condition.wait(lock, [this] { return stop || !requests.empty(); });
20 }
21
14 while (!stop) { 22 while (!stop) {
15 UniqueFunction<void> task; 23 UniqueFunction<void> task;
16 { 24 {
@@ -26,7 +34,9 @@ ThreadWorker::ThreadWorker(std::size_t num_workers, const std::string& name) {
26 requests.pop(); 34 requests.pop();
27 } 35 }
28 task(); 36 task();
37 work_done++;
29 } 38 }
39 workers_stopped++;
30 wait_condition.notify_all(); 40 wait_condition.notify_all();
31 }; 41 };
32 for (size_t i = 0; i < num_workers; ++i) { 42 for (size_t i = 0; i < num_workers; ++i) {
@@ -49,13 +59,15 @@ void ThreadWorker::QueueWork(UniqueFunction<void> work) {
49 { 59 {
50 std::unique_lock lock{queue_mutex}; 60 std::unique_lock lock{queue_mutex};
51 requests.emplace(std::move(work)); 61 requests.emplace(std::move(work));
62 work_scheduled++;
52 } 63 }
53 condition.notify_one(); 64 condition.notify_one();
54} 65}
55 66
56void ThreadWorker::WaitForRequests() { 67void ThreadWorker::WaitForRequests() {
57 std::unique_lock lock{queue_mutex}; 68 std::unique_lock lock{queue_mutex};
58 wait_condition.wait(lock, [this] { return stop || requests.empty(); }); 69 wait_condition.wait(
70 lock, [this] { return workers_stopped >= workers_queued || work_done >= work_scheduled; });
59} 71}
60 72
61} // namespace Common 73} // namespace Common
diff --git a/src/common/thread_worker.h b/src/common/thread_worker.h
index 7e2b04a07..12bbf5fef 100644
--- a/src/common/thread_worker.h
+++ b/src/common/thread_worker.h
@@ -11,6 +11,7 @@
11#include <vector> 11#include <vector>
12#include <queue> 12#include <queue>
13 13
14#include "common/common_types.h"
14#include "common/unique_function.h" 15#include "common/unique_function.h"
15 16
16namespace Common { 17namespace Common {
@@ -29,6 +30,10 @@ private:
29 std::condition_variable condition; 30 std::condition_variable condition;
30 std::condition_variable wait_condition; 31 std::condition_variable wait_condition;
31 std::atomic_bool stop{}; 32 std::atomic_bool stop{};
33 std::atomic<u64> work_scheduled{};
34 std::atomic<u64> work_done{};
35 std::atomic<u64> workers_stopped{};
36 std::atomic<u64> workers_queued{};
32}; 37};
33 38
34} // namespace Common 39} // namespace Common