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 4419 : explicit thread_pool(
137 : [[maybe_unused]] capy::execution_context& ctx, unsigned num_threads = 1)
138 4419 : : num_threads_(num_threads)
139 : {
140 4419 : if (!num_threads)
141 1 : throw std::logic_error("thread_pool requires at least 1 thread");
142 4421 : }
143 :
144 : /** Destroy the pool, joining any worker `shutdown()` did not.
145 :
146 : A context shuts down the pool it owns before destroying it, or
147 : at creation if the context's shutdown already began. A pool
148 : constructed directly gets no such call, so this shuts it down
149 : rather than leave a joinable worker to terminate the process.
150 : */
151 8833 : ~thread_pool() override
152 4418 : {
153 4418 : if (!threads_.empty())
154 1 : shutdown();
155 8833 : }
156 :
157 : thread_pool(thread_pool const&) = delete;
158 : thread_pool& operator=(thread_pool const&) = delete;
159 :
160 : /** Enqueue a work item for execution on the thread pool.
161 :
162 : The first item posted starts the workers. Zero-allocation:
163 : the caller owns the work item's storage.
164 :
165 : A refusal answers with the code the caller reports for the
166 : operation it was starting, so that a system that will not give
167 : the pool a thread is not mistaken for a cancellation.
168 :
169 : @par Thread Safety
170 : Safe. Racing first posts start the workers once.
171 :
172 : @param w The work item to execute. Must remain valid until
173 : its `func_` or `destroy_` has been called.
174 :
175 : @return An empty code if the item was enqueued;
176 : `capy::error::canceled` if the pool has already shut
177 : down; otherwise the code of the thread the system
178 : refused, which left the pool with no worker at all.
179 : */
180 : [[nodiscard]] std::error_code post(pool_work_item* w) noexcept;
181 :
182 : /** Return the number of workers the pool has started.
183 :
184 : Zero until the first `post()`, and zero again once
185 : `shutdown()` has joined them.
186 :
187 : @par Thread Safety
188 : Safe.
189 : */
190 6 : unsigned worker_count() noexcept
191 : {
192 6 : std::lock_guard<std::mutex> lock(mutex_);
193 6 : return static_cast<unsigned>(threads_.size());
194 6 : }
195 :
196 : /** Shut down the thread pool.
197 :
198 : Signals all threads to exit once their current item is done,
199 : joins them, then disposes of every item still queued without
200 : running its blocking work where the item says how.
201 : */
202 : void shutdown() override;
203 : };
204 :
205 : inline void
206 735 : thread_pool::worker_loop(unsigned index)
207 : {
208 : // Name format chosen to fit Linux's 15-char pthread limit:
209 : // "tpool-svc-" (10) + up to 4 digit index leaves "tpool-svc-9999".
210 : char name[16];
211 735 : std::snprintf(name, sizeof(name), "tpool-svc-%u", index);
212 735 : capy::set_current_thread_name(name);
213 :
214 : for (;;)
215 : {
216 : pool_work_item* w;
217 : {
218 1997 : std::unique_lock<std::mutex> lock(mutex_);
219 1997 : cv_.wait(
220 2981 : lock, [this] { return shutdown_ || !work_queue_.empty(); });
221 :
222 : // What is still queued is left to shutdown(), which does not
223 : // make blocking calls whose results nobody will read.
224 1997 : if (shutdown_)
225 1470 : return;
226 1262 : w = work_queue_.pop();
227 1997 : }
228 1262 : w->func_(w);
229 1262 : }
230 : }
231 :
232 : // Called with mutex_ held, so the workers are started once however
233 : // many threads race the first post.
234 : inline std::error_code
235 1314 : thread_pool::start_workers() noexcept
236 : {
237 1314 : if (!threads_.empty())
238 578 : return {};
239 736 : std::error_code ec;
240 : try
241 : {
242 736 : threads_.reserve(num_threads_);
243 1470 : for (unsigned i = 0; i < num_threads_; ++i)
244 1473 : threads_.emplace_back([this, i] { worker_loop(i + 1); });
245 : }
246 4 : catch (std::system_error const& e)
247 : {
248 : // The refusal is carried out, not swallowed: a thread the
249 : // system will not give is a real error and the operation that
250 : // asked for it says so, rather than reporting the cancellation
251 : // that belongs to a stop token.
252 2 : ec = e.code();
253 2 : }
254 2 : catch (...)
255 : {
256 2 : ec = std::make_error_code(std::errc::resource_unavailable_try_again);
257 2 : }
258 : // A pool short of workers still runs everything posted to it, only
259 : // less of it at once, so a partial start is a start. What it does
260 : // not do is come back for the rest: the size is a tuning knob, and
261 : // topping it up would put a thread creation on the initiator's
262 : // path for every operation after a refusal.
263 736 : if (!threads_.empty())
264 732 : return {};
265 4 : return ec;
266 : }
267 :
268 : inline std::error_code
269 1336 : thread_pool::post(pool_work_item* w) noexcept
270 : {
271 : {
272 1336 : std::lock_guard<std::mutex> lock(mutex_);
273 1336 : if (shutdown_)
274 22 : return capy::error::canceled;
275 : // The system can refuse a thread, and an initiator has no way
276 : // to throw; a refused post is the failure the callers already
277 : // report through the operation they were starting.
278 1314 : if (auto ec = start_workers())
279 4 : return ec;
280 1310 : work_queue_.push(w);
281 1336 : }
282 1310 : cv_.notify_one();
283 1310 : return {};
284 : }
285 :
286 : inline void
287 4436 : thread_pool::shutdown()
288 : {
289 : {
290 4436 : std::lock_guard<std::mutex> lock(mutex_);
291 4436 : shutdown_ = true;
292 4436 : }
293 4436 : cv_.notify_all();
294 :
295 : // Unlocked, though a post may add to threads_: the flag above is
296 : // published under the same mutex, so a post that has not taken it
297 : // yet will find it set and start nothing, and one already inside
298 : // released the mutex before this thread acquired it.
299 5171 : for (auto& t : threads_)
300 : {
301 735 : if (t.joinable())
302 735 : t.join();
303 : }
304 4436 : threads_.clear();
305 :
306 4436 : intrusive_queue<pool_work_item> leftover;
307 : {
308 4436 : std::lock_guard<std::mutex> lock(mutex_);
309 4436 : leftover.splice(work_queue_);
310 4436 : }
311 : // Unlocked: an abandoned coroutine's unwind can reach post()
312 : // through the objects its frame owned.
313 4484 : while (auto* w = leftover.pop())
314 : {
315 48 : if (w->destroy_)
316 28 : w->destroy_(w);
317 : else
318 20 : w->func_(w);
319 48 : }
320 4436 : }
321 :
322 : /** A reference to the context's shared thread pool, bound on first use.
323 :
324 : Services that hand blocking work to the pool hold one of these
325 : instead of a reference bound at construction. A service made before
326 : the pool, as one made by the scheduler's constructor would be, would
327 : otherwise create the pool itself, older than the scheduler and so
328 : joined too late; binding on first use puts the pool after it.
329 :
330 : The owning `io_context` creates the pool service during
331 : construction, so by the time any operation can run the binding only
332 : ever finds it. That is what keeps `get()` from constructing
333 : anything on an initiator's thread, and so from throwing where an
334 : initiator may not: the throwing spelling exists for a scheduler
335 : driven without an `io_context`. What the service defers is its
336 : workers, and those are started by `post()`, which reports a refusal
337 : rather than throwing it.
338 :
339 : @par Thread Safety
340 : Distinct objects: Safe.
341 : Shared objects: Safe.
342 :
343 : @see thread_pool
344 : */
345 : class thread_pool_ref
346 : {
347 : capy::execution_context& ctx_;
348 : std::atomic<thread_pool*> pool_{nullptr};
349 :
350 : public:
351 : /** Construct a reference into the given context.
352 :
353 : @param ctx The context whose pool is used.
354 : */
355 907 : explicit thread_pool_ref(capy::execution_context& ctx) noexcept : ctx_(ctx)
356 : {
357 907 : }
358 :
359 : thread_pool_ref(thread_pool_ref const&) = delete;
360 : thread_pool_ref& operator=(thread_pool_ref const&) = delete;
361 :
362 : /** Return the pool, creating it if this is the first use.
363 :
364 : @pre For the throwing clauses below to be unreachable, the owning
365 : context must already hold the pool service. Every `io_context`
366 : constructor installs it — what waits for a first post is the
367 : service's workers, not the service — so the creating branch is
368 : reached only by a scheduler driven without one.
369 :
370 : @par Exception Safety
371 : Strong guarantee.
372 :
373 : @throws std::bad_alloc If the service cannot be allocated.
374 :
375 : @throws std::logic_error If the pool is asked for zero threads.
376 :
377 : @return The context's shared thread pool.
378 : */
379 1263 : thread_pool& get()
380 : {
381 1263 : auto* p = pool_.load(std::memory_order_acquire);
382 1263 : if (!p)
383 : {
384 735 : p = &ctx_.use_service<thread_pool>();
385 735 : pool_.store(p, std::memory_order_release);
386 : }
387 1263 : return *p;
388 : }
389 : };
390 :
391 : } // namespace boost::corosio::detail
392 :
393 : #endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
|
