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