From f28dd32275c1feba4854abad30ff5e21a7b39440 Mon Sep 17 00:00:00 2001 From: ReinUsesLisp Date: Mon, 22 Mar 2021 21:00:48 -0300 Subject: [PATCH 1/7] common/thread_worker: Add wait for requests method --- src/common/thread_worker.cpp | 9 +++++++++ src/common/thread_worker.h | 2 ++ 2 files changed, 11 insertions(+) diff --git a/src/common/thread_worker.cpp b/src/common/thread_worker.cpp index 8f9bf447a..745918c7e 100644 --- a/src/common/thread_worker.cpp +++ b/src/common/thread_worker.cpp @@ -29,6 +29,10 @@ ThreadWorker::ThreadWorker(std::size_t num_workers, const std::string& name) { } task = std::move(requests.front()); requests.pop(); + + if (requests.empty()) { + wait_condition.notify_one(); + } } task(); @@ -55,4 +59,9 @@ void ThreadWorker::QueueWork(std::function&& work) { condition.notify_one(); } +void ThreadWorker::WaitForRequests() { + std::unique_lock lock{queue_mutex}; + wait_condition.wait(lock, [this] { return stop || requests.empty(); }); +} + } // namespace Common diff --git a/src/common/thread_worker.h b/src/common/thread_worker.h index f1859971f..7a6756eb5 100644 --- a/src/common/thread_worker.h +++ b/src/common/thread_worker.h @@ -18,12 +18,14 @@ public: explicit ThreadWorker(std::size_t num_workers, const std::string& name); ~ThreadWorker(); void QueueWork(std::function&& work); + void WaitForRequests(); private: std::vector threads; std::queue> requests; std::mutex queue_mutex; std::condition_variable condition; + std::condition_variable wait_condition; std::atomic_bool stop{}; }; From 2c8d33741889ddffc6dfaf4b2f62e61f496c6b0a Mon Sep 17 00:00:00 2001 From: ReinUsesLisp Date: Thu, 1 Apr 2021 01:03:25 -0300 Subject: [PATCH 2/7] common: Add unique function --- src/common/CMakeLists.txt | 1 + src/common/unique_function.h | 62 +++++++++++++++ src/tests/CMakeLists.txt | 1 + src/tests/common/unique_function.cpp | 108 +++++++++++++++++++++++++++ 4 files changed, 172 insertions(+) create mode 100644 src/common/unique_function.h create mode 100644 src/tests/common/unique_function.cpp diff --git a/src/common/CMakeLists.txt b/src/common/CMakeLists.txt index a6fa9a85d..c05b78cd5 100644 --- a/src/common/CMakeLists.txt +++ b/src/common/CMakeLists.txt @@ -188,6 +188,7 @@ add_library(common STATIC tiny_mt.h tree.h uint128.h + unique_function.h uuid.cpp uuid.h vector_math.h diff --git a/src/common/unique_function.h b/src/common/unique_function.h new file mode 100644 index 000000000..ca0559071 --- /dev/null +++ b/src/common/unique_function.h @@ -0,0 +1,62 @@ +// Copyright 2021 yuzu emulator team +// Licensed under GPLv2 or any later version +// Refer to the license.txt file included. + +#pragma once + +#include +#include + +namespace Common { + +/// General purpose function wrapper similar to std::function. +/// Unlike std::function, the captured values don't have to be copyable. +/// This class can be moved but not copied. +template +class UniqueFunction { + class CallableBase { + public: + virtual ~CallableBase() = default; + virtual ResultType operator()(Args&&...) = 0; + }; + + template + class Callable final : public CallableBase { + public: + Callable(Functor&& functor_) : functor{std::move(functor_)} {} + ~Callable() override = default; + + ResultType operator()(Args&&... args) override { + return functor(std::forward(args)...); + } + + private: + Functor functor; + }; + +public: + UniqueFunction() = default; + + template + UniqueFunction(Functor&& functor) + : callable{std::make_unique>(std::move(functor))} {} + + UniqueFunction& operator=(UniqueFunction&& rhs) noexcept = default; + UniqueFunction(UniqueFunction&& rhs) noexcept = default; + + UniqueFunction& operator=(const UniqueFunction&) = delete; + UniqueFunction(const UniqueFunction&) = delete; + + ResultType operator()(Args&&... args) const { + return (*callable)(std::forward(args)...); + } + + explicit operator bool() const noexcept { + return static_cast(callable); + } + +private: + std::unique_ptr callable; +}; + +} // namespace Common diff --git a/src/tests/CMakeLists.txt b/src/tests/CMakeLists.txt index 96bc30cac..c4c012f3d 100644 --- a/src/tests/CMakeLists.txt +++ b/src/tests/CMakeLists.txt @@ -5,6 +5,7 @@ add_executable(tests common/host_memory.cpp common/param_package.cpp common/ring_buffer.cpp + common/unique_function.cpp core/core_timing.cpp core/network/network.cpp tests.cpp diff --git a/src/tests/common/unique_function.cpp b/src/tests/common/unique_function.cpp new file mode 100644 index 000000000..ac9912738 --- /dev/null +++ b/src/tests/common/unique_function.cpp @@ -0,0 +1,108 @@ +// Copyright 2021 yuzu Emulator Project +// Licensed under GPLv2 or any later version +// Refer to the license.txt file included. + +#include + +#include + +#include "common/unique_function.h" + +namespace { +struct Noisy { + Noisy() : state{"Default constructed"} {} + Noisy(Noisy&& rhs) noexcept : state{"Move constructed"} { + rhs.state = "Moved away"; + } + Noisy& operator=(Noisy&& rhs) noexcept { + state = "Move assigned"; + rhs.state = "Moved away"; + } + Noisy(const Noisy&) : state{"Copied constructed"} {} + Noisy& operator=(const Noisy&) { + state = "Copied assigned"; + } + + std::string state; +}; +} // Anonymous namespace + +TEST_CASE("UniqueFunction", "[common]") { + SECTION("Capture reference") { + int value = 0; + Common::UniqueFunction func = [&value] { value = 5; }; + func(); + REQUIRE(value == 5); + } + SECTION("Capture pointer") { + int value = 0; + int* pointer = &value; + Common::UniqueFunction func = [pointer] { *pointer = 5; }; + func(); + REQUIRE(value == 5); + } + SECTION("Move object") { + Noisy noisy; + REQUIRE(noisy.state == "Default constructed"); + + Common::UniqueFunction func = [noisy = std::move(noisy)] { + REQUIRE(noisy.state == "Move constructed"); + }; + REQUIRE(noisy.state == "Moved away"); + func(); + } + SECTION("Move construct function") { + int value = 0; + Common::UniqueFunction func = [&value] { value = 5; }; + Common::UniqueFunction new_func = std::move(func); + new_func(); + REQUIRE(value == 5); + } + SECTION("Move assign function") { + int value = 0; + Common::UniqueFunction func = [&value] { value = 5; }; + Common::UniqueFunction new_func; + new_func = std::move(func); + new_func(); + REQUIRE(value == 5); + } + SECTION("Default construct then assign function") { + int value = 0; + Common::UniqueFunction func; + func = [&value] { value = 5; }; + func(); + REQUIRE(value == 5); + } + SECTION("Pass arguments") { + int result = 0; + Common::UniqueFunction func = [&result](int a, int b) { result = a + b; }; + func(5, 4); + REQUIRE(result == 9); + } + SECTION("Pass arguments and return value") { + Common::UniqueFunction func = [](int a, int b) { return a + b; }; + REQUIRE(func(5, 4) == 9); + } + SECTION("Destructor") { + int num_destroyed = 0; + struct Foo { + Foo(int* num_) : num{num_} {} + Foo(Foo&& rhs) : num{std::exchange(rhs.num, nullptr)} {} + Foo(const Foo&) = delete; + + ~Foo() { + if (num) { + ++*num; + } + } + + int* num = nullptr; + }; + Foo object{&num_destroyed}; + { + Common::UniqueFunction func = [object = std::move(object)] {}; + REQUIRE(num_destroyed == 0); + } + REQUIRE(num_destroyed == 1); + } +} From bf5b5c1bf43946039d91f78253599c9996f86057 Mon Sep 17 00:00:00 2001 From: ReinUsesLisp Date: Thu, 1 Apr 2021 01:05:45 -0300 Subject: [PATCH 3/7] common/thread_worker: Use unique function --- src/common/thread_worker.cpp | 48 ++++++++++++++++-------------------- src/common/thread_worker.h | 6 +++-- 2 files changed, 25 insertions(+), 29 deletions(-) diff --git a/src/common/thread_worker.cpp b/src/common/thread_worker.cpp index 745918c7e..f4d8bb0f0 100644 --- a/src/common/thread_worker.cpp +++ b/src/common/thread_worker.cpp @@ -8,36 +8,30 @@ namespace Common { ThreadWorker::ThreadWorker(std::size_t num_workers, const std::string& name) { - for (std::size_t i = 0; i < num_workers; ++i) - threads.emplace_back([this, thread_name{std::string{name}}] { - Common::SetCurrentThreadName(thread_name.c_str()); + const auto lambda = [this, thread_name{std::string{name}}] { + Common::SetCurrentThreadName(thread_name.c_str()); - // Wait for first request + while (!stop) { + UniqueFunction task; { std::unique_lock lock{queue_mutex}; - condition.wait(lock, [this] { return stop || !requests.empty(); }); - } - - while (true) { - std::function task; - - { - std::unique_lock lock{queue_mutex}; - condition.wait(lock, [this] { return stop || !requests.empty(); }); - if (stop || requests.empty()) { - return; - } - task = std::move(requests.front()); - requests.pop(); - - if (requests.empty()) { - wait_condition.notify_one(); - } + if (requests.empty()) { + wait_condition.notify_all(); } - - task(); + condition.wait(lock, [this] { return stop || !requests.empty(); }); + if (stop || requests.empty()) { + break; + } + task = std::move(requests.front()); + requests.pop(); } - }); + task(); + } + wait_condition.notify_all(); + }; + for (size_t i = 0; i < num_workers; ++i) { + threads.emplace_back(lambda); + } } ThreadWorker::~ThreadWorker() { @@ -51,10 +45,10 @@ ThreadWorker::~ThreadWorker() { } } -void ThreadWorker::QueueWork(std::function&& work) { +void ThreadWorker::QueueWork(UniqueFunction work) { { std::unique_lock lock{queue_mutex}; - requests.emplace(work); + requests.emplace(std::move(work)); } condition.notify_one(); } diff --git a/src/common/thread_worker.h b/src/common/thread_worker.h index 7a6756eb5..7e2b04a07 100644 --- a/src/common/thread_worker.h +++ b/src/common/thread_worker.h @@ -11,18 +11,20 @@ #include #include +#include "common/unique_function.h" + namespace Common { class ThreadWorker final { public: explicit ThreadWorker(std::size_t num_workers, const std::string& name); ~ThreadWorker(); - void QueueWork(std::function&& work); + void QueueWork(UniqueFunction work); void WaitForRequests(); private: std::vector threads; - std::queue> requests; + std::queue> requests; std::mutex queue_mutex; std::condition_variable condition; std::condition_variable wait_condition; From a10e112e6436b30c9eb5ca2a82c94f83205bbc34 Mon Sep 17 00:00:00 2001 From: FernandoS27 Date: Tue, 6 Apr 2021 04:23:02 +0200 Subject: [PATCH 4/7] common/thread_worker: Fix data race --- src/common/thread_worker.cpp | 14 +++++++++++++- src/common/thread_worker.h | 5 +++++ 2 files changed, 18 insertions(+), 1 deletion(-) 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 @@ namespace Common { ThreadWorker::ThreadWorker(std::size_t num_workers, const std::string& name) { + workers_queued.store(static_cast(num_workers), std::memory_order_release); const auto lambda = [this, thread_name{std::string{name}}] { Common::SetCurrentThreadName(thread_name.c_str()); + // TODO(Blinkhawk): Change the design, this is very prone to data races + // Wait for first request + { + std::unique_lock lock{queue_mutex}; + condition.wait(lock, [this] { return stop || !requests.empty(); }); + } + while (!stop) { UniqueFunction task; { @@ -26,7 +34,9 @@ ThreadWorker::ThreadWorker(std::size_t num_workers, const std::string& name) { requests.pop(); } task(); + work_done++; } + workers_stopped++; wait_condition.notify_all(); }; for (size_t i = 0; i < num_workers; ++i) { @@ -49,13 +59,15 @@ void ThreadWorker::QueueWork(UniqueFunction work) { { std::unique_lock lock{queue_mutex}; requests.emplace(std::move(work)); + work_scheduled++; } condition.notify_one(); } void ThreadWorker::WaitForRequests() { std::unique_lock lock{queue_mutex}; - wait_condition.wait(lock, [this] { return stop || requests.empty(); }); + wait_condition.wait( + lock, [this] { return workers_stopped >= workers_queued || work_done >= work_scheduled; }); } } // 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 @@ #include #include +#include "common/common_types.h" #include "common/unique_function.h" namespace Common { @@ -29,6 +30,10 @@ private: std::condition_variable condition; std::condition_variable wait_condition; std::atomic_bool stop{}; + std::atomic work_scheduled{}; + std::atomic work_done{}; + std::atomic workers_stopped{}; + std::atomic workers_queued{}; }; } // namespace Common From c147e9a90e92ceec17d778d3c6e5cf6f028109b3 Mon Sep 17 00:00:00 2001 From: FernandoS27 Date: Tue, 6 Apr 2021 06:02:44 +0200 Subject: [PATCH 5/7] common/thread_worker: Simplify logic --- src/common/thread_worker.cpp | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/src/common/thread_worker.cpp b/src/common/thread_worker.cpp index fd130dfb4..32be49b15 100644 --- a/src/common/thread_worker.cpp +++ b/src/common/thread_worker.cpp @@ -12,13 +12,6 @@ ThreadWorker::ThreadWorker(std::size_t num_workers, const std::string& name) { const auto lambda = [this, thread_name{std::string{name}}] { Common::SetCurrentThreadName(thread_name.c_str()); - // TODO(Blinkhawk): Change the design, this is very prone to data races - // Wait for first request - { - std::unique_lock lock{queue_mutex}; - condition.wait(lock, [this] { return stop || !requests.empty(); }); - } - while (!stop) { UniqueFunction task; { @@ -27,7 +20,7 @@ ThreadWorker::ThreadWorker(std::size_t num_workers, const std::string& name) { wait_condition.notify_all(); } condition.wait(lock, [this] { return stop || !requests.empty(); }); - if (stop || requests.empty()) { + if (stop) { break; } task = std::move(requests.front()); From da34d3704405665b68d3d992f37a7eeb541238af Mon Sep 17 00:00:00 2001 From: ReinUsesLisp Date: Tue, 25 May 2021 20:37:06 -0300 Subject: [PATCH 6/7] common/thread_worker: Add support for stateful threads --- src/common/CMakeLists.txt | 1 - src/common/thread_worker.cpp | 66 ------------------------ src/common/thread_worker.h | 97 ++++++++++++++++++++++++++++++++---- 3 files changed, 86 insertions(+), 78 deletions(-) delete mode 100644 src/common/thread_worker.cpp 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 thread.cpp thread.h thread_queue_list.h - thread_worker.cpp thread_worker.h threadsafe_queue.h 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 @@ -// Copyright 2020 yuzu emulator team -// Licensed under GPLv2 or any later version -// Refer to the license.txt file included. - -#include "common/thread.h" -#include "common/thread_worker.h" - -namespace Common { - -ThreadWorker::ThreadWorker(std::size_t num_workers, const std::string& name) { - workers_queued.store(static_cast(num_workers), std::memory_order_release); - const auto lambda = [this, thread_name{std::string{name}}] { - Common::SetCurrentThreadName(thread_name.c_str()); - - while (!stop) { - UniqueFunction task; - { - std::unique_lock lock{queue_mutex}; - if (requests.empty()) { - wait_condition.notify_all(); - } - condition.wait(lock, [this] { return stop || !requests.empty(); }); - if (stop) { - break; - } - task = std::move(requests.front()); - requests.pop(); - } - task(); - work_done++; - } - workers_stopped++; - wait_condition.notify_all(); - }; - for (size_t i = 0; i < num_workers; ++i) { - threads.emplace_back(lambda); - } -} - -ThreadWorker::~ThreadWorker() { - { - std::unique_lock lock{queue_mutex}; - stop = true; - } - condition.notify_all(); - for (std::thread& thread : threads) { - thread.join(); - } -} - -void ThreadWorker::QueueWork(UniqueFunction work) { - { - std::unique_lock lock{queue_mutex}; - requests.emplace(std::move(work)); - work_scheduled++; - } - condition.notify_one(); -} - -void ThreadWorker::WaitForRequests() { - std::unique_lock lock{queue_mutex}; - wait_condition.wait( - lock, [this] { return workers_stopped >= workers_queued || work_done >= work_scheduled; }); -} - -} // 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 @@ #include #include #include +#include #include #include -#include "common/common_types.h" +#include "common/thread.h" #include "common/unique_function.h" namespace Common { -class ThreadWorker final { +template +class StatefulThreadWorker { + static constexpr bool with_state = !std::is_same_v; + + struct DummyCallable { + int operator()() const noexcept { + return 0; + } + }; + + using Task = + std::conditional_t, UniqueFunction>; + using StateMaker = std::conditional_t, DummyCallable>; + public: - explicit ThreadWorker(std::size_t num_workers, const std::string& name); - ~ThreadWorker(); - void QueueWork(UniqueFunction work); - void WaitForRequests(); + explicit StatefulThreadWorker(size_t num_workers, std::string name, StateMaker func = {}) + : workers_queued{num_workers}, thread_name{std::move(name)} { + const auto lambda = [this, func] { + Common::SetCurrentThreadName(thread_name.c_str()); + { + std::conditional_t state{func()}; + while (!stop) { + Task task; + { + std::unique_lock lock{queue_mutex}; + if (requests.empty()) { + wait_condition.notify_all(); + } + condition.wait(lock, [this] { return stop || !requests.empty(); }); + if (stop) { + break; + } + task = std::move(requests.front()); + requests.pop(); + } + if constexpr (with_state) { + task(&state); + } else { + task(); + } + ++work_done; + } + } + ++workers_stopped; + wait_condition.notify_all(); + }; + for (size_t i = 0; i < num_workers; ++i) { + threads.emplace_back(lambda); + } + } + + ~StatefulThreadWorker() { + { + std::unique_lock lock{queue_mutex}; + stop = true; + } + condition.notify_all(); + for (std::thread& thread : threads) { + thread.join(); + } + } + + void QueueWork(Task work) { + { + std::unique_lock lock{queue_mutex}; + requests.emplace(std::move(work)); + ++work_scheduled; + } + condition.notify_one(); + } + + void WaitForRequests() { + std::unique_lock lock{queue_mutex}; + wait_condition.wait(lock, [this] { + return workers_stopped >= workers_queued || work_done >= work_scheduled; + }); + } private: std::vector threads; - std::queue> requests; + std::queue requests; std::mutex queue_mutex; std::condition_variable condition; std::condition_variable wait_condition; std::atomic_bool stop{}; - std::atomic work_scheduled{}; - std::atomic work_done{}; - std::atomic workers_stopped{}; - std::atomic workers_queued{}; + std::atomic work_scheduled{}; + std::atomic work_done{}; + std::atomic workers_stopped{}; + std::atomic workers_queued{}; + std::string thread_name; }; +using ThreadWorker = StatefulThreadWorker<>; + } // namespace Common From 0ddbbb64e514ea9bba6a4f8bd6908d654e7f114c Mon Sep 17 00:00:00 2001 From: ReinUsesLisp Date: Tue, 29 Jun 2021 01:47:13 -0300 Subject: [PATCH 7/7] common/thread_worker: Stop workers on stop_token when waiting --- src/common/thread_worker.h | 38 ++++++++++++++++++++------------------ 1 file changed, 20 insertions(+), 18 deletions(-) diff --git a/src/common/thread_worker.h b/src/common/thread_worker.h index 16aa673bd..8272985ff 100644 --- a/src/common/thread_worker.h +++ b/src/common/thread_worker.h @@ -7,7 +7,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -34,19 +36,19 @@ class StatefulThreadWorker { public: explicit StatefulThreadWorker(size_t num_workers, std::string name, StateMaker func = {}) : workers_queued{num_workers}, thread_name{std::move(name)} { - const auto lambda = [this, func] { + const auto lambda = [this, func](std::stop_token stop_token) { Common::SetCurrentThreadName(thread_name.c_str()); { std::conditional_t state{func()}; - while (!stop) { + while (!stop_token.stop_requested()) { Task task; { std::unique_lock lock{queue_mutex}; if (requests.empty()) { wait_condition.notify_all(); } - condition.wait(lock, [this] { return stop || !requests.empty(); }); - if (stop) { + condition.wait(lock, stop_token, [this] { return !requests.empty(); }); + if (stop_token.stop_requested()) { break; } task = std::move(requests.front()); @@ -63,21 +65,17 @@ public: ++workers_stopped; wait_condition.notify_all(); }; + threads.reserve(num_workers); for (size_t i = 0; i < num_workers; ++i) { threads.emplace_back(lambda); } } - ~StatefulThreadWorker() { - { - std::unique_lock lock{queue_mutex}; - stop = true; - } - condition.notify_all(); - for (std::thread& thread : threads) { - thread.join(); - } - } + StatefulThreadWorker& operator=(const StatefulThreadWorker&) = delete; + StatefulThreadWorker(const StatefulThreadWorker&) = delete; + + StatefulThreadWorker& operator=(StatefulThreadWorker&&) = delete; + StatefulThreadWorker(StatefulThreadWorker&&) = delete; void QueueWork(Task work) { { @@ -88,7 +86,12 @@ public: condition.notify_one(); } - void WaitForRequests() { + void WaitForRequests(std::stop_token stop_token = {}) { + std::stop_callback callback(stop_token, [this] { + for (auto& thread : threads) { + thread.request_stop(); + } + }); std::unique_lock lock{queue_mutex}; wait_condition.wait(lock, [this] { return workers_stopped >= workers_queued || work_done >= work_scheduled; @@ -96,17 +99,16 @@ public: } private: - std::vector threads; std::queue requests; std::mutex queue_mutex; - std::condition_variable condition; + std::condition_variable_any condition; std::condition_variable wait_condition; - std::atomic_bool stop{}; std::atomic work_scheduled{}; std::atomic work_done{}; std::atomic workers_stopped{}; std::atomic workers_queued{}; std::string thread_name; + std::vector threads; }; using ThreadWorker = StatefulThreadWorker<>;