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