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

include/boost/corosio/detail/thread_pool.hpp

92.6% Lines (50/54) 100.0% List of functions (8/8) 65.6% Branches (21/32)
[ ] prev/next file n p prev/next uncovered line
thread_pool.hpp
f(x) Functions (8)
Function Calls Lines Branches Blocks
boost::corosio::detail::pool_work_item::pool_work_item() :51 264x 100.0% – 100.0% boost::corosio::detail::thread_pool::thread_pool(boost::capy::execution_context&, unsigned int) :106 1490x 100.0% 75.0% 100.0% boost::corosio::detail::thread_pool::thread_pool(boost::capy::execution_context&, unsigned int)::'lambda'()::operator()() const :115 747x 50.0% 25.0% 100.0% boost::corosio::detail::thread_pool::~thread_pool() :124 2231x 100.0% – 100.0% boost::corosio::detail::thread_pool::worker_loop(unsigned int) :150 747x 100.0% 50.0% 69.0% boost::corosio::detail::thread_pool::worker_loop(unsigned int)::'lambda'()::operator()() const :164 1589x 88.9% 83.3% 100.0% boost::corosio::detail::thread_pool::post(boost::corosio::detail::pool_work_item*) :179 182x 100.0% 75.0% 77.0% boost::corosio::detail::thread_pool::shutdown() :192 748x 100.0% 66.7% 90.0%
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 264x 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 264x 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 745x 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 2235x explicit thread_pool(capy::execution_context& ctx, unsigned num_threads = 1)
107 1490x {
108 745x (void)ctx;
109
2/2
✓ Branch 0 taken 744 times.
✓ Branch 1 taken 1 time.
 745x if (!num_threads)
110
1/2
✓ Branch 0 taken 1 time.
✗ Branch 1 not taken.
 1x throw std::logic_error("thread_pool requires at least 1 thread");
111
1/2
✓ Branch 0 taken 744 times.
✗ Branch 1 not taken.
 744x threads_.reserve(num_threads);
112 try
113 {
114
2/2
✓ Branch 0 taken 744 times.
✓ Branch 1 taken 747 times.
 1491x for (unsigned i = 0; i < num_threads; ++i)
115
1/2
✓ Branch 0 taken 747 times.
✗ Branch 1 not taken.
 1494x threads_.emplace_back([this, i] { worker_loop(i + 1); });
116 744x }
117 catch (...)
118 {
119 ✗ shutdown();
120 ✗ throw;
121 ✗ }
122 1490x }
123
124 2231x ~thread_pool() override = default;
125
126 thread_pool(thread_pool const&) = delete;
127 thread_pool& operator=(thread_pool const&) = delete;
128
129 /** Enqueue a work item for execution on the thread pool.
130
131 Zero-allocation: the caller owns the work item's storage.
132
133 @param w The work item to execute. Must remain valid until
134 its `func_` has been called.
135
136 @return `true` if the item was enqueued, `false` if the
137 pool has already shut down.
138 */
139 bool post(pool_work_item* w) noexcept;
140
141 /** Shut down the thread pool.
142
143 Signals all threads to exit after draining any
144 remaining queued work, then joins them.
145 */
146 void shutdown() override;
147 };
148
149 inline void
150 747x thread_pool::worker_loop(unsigned index)
151 {
152 // Name format chosen to fit Linux's 15-char pthread limit:
153 // "tpool-svc-" (10) + up to 4 digit index leaves "tpool-svc-9999".
154 char name[16];
155 747x std::snprintf(name, sizeof(name), "tpool-svc-%u", index);
156 747x capy::set_current_thread_name(name);
157
158 928x for (;;)
159 {
160 pool_work_item* w;
161 {
162 928x std::unique_lock<std::mutex> lock(mutex_);
163
2/4
✓ Branch 0 taken 928 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 928 times.
✗ Branch 3 not taken.
 1856x cv_.wait(
164
2/2
✓ Branch 0 taken 761 times.
✓ Branch 1 taken 828 times.
 2517x lock, [this] { return shutdown_ || !work_queue_.empty(); });
165
166 928x w = work_queue_.pop();
167
2/2
✓ Branch 0 taken 181 times.
✓ Branch 1 taken 747 times.
 928x if (!w)
168 {
169
1/2
✓ Branch 0 taken 747 times.
✗ Branch 1 not taken.
 747x if (shutdown_)
170 747x return;
171 ✗ continue;
172 }
173 928x }
174 181x w->func_(w);
175 }
176 747x }
177
178 inline bool
179 182x thread_pool::post(pool_work_item* w) noexcept
180 {
181 {
182
1/2
✓ Branch 0 taken 182 times.
✗ Branch 1 not taken.
 182x std::lock_guard<std::mutex> lock(mutex_);
183
2/2
✓ Branch 0 taken 1 time.
✓ Branch 1 taken 181 times.
 182x if (shutdown_)
184 1x return false;
185 181x work_queue_.push(w);
186 182x }
187 181x cv_.notify_one();
188 181x return true;
189 182x }
190
191 inline void
192 748x thread_pool::shutdown()
193 {
194 {
195 748x std::lock_guard<std::mutex> lock(mutex_);
196 748x shutdown_ = true;
197 748x }
198 748x cv_.notify_all();
199
200
2/2
✓ Branch 0 taken 747 times.
✓ Branch 1 taken 748 times.
 1495x for (auto& t : threads_)
201 {
202
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 747 times.
 747x if (t.joinable())
203 747x t.join();
204 }
205 748x threads_.clear();
206
207 {
208 748x std::lock_guard<std::mutex> lock(mutex_);
209
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 748 times.
 748x while (work_queue_.pop())
210 ;
211 748x }
212 748x }
213
214 } // namespace boost::corosio::detail
215
216 #endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
217