. /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 + @par !example example
73 - thread_pool pool(4);  
74 - strand strand(pool.get_executor()); // CTAD deduces the executor type  
75 - // Continuations are linked intrusively into the strand's queue,  
76 - // so each one must outlive its time there. Storage is typically  
77 - // owned by the awaitable or operation state that posted it.  
78 - continuation c1{h1}, c2{h2}, c3{h3};  
79 - strand.post(c1);  
80 - strand.post(c2);  
81 - strand.post(c3);  
82 - @endcode  
83   73  
84   74  
85   @tparam Ex The type of the underlying executor. Must 75   @tparam Ex The type of the underlying executor. Must
86   satisfy the `Executor` concept. 76   satisfy the `Executor` concept.
87   77  
88   @see Executor 78   @see Executor
89   */ 79   */
90   template<typename Ex> 80   template<typename Ex>
91   class strand 81   class strand
92   { 82   {
93   std::shared_ptr<detail::strand_impl> impl_; 83   std::shared_ptr<detail::strand_impl> impl_;
94   Ex ex_; 84   Ex ex_;
95   85  
96   friend struct strand_test; 86   friend struct strand_test;
97   87  
98   public: 88   public:
99   /** Names the executor type this `strand<Ex>` wraps. 89   /** Names the executor type this `strand<Ex>` wraps.
100   */ 90   */
101   using inner_executor_type = Ex; 91   using inner_executor_type = Ex;
102   92  
103   /** Construct a strand for the specified executor. 93   /** Construct a strand for the specified executor.
104   94  
105   Allocates a fresh strand implementation from the service 95   Allocates a fresh strand implementation from the service
106   associated with the executor's context. 96   associated with the executor's context.
107   97  
108   @param ex The inner executor to wrap. Coroutines are 98   @param ex The inner executor to wrap. Coroutines are
109   ultimately dispatched through this executor. 99   ultimately dispatched through this executor.
110   100  
111   @note This constructor is disabled if the argument is a 101   @note This constructor is disabled if the argument is a
112   strand type, to prevent strand-of-strand wrapping. 102   strand type, to prevent strand-of-strand wrapping.
113   */ 103   */
114   template<typename Ex1, 104   template<typename Ex1,
115   typename = std::enable_if_t< 105   typename = std::enable_if_t<
116   !std::is_same_v<std::decay_t<Ex1>, strand> && 106   !std::is_same_v<std::decay_t<Ex1>, strand> &&
117   !detail::is_strand<std::decay_t<Ex1>>::value && 107   !detail::is_strand<std::decay_t<Ex1>>::value &&
118   std::is_convertible_v<Ex1, Ex>>> 108   std::is_convertible_v<Ex1, Ex>>>
119   explicit 109   explicit
HITCBC 120   11446 strand(Ex1&& ex) 110   11446 strand(Ex1&& ex)
HITCBC 121   11446 : impl_(detail::get_strand_service(ex.context()) 111   11446 : impl_(detail::get_strand_service(ex.context())
HITCBC 122   11446 .create_implementation()) 112   11446 .create_implementation())
HITCBC 123   11446 , ex_(std::forward<Ex1>(ex)) 113   11446 , ex_(std::forward<Ex1>(ex))
124   { 114   {
HITCBC 125   11446 } 115   11446 }
126   116  
127   /** Construct a copy. 117   /** Construct a copy.
128   118  
129   Creates a strand that shares serialization state with 119   Creates a strand that shares serialization state with
130   the original. Coroutines dispatched through either strand 120   the original. Coroutines dispatched through either strand
131   are serialized with respect to each other. 121   are serialized with respect to each other.
132   122  
133   @param other The strand to copy. 123   @param other The strand to copy.
134   */ 124   */
HITCBC 135   11 strand(strand const& other) = default; 125   11 strand(strand const& other) = default;
136   126  
137   /** Construct by moving. 127   /** Construct by moving.
138   128  
139   @param other The strand to move from. 129   @param other The strand to move from.
140   130  
141   @note A moved-from strand is only safe to destroy 131   @note A moved-from strand is only safe to destroy
142   or reassign. 132   or reassign.
143   */ 133   */
HITCBC 144   11453 strand(strand&& other) = default; 134   11453 strand(strand&& other) = default;
145   135  
146   /** Assign by copying. 136   /** Assign by copying.
147   137  
148   Shares serialization state with `other`, as the copy 138   Shares serialization state with `other`, as the copy
149   constructor does. 139   constructor does.
150   140  
151   @param other The strand to copy. 141   @param other The strand to copy.
152   142  
153   @return A reference to `*this`. 143   @return A reference to `*this`.
154   */ 144   */
HITCBC 155   1 strand& operator=(strand const& other) = default; 145   1 strand& operator=(strand const& other) = default;
156   146  
157   /** Assign by moving. 147   /** Assign by moving.
158   148  
159   @param other The strand to move from. 149   @param other The strand to move from.
160   150  
161   @return A reference to `*this`. 151   @return A reference to `*this`.
162   152  
163   @note A moved-from strand is only safe to destroy 153   @note A moved-from strand is only safe to destroy
164   or reassign. 154   or reassign.
165   */ 155   */
HITCBC 166   1 strand& operator=(strand&& other) = default; 156   1 strand& operator=(strand&& other) = default;
167   157  
168   /** Return the underlying executor. 158   /** Return the underlying executor.
169   159  
170   @return A const reference to the inner executor. 160   @return A const reference to the inner executor.
171   */ 161   */
172   Ex const& 162   Ex const&
HITCBC 173   1 get_inner_executor() const noexcept 163   1 get_inner_executor() const noexcept
174   { 164   {
HITCBC 175   1 return ex_; 165   1 return ex_;
176   } 166   }
177   167  
178   /** Return the underlying execution context. 168   /** Return the underlying execution context.
179   169  
180   @return A reference to the execution context associated 170   @return A reference to the execution context associated
181   with the inner executor. 171   with the inner executor.
182   */ 172   */
183   auto& 173   auto&
HITCBC 184   6 context() const noexcept 174   6 context() const noexcept
185   { 175   {
HITCBC 186   6 return ex_.context(); 176   6 return ex_.context();
187   } 177   }
188   178  
189   /** Notify that work has started. 179   /** Notify that work has started.
190   180  
191   Delegates to the inner executor's `on_work_started()`. For a 181   Delegates to the inner executor's `on_work_started()`. For a
192   `thread_pool` inner executor, this increments the count that 182   `thread_pool` inner executor, this increments the count that
193   `join()` blocks on. 183   `join()` blocks on.
194   */ 184   */
195   void 185   void
HITCBC 196   7 on_work_started() const noexcept 186   7 on_work_started() const noexcept
197   { 187   {
HITCBC 198   7 ex_.on_work_started(); 188   7 ex_.on_work_started();
HITCBC 199   7 } 189   7 }
200   190  
201   /** Notify that work has finished. 191   /** Notify that work has finished.
202   192  
203   Delegates to the inner executor's `on_work_finished()`. For a 193   Delegates to the inner executor's `on_work_finished()`. For a
204   `thread_pool` inner executor, this decrements the count that 194   `thread_pool` inner executor, this decrements the count that
205   `join()` blocks on. 195   `join()` blocks on.
206   */ 196   */
207   void 197   void
HITCBC 208   7 on_work_finished() const noexcept 198   7 on_work_finished() const noexcept
209   { 199   {
HITCBC 210   7 ex_.on_work_finished(); 200   7 ex_.on_work_finished();
HITCBC 211   7 } 201   7 }
212   202  
213   /** Determine whether the strand is running in the current thread. 203   /** Determine whether the strand is running in the current thread.
214   204  
215   @return true if the current thread is executing a coroutine 205   @return true if the current thread is executing a coroutine
216   within this strand's dispatch loop. 206   within this strand's dispatch loop.
217   */ 207   */
218   bool 208   bool
HITCBC 219   4 running_in_this_thread() const noexcept 209   4 running_in_this_thread() const noexcept
220   { 210   {
HITCBC 221   4 return detail::strand_service::running_in_this_thread(*impl_); 211   4 return detail::strand_service::running_in_this_thread(*impl_);
222   } 212   }
223   213  
224   /** Compare two strands for equality. 214   /** Compare two strands for equality.
225   215  
226   Two strands are equal if they share the same internal 216   Two strands are equal if they share the same internal
227   serialization state. Equal strands serialize coroutines 217   serialization state. Equal strands serialize coroutines
228   with respect to each other. 218   with respect to each other.
229   219  
230   @param other The strand to compare against. 220   @param other The strand to compare against.
231   @return true if both strands share the same implementation. 221   @return true if both strands share the same implementation.
232   */ 222   */
233   bool 223   bool
HITCBC 234   499505 operator==(strand const& other) const noexcept 224   499505 operator==(strand const& other) const noexcept
235   { 225   {
HITCBC 236   499505 return impl_.get() == other.impl_.get(); 226   499505 return impl_.get() == other.impl_.get();
237   } 227   }
238   228  
239   /** Post a continuation to the strand. 229   /** Post a continuation to the strand.
240   230  
241   The continuation is always queued for execution, never resumed 231   The continuation is always queued for execution, never resumed
242   immediately. When the strand becomes available, queued 232   immediately. When the strand becomes available, queued
243   work executes in FIFO order on the underlying executor. 233   work executes in FIFO order on the underlying executor.
244   234  
245   @par Ordering 235   @par Ordering
246   Guarantees strict FIFO ordering relative to other post() calls. 236   Guarantees strict FIFO ordering relative to other post() calls.
247   Use this instead of dispatch() when ordering matters. 237   Use this instead of dispatch() when ordering matters.
248   238  
249   @param c The continuation to post. The caller retains 239   @param c The continuation to post. The caller retains
250   ownership; the continuation must remain valid until 240   ownership; the continuation must remain valid until
251   it is dequeued and resumed. 241   it is dequeued and resumed.
252   242  
253   @par Preconditions 243   @par Preconditions
254   The strand's execution context must outlive this call. Posting 244   The strand's execution context must outlive this call. Posting
255   concurrently with, or after, that context's destruction is 245   concurrently with, or after, that context's destruction is
256   undefined behavior. 246   undefined behavior.
257   */ 247   */
258   void 248   void
HITCBC 259   30336 post(continuation& c) const 249   30336 post(continuation& c) const
260   { 250   {
HITCBC 261   30336 detail::strand_service::post(impl_, executor_ref(ex_), c); 251   30336 detail::strand_service::post(impl_, executor_ref(ex_), c);
HITCBC 262   30336 } 252   30336 }
263   253  
264   /** Dispatch a continuation through the strand. 254   /** Dispatch a continuation through the strand.
265   255  
266   Returns a handle for symmetric transfer. If the calling 256   Returns a handle for symmetric transfer. If the calling
267   thread is already executing within this strand, returns `c.h`. 257   thread is already executing within this strand, returns `c.h`.
268   Otherwise, the continuation is queued and 258   Otherwise, the continuation is queued and
269   `std::noop_coroutine()` is returned. 259   `std::noop_coroutine()` is returned.
270   260  
271   @par Ordering 261   @par Ordering
272   Callers requiring strict FIFO ordering should use post() 262   Callers requiring strict FIFO ordering should use post()
273   instead, which always queues the continuation. 263   instead, which always queues the continuation.
274   264  
275   @param c The continuation to dispatch. The caller retains 265   @param c The continuation to dispatch. The caller retains
276   ownership; the continuation must remain valid until 266   ownership; the continuation must remain valid until
277   it is dequeued and resumed. 267   it is dequeued and resumed.
278   268  
279   @return A handle for symmetric transfer or `std::noop_coroutine()`. 269   @return A handle for symmetric transfer or `std::noop_coroutine()`.
280   270  
281   @par Preconditions 271   @par Preconditions
282   The strand's execution context must outlive this call. 272   The strand's execution context must outlive this call.
283   Dispatching concurrently with, or after, that context's 273   Dispatching concurrently with, or after, that context's
284   destruction is undefined behavior. 274   destruction is undefined behavior.
285   */ 275   */
286   std::coroutine_handle<> 276   std::coroutine_handle<>
HITCBC 287   9 dispatch(continuation& c) const 277   9 dispatch(continuation& c) const
288   { 278   {
HITCBC 289   9 return detail::strand_service::dispatch(impl_, executor_ref(ex_), c); 279   9 return detail::strand_service::dispatch(impl_, executor_ref(ex_), c);
290   } 280   }
291   }; 281   };
292   282  
293   /** Deduce the executor type from the constructor argument. 283   /** Deduce the executor type from the constructor argument.
294   284  
295   @tparam Ex The wrapped executor type. 285   @tparam Ex The wrapped executor type.
296   */ 286   */
297   template<typename Ex> 287   template<typename Ex>
298   strand(Ex) -> strand<Ex>; 288   strand(Ex) -> strand<Ex>;
299   289  
300   } // namespace capy 290   } // namespace capy
301   } // namespace boost 291   } // namespace boost
302   292  
303   #endif 293   #endif