TLA Line data 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/error.hpp>
16 : #include <boost/capy/ex/execution_context.hpp>
17 : #include <boost/capy/test/thread_name.hpp>
18 :
19 : #include <atomic>
20 : #include <condition_variable>
21 : #include <cstdio>
22 : #include <mutex>
23 : #include <stdexcept>
24 : #include <system_error>
25 : #include <thread>
26 : #include <vector>
27 :
28 : namespace boost::corosio::detail {
29 :
30 : /** Base class for thread pool work items.
31 :
32 : Derive from this to create work that can be posted to a
33 : @ref thread_pool. Uses static function pointer dispatch,
34 : consistent with the IOCP `op` pattern.
35 :
36 : @par Example
37 : @code
38 : struct my_work : pool_work_item
39 : {
40 : int* result;
41 : static void execute( pool_work_item* w ) noexcept
42 : {
43 : auto* self = static_cast<my_work*>( w );
44 : *self->result = 42;
45 : }
46 : };
47 :
48 : my_work w;
49 : w.func_ = &my_work::execute;
50 : w.result = &r;
51 : auto ec = pool.post( &w );
52 : @endcode
53 : */
54 : struct pool_work_item : intrusive_queue<pool_work_item>::node
55 : {
56 : /// Static dispatch function signature.
57 : using func_type = void (*)(pool_work_item*) noexcept;
58 :
59 : /// Completion handler invoked by the worker thread.
60 : func_type func_ = nullptr;
61 : };
62 :
63 : /** Shared thread pool for dispatching blocking operations.
64 :
65 : Provides a fixed pool of reusable worker threads for operations
66 : that cannot be integrated with async I/O (e.g. blocking DNS
67 : calls). Registered as an `execution_context::service` so it
68 : is a singleton per io_context.
69 :
70 : The service is created with its context, but the workers start on
71 : the first `post()`: a context that never opens a file and never
72 : resolves a name never pays for a thread. The default thread count
73 : is 1.
74 :
75 : @par Thread Safety
76 : All public member functions are thread-safe.
77 :
78 : @par Shutdown
79 : Sets a shutdown flag, notifies all threads, and joins them.
80 : In-flight blocking calls complete naturally before the thread
81 : exits.
82 :
83 : @note Create this service after the scheduler its work items post
84 : completions to. Services shut down newest first, so a pool created
85 : earlier joins its workers only after the scheduler has drained its
86 : completion queue, and the completion the last worker posts is then
87 : neither run nor destroyed.
88 :
89 : @note The type is symbol-visible because services are keyed by type
90 : identity: with RTTI, hidden behind a shared library boundary, a
91 : module that asks for the pool would look up, and create, one of its
92 : own (the no-RTTI key is a template static whose visibility follows
93 : the template it is instantiated from).
94 : */
95 : class BOOST_COROSIO_SYMBOL_VISIBLE thread_pool final
96 : : public capy::execution_context::service
97 : {
98 : std::mutex mutex_;
99 : std::condition_variable cv_;
100 : intrusive_queue<pool_work_item> work_queue_;
101 : std::vector<std::thread> threads_;
102 : unsigned num_threads_;
103 : bool shutdown_ = false;
104 :
105 : void worker_loop(unsigned index);
106 : std::error_code start_workers() noexcept;
107 :
108 : public:
109 : using key_type = thread_pool;
110 :
111 : /** Construct the thread pool service.
112 :
113 : Records the worker count. The workers themselves start on the
114 : first `post()`.
115 :
116 : @par Exception Safety
117 : Strong guarantee.
118 :
119 : @param ctx Reference to the owning execution_context.
120 : @param num_threads Number of worker threads. Must be
121 : at least 1.
122 :
123 : @throws std::logic_error If `num_threads` is 0.
124 : */
125 HIT 2109 : explicit thread_pool(
126 : [[maybe_unused]] capy::execution_context& ctx, unsigned num_threads = 1)
127 2109 : : num_threads_(num_threads)
128 : {
129 2109 : if (!num_threads)
130 1 : throw std::logic_error("thread_pool requires at least 1 thread");
131 2111 : }
132 :
133 : /** Destroy the pool, joining any worker `shutdown()` never reached.
134 :
135 : The context's shutdown walk is the normal path; this only
136 : catches a pool created after that walk, whose `shutdown()` is
137 : therefore never called and whose joinable threads would
138 : otherwise terminate the process. A pool that was never posted
139 : to holds no thread and needs neither.
140 : */
141 4214 : ~thread_pool() override
142 2108 : {
143 2108 : if (!threads_.empty())
144 1 : shutdown();
145 4214 : }
146 :
147 : thread_pool(thread_pool const&) = delete;
148 : thread_pool& operator=(thread_pool const&) = delete;
149 :
150 : /** Enqueue a work item for execution on the thread pool.
151 :
152 : The first item posted starts the workers. Zero-allocation:
153 : the caller owns the work item's storage.
154 :
155 : A refusal answers with the code the caller reports for the
156 : operation it was starting, so that a system that will not give
157 : the pool a thread is not mistaken for a cancellation.
158 :
159 : @par Thread Safety
160 : Safe. Racing first posts start the workers once.
161 :
162 : @param w The work item to execute. Must remain valid until
163 : its `func_` has been called.
164 :
165 : @return An empty code if the item was enqueued;
166 : `capy::error::canceled` if the pool has already shut
167 : down; otherwise the code of the thread the system
168 : refused, which left the pool with no worker at all.
169 : */
170 : [[nodiscard]] std::error_code post(pool_work_item* w) noexcept;
171 :
172 : /** Return the number of workers the pool has started.
173 :
174 : Zero until the first `post()`, and zero again once
175 : `shutdown()` has joined them.
176 :
177 : @par Thread Safety
178 : Safe.
179 : */
180 6 : unsigned worker_count() noexcept
181 : {
182 6 : std::lock_guard<std::mutex> lock(mutex_);
183 6 : return static_cast<unsigned>(threads_.size());
184 6 : }
185 :
186 : /** Shut down the thread pool.
187 :
188 : Signals all threads to exit after draining any
189 : remaining queued work, then joins them.
190 : */
191 : void shutdown() override;
192 : };
193 :
194 : inline void
195 185 : thread_pool::worker_loop(unsigned index)
196 : {
197 : // Name format chosen to fit Linux's 15-char pthread limit:
198 : // "tpool-svc-" (10) + up to 4 digit index leaves "tpool-svc-9999".
199 : char name[16];
200 185 : std::snprintf(name, sizeof(name), "tpool-svc-%u", index);
201 185 : capy::set_current_thread_name(name);
202 :
203 : for (;;)
204 : {
205 : pool_work_item* w;
206 : {
207 697 : std::unique_lock<std::mutex> lock(mutex_);
208 697 : cv_.wait(
209 916 : lock, [this] { return shutdown_ || !work_queue_.empty(); });
210 :
211 697 : w = work_queue_.pop();
212 697 : if (!w)
213 : {
214 185 : if (shutdown_)
215 370 : return;
216 MIS 0 : continue;
217 : }
218 HIT 697 : }
219 512 : w->func_(w);
220 512 : }
221 : }
222 :
223 : // Called with mutex_ held, so the workers are started once however
224 : // many threads race the first post.
225 : inline std::error_code
226 516 : thread_pool::start_workers() noexcept
227 : {
228 516 : if (!threads_.empty())
229 330 : return {};
230 186 : std::error_code ec;
231 : try
232 : {
233 186 : threads_.reserve(num_threads_);
234 370 : for (unsigned i = 0; i < num_threads_; ++i)
235 373 : threads_.emplace_back([this, i] { worker_loop(i + 1); });
236 : }
237 4 : catch (std::system_error const& e)
238 : {
239 : // The refusal is carried out, not swallowed: a thread the
240 : // system will not give is a real error and the operation that
241 : // asked for it says so, rather than reporting the cancellation
242 : // that belongs to a stop token.
243 2 : ec = e.code();
244 2 : }
245 2 : catch (...)
246 : {
247 2 : ec = std::make_error_code(std::errc::resource_unavailable_try_again);
248 2 : }
249 : // A pool short of workers still runs everything posted to it, only
250 : // less of it at once, so a partial start is a start. What it does
251 : // not do is come back for the rest: the size is a tuning knob, and
252 : // topping it up would put a thread creation on the initiator's
253 : // path for every operation after a refusal.
254 186 : if (!threads_.empty())
255 182 : return {};
256 4 : return ec;
257 : }
258 :
259 : inline std::error_code
260 527 : thread_pool::post(pool_work_item* w) noexcept
261 : {
262 : {
263 527 : std::lock_guard<std::mutex> lock(mutex_);
264 527 : if (shutdown_)
265 11 : return capy::error::canceled;
266 : // The system can refuse a thread, and an initiator has no way
267 : // to throw; a refused post is the failure the callers already
268 : // report through the operation they were starting.
269 516 : if (auto ec = start_workers())
270 4 : return ec;
271 512 : work_queue_.push(w);
272 527 : }
273 512 : cv_.notify_one();
274 512 : return {};
275 : }
276 :
277 : inline void
278 2118 : thread_pool::shutdown()
279 : {
280 : {
281 2118 : std::lock_guard<std::mutex> lock(mutex_);
282 2118 : shutdown_ = true;
283 2118 : }
284 2118 : cv_.notify_all();
285 :
286 : // Unlocked, though a post may add to threads_: the flag above is
287 : // published under the same mutex, so a post that has not taken it
288 : // yet will find it set and start nothing, and one already inside
289 : // released the mutex before this thread acquired it.
290 2303 : for (auto& t : threads_)
291 : {
292 185 : if (t.joinable())
293 185 : t.join();
294 : }
295 2118 : threads_.clear();
296 :
297 : {
298 2118 : std::lock_guard<std::mutex> lock(mutex_);
299 2118 : while (work_queue_.pop())
300 : ;
301 2118 : }
302 2118 : }
303 :
304 : /** A reference to the context's shared thread pool, bound on first use.
305 :
306 : Services that hand blocking work to the pool hold one of these
307 : instead of a reference bound at construction. They are constructed
308 : from the scheduler's constructor, where the pool they created would
309 : be older than the scheduler and would join too late; binding on
310 : first use puts the pool after it instead.
311 :
312 : The owning `io_context` creates the pool service during
313 : construction, so by the time any operation can run the binding only
314 : ever finds it. That is what keeps `get()` from constructing
315 : anything on an initiator's thread, and so from throwing where an
316 : initiator may not: the throwing spelling exists for a scheduler
317 : driven without an `io_context`. What the service defers is its
318 : workers, and those are started by `post()`, which reports a refusal
319 : rather than throwing it.
320 :
321 : @par Thread Safety
322 : Distinct objects: Safe.
323 : Shared objects: Safe.
324 :
325 : @see thread_pool
326 : */
327 : class thread_pool_ref
328 : {
329 : capy::execution_context& ctx_;
330 : std::atomic<thread_pool*> pool_{nullptr};
331 :
332 : public:
333 : /** Construct a reference into the given context.
334 :
335 : @param ctx The context whose pool is used.
336 : */
337 6318 : explicit thread_pool_ref(capy::execution_context& ctx) noexcept : ctx_(ctx)
338 : {
339 6318 : }
340 :
341 : thread_pool_ref(thread_pool_ref const&) = delete;
342 : thread_pool_ref& operator=(thread_pool_ref const&) = delete;
343 :
344 : /** Return the pool, creating it if this is the first use.
345 :
346 : @par Preconditions
347 : For the throwing clauses below to be unreachable, the owning
348 : context must already hold the pool service. Every `io_context`
349 : constructor installs it — what waits for a first post is the
350 : service's workers, not the service — so the creating branch is
351 : reached only by a scheduler driven without one.
352 :
353 : @par Exception Safety
354 : Strong guarantee.
355 :
356 : @throws std::bad_alloc If the service cannot be allocated.
357 :
358 : @throws std::logic_error If the pool is asked for zero threads.
359 :
360 : @return The context's shared thread pool.
361 : */
362 503 : thread_pool& get()
363 : {
364 503 : auto* p = pool_.load(std::memory_order_acquire);
365 503 : if (!p)
366 : {
367 186 : p = &ctx_.use_service<thread_pool>();
368 186 : pool_.store(p, std::memory_order_release);
369 : }
370 503 : return *p;
371 : }
372 : };
373 :
374 : } // namespace boost::corosio::detail
375 :
376 : #endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
|
