| Taskflow: A General-purpose Task-parallel Programming System

Loading...
Searching...
No Matches
worker.hpp
1#pragma once
2
3#include "declarations.hpp"
4#include "wsq.hpp"
5#include "nonblocking_notifier.hpp"
6#include "atomic_notifier.hpp"
7
13 namespace tf {
14
15// ----------------------------------------------------------------------------
16// Default Notifier
17// Our experiments show that NonblockingNotifier has the most stable performance
18// ----------------------------------------------------------------------------
19
35#ifdef TF_ENABLE_ATOMIC_NOTIFIER
36 using DefaultNotifier = AtomicNotifier;
37#else
38 using DefaultNotifier = NonblockingNotifier;
39#endif
40
41// ----------------------------------------------------------------------------
42// Class Definition: Worker
43// ----------------------------------------------------------------------------
44
55 class Worker {
56
57 friend class Executor;
58 friend class Runtime;
59 friend class WorkerView;
60
61 using wsq_type = BoundedWSQ<Node*>;
62
63 public:
64
72 inline size_t id() const { return _id; }
73
78 inline size_t queue_size() const { return _wsq.size(); }
79
83 inline size_t queue_capacity() const { return static_cast<size_t>(_wsq.capacity()); }
84
88 std::thread& thread() { return _thread; }
89
90 private:
91
92 alignas(TF_CACHELINE_SIZE) std::atomic_flag _done = ATOMIC_FLAG_INIT;
93
94 size_t _id;
95 size_t _sticky_victim;
96 Xorshift<uint32_t> _rdgen;
97 std::thread _thread;
98 wsq_type _wsq;
99};
100
101// ----------------------------------------------------------------------------
102// Class Definition: WorkerView
103// ----------------------------------------------------------------------------
104
115 class WorkerView {
116
117 friend class Executor;
118
119 public:
120
128 size_t id() const;
129
134 size_t queue_size() const;
135
139 size_t queue_capacity() const;
140
141 private:
142
143 WorkerView(const Worker&);
144 WorkerView(const WorkerView&) = default;
145
146 const Worker& _worker;
147};
148
149// Constructor
150 inline WorkerView::WorkerView(const Worker& w) : _worker{w} {
151}
152
153// function: id
154 inline size_t WorkerView::id() const {
155 return _worker._id;
156}
157
158// Function: queue_size
159 inline size_t WorkerView::queue_size() const {
160 return _worker._wsq.size();
161}
162
163// Function: queue_capacity
164 inline size_t WorkerView::queue_capacity() const {
165 return static_cast<size_t>(_worker._wsq.capacity());
166}
167
168// ----------------------------------------------------------------------------
169// Class Definition: WorkerInterface
170// ----------------------------------------------------------------------------
171
273 class WorkerInterface {
274
275 public:
276
280 virtual ~WorkerInterface() = default;
281
288 virtual void scheduler_prologue(Worker& worker) = 0;
289
299 virtual void scheduler_epilogue(Worker& worker, std::exception_ptr ptr) = 0;
300
301};
302
311 template <typename T, typename... ArgsT>
312std::shared_ptr<T> make_worker_interface(ArgsT&&... args) {
313 static_assert(
314 std::is_base_of_v<WorkerInterface, T>,
315 "T must be derived from WorkerInterface"
316 );
317 return std::make_shared<T>(std::forward<ArgsT>(args)...);
318}
319
320
321
322
323} // end of namespact tf ------------------------------------------------------
324
325
tf::BoundedWSQ< Node * >
tf::BoundedWSQ::capacity
constexpr size_t capacity() const
queries the capacity of the queue
Definition wsq.hpp:879
tf::BoundedWSQ::size
size_t size() const noexcept
queries the number of items at the time of this call
Definition wsq.hpp:764
tf::Executor
class to create an executor
Definition executor.hpp:62
tf::NonblockingNotifier
class to create a non-blocking notifier
Definition nonblocking_notifier.hpp:84
tf::Runtime
class to create a runtime task
Definition runtime.hpp:47
tf::WorkerInterface
class to configure worker behavior in an executor
Definition worker.hpp:273
tf::WorkerInterface::scheduler_epilogue
virtual void scheduler_epilogue(Worker &worker, std::exception_ptr ptr)=0
method to call after a worker leaves the scheduling loop
tf::WorkerInterface::scheduler_prologue
virtual void scheduler_prologue(Worker &worker)=0
method to call before a worker enters the scheduling loop
tf::WorkerInterface::~WorkerInterface
virtual ~WorkerInterface()=default
default destructor
tf::WorkerView
class to create an immutable view of a worker
Definition worker.hpp:115
tf::WorkerView::id
size_t id() const
queries the worker id associated with its parent executor
Definition worker.hpp:154
tf::WorkerView::queue_capacity
size_t queue_capacity() const
queries the current capacity of the queue
Definition worker.hpp:164
tf::WorkerView::queue_size
size_t queue_size() const
queries the size of the queue (i.e., number of pending tasks to run) associated with the worker
Definition worker.hpp:159
tf::Worker
class to create a worker in an executor
Definition worker.hpp:55
tf::Worker::id
size_t id() const
queries the worker id associated with its parent executor
Definition worker.hpp:72
tf::Worker::queue_capacity
size_t queue_capacity() const
queries the current capacity of the queue
Definition worker.hpp:83
tf::Worker::queue_size
size_t queue_size() const
queries the size of the queue (i.e., number of enqueued tasks to run) associated with the worker
Definition worker.hpp:78
tf::Worker::thread
std::thread & thread()
acquires the associated thread
Definition worker.hpp:88
tf::Xorshift
class to create a fast xorshift-based pseudo-random number generator
Definition math.hpp:412
tf
taskflow namespace
Definition small_vector.hpp:20
tf::make_worker_interface
std::shared_ptr< T > make_worker_interface(ArgsT &&... args)
helper function to create an instance derived from tf::WorkerInterface
Definition worker.hpp:312
tf::DefaultNotifier
NonblockingNotifier DefaultNotifier
the default notifier type used by Taskflow
Definition worker.hpp:38