TLA Line data Source code
1 : //
2 : // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3 : // Copyright (c) 2026 Michael Vandeberg
4 : //
5 : // Distributed under the Boost Software License, Version 1.0. (See accompanying
6 : // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7 : //
8 : // Official repository: https://github.com/cppalliance/capy
9 : //
10 :
11 : #ifndef BOOST_CAPY_EX_STRAND_HPP
12 : #define BOOST_CAPY_EX_STRAND_HPP
13 :
14 : #include <boost/capy/detail/config.hpp>
15 : #include <boost/capy/continuation.hpp>
16 : #include <coroutine>
17 : #include <boost/capy/ex/detail/strand_service.hpp>
18 :
19 : #include <type_traits>
20 :
21 : namespace boost {
22 : namespace capy {
23 :
24 : /** Provides serialized coroutine execution for any executor type.
25 :
26 : A strand wraps an inner executor and ensures that coroutines
27 : dispatched through it never run concurrently. At most one
28 : coroutine executes at a time within a strand, even when the
29 : underlying executor runs on multiple threads.
30 :
31 : Strands are lightweight handles that can be copied freely.
32 : Copies share the same internal serialization state, so
33 : coroutines dispatched through any copy are serialized with
34 : respect to all other copies.
35 :
36 : @par Invariant
37 : Coroutines resumed through a strand shall not run concurrently.
38 :
39 : @par Implementation
40 : Each strand allocates a private serialization state. Strands
41 : constructed from the same execution context share a small pool
42 : of mutexes (193 entries) selected by hash. Mutex sharing causes
43 : only brief contention on the push/pop critical section, never
44 : cross-strand state sharing. Construction cost: one
45 : `std::make_shared` per strand.
46 :
47 : @par Executor Concept
48 : This class satisfies the `Executor` concept, providing:
49 : - `context()` - Returns the underlying execution context
50 : - `on_work_started()` / `on_work_finished()` - Work tracking
51 : - `dispatch(continuation&)` - May run immediately if already executing in this strand
52 : - `post(continuation&)` - Always queues for later execution
53 :
54 : @par Preconditions
55 : A strand holds only a non-owning reference to its inner executor's
56 : execution context (for example a `thread_pool`). That context must
57 : outlive every post() and dispatch() call; posting or dispatching
58 : concurrently with, or after, the context's destruction is undefined
59 : behavior. To guarantee this, submit work through @ref run_async or
60 : @ref run. Their operations are work-tracked, so the context's
61 : `join()` waits for them. Call `join()` on the context before
62 : destroying it, rather than posting to a strand from an external
63 : thread the context does not track. Destroying the strand handle
64 : itself is always safe, including after the context is
65 : destroyed.
66 :
67 : @par Thread Safety
68 : Distinct objects: Safe.
69 : Shared objects: Safe.
70 :
71 : @par Example
72 : @par !example example
73 :
74 :
75 : @tparam Ex The type of the underlying executor. Must
76 : satisfy the `Executor` concept.
77 :
78 : @see Executor
79 : */
80 : template<typename Ex>
81 : class strand
82 : {
83 : std::shared_ptr<detail::strand_impl> impl_;
84 : Ex ex_;
85 :
86 : friend struct strand_test;
87 :
88 : public:
89 : /** Names the executor type this `strand<Ex>` wraps.
90 : */
91 : using inner_executor_type = Ex;
92 :
93 : /** Construct a strand for the specified executor.
94 :
95 : Allocates a fresh strand implementation from the service
96 : associated with the executor's context.
97 :
98 : @param ex The inner executor to wrap. Coroutines are
99 : ultimately dispatched through this executor.
100 :
101 : @note This constructor is disabled if the argument is a
102 : strand type, to prevent strand-of-strand wrapping.
103 : */
104 : template<typename Ex1,
105 : typename = std::enable_if_t<
106 : !std::is_same_v<std::decay_t<Ex1>, strand> &&
107 : !detail::is_strand<std::decay_t<Ex1>>::value &&
108 : std::is_convertible_v<Ex1, Ex>>>
109 : explicit
110 HIT 11446 : strand(Ex1&& ex)
111 11446 : : impl_(detail::get_strand_service(ex.context())
112 11446 : .create_implementation())
113 11446 : , ex_(std::forward<Ex1>(ex))
114 : {
115 11446 : }
116 :
117 : /** Construct a copy.
118 :
119 : Creates a strand that shares serialization state with
120 : the original. Coroutines dispatched through either strand
121 : are serialized with respect to each other.
122 :
123 : @param other The strand to copy.
124 : */
125 11 : strand(strand const& other) = default;
126 :
127 : /** Construct by moving.
128 :
129 : @param other The strand to move from.
130 :
131 : @note A moved-from strand is only safe to destroy
132 : or reassign.
133 : */
134 11453 : strand(strand&& other) = default;
135 :
136 : /** Assign by copying.
137 :
138 : Shares serialization state with `other`, as the copy
139 : constructor does.
140 :
141 : @param other The strand to copy.
142 :
143 : @return A reference to `*this`.
144 : */
145 1 : strand& operator=(strand const& other) = default;
146 :
147 : /** Assign by moving.
148 :
149 : @param other The strand to move from.
150 :
151 : @return A reference to `*this`.
152 :
153 : @note A moved-from strand is only safe to destroy
154 : or reassign.
155 : */
156 1 : strand& operator=(strand&& other) = default;
157 :
158 : /** Return the underlying executor.
159 :
160 : @return A const reference to the inner executor.
161 : */
162 : Ex const&
163 1 : get_inner_executor() const noexcept
164 : {
165 1 : return ex_;
166 : }
167 :
168 : /** Return the underlying execution context.
169 :
170 : @return A reference to the execution context associated
171 : with the inner executor.
172 : */
173 : auto&
174 6 : context() const noexcept
175 : {
176 6 : return ex_.context();
177 : }
178 :
179 : /** Notify that work has started.
180 :
181 : Delegates to the inner executor's `on_work_started()`. For a
182 : `thread_pool` inner executor, this increments the count that
183 : `join()` blocks on.
184 : */
185 : void
186 7 : on_work_started() const noexcept
187 : {
188 7 : ex_.on_work_started();
189 7 : }
190 :
191 : /** Notify that work has finished.
192 :
193 : Delegates to the inner executor's `on_work_finished()`. For a
194 : `thread_pool` inner executor, this decrements the count that
195 : `join()` blocks on.
196 : */
197 : void
198 7 : on_work_finished() const noexcept
199 : {
200 7 : ex_.on_work_finished();
201 7 : }
202 :
203 : /** Determine whether the strand is running in the current thread.
204 :
205 : @return true if the current thread is executing a coroutine
206 : within this strand's dispatch loop.
207 : */
208 : bool
209 4 : running_in_this_thread() const noexcept
210 : {
211 4 : return detail::strand_service::running_in_this_thread(*impl_);
212 : }
213 :
214 : /** Compare two strands for equality.
215 :
216 : Two strands are equal if they share the same internal
217 : serialization state. Equal strands serialize coroutines
218 : with respect to each other.
219 :
220 : @param other The strand to compare against.
221 : @return true if both strands share the same implementation.
222 : */
223 : bool
224 499505 : operator==(strand const& other) const noexcept
225 : {
226 499505 : return impl_.get() == other.impl_.get();
227 : }
228 :
229 : /** Post a continuation to the strand.
230 :
231 : The continuation is always queued for execution, never resumed
232 : immediately. When the strand becomes available, queued
233 : work executes in FIFO order on the underlying executor.
234 :
235 : @par Ordering
236 : Guarantees strict FIFO ordering relative to other post() calls.
237 : Use this instead of dispatch() when ordering matters.
238 :
239 : @param c The continuation to post. The caller retains
240 : ownership; the continuation must remain valid until
241 : it is dequeued and resumed.
242 :
243 : @par Preconditions
244 : The strand's execution context must outlive this call. Posting
245 : concurrently with, or after, that context's destruction is
246 : undefined behavior.
247 : */
248 : void
249 30336 : post(continuation& c) const
250 : {
251 30336 : detail::strand_service::post(impl_, executor_ref(ex_), c);
252 30336 : }
253 :
254 : /** Dispatch a continuation through the strand.
255 :
256 : Returns a handle for symmetric transfer. If the calling
257 : thread is already executing within this strand, returns `c.h`.
258 : Otherwise, the continuation is queued and
259 : `std::noop_coroutine()` is returned.
260 :
261 : @par Ordering
262 : Callers requiring strict FIFO ordering should use post()
263 : instead, which always queues the continuation.
264 :
265 : @param c The continuation to dispatch. The caller retains
266 : ownership; the continuation must remain valid until
267 : it is dequeued and resumed.
268 :
269 : @return A handle for symmetric transfer or `std::noop_coroutine()`.
270 :
271 : @par Preconditions
272 : The strand's execution context must outlive this call.
273 : Dispatching concurrently with, or after, that context's
274 : destruction is undefined behavior.
275 : */
276 : std::coroutine_handle<>
277 9 : dispatch(continuation& c) const
278 : {
279 9 : return detail::strand_service::dispatch(impl_, executor_ref(ex_), c);
280 : }
281 : };
282 :
283 : /** Deduce the executor type from the constructor argument.
284 :
285 : @tparam Ex The wrapped executor type.
286 : */
287 : template<typename Ex>
288 : strand(Ex) -> strand<Ex>;
289 :
290 : } // namespace capy
291 : } // namespace boost
292 :
293 : #endif
|
