From 4bf94458d5d7abafb154e914620dc758d26719c2 Mon Sep 17 00:00:00 2001 From: Dan Goodliffe Date: Fri, 14 Apr 2023 10:24:10 +0100 Subject: Global worker instance --- lib/worker.cpp | 12 ++++++------ lib/worker.h | 35 ++++++++++------------------------- 2 files changed, 16 insertions(+), 31 deletions(-) (limited to 'lib') diff --git a/lib/worker.cpp b/lib/worker.cpp index fd255c7..cf59f56 100644 --- a/lib/worker.cpp +++ b/lib/worker.cpp @@ -1,9 +1,10 @@ #include "worker.h" -#if __cpp_lib_semaphore -# include "work.h" -# include -# include -# include +#include "work.h" +#include +#include +#include + +Worker Worker::instance; Worker::Worker() : todoLen {0} { @@ -45,4 +46,3 @@ Worker::worker() j->doWork(); } } -#endif diff --git a/lib/worker.h b/lib/worker.h index 7cd06f9..136c50e 100644 --- a/lib/worker.h +++ b/lib/worker.h @@ -1,19 +1,17 @@ #pragma once +#include +#include +#include +#include +#include +#include #include -class Work; - -#if __cpp_lib_semaphore -# include -# include -# include -# include -# include -# include -# include +#include +class Work; class Worker { -public: +private: Worker(); ~Worker(); @@ -42,19 +40,6 @@ private: ToDo todo; std::counting_semaphore<16> todoLen; std::mutex todoMutex; -}; - -#else -class Worker { -public: - template - void - addWork(Params &&... params) - requires std::is_base_of_v - { - T(std::forward(params)...).doWork(); - } + static Worker instance; }; - -#endif -- cgit v1.2.3 From 1cbf18ff8463946aa5a09ccf1d6873b222b912b6 Mon Sep 17 00:00:00 2001 From: Dan Goodliffe Date: Fri, 14 Apr 2023 10:26:51 +0100 Subject: Simplify worker with jthread Moves thread collection to bottom of class so threads are joined before job storage is destroyed. --- lib/work.h | 12 ------------ lib/worker.cpp | 5 +---- lib/worker.h | 4 ++-- 3 files changed, 3 insertions(+), 18 deletions(-) delete mode 100644 lib/work.h (limited to 'lib') diff --git a/lib/work.h b/lib/work.h deleted file mode 100644 index c780e13..0000000 --- a/lib/work.h +++ /dev/null @@ -1,12 +0,0 @@ -#pragma once - -#include - -class Work { -public: - virtual ~Work() = default; - NO_COPY(Work); - NO_MOVE(Work); - - virtual void doWork() = 0; -}; diff --git a/lib/worker.cpp b/lib/worker.cpp index cf59f56..4f1352d 100644 --- a/lib/worker.cpp +++ b/lib/worker.cpp @@ -9,16 +9,13 @@ Worker Worker::instance; Worker::Worker() : todoLen {0} { std::generate_n(std::back_inserter(threads), std::thread::hardware_concurrency(), [this]() { - return std::thread {&Worker::worker, this}; + return std::jthread {&Worker::worker, this}; }); } Worker::~Worker() { todoLen.release(std::thread::hardware_concurrency()); - std::for_each(threads.begin(), threads.end(), [](auto & th) { - th.join(); - }); } void diff --git a/lib/worker.h b/lib/worker.h index 136c50e..96593d9 100644 --- a/lib/worker.h +++ b/lib/worker.h @@ -33,13 +33,13 @@ private: private: void worker(); - using Threads = std::vector; + using Threads = std::vector; using ToDo = std::deque; - Threads threads; ToDo todo; std::counting_semaphore<16> todoLen; std::mutex todoMutex; + Threads threads; static Worker instance; }; -- cgit v1.2.3 From aa77548bd637ffd1a7461136e6206906dc4c61c7 Mon Sep 17 00:00:00 2001 From: Dan Goodliffe Date: Fri, 14 Apr 2023 11:54:45 +0100 Subject: New WorkItem/job/promise/future based interface --- lib/worker.cpp | 3 +- lib/worker.h | 87 +++++++++++++++++++++++++++++++++++++++++++++++++++------- 2 files changed, 78 insertions(+), 12 deletions(-) (limited to 'lib') diff --git a/lib/worker.cpp b/lib/worker.cpp index 4f1352d..7e7f296 100644 --- a/lib/worker.cpp +++ b/lib/worker.cpp @@ -1,5 +1,4 @@ #include "worker.h" -#include "work.h" #include #include #include @@ -19,7 +18,7 @@ Worker::~Worker() } void -Worker::addWork(WorkPtr j) +Worker::addWorkPtr(WorkPtr j) { std::lock_guard lck {todoMutex}; todoLen.release(); diff --git a/lib/worker.h b/lib/worker.h index 96593d9..5356606 100644 --- a/lib/worker.h +++ b/lib/worker.h @@ -1,6 +1,8 @@ #pragma once #include +#include +#include #include #include #include @@ -9,28 +11,93 @@ #include #include -class Work; class Worker { +public: + class WorkItem { + public: + WorkItem() = default; + virtual ~WorkItem() = default; + NO_MOVE(WorkItem); + NO_COPY(WorkItem); + + virtual void doWork() = 0; + }; + + template class WorkItemT : public WorkItem { + public: + T + get() + { + return future.get(); + } + + protected: + std::promise promise; + std::future future {promise.get_future()}; + friend Worker; + }; + + template + static auto + addWork(Params &&... params) + { + return instance.addWorkImpl(std::forward(params)...); + } + template using WorkPtrT = std::shared_ptr>; + private: + template class WorkItemTImpl : public WorkItemT { + public: + WorkItemTImpl(Params &&... params) : params {std::forward(params)...} { } + + private: + void + doWork() override + { + try { + if constexpr (std::is_void_v) { + std::apply( + [](auto &&... p) { + return std::invoke(p...); + }, + params); + WorkItemT::promise.set_value(); + } + else { + WorkItemT::promise.set_value(std::apply( + [](auto &&... p) { + return std::invoke(p...); + }, + params)); + } + } + catch (...) { + WorkItemT::promise.set_exception(std::current_exception()); + } + } + + std::tuple params; + }; + Worker(); ~Worker(); NO_COPY(Worker); NO_MOVE(Worker); - using WorkPtr = std::unique_ptr; + using WorkPtr = std::shared_ptr; - template - void - addWork(Params &&... params) - requires std::is_base_of_v + template + auto + addWorkImpl(Params &&... params) { - addWork(std::make_unique(std::forward(params)...)); + using T = decltype(std::invoke(std::forward(params)...)); + auto work = std::make_shared>(std::forward(params)...); + addWorkPtr(work); + return work; } - void addWork(WorkPtr w); - -private: + void addWorkPtr(WorkPtr w); void worker(); using Threads = std::vector; -- cgit v1.2.3 From e671adba5a57e1d4e848eb4d6de0f480e7b3709a Mon Sep 17 00:00:00 2001 From: Dan Goodliffe Date: Fri, 14 Apr 2023 12:29:24 +0100 Subject: Simplify doWork, add tests for various interface uses --- lib/worker.h | 12 ++---------- 1 file changed, 2 insertions(+), 10 deletions(-) (limited to 'lib') diff --git a/lib/worker.h b/lib/worker.h index 5356606..1bc7c14 100644 --- a/lib/worker.h +++ b/lib/worker.h @@ -56,19 +56,11 @@ private: { try { if constexpr (std::is_void_v) { - std::apply( - [](auto &&... p) { - return std::invoke(p...); - }, - params); + std::apply(std::invoke, params); WorkItemT::promise.set_value(); } else { - WorkItemT::promise.set_value(std::apply( - [](auto &&... p) { - return std::invoke(p...); - }, - params)); + WorkItemT::promise.set_value(std::apply(std::invoke, params)); } } catch (...) { -- cgit v1.2.3 From f8ee56f6dc8cf7e92c4bfc5930d32b14f634141c Mon Sep 17 00:00:00 2001 From: Dan Goodliffe Date: Fri, 14 Apr 2023 13:13:47 +0100 Subject: Current thread partakes in work effort while waiting This will prevent deadlock if the work pool is otherwise busy by ensuring work is always being done --- lib/worker.cpp | 21 +++++++++++++++++++++ lib/worker.h | 26 ++++++++++++++++++++++---- 2 files changed, 43 insertions(+), 4 deletions(-) (limited to 'lib') diff --git a/lib/worker.cpp b/lib/worker.cpp index 7e7f296..45fb6df 100644 --- a/lib/worker.cpp +++ b/lib/worker.cpp @@ -42,3 +42,24 @@ Worker::worker() j->doWork(); } } + +void +Worker::assist() +{ + auto job = [this]() { + using namespace std::chrono_literals; + if (todoLen.try_acquire_for(100us)) { + if (std::lock_guard lck {todoMutex}; todo.size()) { + WorkPtr x = std::move(todo.front()); + if (x) { + todo.pop_front(); + } + return x; + } + } + return WorkPtr {}; + }; + if (auto j = job()) { + j->doWork(); + } +} diff --git a/lib/worker.h b/lib/worker.h index 1bc7c14..d9a5a6f 100644 --- a/lib/worker.h +++ b/lib/worker.h @@ -14,12 +14,20 @@ class Worker { public: class WorkItem { - public: - WorkItem() = default; + protected: + WorkItem(Worker * worker) : worker {worker} { } virtual ~WorkItem() = default; NO_MOVE(WorkItem); NO_COPY(WorkItem); + void + assist() const + { + worker->assist(); + } + Worker * worker; + + public: virtual void doWork() = 0; }; @@ -28,10 +36,16 @@ public: T get() { + using namespace std::chrono_literals; + while (future.wait_for(0s) == std::future_status::timeout) { + assist(); + } return future.get(); } protected: + WorkItemT(Worker * worker) : WorkItem {worker} { } + std::promise promise; std::future future {promise.get_future()}; friend Worker; @@ -48,7 +62,10 @@ public: private: template class WorkItemTImpl : public WorkItemT { public: - WorkItemTImpl(Params &&... params) : params {std::forward(params)...} { } + WorkItemTImpl(Worker * worker, Params &&... params) : + WorkItemT {worker}, params {std::forward(params)...} + { + } private: void @@ -84,13 +101,14 @@ private: addWorkImpl(Params &&... params) { using T = decltype(std::invoke(std::forward(params)...)); - auto work = std::make_shared>(std::forward(params)...); + auto work = std::make_shared>(this, std::forward(params)...); addWorkPtr(work); return work; } void addWorkPtr(WorkPtr w); void worker(); + void assist(); using Threads = std::vector; using ToDo = std::deque; -- cgit v1.2.3