Index/include / boost / corosio / detail / thread_pool.hpp

include/boost/corosio/detail/thread_pool.hpp

89.6% Lines (43/48) 100.0% List of functions (7/7) 86.7% Branches (26/30)
[ ] prev/next file n p prev/next uncovered line
thread_pool.hpp
f(x) Functions (7)
Function Calls Lines Branches Blocks
boost::corosio::detail::thread_pool::thread_pool(boost::capy::execution_context&, unsigned int) :106 1123x 63.6% 85.7% 71.9% boost::corosio::detail::thread_pool::thread_pool(boost::capy::execution_context&, unsigned int)::{lambda()#1}::operator()() const :116 1125x 100.0% 100.0% 100.0% boost::corosio::detail::thread_pool::~thread_pool() :125 2243x 100.0% – 100.0% boost::corosio::detail::thread_pool::worker_loop(unsigned int) :151 1125x 92.3% 85.7% 77.3% boost::corosio::detail::thread_pool::worker_loop(unsigned int)::{lambda()#1}::operator()() const :165 2288x 100.0% 100.0% 100.0% boost::corosio::detail::thread_pool::post(boost::corosio::detail::pool_work_item*) :180 29x 100.0% 100.0% 100.0% boost::corosio::detail::thread_pool::shutdown() :193 1126x 100.0% 77.8% 90.9%
Line Branch TLA Hits Source Code
1 //
2 // Copyright (c) 2026 Steve Gerbino
3 //
4 // Distributed under the Boost Software License, Version 1.0. (See accompanying
5 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6 //
7 // Official repository: https://github.com/cppalliance/corosio
8 //
9
10 #ifndef BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
11 #define BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
12
13 #include <boost/corosio/detail/config.hpp>
14 #include <boost/corosio/detail/intrusive.hpp>
15 #include <boost/capy/ex/execution_context.hpp>
16 #include <boost/capy/test/thread_name.hpp>
17
18 #include <condition_variable>
19 #include <cstdio>
20 #include <mutex>
21 #include <stdexcept>
22 #include <thread>
23 #include <vector>
24
25 namespace boost::corosio::detail {
26
27 /** Base class for thread pool work items.
28
29 Derive from this to create work that can be posted to a
30 @ref thread_pool. Uses static function pointer dispatch,
31 consistent with the IOCP `op` pattern.
32
33 @par Example
34 @code
35 struct my_work : pool_work_item
36 {
37 int* result;
38 static void execute( pool_work_item* w ) noexcept
39 {
40 auto* self = static_cast<my_work*>( w );
41 *self->result = 42;
42 }
43 };
44
45 my_work w;
46 w.func_ = &my_work::execute;
47 w.result = &r;
48 pool.post( &w );
49 @endcode
50 */
51 struct pool_work_item : intrusive_queue<pool_work_item>::node
52 {
53 /// Static dispatch function signature.
54 using func_type = void (*)(pool_work_item*) noexcept;
55
56 /// Completion handler invoked by the worker thread.
57 func_type func_ = nullptr;
58 };
59
60 /** Shared thread pool for dispatching blocking operations.
61
62 Provides a fixed pool of reusable worker threads for operations
63 that cannot be integrated with async I/O (e.g. blocking DNS
64 calls). Registered as an `execution_context::service` so it
65 is a singleton per io_context.
66
67 Threads are created eagerly in the constructor. The default
68 thread count is 1.
69
70 @par Thread Safety
71 All public member functions are thread-safe.
72
73 @par Shutdown
74 Sets a shutdown flag, notifies all threads, and joins them.
75 In-flight blocking calls complete naturally before the thread
76 exits.
77 */
78 class thread_pool final : public capy::execution_context::service
79 {
80 std::mutex mutex_;
81 std::condition_variable cv_;
82 intrusive_queue<pool_work_item> work_queue_;
83 std::vector<std::thread> threads_;
84 bool shutdown_ = false;
85
86 void worker_loop(unsigned index);
87
88 public:
89 using key_type = thread_pool;
90
91 /** Construct the thread pool service.
92
93 Eagerly creates all worker threads.
94
95 @par Exception Safety
96 Strong guarantee. If thread creation fails, all
97 already-created threads are shut down and joined
98 before the exception propagates.
99
100 @param ctx Reference to the owning execution_context.
101 @param num_threads Number of worker threads. Must be
102 at least 1.
103
104 @throws std::logic_error If `num_threads` is 0.
105 */
106 1123x explicit thread_pool(
107 [[maybe_unused]] capy::execution_context& ctx,
108 unsigned num_threads = 1)
109 1123x {
110
2/2
✓ Branch 7 → 8 taken 1 time.
✓ Branch 7 → 11 taken 1122 times.
 1123x if (!num_threads)
111
1/1
✓ Branch 9 → 10 taken 1 time.
 1x throw std::logic_error("thread_pool requires at least 1 thread");
112
1/1
✓ Branch 11 → 12 taken 1122 times.
 1122x threads_.reserve(num_threads);
113 try
114 {
115
2/2
✓ Branch 15 → 13 taken 1125 times.
✓ Branch 15 → 16 taken 1122 times.
 2247x for (unsigned i = 0; i < num_threads; ++i)
116
1/1
✓ Branch 13 → 14 taken 1125 times.
 2250x threads_.emplace_back([this, i] { worker_loop(i + 1); });
117 }
118 ✗ catch (...)
119 {
120 ✗ shutdown();
121 ✗ throw;
122 ✗ }
123 1126x }
124
125 2243x ~thread_pool() override = default;
126
127 thread_pool(thread_pool const&) = delete;
128 thread_pool& operator=(thread_pool const&) = delete;
129
130 /** Enqueue a work item for execution on the thread pool.
131
132 Zero-allocation: the caller owns the work item's storage.
133
134 @param w The work item to execute. Must remain valid until
135 its `func_` has been called.
136
137 @return `true` if the item was enqueued, `false` if the
138 pool has already shut down.
139 */
140 bool post(pool_work_item* w) noexcept;
141
142 /** Shut down the thread pool.
143
144 Signals all threads to exit after draining any
145 remaining queued work, then joins them.
146 */
147 void shutdown() override;
148 };
149
150 inline void
151 1125x thread_pool::worker_loop(unsigned index)
152 {
153 // Name format chosen to fit Linux's 15-char pthread limit:
154 // "tpool-svc-" (10) + up to 4 digit index leaves "tpool-svc-9999".
155 char name[16];
156
1/1
✓ Branch 2 → 3 taken 1125 times.
 1125x std::snprintf(name, sizeof(name), "tpool-svc-%u", index);
157 1125x capy::set_current_thread_name(name);
158
159 for (;;)
160 {
161 pool_work_item* w;
162 {
163
1/1
✓ Branch 4 → 5 taken 1153 times.
 1153x std::unique_lock<std::mutex> lock(mutex_);
164
1/1
✓ Branch 5 → 6 taken 1153 times.
 1153x cv_.wait(
165
4/4
✓ Branch 2 → 3 taken 1149 times.
✓ Branch 2 → 5 taken 1139 times.
✓ Branch 4 → 5 taken 14 times.
✓ Branch 4 → 6 taken 1135 times.
 2288x lock, [this] { return shutdown_ || !work_queue_.empty(); });
166
167 1153x w = work_queue_.pop();
168
2/2
✓ Branch 7 → 8 taken 1125 times.
✓ Branch 7 → 11 taken 28 times.
 1153x if (!w)
169 {
170
1/2
✓ Branch 8 → 9 taken 1125 times.
✗ Branch 8 → 10 not taken.
 1125x if (shutdown_)
171 2250x return;
172 ✗ continue;
173 }
174 1153x }
175 28x w->func_(w);
176 28x }
177 }
178
179 inline bool
180 29x thread_pool::post(pool_work_item* w) noexcept
181 {
182 {
183 29x std::lock_guard<std::mutex> lock(mutex_);
184
2/2
✓ Branch 3 → 4 taken 1 time.
✓ Branch 3 → 5 taken 28 times.
 29x if (shutdown_)
185 1x return false;
186 28x work_queue_.push(w);
187 29x }
188 28x cv_.notify_one();
189 28x return true;
190 }
191
192 inline void
193 1126x thread_pool::shutdown()
194 {
195 {
196
1/1
✓ Branch 2 → 3 taken 1126 times.
 1126x std::lock_guard<std::mutex> lock(mutex_);
197 1126x shutdown_ = true;
198 1126x }
199 1126x cv_.notify_all();
200
201
2/2
✓ Branch 13 → 7 taken 1125 times.
✓ Branch 13 → 14 taken 1126 times.
 2251x for (auto& t : threads_)
202 {
203
1/2
✓ Branch 9 → 10 taken 1125 times.
✗ Branch 9 → 11 not taken.
 1125x if (t.joinable())
204
1/1
✓ Branch 10 → 11 taken 1125 times.
 1125x t.join();
205 }
206 1126x threads_.clear();
207
208 {
209
1/1
✓ Branch 15 → 16 taken 1126 times.
 1126x std::lock_guard<std::mutex> lock(mutex_);
210
1/2
✗ Branch 18 → 17 not taken.
✓ Branch 18 → 19 taken 1126 times.
 1126x while (work_queue_.pop())
211 ;
212 1126x }
213 1126x }
214
215 } // namespace boost::corosio::detail
216
217 #endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
218