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