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