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