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)
[ ] prev/next file n p prev/next uncovered line
thread_pool.hpp
f(x) Functions (7)
Function Calls Lines Blocks
boost::corosio::detail::thread_pool::thread_pool(boost::capy::execution_context&, unsigned int) :106 607x 100.0% 70.0% boost::corosio::detail::thread_pool::thread_pool(boost::capy::execution_context&, unsigned int)::{lambda()#1}::operator()() const :115 609x 33.3% 100.0% boost::corosio::detail::thread_pool::~thread_pool() :124 1211x 100.0% 100.0% boost::corosio::detail::thread_pool::worker_loop(unsigned int) :150 609x 100.0% 76.0% boost::corosio::detail::thread_pool::worker_loop(unsigned int)::{lambda()#1}::operator()() const :164 1330x 88.9% 100.0% boost::corosio::detail::thread_pool::post(boost::corosio::detail::pool_work_item*) :179 180x 100.0% 100.0% boost::corosio::detail::thread_pool::shutdown() :192 610x 100.0% 87.0%
Line 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 607x explicit thread_pool(capy::execution_context& ctx, unsigned num_threads = 1)
107 607x {
108 (void)ctx;
109 607x if (!num_threads)
110 1x throw std::logic_error("thread_pool requires at least 1 thread");
111 606x threads_.reserve(num_threads);
112 try
113 {
114 1215x for (unsigned i = 0; i < num_threads; ++i)
115 1218x threads_.emplace_back([this, i] { worker_loop(i + 1); });
116 }
117 ✗ catch (...)
118 {
119 ✗ shutdown();
120 ✗ throw;
121 ✗ }
122 609x }
123
124 1211x ~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 609x 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 609x std::snprintf(name, sizeof(name), "tpool-svc-%u", index);
156 609x capy::set_current_thread_name(name);
157
158 for (;;)
159 {
160 pool_work_item* w;
161 {
162 788x std::unique_lock<std::mutex> lock(mutex_);
163 788x cv_.wait(
164 1330x lock, [this] { return shutdown_ || !work_queue_.empty(); });
165
166 788x w = work_queue_.pop();
167 788x if (!w)
168 {
169 609x if (shutdown_)
170 1218x return;
171 ✗ continue;
172 }
173 788x }
174 179x w->func_(w);
175 179x }
176 }
177
178 inline bool
179 180x thread_pool::post(pool_work_item* w) noexcept
180 {
181 {
182 180x std::lock_guard<std::mutex> lock(mutex_);
183 180x if (shutdown_)
184 1x return false;
185 179x work_queue_.push(w);
186 180x }
187 179x cv_.notify_one();
188 179x return true;
189 }
190
191 inline void
192 610x thread_pool::shutdown()
193 {
194 {
195 610x std::lock_guard<std::mutex> lock(mutex_);
196 610x shutdown_ = true;
197 610x }
198 610x cv_.notify_all();
199
200 1219x for (auto& t : threads_)
201 {
202 609x if (t.joinable())
203 609x t.join();
204 }
205 610x threads_.clear();
206
207 {
208 610x std::lock_guard<std::mutex> lock(mutex_);
209 610x while (work_queue_.pop())
210 ;
211 610x }
212 610x }
213
214 } // namespace boost::corosio::detail
215
216 #endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
217