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