. /include/boost/corosio /tcp_server.hpp
98.13% Lines (210/214) 97.96% Functions (48/49)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1   // 1   //
2   // Copyright (c) 2026 Vinnie Falco (vinnie.falco@gmail.com) 2   // Copyright (c) 2026 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/corosio 8   // Official repository: https://github.com/cppalliance/corosio
9   // 9   //
10   10  
11   #ifndef BOOST_COROSIO_TCP_SERVER_HPP 11   #ifndef BOOST_COROSIO_TCP_SERVER_HPP
12   #define BOOST_COROSIO_TCP_SERVER_HPP 12   #define BOOST_COROSIO_TCP_SERVER_HPP
13   13  
14   #include <boost/corosio/detail/config.hpp> 14   #include <boost/corosio/detail/config.hpp>
15   #include <boost/corosio/detail/except.hpp> 15   #include <boost/corosio/detail/except.hpp>
16   #include <boost/corosio/tcp_acceptor.hpp> 16   #include <boost/corosio/tcp_acceptor.hpp>
17   #include <boost/corosio/tcp_socket.hpp> 17   #include <boost/corosio/tcp_socket.hpp>
18   #include <boost/corosio/io_context.hpp> 18   #include <boost/corosio/io_context.hpp>
19   #include <boost/corosio/endpoint.hpp> 19   #include <boost/corosio/endpoint.hpp>
20   #include <boost/capy/task.hpp> 20   #include <boost/capy/task.hpp>
21   #include <boost/capy/concept/execution_context.hpp> 21   #include <boost/capy/concept/execution_context.hpp>
22   #include <boost/capy/concept/context_awaitable.hpp> 22   #include <boost/capy/concept/context_awaitable.hpp>
23   #include <boost/capy/concept/executor.hpp> 23   #include <boost/capy/concept/executor.hpp>
24   #include <boost/capy/ex/any_executor.hpp> 24   #include <boost/capy/ex/any_executor.hpp>
25   #include <boost/capy/detail/abandonable.hpp> 25   #include <boost/capy/detail/abandonable.hpp>
26   #include <boost/capy/detail/await_suspend_helper.hpp> 26   #include <boost/capy/detail/await_suspend_helper.hpp>
27   #include <boost/capy/ex/abandon.hpp> 27   #include <boost/capy/ex/abandon.hpp>
28   #include <boost/capy/ex/frame_alloc_mixin.hpp> 28   #include <boost/capy/ex/frame_alloc_mixin.hpp>
29   #include <boost/capy/ex/frame_allocator.hpp> 29   #include <boost/capy/ex/frame_allocator.hpp>
30   #include <boost/capy/ex/await_context.hpp> 30   #include <boost/capy/ex/await_context.hpp>
31   #include <boost/capy/ex/recycling_memory_resource.hpp> 31   #include <boost/capy/ex/recycling_memory_resource.hpp>
32   #include <boost/capy/ex/run_async.hpp> 32   #include <boost/capy/ex/run_async.hpp>
33   #include <boost/capy/ex/work_guard.hpp> 33   #include <boost/capy/ex/work_guard.hpp>
34   34  
35   #include <atomic> 35   #include <atomic>
36   #include <coroutine> 36   #include <coroutine>
37   #include <memory> 37   #include <memory>
38   #include <memory_resource> 38   #include <memory_resource>
39   #include <mutex> 39   #include <mutex>
40   #include <ranges> 40   #include <ranges>
41   #include <utility> 41   #include <utility>
42   #include <vector> 42   #include <vector>
43   43  
44   namespace boost::corosio { 44   namespace boost::corosio {
45   45  
46   #ifdef _MSC_VER 46   #ifdef _MSC_VER
47   #pragma warning(push) 47   #pragma warning(push)
48   #pragma warning(disable : 4251) // class needs to have dll-interface 48   #pragma warning(disable : 4251) // class needs to have dll-interface
49   #endif 49   #endif
50   50  
51   /** Manages a pool of reusable workers that handle incoming TCP connections. 51   /** Manages a pool of reusable workers that handle incoming TCP connections.
52   52  
53   This class manages a pool of reusable worker objects that handle 53   This class manages a pool of reusable worker objects that handle
54   incoming connections. When a connection arrives, an idle worker 54   incoming connections. When a connection arrives, an idle worker
55   is dispatched to handle it. After the connection completes, the 55   is dispatched to handle it. After the connection completes, the
56   worker returns to the pool for reuse, avoiding allocation overhead 56   worker returns to the pool for reuse, avoiding allocation overhead
57   per connection. 57   per connection.
58   58  
59   Workers are set via @ref set_workers as a forward range of 59   Workers are set via @ref set_workers as a forward range of
60   pointer-like objects (e.g., `unique_ptr<worker_base>`). The server 60   pointer-like objects (e.g., `unique_ptr<worker_base>`). The server
61   takes ownership of the container via type erasure. 61   takes ownership of the container via type erasure.
62   62  
63   @par Thread Safety 63   @par Thread Safety
64   Distinct objects: Safe. 64   Distinct objects: Safe.
65   Shared objects: Unsafe. 65   Shared objects: Unsafe.
66   66  
67   @par Lifecycle 67   @par Lifecycle
68   The server operates in three states: 68   The server operates in three states:
69   69  
70   - **Stopped**: Initial state, or after @ref join completes. 70   - **Stopped**: Initial state, or after @ref join completes.
71   - **Running**: After @ref start, actively accepting connections. 71   - **Running**: After @ref start, actively accepting connections.
72   - **Stopping**: After @ref stop, draining active work. 72   - **Stopping**: After @ref stop, draining active work.
73   73  
74   State transitions: 74   State transitions:
75   @code 75   @code
76   [Stopped] --start()--> [Running] --stop()--> [Stopping] --join()--> [Stopped] 76   [Stopped] --start()--> [Running] --stop()--> [Stopping] --join()--> [Stopped]
77   @endcode 77   @endcode
78   78  
79   @par Running the Server 79   @par Running the Server
80   @par !example running_the_server 80   @par !example running_the_server
81   81  
82   @par Graceful Shutdown 82   @par Graceful Shutdown
83   To shut down gracefully, call @ref stop then drain the `io_context`: 83   To shut down gracefully, call @ref stop then drain the `io_context`:
84   @par !example graceful_shutdown 84   @par !example graceful_shutdown
85   85  
86   @par Restart After Stop 86   @par Restart After Stop
87   The server can be restarted after a complete shutdown cycle. 87   The server can be restarted after a complete shutdown cycle.
88   You must drain the `io_context`, call @ref join, and restart the 88   You must drain the `io_context`, call @ref join, and restart the
89   `io_context` itself (`ioc.restart()`) before restarting: 89   `io_context` itself (`ioc.restart()`) before restarting:
90   @par !example restart_after_stop 90   @par !example restart_after_stop
91   91  
92   @par WARNING: What NOT to Do 92   @par WARNING: What NOT to Do
93   - Do NOT call @ref join from inside a worker coroutine (deadlock). 93   - Do NOT call @ref join from inside a worker coroutine (deadlock).
94   - Do NOT call @ref join from a thread running `ioc.run()` (deadlock). 94   - Do NOT call @ref join from a thread running `ioc.run()` (deadlock).
95   - Do NOT call @ref start without completing @ref join after @ref stop. 95   - Do NOT call @ref start without completing @ref join after @ref stop.
96   - Do NOT call `ioc.stop()` for graceful shutdown; use @ref stop instead. 96   - Do NOT call `ioc.stop()` for graceful shutdown; use @ref stop instead.
97   97  
98   @par Example 98   @par Example
99   @par !example custom_worker 99   @par !example custom_worker
100   100  
101   @see worker_base, set_workers, launcher 101   @see worker_base, set_workers, launcher
102   */ 102   */
103   class BOOST_COROSIO_DECL tcp_server 103   class BOOST_COROSIO_DECL tcp_server
104   { 104   {
105   public: 105   public:
106   class worker_base; ///< Abstract base for connection handlers. 106   class worker_base; ///< Abstract base for connection handlers.
107   class launcher; ///< Move-only handle to launch worker coroutines. 107   class launcher; ///< Move-only handle to launch worker coroutines.
108   108  
109   private: 109   private:
110   struct waiter 110   struct waiter
111   { 111   {
112   waiter* next; 112   waiter* next;
113   capy::continuation cont; 113   capy::continuation cont;
114   worker_base* w; 114   worker_base* w;
115   }; 115   };
116   116  
117   struct impl; 117   struct impl;
118   118  
119   /** Everything the server's coroutines reach. 119   /** Everything the server's coroutines reach.
120   120  
121   Accept loops, launchers and the coroutines that start 121   Accept loops, launchers and the coroutines that start
122   connections each hold a reference, so moving or destroying the 122   connections each hold a reference, so moving or destroying the
123   server never strands them. The lists are guarded by `mutex`, 123   server never strands them. The lists are guarded by `mutex`,
124   because workers return from whichever thread their coroutine 124   because workers return from whichever thread their coroutine
125   finished on. 125   finished on.
126   */ 126   */
127   struct state 127   struct state
128   { 128   {
129   std::atomic<std::size_t> refs{1}; 129   std::atomic<std::size_t> refs{1};
130   impl* impl_; 130   impl* impl_;
131   capy::any_executor ex; 131   capy::any_executor ex;
132   std::mutex mutex; 132   std::mutex mutex;
133   waiter* waiters = nullptr; 133   waiter* waiters = nullptr;
134   worker_base* idle_head = nullptr; // forward list 134   worker_base* idle_head = nullptr; // forward list
135   worker_base* active_head = nullptr; // doubly linked 135   worker_base* active_head = nullptr; // doubly linked
136   worker_base* active_tail = nullptr; 136   worker_base* active_tail = nullptr;
137   std::size_t active_accepts = 0; // guarded by the join mutex 137   std::size_t active_accepts = 0; // guarded by the join mutex
138   std::size_t lent = 0; // borrowed workers, guarded by mutex 138   std::size_t lent = 0; // borrowed workers, guarded by mutex
139   std::shared_ptr<void> storage; // owns the worker container 139   std::shared_ptr<void> storage; // owns the worker container
140   bool running = false; 140   bool running = false;
141   // Set by stop(); accept loops waiting for a worker give up. 141   // Set by stop(); accept loops waiting for a worker give up.
142   bool stopping = false; // guarded by mutex 142   bool stopping = false; // guarded by mutex
143   143  
HITCBC 144   169 state(impl* i, capy::any_executor const& e) noexcept : impl_(i), ex(e) 144   169 state(impl* i, capy::any_executor const& e) noexcept : impl_(i), ex(e)
145   { 145   {
HITCBC 146   169 } 146   169 }
147   147  
148   // The list operations below require `mutex`. 148   // The list operations below require `mutex`.
149   149  
HITCBC 150   1150 void idle_push(worker_base* w) noexcept 150   1198 void idle_push(worker_base* w) noexcept
151   { 151   {
HITCBC 152   1150 w->next_ = idle_head; 152   1198 w->next_ = idle_head;
HITCBC 153   1150 idle_head = w; 153   1198 idle_head = w;
HITCBC 154   1150 } 154   1198 }
155   155  
HITCBC 156   1049 worker_base* idle_pop() noexcept 156   1158 worker_base* idle_pop() noexcept
157   { 157   {
HITCBC 158   1049 auto* w = idle_head; 158   1158 auto* w = idle_head;
HITCBC 159   1049 if (w) 159   1158 if (w)
160   { 160   {
HITCBC 161   739 idle_head = w->next_; 161   787 idle_head = w->next_;
HITCBC 162   739 ++lent; 162   787 ++lent;
163   } 163   }
HITCBC 164   1049 return w; 164   1158 return w;
165   } 165   }
166   166  
HITCBC 167   892 void active_push(worker_base* w) noexcept 167   1000 void active_push(worker_base* w) noexcept
168   { 168   {
HITCBC 169   892 w->next_ = nullptr; 169   1000 w->next_ = nullptr;
HITCBC 170   892 w->prev_ = active_tail; 170   1000 w->prev_ = active_tail;
HITCBC 171   892 if (active_tail) 171   1000 if (active_tail)
HITCBC 172   601 active_tail->next_ = w; 172   703 active_tail->next_ = w;
173   else 173   else
HITCBC 174   291 active_head = w; 174   297 active_head = w;
HITCBC 175   892 active_tail = w; 175   1000 active_tail = w;
HITCBC 176   892 } 176   1000 }
177   177  
HITCBC 178   1020 void active_remove(worker_base* w) noexcept 178   1130 void active_remove(worker_base* w) noexcept
179   { 179   {
180   // Skip if not in active list (e.g., after failed accept) 180   // Skip if not in active list (e.g., after failed accept)
HITCBC 181   1020 if (w != active_head && w->prev_ == nullptr) 181   1130 if (w != active_head && w->prev_ == nullptr)
HITCBC 182   128 return; 182   130 return;
HITCBC 183   892 if (w->prev_) 183   1000 if (w->prev_)
HITCBC 184   275 w->prev_->next_ = w->next_; 184   268 w->prev_->next_ = w->next_;
185   else 185   else
HITCBC 186   617 active_head = w->next_; 186   732 active_head = w->next_;
HITCBC 187   892 if (w->next_) 187   1000 if (w->next_)
HITCBC 188   479 w->next_->prev_ = w->prev_; 188   608 w->next_->prev_ = w->prev_;
189   else 189   else
HITCBC 190   413 active_tail = w->prev_; 190   392 active_tail = w->prev_;
HITCBC 191   892 w->prev_ = nullptr; // Mark as not in active list 191   1000 w->prev_ = nullptr; // Mark as not in active list
192   } 192   }
193   193  
194   /// Return @p w to the pool, handing it to a waiting accept loop 194   /// Return @p w to the pool, handing it to a waiting accept loop
195   /// if there is one. 195   /// if there is one.
HITCBC 196   1020 void push(worker_base& w) noexcept 196   1130 void push(worker_base& w) noexcept
197   { 197   {
HITCBC 198   1020 waiter* wake = nullptr; 198   1130 waiter* wake = nullptr;
199   { 199   {
HITCBC 200   1020 std::lock_guard<std::mutex> lock(mutex); 200   1130 std::lock_guard<std::mutex> lock(mutex);
HITCBC 201   1020 active_remove(&w); 201   1130 active_remove(&w);
HITCBC 202   1020 if (waiters) 202   1130 if (waiters)
203   { 203   {
HITCBC 204   283 wake = waiters; 204   345 wake = waiters;
HITCBC 205   283 waiters = wake->next; 205   345 waiters = wake->next;
HITCBC 206   283 wake->w = &w; 206   345 wake->w = &w;
207   } 207   }
208   else 208   else
209   { 209   {
HITCBC 210   737 idle_push(&w); 210   785 idle_push(&w);
HITCBC 211   737 --lent; 211   785 --lent;
212   } 212   }
HITCBC 213   1020 } 213   1130 }
HITCBC 214   1020 if (wake) 214   1130 if (wake)
HITCBC 215   283 ex.post(wake->cont); 215   345 ex.post(wake->cont);
HITCBC 216   1020 } 216   1130 }
217   217  
218   /// Drop one hold on @p w, returning it to the pool with the last. 218   /// Drop one hold on @p w, returning it to the pool with the last.
HITCBC 219   1926 void drop_hold(worker_base& w) noexcept 219   2144 void drop_hold(worker_base& w) noexcept
220   { 220   {
HITCBC 221   3852 if (w.holds_.fetch_sub(1, std::memory_order_acq_rel) == 1) 221   4288 if (w.holds_.fetch_sub(1, std::memory_order_acq_rel) == 1)
HITCBC 222   1020 push(w); 222   1130 push(w);
HITCBC 223   1926 } 223   2144 }
224   224  
225   /// Mark the pool as stopping and release every accept loop 225   /// Mark the pool as stopping and release every accept loop
226   /// waiting for a worker, with none. 226   /// waiting for a worker, with none.
HITCBC 227   157 void stop_waiters() noexcept 227   157 void stop_waiters() noexcept
228   { 228   {
HITCBC 229   157 waiter* list = nullptr; 229   157 waiter* list = nullptr;
230   { 230   {
HITCBC 231   157 std::lock_guard<std::mutex> lock(mutex); 231   157 std::lock_guard<std::mutex> lock(mutex);
HITCBC 232   157 stopping = true; 232   157 stopping = true;
HITCBC 233   157 list = std::exchange(waiters, nullptr); 233   157 list = std::exchange(waiters, nullptr);
HITCBC 234   157 } 234   157 }
HITCBC 235   184 while (list) 235   183 while (list)
236   { 236   {
HITCBC 237   27 auto* next = list->next; 237   26 auto* next = list->next;
HITCBC 238   27 ex.post(list->cont); 238   26 ex.post(list->cont);
HITCBC 239   27 list = next; 239   26 list = next;
240   } 240   }
HITCBC 241   157 } 241   157 }
242   }; 242   };
243   243  
244   /// Add a reference to @p s. 244   /// Add a reference to @p s.
HITCBC 245   2273 static void add_ref(state* s) noexcept 245   2489 static void add_ref(state* s) noexcept
246   { 246   {
HITCBC 247   2273 s->refs.fetch_add(1, std::memory_order_relaxed); 247   2489 s->refs.fetch_add(1, std::memory_order_relaxed);
HITCBC 248   2273 } 248   2489 }
249   249  
250   /// Drop a reference to @p s, freeing it with the last one. 250   /// Drop a reference to @p s, freeing it with the last one.
251   static void release(state* s) noexcept; 251   static void release(state* s) noexcept;
252   252  
253   /// An owned reference to the server's state. 253   /// An owned reference to the server's state.
254   class state_ref 254   class state_ref
255   { 255   {
256   state* p_ = nullptr; 256   state* p_ = nullptr;
257   257  
258   public: 258   public:
HITCBC 259   1367 explicit state_ref(state* p) noexcept : p_(p) 259   1475 explicit state_ref(state* p) noexcept : p_(p)
260   { 260   {
HITCBC 261   1367 add_ref(p_); 261   1475 add_ref(p_);
HITCBC 262   1367 } 262   1475 }
263   263  
264   state_ref(state_ref const& o) noexcept : p_(o.p_) 264   state_ref(state_ref const& o) noexcept : p_(o.p_)
265   { 265   {
266   if (p_) 266   if (p_)
267   add_ref(p_); 267   add_ref(p_);
268   } 268   }
269   269  
HITCBC 270   3620 state_ref(state_ref&& o) noexcept : p_(std::exchange(o.p_, nullptr)) {} 270   3944 state_ref(state_ref&& o) noexcept : p_(std::exchange(o.p_, nullptr)) {}
271   271  
272   state_ref& operator=(state_ref const&) = delete; 272   state_ref& operator=(state_ref const&) = delete;
273   state_ref& operator=(state_ref&&) = delete; 273   state_ref& operator=(state_ref&&) = delete;
274   274  
HITCBC 275   4987 ~state_ref() 275   5419 ~state_ref()
276   { 276   {
HITCBC 277   4987 if (p_) 277   5419 if (p_)
HITCBC 278   1367 release(p_); 278   1475 release(p_);
HITCBC 279   4987 } 279   5419 }
280   280  
HITCBC 281   2522 state* operator->() const noexcept 281   2742 state* operator->() const noexcept
282   { 282   {
HITCBC 283   2522 return p_; 283   2742 return p_;
284   } 284   }
285   285  
HITCBC 286   2949 state& operator*() const noexcept 286   3276 state& operator*() const noexcept
287   { 287   {
HITCBC 288   2949 return *p_; 288   3276 return *p_;
289   } 289   }
290   }; 290   };
291   291  
292   static state* 292   static state*
293   make_state(capy::execution_context& ctx, capy::any_executor const& ex); 293   make_state(capy::execution_context& ctx, capy::any_executor const& ex);
294   294  
295   state* st_; 295   state* st_;
296   296  
297   template<capy::Executor Ex> 297   template<capy::Executor Ex>
298   struct launch_wrapper 298   struct launch_wrapper
299   { 299   {
300   // frame_alloc_mixin routes the frame through the thread-local 300   // frame_alloc_mixin routes the frame through the thread-local
301   // recycling allocator, so a warmed per-connection launch does 301   // recycling allocator, so a warmed per-connection launch does
302   // not hit the global allocator. 302   // not hit the global allocator.
303   struct promise_type : capy::frame_alloc_mixin 303   struct promise_type : capy::frame_alloc_mixin
304   { 304   {
305   Ex ex; // Executor stored directly in frame (outlives child tasks) 305   Ex ex; // Executor stored directly in frame (outlives child tasks)
306   // Owns the connection's chain: abandoning it resumes, or has 306   // Owns the connection's chain: abandoning it resumes, or has
307   // the executor destroy, this frame to unwind the task. 307   // the executor destroy, this frame to unwind the task.
308   capy::chain_owner owner_; 308   capy::chain_owner owner_;
309   capy::await_context env_; 309   capy::await_context env_;
310   /// Embedded post node: the start posts this instead of the 310   /// Embedded post node: the start posts this instead of the
311   /// bare handle, which would heap-allocate a wrapper op. 311   /// bare handle, which would heap-allocate a wrapper op.
312   capy::continuation cont_; 312   capy::continuation cont_;
313   313  
314   // For regular coroutines: first arg is executor, second is stop token 314   // For regular coroutines: first arg is executor, second is stop token
315   template<class E, class S, class... Args> 315   template<class E, class S, class... Args>
316   requires capy::Executor<std::decay_t<E>> 316   requires capy::Executor<std::decay_t<E>>
317   promise_type(E e, S s, Args&&...) 317   promise_type(E e, S s, Args&&...)
318   : ex(std::move(e)) 318   : ex(std::move(e))
319   , env_{ 319   , env_{
320   capy::executor_ref(ex), std::move(s), 320   capy::executor_ref(ex), std::move(s),
321   capy::get_current_frame_allocator(), &owner_} 321   capy::get_current_frame_allocator(), &owner_}
322   { 322   {
323   } 323   }
324   324  
325   // For lambda coroutines: first arg is closure, second is executor, third is stop token 325   // For lambda coroutines: first arg is closure, second is executor, third is stop token
326   template<class Closure, class E, class S, class... Args> 326   template<class Closure, class E, class S, class... Args>
327   requires(!capy::Executor<std::decay_t<Closure>> && 327   requires(!capy::Executor<std::decay_t<Closure>> &&
328   capy::Executor<std::decay_t<E>>) 328   capy::Executor<std::decay_t<E>>)
HITCBC 329   892 promise_type(Closure&&, E e, S s, Args&&...) 329   1000 promise_type(Closure&&, E e, S s, Args&&...)
HITCBC 330   892 : ex(std::move(e)) 330   1000 : ex(std::move(e))
HITCBC 331   892 , env_{ 331   1000 , env_{
HITCBC 332   892 capy::executor_ref(ex), std::move(s), 332   1000 capy::executor_ref(ex), std::move(s),
HITCBC 333   1784 capy::get_current_frame_allocator(), &owner_} 333   2000 capy::get_current_frame_allocator(), &owner_}
334   { 334   {
HITCBC 335   892 } 335   1000 }
336   336  
HITCBC 337   892 launch_wrapper get_return_object() noexcept 337   1000 launch_wrapper get_return_object() noexcept
338   { 338   {
339   return { 339   return {
HITCBC 340   892 std::coroutine_handle<promise_type>::from_promise(*this)}; 340   1000 std::coroutine_handle<promise_type>::from_promise(*this)};
341   } 341   }
HITCBC 342   892 std::suspend_always initial_suspend() noexcept 342   1000 std::suspend_always initial_suspend() noexcept
343   { 343   {
HITCBC 344   892 return {}; 344   1000 return {};
345   } 345   }
HITCBC 346   875 std::suspend_never final_suspend() noexcept 346   983 std::suspend_never final_suspend() noexcept
347   { 347   {
HITCBC 348   875 return {}; 348   983 return {};
349   } 349   }
HITCBC 350   875 void return_void() noexcept {} 350   983 void return_void() noexcept {}
MISUBC 351   ✗ void unhandled_exception() 351   ✗ void unhandled_exception()
352   { 352   {
353   // LCOV_EXCL_START: terminating by contract is not a 353   // LCOV_EXCL_START: terminating by contract is not a
354   // coverable outcome. 354   // coverable outcome.
355   std::terminate(); 355   std::terminate();
356   // LCOV_EXCL_STOP 356   // LCOV_EXCL_STOP
357   } 357   }
358   358  
359   // Inject await_context for ContextAwaitable 359   // Inject await_context for ContextAwaitable
360   template<capy::ContextAwaitable Awaitable> 360   template<capy::ContextAwaitable Awaitable>
HITCBC 361   1762 auto await_transform(Awaitable&& a) 361   1978 auto await_transform(Awaitable&& a)
362   { 362   {
363   using AwaitableT = std::decay_t<Awaitable>; 363   using AwaitableT = std::decay_t<Awaitable>;
364   struct adapter 364   struct adapter
365   { 365   {
366   AwaitableT aw; 366   AwaitableT aw;
367   capy::await_context const* env; 367   capy::await_context const* env;
368   368  
HITCBC 369   1762 bool await_ready() 369   1978 bool await_ready()
370   { 370   {
HITCBC 371   1762 return aw.await_ready(); 371   1978 return aw.await_ready();
372   } 372   }
HITCBC 373   1753 decltype(auto) await_resume() 373   1969 decltype(auto) await_resume()
374   { 374   {
HITCBC 375   1753 return aw.await_resume(); 375   1969 return aw.await_resume();
376   } 376   }
377   377  
378   // Where the returned handle would be stored in a frame 378   // Where the returned handle would be stored in a frame
379   // another thread may already have resumed and freed, 379   // another thread may already have resumed and freed,
380   // it is resumed here instead, as a task does. 380   // it is resumed here instead, as a task does.
HITCBC 381   1762 auto await_suspend(std::coroutine_handle<promise_type> h) 381   1978 auto await_suspend(std::coroutine_handle<promise_type> h)
382   { 382   {
HITCBC 383   1762 return capy::detail::symmetric_transfer( 383   1978 return capy::detail::symmetric_transfer(
HITCBC 384   1762 capy::detail::call_await_suspend(&aw, h, env)); 384   1978 capy::detail::call_await_suspend(&aw, h, env));
385   } 385   }
386   }; 386   };
HITCBC 387   1762 return adapter{std::forward<Awaitable>(a), &env_}; 387   1978 return adapter{std::forward<Awaitable>(a), &env_};
388   } 388   }
389   }; 389   };
390   390  
391   std::coroutine_handle<promise_type> h; 391   std::coroutine_handle<promise_type> h;
392   392  
HITCBC 393   892 launch_wrapper(std::coroutine_handle<promise_type> handle) noexcept 393   1000 launch_wrapper(std::coroutine_handle<promise_type> handle) noexcept
HITCBC 394   892 : h(handle) 394   1000 : h(handle)
395   { 395   {
HITCBC 396   892 } 396   1000 }
397   397  
HITCBC 398   892 ~launch_wrapper() 398   1000 ~launch_wrapper()
399   { 399   {
400   // An unstarted root owns no chain yet: destroying it is 400   // An unstarted root owns no chain yet: destroying it is
401   // the whole teardown. 401   // the whole teardown.
HITCBC 402   892 if (h) 402   1000 if (h)
HITCBC 403   4 h.destroy(); 403   4 h.destroy();
HITCBC 404   892 } 404   1000 }
405   405  
406   launch_wrapper(launch_wrapper&& o) noexcept 406   launch_wrapper(launch_wrapper&& o) noexcept
407   : h(std::exchange(o.h, nullptr)) 407   : h(std::exchange(o.h, nullptr))
408   { 408   {
409   } 409   }
410   410  
411   launch_wrapper(launch_wrapper const&) = delete; 411   launch_wrapper(launch_wrapper const&) = delete;
412   launch_wrapper& operator=(launch_wrapper const&) = delete; 412   launch_wrapper& operator=(launch_wrapper const&) = delete;
413   launch_wrapper& operator=(launch_wrapper&&) = delete; 413   launch_wrapper& operator=(launch_wrapper&&) = delete;
414   }; 414   };
415   415  
416   /// One hold on a worker, dropped when this is destroyed. 416   /// One hold on a worker, dropped when this is destroyed.
417   class worker_hold 417   class worker_hold
418   { 418   {
419   state_ref st_; 419   state_ref st_;
420   worker_base* w_; 420   worker_base* w_;
421   421  
422   public: 422   public:
HITCBC 423   892 worker_hold(state& st, worker_base& w) noexcept : st_(&st), w_(&w) {} 423   1000 worker_hold(state& st, worker_base& w) noexcept : st_(&st), w_(&w) {}
424   424  
HITCBC 425   2668 worker_hold(worker_hold&& o) noexcept 425   2992 worker_hold(worker_hold&& o) noexcept
HITCBC 426   2668 : st_(std::move(o.st_)) 426   2992 : st_(std::move(o.st_))
HITCBC 427   2668 , w_(std::exchange(o.w_, nullptr)) 427   2992 , w_(std::exchange(o.w_, nullptr))
428   { 428   {
HITCBC 429   2668 } 429   2992 }
430   430  
431   worker_hold(worker_hold const&) = delete; 431   worker_hold(worker_hold const&) = delete;
432   worker_hold& operator=(worker_hold const&) = delete; 432   worker_hold& operator=(worker_hold const&) = delete;
433   worker_hold& operator=(worker_hold&&) = delete; 433   worker_hold& operator=(worker_hold&&) = delete;
434   434  
HITCBC 435   3560 ~worker_hold() 435   3992 ~worker_hold()
436   { 436   {
HITCBC 437   3560 if (w_) 437   3992 if (w_)
HITCBC 438   892 st_->drop_hold(*w_); 438   1000 st_->drop_hold(*w_);
HITCBC 439   3560 } 439   3992 }
440   440  
HITCBC 441   878 state& server() const noexcept 441   986 state& server() const noexcept
442   { 442   {
HITCBC 443   878 return *st_; 443   986 return *st_;
444   } 444   }
445   }; 445   };
446   446  
447   // The holds travel in the frame so that they are also released 447   // The holds travel in the frame so that they are also released
448   // when the frame is destroyed before it starts or finishes. One 448   // when the frame is destroyed before it starts or finishes. One
449   // parameter, not three: compilers disagree on the order parameter 449   // parameter, not three: compilers disagree on the order parameter
450   // copies are destroyed, but members always go in reverse, so the 450   // copies are destroyed, but members always go in reverse, so the
451   // task, whose parameters may still use the worker, goes first. 451   // task, whose parameters may still use the worker, goes first.
452   template<class Executor> 452   template<class Executor>
453   struct session 453   struct session
454   { 454   {
455   worker_hold hold; 455   worker_hold hold;
456   // Keeps the session's context running until the connection ends. 456   // Keeps the session's context running until the connection ends.
457   capy::work_guard<Executor> work; 457   capy::work_guard<Executor> work;
458   capy::task<void> t; 458   capy::task<void> t;
459   }; 459   };
460   460  
461   // Named functor to avoid incomplete lambda type in coroutine promise 461   // Named functor to avoid incomplete lambda type in coroutine promise
462   template<class Executor> 462   template<class Executor>
463   struct launch_coro 463   struct launch_coro
464   { 464   {
465   launch_wrapper<Executor> 465   launch_wrapper<Executor>
HITCBC 466   892 operator()(Executor, std::stop_token, session<Executor> in) 466   1000 operator()(Executor, std::stop_token, session<Executor> in)
467   { 467   {
468   // Executor and stop token stored in promise via constructor. 468   // Executor and stop token stored in promise via constructor.
469   // A local, so a frame destroyed unfinished destroys the task 469   // A local, so a frame destroyed unfinished destroys the task
470   // while the promise and its environment still exist. 470   // while the promise and its environment still exist.
471   auto s = std::move(in); 471   auto s = std::move(in);
472   // Resumed with the task unfinished: the chain was abandoned, 472   // Resumed with the task unfinished: the chain was abandoned,
473   // and leaving destroys the task, which unwinds it. 473   // and leaving destroys the task, which unwinds it.
474   if (!co_await capy::detail::abandonable<capy::task<void>>{s.t}) 474   if (!co_await capy::detail::abandonable<capy::task<void>>{s.t})
475   co_return; 475   co_return;
476   // The connection's work ends with its task: the hop below is 476   // The connection's work ends with its task: the hop below is
477   // the server's, and its context may no longer be running. 477   // the server's, and its context may no longer be running.
478   s.work.reset(); 478   s.work.reset();
479   // The hold returns the worker from the server's executor. 479   // The hold returns the worker from the server's executor.
480   co_await hop_awaitable{s.hold.server(), true}; 480   co_await hop_awaitable{s.hold.server(), true};
HITCBC 481   1784 } 481   2000 }
482   }; 482   };
483   483  
484   /// Resume on the server's executor. 484   /// Resume on the server's executor.
485   class hop_awaitable 485   class hop_awaitable
486   { 486   {
487   state& st_; 487   state& st_;
488   bool root_; 488   bool root_;
489   capy::continuation cont_; 489   capy::continuation cont_;
490   490  
491   public: 491   public:
492   // `root` is true when the awaiting coroutine is its chain's root. 492   // `root` is true when the awaiting coroutine is its chain's root.
HITCBC 493   994 explicit hop_awaitable(state& st, bool root = false) noexcept 493   1104 explicit hop_awaitable(state& st, bool root = false) noexcept
HITCBC 494   994 : st_(st) 494   1104 : st_(st)
HITCBC 495   994 , root_(root) 495   1104 , root_(root)
496   { 496   {
HITCBC 497   994 } 497   1104 }
498   498  
HITCBC 499   994 bool await_ready() const noexcept 499   1104 bool await_ready() const noexcept
500   { 500   {
HITCBC 501   994 return false; 501   1104 return false;
502   } 502   }
503   503  
504   std::coroutine_handle<> 504   std::coroutine_handle<>
HITCBC 505   994 await_suspend( 505   1104 await_suspend(
506   std::coroutine_handle<> h, capy::await_context const* env) noexcept 506   std::coroutine_handle<> h, capy::await_context const* env) noexcept
507   { 507   {
HITCBC 508   994 cont_.h = h; 508   1104 cont_.h = h;
509   // A root has no owner: the launcher's would unwind it on the 509   // A root has no owner: the launcher's would unwind it on the
510   // session's executor, which may already be gone. Given up 510   // session's executor, which may already be gone. Given up
511   // here, it is destroyed in place. 511   // here, it is destroyed in place.
HITCBC 512   994 if (!root_) 512   1104 if (!root_)
HITCBC 513   116 cont_.owner = env->owner; 513   118 cont_.owner = env->owner;
HITCBC 514   994 return st_.ex.dispatch(cont_); 514   1104 return st_.ex.dispatch(cont_);
515   } 515   }
516   516  
HITCBC 517   989 void await_resume() noexcept {} 517   1099 void await_resume() noexcept {}
518   }; 518   };
519   519  
520   class pop_awaitable 520   class pop_awaitable
521   { 521   {
522   state& st_; 522   state& st_;
523   waiter wait_; 523   waiter wait_;
524   524  
525   public: 525   public:
HITCBC 526   1049 pop_awaitable(state& st) noexcept : st_(st), wait_{} {} 526   1158 pop_awaitable(state& st) noexcept : st_(st), wait_{} {}
527   527  
HITCBC 528   1049 bool await_ready() const noexcept 528   1158 bool await_ready() const noexcept
529   { 529   {
HITCBC 530   1049 return false; 530   1158 return false;
531   } 531   }
532   532  
533   bool 533   bool
HITCBC 534   1049 await_suspend(std::coroutine_handle<> h, capy::await_context const* env) noexcept 534   1158 await_suspend(std::coroutine_handle<> h, capy::await_context const* env) noexcept
535   { 535   {
HITCBC 536   1049 std::lock_guard<std::mutex> lock(st_.mutex); 536   1158 std::lock_guard<std::mutex> lock(st_.mutex);
HITCBC 537   1049 if (st_.stopping) 537   1158 if (st_.stopping)
538   { 538   {
MISUBC 539   ✗ wait_.w = nullptr; 539   ✗ wait_.w = nullptr;
MISUBC 540   ✗ return false; 540   ✗ return false;
541   } 541   }
HITCBC 542   1049 if (auto* w = st_.idle_pop()) 542   1158 if (auto* w = st_.idle_pop())
543   { 543   {
HITCBC 544   739 wait_.w = w; 544   787 wait_.w = w;
HITCBC 545   739 return false; 545   787 return false;
546   } 546   }
HITCBC 547   310 wait_.cont.h = h; 547   371 wait_.cont.h = h;
HITCBC 548   310 wait_.cont.owner = env->owner; 548   371 wait_.cont.owner = env->owner;
HITCBC 549   310 wait_.w = nullptr; 549   371 wait_.w = nullptr;
HITCBC 550   310 wait_.next = st_.waiters; 550   371 wait_.next = st_.waiters;
HITCBC 551   310 st_.waiters = &wait_; 551   371 st_.waiters = &wait_;
HITCBC 552   310 return true; 552   371 return true;
HITCBC 553   1049 } 553   1158 }
554   554  
555   /// Return the worker, or null once the server is stopping. 555   /// Return the worker, or null once the server is stopping.
HITCBC 556   1045 worker_base* await_resume() noexcept 556   1154 worker_base* await_resume() noexcept
557   { 557   {
558   // Set by await_suspend, or by the push that woke us. 558   // Set by await_suspend, or by the push that woke us.
HITCBC 559   1045 return wait_.w; 559   1154 return wait_.w;
560   } 560   }
561   }; 561   };
562   562  
563   static capy::task<void> do_accept(state_ref st, tcp_acceptor& acc); 563   static capy::task<void> do_accept(state_ref st, tcp_acceptor& acc);
564   static capy::task<> do_stop(state_ref st); 564   static capy::task<> do_stop(state_ref st);
565   565  
566   public: 566   public:
567   /** Handles one accepted connection using a socket the derived class owns. 567   /** Handles one accepted connection using a socket the derived class owns.
568   568  
569   Derive from this class to implement custom connection handling. 569   Derive from this class to implement custom connection handling.
570   Each worker owns a socket and is reused across multiple 570   Each worker owns a socket and is reused across multiple
571   connections to avoid per-connection allocation. 571   connections to avoid per-connection allocation.
572   572  
573   @par Thread Safety 573   @par Thread Safety
574   run() and socket() execute on the server's executor. 574   run() and socket() execute on the server's executor.
575   575  
576   @see tcp_server, launcher 576   @see tcp_server, launcher
577   */ 577   */
578   class BOOST_COROSIO_DECL worker_base 578   class BOOST_COROSIO_DECL worker_base
579   { 579   {
580   // Ordered largest to smallest for optimal packing 580   // Ordered largest to smallest for optimal packing
581   std::stop_source stop_; // ~16 bytes 581   std::stop_source stop_; // ~16 bytes
582   worker_base* next_ = nullptr; // 8 bytes - used by idle and active lists 582   worker_base* next_ = nullptr; // 8 bytes - used by idle and active lists
583   worker_base* prev_ = nullptr; // 8 bytes - used only by active list 583   worker_base* prev_ = nullptr; // 8 bytes - used only by active list
584   // One for the accept loop until run() returns, one for the 584   // One for the accept loop until run() returns, one for the
585   // launcher and its connection; the last one returns the worker. 585   // launcher and its connection; the last one returns the worker.
586   std::atomic<unsigned> holds_{0}; 586   std::atomic<unsigned> holds_{0};
587   587  
588   friend class tcp_server; 588   friend class tcp_server;
589   589  
590   public: 590   public:
591   /// Construct a worker. 591   /// Construct a worker.
592   worker_base(); 592   worker_base();
593   593  
594   /// Destroy the worker. 594   /// Destroy the worker.
595   virtual ~worker_base(); 595   virtual ~worker_base();
596   596  
597   /** Handle an accepted connection. 597   /** Handle an accepted connection.
598   598  
599   Called when this worker is dispatched to handle a new 599   Called when this worker is dispatched to handle a new
600   connection. The implementation must invoke the launcher 600   connection. The implementation must invoke the launcher
601   exactly once to start the handling coroutine. 601   exactly once to start the handling coroutine.
602   602  
603   @par Exception Safety 603   @par Exception Safety
604   An exception escaping this function ends the process. A 604   An exception escaping this function ends the process. A
605   worker that cannot start the connection should handle the 605   worker that cannot start the connection should handle the
606   failure here; destroying the launcher without invoking it 606   failure here; destroying the launcher without invoking it
607   returns the worker to the pool. 607   returns the worker to the pool.
608   608  
609   The worker is not given another connection before this 609   The worker is not given another connection before this
610   returns, even if the connection it started has finished. 610   returns, even if the connection it started has finished.
611   611  
612   @param launch Handle to start the connection coroutine. 612   @param launch Handle to start the connection coroutine.
613   */ 613   */
614   virtual void run(launcher launch) = 0; 614   virtual void run(launcher launch) = 0;
615   615  
616   /// Return the socket used for connections. 616   /// Return the socket used for connections.
617   virtual corosio::tcp_socket& socket() = 0; 617   virtual corosio::tcp_socket& socket() = 0;
618   }; 618   };
619   619  
620   /** Starts a worker's connection-handling coroutine and returns the 620   /** Starts a worker's connection-handling coroutine and returns the
621   worker to the idle pool automatically. 621   worker to the idle pool automatically.
622   622  
623   Passed to @ref worker_base::run to start the connection-handling 623   Passed to @ref worker_base::run to start the connection-handling
624   coroutine. The launcher ensures the worker returns to the idle 624   coroutine. The launcher ensures the worker returns to the idle
625   pool when the coroutine completes or if starting fails. 625   pool when the coroutine completes or if starting fails.
626   626  
627   The launcher must be invoked exactly once via `operator()`. 627   The launcher must be invoked exactly once via `operator()`.
628   If destroyed without invoking, the worker is returned to the 628   If destroyed without invoking, the worker is returned to the
629   idle pool automatically. So is the worker of a coroutine 629   idle pool automatically. So is the worker of a coroutine
630   abandoned before it completes, as when the execution context 630   abandoned before it completes, as when the execution context
631   holding it is destroyed. 631   holding it is destroyed.
632   632  
633   @see worker_base::run 633   @see worker_base::run
634   */ 634   */
635   class BOOST_COROSIO_DECL launcher 635   class BOOST_COROSIO_DECL launcher
636   { 636   {
637   state* st_; 637   state* st_;
638   worker_base* w_; 638   worker_base* w_;
639   639  
640   friend class tcp_server; 640   friend class tcp_server;
641   641  
HITCBC 642   906 launcher(state& st, worker_base& w) noexcept : st_(&st), w_(&w) 642   1014 launcher(state& st, worker_base& w) noexcept : st_(&st), w_(&w)
643   { 643   {
HITCBC 644   906 add_ref(st_); 644   1014 add_ref(st_);
HITCBC 645   906 } 645   1014 }
646   646  
647   public: 647   public:
648   /// Return the worker to the pool if not started. 648   /// Return the worker to the pool if not started.
HITCBC 649   918 ~launcher() 649   1026 ~launcher()
650   { 650   {
HITCBC 651   918 if (w_) 651   1026 if (w_)
HITCBC 652   14 st_->drop_hold(*w_); 652   14 st_->drop_hold(*w_);
HITCBC 653   918 if (st_) 653   1026 if (st_)
HITCBC 654   906 release(st_); 654   1014 release(st_);
HITCBC 655   918 } 655   1026 }
656   656  
657   /** Move construct, transferring the borrowed worker. 657   /** Move construct, transferring the borrowed worker.
658   658  
659   @param o The launcher to take the worker from. It is left 659   @param o The launcher to take the worker from. It is left
660   holding none, so only one of the two returns it. 660   holding none, so only one of the two returns it.
661   */ 661   */
HITCBC 662   12 launcher(launcher&& o) noexcept 662   12 launcher(launcher&& o) noexcept
HITCBC 663   12 : st_(std::exchange(o.st_, nullptr)) 663   12 : st_(std::exchange(o.st_, nullptr))
HITCBC 664   12 , w_(std::exchange(o.w_, nullptr)) 664   12 , w_(std::exchange(o.w_, nullptr))
665   { 665   {
HITCBC 666   12 } 666   12 }
667   /// Copy construction is disabled; a launcher holds a borrowed worker it must return exactly once. 667   /// Copy construction is disabled; a launcher holds a borrowed worker it must return exactly once.
668   launcher(launcher const&) = delete; 668   launcher(launcher const&) = delete;
669   /// Copy assignment is disabled; a launcher holds a borrowed worker it must return exactly once. 669   /// Copy assignment is disabled; a launcher holds a borrowed worker it must return exactly once.
670   launcher& operator=(launcher const&) = delete; 670   launcher& operator=(launcher const&) = delete;
671   /// Move assignment is disabled; a launcher is moved, never reassigned. 671   /// Move assignment is disabled; a launcher is moved, never reassigned.
672   launcher& operator=(launcher&&) = delete; 672   launcher& operator=(launcher&&) = delete;
673   673  
674   /** Start the connection-handling coroutine. 674   /** Start the connection-handling coroutine.
675   675  
676   Starts the given coroutine on the specified executor, which 676   Starts the given coroutine on the specified executor, which
677   counts it as work until it ends. That execution context's 677   counts it as work until it ends. That execution context's
678   `run()` therefore does not return while a connection is 678   `run()` therefore does not return while a connection is
679   active. When the coroutine completes, or is abandoned before 679   active. When the coroutine completes, or is abandoned before
680   completing, the worker is automatically returned to the idle 680   completing, the worker is automatically returned to the idle
681   pool. 681   pool.
682   682  
683   @pre When @p ex belongs to another execution context, the 683   @pre When @p ex belongs to another execution context, the
684   server's context is not destroyed until the coroutine has 684   server's context is not destroyed until the coroutine has
685   finished. The coroutine holds the worker, whose socket 685   finished. The coroutine holds the worker, whose socket
686   belongs to the server's context. The coroutine's own 686   belongs to the server's context. The coroutine's own
687   context may be destroyed first, but only while the 687   context may be destroyed first, but only while the
688   coroutine is not suspended in an operation of the server's 688   coroutine is not suspended in an operation of the server's
689   context. An operation on the worker's socket is one such 689   context. An operation on the worker's socket is one such
690   operation. Destroying 690   operation. Destroying
691   that context then abandons the coroutine and returns its 691   that context then abandons the coroutine and returns its
692   worker. Otherwise stop the server and run both contexts 692   worker. Otherwise stop the server and run both contexts
693   until their `run()` returns first. 693   until their `run()` returns first.
694   694  
695   @pre @p task was created on a chain whose frame allocator 695   @pre @p task was created on a chain whose frame allocator
696   outlives the connection. The task's frame is allocated 696   outlives the connection. The task's frame is allocated
697   where it is created, and it lives until the connection 697   where it is created, and it lives until the connection
698   ends. Inside @ref worker_base::run that allocator is the 698   ends. Inside @ref worker_base::run that allocator is the
699   accept loop's, the frame allocator of the server's 699   accept loop's, the frame allocator of the server's
700   execution context. By default that is the process-wide 700   execution context. By default that is the process-wide
701   recycling resource, so a task created there meets this. 701   recycling resource, so a task created there meets this.
702   702  
703   @tparam Executor Executor type satisfying capy::Executor. 703   @tparam Executor Executor type satisfying capy::Executor.
704   704  
705   @param ex The executor to run the coroutine on. 705   @param ex The executor to run the coroutine on.
706   @param task The coroutine to execute. 706   @param task The coroutine to execute.
707   707  
708   @throws std::logic_error If this launcher was already invoked. 708   @throws std::logic_error If this launcher was already invoked.
709   */ 709   */
710   template<class Executor> 710   template<class Executor>
HITCBC 711   894 void operator()(Executor const& ex, capy::task<void> task) 711   1002 void operator()(Executor const& ex, capy::task<void> task)
712   { 712   {
HITCBC 713   894 if (!w_) 713   1002 if (!w_)
HITCBC 714   2 detail::throw_logic_error(); // launcher already invoked 714   2 detail::throw_logic_error(); // launcher already invoked
715   715  
HITCBC 716   892 auto* w = std::exchange(w_, nullptr); 716   1000 auto* w = std::exchange(w_, nullptr);
717   717  
718   // A stop_source allocates shared state on construction; 718   // A stop_source allocates shared state on construction;
719   // reuse the worker's across connections and replace it 719   // reuse the worker's across connections and replace it
720   // only once a stop has actually been delivered (the state 720   // only once a stop has actually been delivered (the state
721   // is then latched for good). Replaced before the worker is 721   // is then latched for good). Replaced before the worker is
722   // listed active, under the lock stop() copies it under, so 722   // listed active, under the lock stop() copies it under, so
723   // that copy never races the replacement or misses it. 723   // that copy never races the replacement or misses it.
HITCBC 724   892 std::stop_token st; 724   1000 std::stop_token st;
HITCBC 725   892 bool stopped_before = false; 725   1000 bool stopped_before = false;
726   { 726   {
HITCBC 727   892 std::lock_guard<std::mutex> lock(st_->mutex); 727   1000 std::lock_guard<std::mutex> lock(st_->mutex);
HITCBC 728   892 if (w->stop_.stop_requested()) 728   1000 if (w->stop_.stop_requested())
MISUBC 729   ✗ w->stop_ = {}; 729   ✗ w->stop_ = {};
HITCBC 730   892 st = w->stop_.get_token(); 730   1000 st = w->stop_.get_token();
HITCBC 731   892 st_->active_push(w); 731   1000 st_->active_push(w);
HITCBC 732   892 stopped_before = st_->stopping; 732   1000 stopped_before = st_->stopping;
HITCBC 733   892 } 733   1000 }
734   // A stop that came before this launch never saw the worker 734   // A stop that came before this launch never saw the worker
735   // listed. The token is not handed out yet, so no callback 735   // listed. The token is not handed out yet, so no callback
736   // runs here. 736   // runs here.
HITCBC 737   892 if (stopped_before) 737   1000 if (stopped_before)
HITCBC 738   13 w->stop_.request_stop(); 738   10 w->stop_.request_stop();
739   739  
HITCBC 740   892 worker_hold hold(*st_, *w); 740   1000 worker_hold hold(*st_, *w);
741   741  
742   // The connection can outlive whatever chain invoked this 742   // The connection can outlive whatever chain invoked this
743   // launcher, so its frame comes from the process-wide 743   // launcher, so its frame comes from the process-wide
744   // recycling resource rather than that chain's allocator. 744   // recycling resource rather than that chain's allocator.
745   struct allocator_scope 745   struct allocator_scope
746   { 746   {
747   std::pmr::memory_resource* prev = 747   std::pmr::memory_resource* prev =
748   capy::get_current_frame_allocator(); 748   capy::get_current_frame_allocator();
HITCBC 749   892 allocator_scope() noexcept 749   1000 allocator_scope() noexcept
HITCBC 750   892 { 750   1000 {
HITCBC 751   892 capy::set_current_frame_allocator( 751   1000 capy::set_current_frame_allocator(
752   capy::get_recycling_memory_resource()); 752   capy::get_recycling_memory_resource());
HITCBC 753   892 } 753   1000 }
HITCBC 754   892 ~allocator_scope() 754   1000 ~allocator_scope()
755   { 755   {
HITCBC 756   892 capy::set_current_frame_allocator(prev); 756   1000 capy::set_current_frame_allocator(prev);
HITCBC 757   892 } 757   1000 }
758   }; 758   };
HITCBC 759   2676 launch_wrapper<Executor> wrapper = [&] { 759   3000 launch_wrapper<Executor> wrapper = [&] {
HITCBC 760   892 allocator_scope scope; 760   1000 allocator_scope scope;
761   return launch_coro<Executor>{}( 761   return launch_coro<Executor>{}(
762   ex, st, 762   ex, st,
763   session<Executor>{ 763   session<Executor>{
HITCBC 764   892 std::move(hold), capy::work_guard<Executor>(ex), 764   1000 std::move(hold), capy::work_guard<Executor>(ex),
HITCBC 765   2676 std::move(task)}); 765   3000 std::move(task)});
HITCBC 766   892 }(); 766   1000 }();
767   767  
768   // Executor and stop token stored in promise via 768   // Executor and stop token stored in promise via
769   // constructor. Post through the frame-embedded 769   // constructor. Post through the frame-embedded
770   // continuation — posting the bare handle would allocate a 770   // continuation — posting the bare handle would allocate a
771   // wrapper op. The frame stays suspended until the 771   // wrapper op. The frame stays suspended until the
772   // executor resumes it, so the node outlives the queue. 772   // executor resumes it, so the node outlives the queue.
HITCBC 773   892 auto h = wrapper.h; 773   1000 auto h = wrapper.h;
HITCBC 774   892 h.promise().owner_.bind(h, h.promise().ex); 774   1000 h.promise().owner_.bind(h, h.promise().ex);
HITCBC 775   892 h.promise().cont_.h = h; 775   1000 h.promise().cont_.h = h;
776   // Released only once posted: a post that throws leaves the 776   // Released only once posted: a post that throws leaves the
777   // frame to the wrapper, which destroys it and with it the 777   // frame to the wrapper, which destroys it and with it the
778   // hold, the work, and the task. 778   // hold, the work, and the task.
HITCBC 779   892 ex.post(h.promise().cont_); 779   1000 ex.post(h.promise().cont_);
HITCBC 780   888 wrapper.h = nullptr; 780   996 wrapper.h = nullptr;
HITCBC 781   900 } 781   1008 }
782   }; 782   };
783   783  
784   /** Construct a TCP server. 784   /** Construct a TCP server.
785   785  
786   @tparam Ctx Execution context type satisfying ExecutionContext. 786   @tparam Ctx Execution context type satisfying ExecutionContext.
787   @tparam Ex Executor type satisfying Executor. 787   @tparam Ex Executor type satisfying Executor.
788   788  
789   @param ctx The execution context for socket operations. 789   @param ctx The execution context for socket operations.
790   @param ex The executor for dispatching coroutines. 790   @param ex The executor for dispatching coroutines.
791   791  
792   @par Example 792   @par Example
793   @par !example tcp_server 793   @par !example tcp_server
794   */ 794   */
795   template<capy::ExecutionContext Ctx, capy::Executor Ex> 795   template<capy::ExecutionContext Ctx, capy::Executor Ex>
HITCBC 796   169 tcp_server(Ctx& ctx, Ex ex) 796   169 tcp_server(Ctx& ctx, Ex ex)
HITCBC 797   169 : st_(make_state(ctx, capy::any_executor(std::move(ex)))) 797   169 : st_(make_state(ctx, capy::any_executor(std::move(ex))))
798   { 798   {
HITCBC 799   169 } 799   169 }
800   800  
801   public: 801   public:
802   /** Destroy the server, stopping it. 802   /** Destroy the server, stopping it.
803   803  
804   A running server is stopped and its listeners closed, so its 804   A running server is stopped and its listeners closed, so its
805   accept loops end. The loops and any active workers finish on 805   accept loops end. The loops and any active workers finish on
806   the server's executor without reaching the destroyed object; 806   the server's executor without reaching the destroyed object;
807   the workers stay alive until they do. 807   the workers stay alive until they do.
808   808  
809   @par Thread Safety 809   @par Thread Safety
810   Not thread safe. 810   Not thread safe.
811   */ 811   */
812   ~tcp_server(); 812   ~tcp_server();
813   813  
814   /// Copy construction is disabled; the server owns its worker storage. 814   /// Copy construction is disabled; the server owns its worker storage.
815   tcp_server(tcp_server const&) = delete; 815   tcp_server(tcp_server const&) = delete;
816   /// Copy assignment is disabled; the server owns its worker storage. 816   /// Copy assignment is disabled; the server owns its worker storage.
817   tcp_server& operator=(tcp_server const&) = delete; 817   tcp_server& operator=(tcp_server const&) = delete;
818   818  
819   /** Move construct from another server. 819   /** Move construct from another server.
820   820  
821   @param o The source server. After the move, @p o is 821   @param o The source server. After the move, @p o is
822   in a valid but unspecified state. 822   in a valid but unspecified state.
823   */ 823   */
824   tcp_server(tcp_server&& o) noexcept; 824   tcp_server(tcp_server&& o) noexcept;
825   825  
826   /** Move assign from another server. 826   /** Move assign from another server.
827   827  
828   @param o The source server. After the move, @p o is 828   @param o The source server. After the move, @p o is
829   in a valid but unspecified state. 829   in a valid but unspecified state.
830   830  
831   @return `*this`. 831   @return `*this`.
832   */ 832   */
833   tcp_server& operator=(tcp_server&& o) noexcept; 833   tcp_server& operator=(tcp_server&& o) noexcept;
834   834  
835   /** Bind to a local endpoint. 835   /** Bind to a local endpoint.
836   836  
837   Creates an acceptor listening on the specified endpoint. 837   Creates an acceptor listening on the specified endpoint.
838   Multiple endpoints can be bound by calling this method 838   Multiple endpoints can be bound by calling this method
839   multiple times before @ref start. 839   multiple times before @ref start.
840   840  
841   @pre No session is active: the server was never started, or 841   @pre No session is active: the server was never started, or
842   it was stopped and joined. 842   it was stopped and joined.
843   843  
844   @param ep The local endpoint to bind to. 844   @param ep The local endpoint to bind to.
845   845  
846   @return An error code indicating success, or the reason binding 846   @return An error code indicating success, or the reason binding
847   failed. 847   failed.
848   */ 848   */
849   [[nodiscard]] std::error_code bind(endpoint ep); 849   [[nodiscard]] std::error_code bind(endpoint ep);
850   850  
851   /** Set the worker pool. 851   /** Set the worker pool.
852   852  
853   Replaces any existing workers with the given range. Any 853   Replaces any existing workers with the given range. Any
854   previous workers are released and the idle/active lists 854   previous workers are released and the idle/active lists
855   are cleared before populating with new workers. 855   are cleared before populating with new workers.
856   856  
857   @pre No worker is waiting on or serving a connection: the 857   @pre No worker is waiting on or serving a connection: the
858   server was never started, or it was stopped and joined 858   server was never started, or it was stopped and joined
859   and every connection has finished. 859   and every connection has finished.
860   860  
861   @par Exception Safety 861   @par Exception Safety
862   The server keeps its current workers if this throws. 862   The server keeps its current workers if this throws.
863   863  
864   @throws std::logic_error If a worker is waiting on or serving 864   @throws std::logic_error If a worker is waiting on or serving
865   a connection. 865   a connection.
866   866  
867   @tparam Range Forward range of pointer-like objects to worker_base. 867   @tparam Range Forward range of pointer-like objects to worker_base.
868   868  
869   @param workers Range of workers to manage. Each element must 869   @param workers Range of workers to manage. Each element must
870   support `std::to_address()` yielding `worker_base*`. 870   support `std::to_address()` yielding `worker_base*`.
871   871  
872   @par Example 872   @par Example
873   @par !example set_workers 873   @par !example set_workers
874   */ 874   */
875   template<std::ranges::forward_range Range> 875   template<std::ranges::forward_range Range>
876   requires std::convertible_to< 876   requires std::convertible_to<
877   decltype(std::to_address( 877   decltype(std::to_address(
878   std::declval<std::ranges::range_value_t<Range>&>())), 878   std::declval<std::ranges::range_value_t<Range>&>())),
879   worker_base*> 879   worker_base*>
HITCBC 880   183 void set_workers(Range&& workers) 880   183 void set_workers(Range&& workers)
881   { 881   {
882   // The previous workers are released after the lock. 882   // The previous workers are released after the lock.
HITCBC 883   183 std::shared_ptr<void> previous; 883   183 std::shared_ptr<void> previous;
HITCBC 884   183 std::lock_guard<std::mutex> lock(st_->mutex); 884   183 std::lock_guard<std::mutex> lock(st_->mutex);
HITCBC 885   183 if (st_->lent != 0) 885   183 if (st_->lent != 0)
HITCBC 886   6 detail::throw_logic_error( 886   6 detail::throw_logic_error(
887   "tcp_server::set_workers: a worker is still in use"); 887   "tcp_server::set_workers: a worker is still in use");
888   888  
889   using StorageType = std::decay_t<Range>; 889   using StorageType = std::decay_t<Range>;
HITCBC 890   177 auto* p = new StorageType(std::forward<Range>(workers)); 890   177 auto* p = new StorageType(std::forward<Range>(workers));
HITCBC 891   177 std::shared_ptr<void> storage( 891   177 std::shared_ptr<void> storage(
HITCBC 892   177 p, [](void* ptr) { delete static_cast<StorageType*>(ptr); }); 892   177 p, [](void* ptr) { delete static_cast<StorageType*>(ptr); });
893   893  
HITCBC 894   177 st_->idle_head = nullptr; 894   177 st_->idle_head = nullptr;
HITCBC 895   177 st_->active_head = nullptr; 895   177 st_->active_head = nullptr;
HITCBC 896   177 st_->active_tail = nullptr; 896   177 st_->active_tail = nullptr;
HITCBC 897   177 previous = std::exchange(st_->storage, std::move(storage)); 897   177 previous = std::exchange(st_->storage, std::move(storage));
HITCBC 898   590 for (auto&& elem : *p) 898   590 for (auto&& elem : *p)
HITCBC 899   413 st_->idle_push(std::to_address(elem)); 899   413 st_->idle_push(std::to_address(elem));
HITCBC 900   189 } 900   189 }
901   901  
902   /** Start accepting connections. 902   /** Start accepting connections.
903   903  
904   Starts accept loops for all bound endpoints. Incoming 904   Starts accept loops for all bound endpoints. Incoming
905   connections are dispatched to idle workers from the pool. 905   connections are dispatched to idle workers from the pool.
906   906  
907   Calling `start()` on an already-running server has no effect. 907   Calling `start()` on an already-running server has no effect.
908   908  
909   @pre At least one endpoint bound via @ref bind. 909   @pre At least one endpoint bound via @ref bind.
910   @pre Workers provided via @ref set_workers. 910   @pre Workers provided via @ref set_workers.
911   @pre If restarting, @ref join must have completed first, and the 911   @pre If restarting, @ref join must have completed first, and the
912   `io_context` must be restarted (`ioc.restart()`). 912   `io_context` must be restarted (`ioc.restart()`).
913   913  
914   @par Effects 914   @par Effects
915   Creates one accept coroutine per bound endpoint. Each coroutine 915   Creates one accept coroutine per bound endpoint. Each coroutine
916   runs on the server's executor, waiting for connections and 916   runs on the server's executor, waiting for connections and
917   dispatching them to idle workers. 917   dispatching them to idle workers.
918   918  
919   @par Restart Sequence 919   @par Restart Sequence
920   To restart after stopping, complete the full shutdown cycle: 920   To restart after stopping, complete the full shutdown cycle:
921   @par !example start 921   @par !example start
922   922  
923   @par Thread Safety 923   @par Thread Safety
924   Not thread safe. 924   Not thread safe.
925   925  
926   @throws std::logic_error If a previous session has not been 926   @throws std::logic_error If a previous session has not been
927   joined (accept loops still active). 927   joined (accept loops still active).
928   */ 928   */
929   void start(); 929   void start();
930   930  
931   /** Return the local endpoint for the i-th bound port. 931   /** Return the local endpoint for the i-th bound port.
932   932  
933   @param index Zero-based index into the list of bound ports. 933   @param index Zero-based index into the list of bound ports.
934   934  
935   @return The local endpoint, or a default-constructed endpoint 935   @return The local endpoint, or a default-constructed endpoint
936   if @p index is out of range or the acceptor is not open. 936   if @p index is out of range or the acceptor is not open.
937   */ 937   */
938   endpoint local_endpoint(std::size_t index = 0) const noexcept; 938   endpoint local_endpoint(std::size_t index = 0) const noexcept;
939   939  
940   /** Stop accepting connections. 940   /** Stop accepting connections.
941   941  
942   Requests the accept loops' stop token and requests cancellation 942   Requests the accept loops' stop token and requests cancellation
943   of active workers via their stop tokens. The acceptors are not 943   of active workers via their stop tokens. The acceptors are not
944   closed. A suspended accept completes once more before its loop 944   closed. A suspended accept completes once more before its loop
945   observes the stop token and ends. 945   observes the stop token and ends.
946   946  
947   This function returns immediately; it does not wait for workers 947   This function returns immediately; it does not wait for workers
948   to finish. Pending I/O operations complete asynchronously. 948   to finish. Pending I/O operations complete asynchronously.
949   949  
950   Calling `stop()` on a non-running server has no effect. 950   Calling `stop()` on a non-running server has no effect.
951   951  
952   @par Effects 952   @par Effects
953   - Requests stop on the accept loops' stop token. The acceptors 953   - Requests stop on the accept loops' stop token. The acceptors
954   are not closed; a pending accept completes once more before 954   are not closed; a pending accept completes once more before
955   the accept loop ends. 955   the accept loop ends.
956   - Requests stop on each active worker's stop token. 956   - Requests stop on each active worker's stop token.
957   - Workers observing their stop token should exit promptly. 957   - Workers observing their stop token should exit promptly.
958   - A worker reuses its stop token across connections until a stop 958   - A worker reuses its stop token across connections until a stop
959   is requested. Work that keeps a connection's token past that 959   is requested. Work that keeps a connection's token past that
960   connection's end therefore also observes this stop. 960   connection's end therefore also observes this stop.
961   961  
962   @par Postconditions 962   @par Postconditions
963   The server accepts no new connections. Active workers continue 963   The server accepts no new connections. Active workers continue
964   until they observe their stop token or complete naturally. 964   until they observe their stop token or complete naturally.
965   965  
966   @par What Happens Next 966   @par What Happens Next
967   After calling `stop()`: 967   After calling `stop()`:
968   1. Let `ioc.run()` return (drains pending completions). 968   1. Let `ioc.run()` return (drains pending completions).
969   2. Call @ref join to wait for accept loops to finish. 969   2. Call @ref join to wait for accept loops to finish.
970   3. Only then may the server be restarted. It may be destroyed at 970   3. Only then may the server be restarted. It may be destroyed at
971   any time; see @ref ~tcp_server. 971   any time; see @ref ~tcp_server.
972   972  
973   @par Thread Safety 973   @par Thread Safety
974   Not thread safe. 974   Not thread safe.
975   975  
976   @see join, start 976   @see join, start
977   */ 977   */
978   void stop(); 978   void stop();
979   979  
980   /** Block until all accept loops complete. 980   /** Block until all accept loops complete.
981   981  
982   Blocks the calling thread until all accept coroutines started 982   Blocks the calling thread until all accept coroutines started
983   by @ref start have finished executing. This synchronizes the 983   by @ref start have finished executing. This synchronizes the
984   shutdown sequence, ensuring the server is fully stopped before 984   shutdown sequence, ensuring the server is fully stopped before
985   restarting or destroying it. 985   restarting or destroying it.
986   986  
987   @pre @ref stop was called and `ioc.run()` returned. 987   @pre @ref stop was called and `ioc.run()` returned.
988   988  
989   @par Postconditions 989   @par Postconditions
990   All accept loops have completed. The server is in the stopped 990   All accept loops have completed. The server is in the stopped
991   state and may be restarted via @ref start. 991   state and may be restarted via @ref start.
992   992  
993   @par Example (Correct Usage) 993   @par Example (Correct Usage)
994   @par !example correct_usage 994   @par !example correct_usage
995   995  
996   @par WARNING: Deadlock Scenario 996   @par WARNING: Deadlock Scenario
997   Calling `join()` from inside a worker coroutine deadlocks: 997   Calling `join()` from inside a worker coroutine deadlocks:
998   998  
999   @par !example deadlock_scenarios 999   @par !example deadlock_scenarios
1000   1000  
1001   @par Thread Safety 1001   @par Thread Safety
1002   May be called from any thread. It deadlocks if called 1002   May be called from any thread. It deadlocks if called
1003   from within the `io_context` event loop or from a worker coroutine. 1003   from within the `io_context` event loop or from a worker coroutine.
1004   1004  
1005   @see stop, start 1005   @see stop, start
1006   */ 1006   */
1007   void join(); 1007   void join();
1008   1008  
1009   private: 1009   private:
1010   void discard() noexcept; 1010   void discard() noexcept;
1011   }; 1011   };
1012   1012  
1013   #ifdef _MSC_VER 1013   #ifdef _MSC_VER
1014   #pragma warning(pop) 1014   #pragma warning(pop)
1015   #endif 1015   #endif
1016   1016  
1017   } // namespace boost::corosio 1017   } // namespace boost::corosio
1018   1018  
1019   #endif 1019   #endif