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/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 HIT 1605 : explicit thread_pool(
107 : [[maybe_unused]] capy::execution_context& ctx,
108 : unsigned num_threads = 1)
109 1605 : {
110 1605 : if (!num_threads)
111 1 : throw std::logic_error("thread_pool requires at least 1 thread");
112 1604 : threads_.reserve(num_threads);
113 : try
114 : {
115 3217 : for (unsigned i = 0; i < num_threads; ++i)
116 3226 : threads_.emplace_back([this, i] { worker_loop(i + 1); });
117 : }
118 MIS 0 : catch (...)
119 : {
120 0 : shutdown();
121 0 : throw;
122 0 : }
123 HIT 1607 : }
124 :
125 3207 : ~thread_pool() override = default;
126 :
127 : thread_pool(thread_pool const&) = delete;
128 : thread_pool& operator=(thread_pool const&) = delete;
129 :
130 : /** Enqueue a work item for execution on the thread pool.
131 :
132 : Zero-allocation: the caller owns the work item's storage.
133 :
134 : @param w The work item to execute. Must remain valid until
135 : its `func_` has been called.
136 :
137 : @return `true` if the item was enqueued, `false` if the
138 : pool has already shut down.
139 : */
140 : bool post(pool_work_item* w) noexcept;
141 :
142 : /** Shut down the thread pool.
143 :
144 : Signals all threads to exit after draining any
145 : remaining queued work, then joins them.
146 : */
147 : void shutdown() override;
148 : };
149 :
150 : inline void
151 1613 : thread_pool::worker_loop(unsigned index)
152 : {
153 : // Name format chosen to fit Linux's 15-char pthread limit:
154 : // "tpool-svc-" (10) + up to 4 digit index leaves "tpool-svc-9999".
155 : char name[16];
156 1613 : std::snprintf(name, sizeof(name), "tpool-svc-%u", index);
157 1613 : capy::set_current_thread_name(name);
158 :
159 : for (;;)
160 : {
161 : pool_work_item* w;
162 : {
163 2005 : std::unique_lock<std::mutex> lock(mutex_);
164 2005 : cv_.wait(
165 3760 : lock, [this] { return shutdown_ || !work_queue_.empty(); });
166 :
167 2005 : w = work_queue_.pop();
168 2005 : if (!w)
169 : {
170 1613 : if (shutdown_)
171 3226 : return;
172 MIS 0 : continue;
173 : }
174 HIT 2005 : }
175 392 : w->func_(w);
176 392 : }
177 : }
178 :
179 : inline bool
180 393 : thread_pool::post(pool_work_item* w) noexcept
181 : {
182 : {
183 393 : std::lock_guard<std::mutex> lock(mutex_);
184 393 : if (shutdown_)
185 1 : return false;
186 392 : work_queue_.push(w);
187 393 : }
188 392 : cv_.notify_one();
189 392 : return true;
190 : }
191 :
192 : inline void
193 1608 : thread_pool::shutdown()
194 : {
195 : {
196 1608 : std::lock_guard<std::mutex> lock(mutex_);
197 1608 : shutdown_ = true;
198 1608 : }
199 1608 : cv_.notify_all();
200 :
201 3221 : for (auto& t : threads_)
202 : {
203 1613 : if (t.joinable())
204 1613 : t.join();
205 : }
206 1608 : threads_.clear();
207 :
208 : {
209 1608 : std::lock_guard<std::mutex> lock(mutex_);
210 1608 : while (work_queue_.pop())
211 : ;
212 1608 : }
213 1608 : }
214 :
215 : } // namespace boost::corosio::detail
216 :
217 : #endif // BOOST_COROSIO_DETAIL_THREAD_POOL_HPP
|
