25 #ifndef GUL14_THREADPOOL_H_
26 #define GUL14_THREADPOOL_H_
29 #include <condition_variable>
49 std::shared_ptr<ThreadPool> lock_pool_or_throw(std::weak_ptr<ThreadPool> pool);
110 class ThreadPool :
public std::enable_shared_from_this<ThreadPool>
117 using ThreadId = std::vector<std::thread>::size_type;
135 template <
typename T>
161 : future_{ std::move(future) }
163 , pool_{ std::move(pool) }
180 return detail::lock_pool_or_throw(pool_)->cancel_pending_task(id_);
191 if (not future_.valid())
192 throw std::logic_error(
"Canceled task has no result");
193 return future_.get();
210 if (not future_.valid())
213 return future_.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
230 const auto state = detail::lock_pool_or_throw(pool_)->get_task_state(id_);
232 if (state == InternalTaskState::unknown)
243 std::future<T> future_;
245 std::weak_ptr<ThreadPool> pool_;
249 using TimePoint = std::chrono::time_point<std::chrono::system_clock>;
250 using Duration = TimePoint::duration;
307 template <
typename Function>
308 TaskHandle<invoke_result_t<Function, ThreadPool&>>
309 add_task(Function fct, TimePoint start_time = {}, std::string name = {})
312 is_invocable<Function, ThreadPool&>::value
313 || is_invocable<Function>::value,
314 "Invalid function signature: Must be T fct() or T fct(ThreadPool&)");
316 using Result = invoke_result_t<Function, ThreadPool&>;
318 TaskHandle<Result> task_handle = [
this, &fct, start_time, &name]()
320 std::lock_guard<std::mutex> lock(mutex_);
324 throw std::runtime_error(
cat(
325 "Cannot add task: Pending queue has reached capacity (",
326 pending_tasks_.size(),
')'));
329 using PackagedTask = std::packaged_task<Result(ThreadPool&)>;
331 auto named_task_ptr = std::make_unique<NamedTaskImpl<PackagedTask>>(
332 PackagedTask{ std::move(fct) }, std::move(name));
334 TaskHandle<Result> handle{
335 next_task_id_, named_task_ptr->fct_.get_future(), shared_from_this() };
337 pending_tasks_.emplace_back(
338 next_task_id_, std::move(named_task_ptr), start_time);
350 template <
typename Function,
351 std::enable_if_t<is_invocable<Function>::value,
bool> =
true>
352 TaskHandle<invoke_result_t<Function>>
353 add_task(Function fct, TimePoint start_time = {}, std::string name = {})
356 [f = std::move(fct)](ThreadPool&)
mutable {
return f(); },
357 start_time, std::move(name));
360 template <
typename Function,
361 std::enable_if_t<is_invocable<Function, ThreadPool&>::value,
bool> =
true>
362 TaskHandle<invoke_result_t<Function, ThreadPool&>>
363 add_task(Function fct, Duration delay_before_start, std::string name = {})
366 std::chrono::system_clock::now() + delay_before_start, std::move(name));
369 template <
typename Function,
370 std::enable_if_t<is_invocable<Function>::value,
bool> =
true>
371 TaskHandle<invoke_result_t<Function>>
372 add_task(Function fct, Duration delay_before_start, std::string name = {})
375 std::chrono::system_clock::now() + delay_before_start, std::move(name));
378 template <
typename Function,
379 std::enable_if_t<is_invocable<Function, ThreadPool&>::value,
bool> =
true>
380 TaskHandle<invoke_result_t<Function, ThreadPool&>>
381 add_task(Function fct, std::string name)
383 return add_task(std::move(fct), TimePoint{}, std::move(name));
386 template <
typename Function,
387 std::enable_if_t<is_invocable<Function>::value,
bool> =
true>
388 TaskHandle<invoke_result_t<Function>>
389 add_task(Function fct, std::string name)
391 return add_task(std::move(fct), TimePoint{}, std::move(name));
407 std::size_t
capacity() const noexcept {
return capacity_; }
474 enum class InternalTaskState
486 NamedTask(std::string name)
487 : name_{ std::move(name) }
490 virtual ~NamedTask() =
default;
491 virtual void operator()(ThreadPool& pool) = 0;
496 template <
typename FunctionType>
497 struct NamedTaskImpl :
public NamedTask
500 NamedTaskImpl(FunctionType fct, std::string name)
501 : NamedTask{ std::move(name) }
502 , fct_{ std::move(fct) }
505 void operator()(ThreadPool& pool)
override { fct_(pool); }
513 std::unique_ptr<NamedTask> named_task_;
514 TimePoint start_time_{};
518 Task(
TaskId task_id, std::unique_ptr<NamedTask> named_task, TimePoint start_time)
520 , named_task_{ std::move(named_task) }
521 , start_time_{ start_time }
525 std::size_t capacity_{ 0 };
531 std::vector<std::thread> threads_;
538 thread_local
static ThreadId thread_id_;
544 std::condition_variable cv_;
546 mutable std::mutex mutex_;
547 std::vector<Task> pending_tasks_;
548 std::vector<TaskId> running_task_ids_;
549 std::vector<std::string> running_task_names_;
551 bool shutdown_requested_{
false };
571 ThreadPool(std::size_t num_threads, std::size_t
capacity);
586 bool cancel_pending_task(
TaskId task_id);
598 InternalTaskState get_task_state(
TaskId task_id)
const;
605 bool is_full_i() const noexcept;
613 void perform_work(std::
size_t thread_index);
Declaration of the overload set for cat() and of the associated class ConvertingStringView.
A handle for a task that has (or had) been enqueued on a ThreadPool.
Definition: ThreadPool.h:137
T get_result()
Block until the task has finished and return its result.
Definition: ThreadPool.h:189
bool cancel()
Remove the task from the queue if it is still pending.
Definition: ThreadPool.h:177
TaskHandle()
Default-construct an invalid TaskHandle.
Definition: ThreadPool.h:145
bool is_complete() const
Determine whether the task has completed.
Definition: ThreadPool.h:208
TaskState get_state() const
Determine if the task is running, waiting to be started, completed, or has been canceled.
Definition: ThreadPool.h:228
TaskHandle(TaskId id, std::future< T > future, std::shared_ptr< ThreadPool > pool)
Construct a TaskHandle.
Definition: ThreadPool.h:160
A pool of worker threads with a task queue.
Definition: ThreadPool.h:111
std::uint64_t TaskId
A unique identifier for a task.
Definition: ThreadPool.h:114
~ThreadPool()
Destruct the ThreadPool and join all threads.
Definition: ThreadPool.cc:69
static GUL_EXPORT std::shared_ptr< ThreadPool > make_shared(std::size_t num_threads, std::size_t capacity=default_capacity)
Create a thread pool with the desired number of threads and the specified capacity for enqueuing task...
Definition: ThreadPool.cc:187
GUL_EXPORT std::size_t cancel_pending_tasks()
Remove all pending tasks from the queue.
Definition: ThreadPool.cc:98
GUL_EXPORT bool is_idle() const
Return true if the pool has neither pending tasks nor tasks that are currently being executed.
Definition: ThreadPool.cc:175
GUL_EXPORT std::vector< std::string > get_running_task_names() const
Return a vector with the names of the tasks that are currently running.
Definition: ThreadPool.cc:132
GUL_EXPORT std::vector< std::string > get_pending_task_names() const
Return a vector with the names of the tasks that are waiting to be executed.
Definition: ThreadPool.cc:119
GUL_EXPORT ThreadId get_thread_id() const
Return the thread pool ID of the current thread.
Definition: ThreadPool.cc:156
GUL_EXPORT std::size_t capacity() const noexcept
Return the maximum number of pending tasks that can be queued.
Definition: ThreadPool.h:407
constexpr static std::size_t default_capacity
Default capacity for the task queue.
Definition: ThreadPool.h:253
GUL_EXPORT std::size_t count_pending() const
Return the number of pending tasks.
Definition: ThreadPool.cc:108
std::vector< std::thread >::size_type ThreadId
A unique identifier for a thread in the pool in the range of [0, count_threads()).
Definition: ThreadPool.h:117
TaskHandle< invoke_result_t< Function, ThreadPool & > > add_task(Function fct, TimePoint start_time={}, std::string name={})
Enqueue a task.
Definition: ThreadPool.h:309
constexpr static std::size_t max_capacity
Maximum possible capacity for the task queue.
Definition: ThreadPool.h:256
GUL_EXPORT std::size_t count_threads() const noexcept
Return the number of threads in the pool.
Definition: ThreadPool.cc:114
constexpr static std::size_t max_threads
Maximum possible number of threads.
Definition: ThreadPool.h:259
GUL_EXPORT bool is_shutdown_requested() const
Determine whether the thread pool has been requested to shut down.
Definition: ThreadPool.cc:181
GUL_EXPORT bool is_full() const noexcept
Determine whether the queue for pending tasks is full (at capacity).
Definition: ThreadPool.cc:164
std::shared_ptr< ThreadPool > make_thread_pool(std::size_t num_threads, std::size_t capacity=ThreadPool::default_capacity)
Create a thread pool with the desired number of threads and the specified capacity for queuing tasks.
Definition: ThreadPool.h:625
TaskState
An enum describing the state of an individual task.
Definition: ThreadPool.h:66
@ running
The task is currently being executed.
@ pending
The task is waiting to be started.
@ canceled
The task was removed from the queue before it was started.
@ complete
The task has finished (successfully or by throwing an exception).
std::string cat()
Efficiently concatenate an arbitrary number of strings and numbers.
Definition: cat.h:97
Namespace gul14 contains all functions and classes of the General Utility Library.
Definition: doxygen.h:26
Some metaprogramming traits for the General Utility Library.