1+ #pragma once
2+
3+ #include < thread_pool/slotted_bag.hpp>
4+ #include < thread_pool/thread_pool_options.hpp>
5+ #include < thread_pool/worker.hpp>
6+
7+ #include < atomic>
8+ #include < thread>
9+ #include < limits>
10+ #include < mutex>
11+ #include < chrono>
12+ #include < condition_variable>
13+
14+ namespace tp
15+ {
16+
17+ /* *
18+ * @brief Rouser is a specialized worker that periodically wakes other workers.
19+ * @detail This serves two purposes:
20+ * The first is that it emplaces an upper bound on the processing time of tasks in the thread pool, since
21+ * it is otherwise possible for the thread pool to enter a state where all threads are asleep, and tasks exist
22+ * in worker queues. The second is that it increases the likelihood of at least one worker busy-waiting at any
23+ * point in time, which speeds up task processing response time.
24+ */
25+ class Rouser final
26+ {
27+ public:
28+ /* *
29+ * @brief Worker Constructor.
30+ */
31+ Rouser (std::chrono::microseconds rouse_period);
32+
33+ /* *
34+ * @brief Copy ctor implementation.
35+ */
36+ Rouser (Rouser const &) = delete ;
37+
38+ /* *
39+ * @brief Copy assignment implementation.
40+ */
41+ Rouser& operator =(Rouser const & rhs) = delete ;
42+
43+ /* *
44+ * @brief Move ctor implementation.
45+ * @note Be very careful when invoking this while the thread pool is
46+ * active, or in an otherwise undefined state.
47+ */
48+ Rouser (Rouser&& rhs) noexcept ;
49+
50+ /* *
51+ * @brief Move assignment implementaion.
52+ * @note Be very careful when invoking this while the thread pool is
53+ * active, or in an otherwise undefined state.
54+ */
55+ Rouser& operator =(Rouser&& rhs) noexcept ;
56+
57+ /* *
58+ * @brief Destructor implementation.
59+ */
60+ ~Rouser ();
61+
62+ /* *
63+ * @brief start Create the executing thread and start tasks execution.
64+ * @param workers A pointer to the vector containing sibling workers for performing round robin work stealing.
65+ * @param idle_workers A pointer to the slotted bag containing all idle workers.
66+ * @param num_busy_waiters A pointer to the atomic busy waiter counter.
67+ * @note The parameters passed into this function generally relate to the global thread pool state.
68+ */
69+ template <typename Task, template <typename > class Queue >
70+ void start (std::vector<std::unique_ptr<Worker<Task, Queue>>>& workers, SlottedBag<Queue>& idle_workers, std::atomic<size_t >& num_busy_waiters);
71+
72+ /* *
73+ * @brief stop Stop all worker's thread and stealing activity.
74+ * Waits until the executing thread becomes finished.
75+ */
76+ void stop ();
77+
78+ private:
79+
80+ /* *
81+ * @brief threadFunc Executing thread function.
82+ * @param workers A pointer to the vector containing sibling workers for performing round robin work stealing.
83+ * @param idle_workers A pointer to the slotted bag containing all idle workers.
84+ * @param num_busy_waiters A pointer to the atomic busy waiter counter.
85+ */
86+ template <typename Task, template <typename > class Queue >
87+ void threadFunc (std::vector<std::unique_ptr<Worker<Task, Queue>>>& workers, SlottedBag<Queue>& idle_workers, std::atomic<size_t >& num_busy_waiters);
88+
89+ std::atomic<bool > m_running_flag;
90+ std::atomic<bool > m_started_flag;
91+ std::thread m_thread;
92+ std::chrono::microseconds m_rouse_period;
93+ };
94+
95+ inline Rouser::Rouser (std::chrono::microseconds rouse_period)
96+ : m_running_flag(false )
97+ , m_started_flag(false )
98+ , m_rouse_period(std::move(rouse_period))
99+ {
100+ }
101+
102+ inline Rouser::Rouser (Rouser&& rhs) noexcept
103+ {
104+ *this = std::move (rhs);
105+ }
106+
107+ inline Rouser& Rouser::operator =(Rouser&& rhs) noexcept
108+ {
109+ if (this != &rhs)
110+ {
111+ m_running_flag = rhs.m_running_flag .load ();
112+ m_started_flag = rhs.m_started_flag .load ();
113+ m_thread = std::move (rhs.m_thread );
114+ m_rouse_period = std::move (rhs.m_rouse_period );
115+ }
116+
117+ return *this ;
118+ }
119+
120+ inline Rouser::~Rouser ()
121+ {
122+ stop ();
123+ }
124+
125+ template <typename Task, template <typename > class Queue >
126+ inline void Rouser::start (std::vector<std::unique_ptr<Worker<Task, Queue>>>& workers, SlottedBag<Queue>& idle_workers, std::atomic<size_t >& num_busy_waiters)
127+ {
128+ if (m_started_flag.exchange (true , std::memory_order_acq_rel))
129+ throw std::runtime_error (" The Rouser has already been started."
130+
131+ m_running_flag.store (true , std::memory_order_release);
132+ m_thread = std::thread (&Rouser::threadFunc<Task, Queue>, this , std::ref (workers), std::ref (idle_workers), std::ref (num_busy_waiters));
133+ }
134+
135+ inline void Rouser::stop ()
136+ {
137+ if (m_running_flag.exchange (false , std::memory_order_acq_rel))
138+ m_thread.join ();
139+ }
140+
141+
142+ template <typename Task, template <typename > class Queue >
143+ inline void Rouser::threadFunc (std::vector<std::unique_ptr<Worker<Task, Queue>>>& workers, SlottedBag<Queue>& idle_workers, std::atomic<size_t >& num_busy_waiters)
144+ {
145+ while (m_running_flag.load (std::memory_order_acquire))
146+ {
147+ // Try to wake up a thread if there are no current busy waiters.
148+ if (num_busy_waiters.load (std::memory_order_acquire) == 0 )
149+ {
150+ auto result = idle_workers.tryEmptyAny ();
151+ if (result.first )
152+ workers[result.second ]->wake ();
153+ }
154+
155+ // Sleep.
156+ std::this_thread::sleep_for (m_rouse_period);
157+ }
158+ }
159+
160+ }
0 commit comments