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