. /include/boost/corosio /tcp_server.hpp
93.79% Lines (136/145) 97.22% Functions (35/36)
[ ] 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   // 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/corosio 7   // Official repository: https://github.com/cppalliance/corosio
8   // 8   //
9   9  
10   #ifndef BOOST_COROSIO_TCP_SERVER_HPP 10   #ifndef BOOST_COROSIO_TCP_SERVER_HPP
11   #define BOOST_COROSIO_TCP_SERVER_HPP 11   #define BOOST_COROSIO_TCP_SERVER_HPP
12   12  
13   #include <boost/corosio/detail/config.hpp> 13   #include <boost/corosio/detail/config.hpp>
14   #include <boost/corosio/detail/except.hpp> 14   #include <boost/corosio/detail/except.hpp>
15   #include <boost/corosio/tcp_acceptor.hpp> 15   #include <boost/corosio/tcp_acceptor.hpp>
16   #include <boost/corosio/tcp_socket.hpp> 16   #include <boost/corosio/tcp_socket.hpp>
17   #include <boost/corosio/io_context.hpp> 17   #include <boost/corosio/io_context.hpp>
18   #include <boost/corosio/endpoint.hpp> 18   #include <boost/corosio/endpoint.hpp>
19   #include <boost/capy/task.hpp> 19   #include <boost/capy/task.hpp>
20   #include <boost/capy/concept/execution_context.hpp> 20   #include <boost/capy/concept/execution_context.hpp>
21   #include <boost/capy/concept/io_awaitable.hpp> 21   #include <boost/capy/concept/io_awaitable.hpp>
22   #include <boost/capy/concept/executor.hpp> 22   #include <boost/capy/concept/executor.hpp>
23   #include <boost/capy/ex/any_executor.hpp> 23   #include <boost/capy/ex/any_executor.hpp>
24   #include <boost/capy/ex/frame_allocator.hpp> 24   #include <boost/capy/ex/frame_allocator.hpp>
25   #include <boost/capy/ex/io_env.hpp> 25   #include <boost/capy/ex/io_env.hpp>
26   #include <boost/capy/ex/run_async.hpp> 26   #include <boost/capy/ex/run_async.hpp>
27   27  
28   #include <coroutine> 28   #include <coroutine>
29   #include <memory> 29   #include <memory>
30   #include <ranges> 30   #include <ranges>
31   #include <vector> 31   #include <vector>
32   32  
33   namespace boost::corosio { 33   namespace boost::corosio {
34   34  
35   #ifdef _MSC_VER 35   #ifdef _MSC_VER
36   #pragma warning(push) 36   #pragma warning(push)
37   #pragma warning(disable : 4251) // class needs to have dll-interface 37   #pragma warning(disable : 4251) // class needs to have dll-interface
38   #endif 38   #endif
39   39  
40   /** TCP server with pooled workers. 40   /** TCP server with pooled workers.
41   41  
42   This class manages a pool of reusable worker objects that handle 42   This class manages a pool of reusable worker objects that handle
43   incoming connections. When a connection arrives, an idle worker 43   incoming connections. When a connection arrives, an idle worker
44   is dispatched to handle it. After the connection completes, the 44   is dispatched to handle it. After the connection completes, the
45   worker returns to the pool for reuse, avoiding allocation overhead 45   worker returns to the pool for reuse, avoiding allocation overhead
46   per connection. 46   per connection.
47   47  
48   Workers are set via @ref set_workers as a forward range of 48   Workers are set via @ref set_workers as a forward range of
49   pointer-like objects (e.g., `unique_ptr<worker_base>`). The server 49   pointer-like objects (e.g., `unique_ptr<worker_base>`). The server
50   takes ownership of the container via type erasure. 50   takes ownership of the container via type erasure.
51   51  
52   @par Thread Safety 52   @par Thread Safety
53   Distinct objects: Safe. 53   Distinct objects: Safe.
54   Shared objects: Unsafe. 54   Shared objects: Unsafe.
55   55  
56   @par Lifecycle 56   @par Lifecycle
57   The server operates in three states: 57   The server operates in three states:
58   58  
59   - **Stopped**: Initial state, or after @ref join completes. 59   - **Stopped**: Initial state, or after @ref join completes.
60   - **Running**: After @ref start, actively accepting connections. 60   - **Running**: After @ref start, actively accepting connections.
61   - **Stopping**: After @ref stop, draining active work. 61   - **Stopping**: After @ref stop, draining active work.
62   62  
63   State transitions: 63   State transitions:
64   @code 64   @code
65   [Stopped] --start()--> [Running] --stop()--> [Stopping] --join()--> [Stopped] 65   [Stopped] --start()--> [Running] --stop()--> [Stopping] --join()--> [Stopped]
66   @endcode 66   @endcode
67   67  
68   @par Running the Server 68   @par Running the Server
69   @code 69   @code
70   io_context ioc; 70   io_context ioc;
71   tcp_server srv(ioc, ioc.get_executor()); 71   tcp_server srv(ioc, ioc.get_executor());
72   srv.set_workers(make_workers(ioc, 100)); 72   srv.set_workers(make_workers(ioc, 100));
73   if (auto ec = srv.bind(endpoint{ipv4_address::any(), 8080})) 73   if (auto ec = srv.bind(endpoint{ipv4_address::any(), 8080}))
74   return; 74   return;
75   srv.start(); 75   srv.start();
76   ioc.run(); // Blocks until all work completes 76   ioc.run(); // Blocks until all work completes
77   @endcode 77   @endcode
78   78  
79   @par Graceful Shutdown 79   @par Graceful Shutdown
80   To shut down gracefully, call @ref stop then drain the io_context: 80   To shut down gracefully, call @ref stop then drain the io_context:
81   @code 81   @code
82   // From a signal handler or timer callback: 82   // From a signal handler or timer callback:
83   srv.stop(); 83   srv.stop();
84   84  
85   // ioc.run() returns after pending work drains. 85   // ioc.run() returns after pending work drains.
86   // Then from the thread that called ioc.run(): 86   // Then from the thread that called ioc.run():
87   srv.join(); // Wait for accept loops to finish 87   srv.join(); // Wait for accept loops to finish
88   @endcode 88   @endcode
89   89  
90   @par Restart After Stop 90   @par Restart After Stop
91   The server can be restarted after a complete shutdown cycle. 91   The server can be restarted after a complete shutdown cycle.
92   You must drain the io_context and call @ref join before restarting: 92   You must drain the io_context and call @ref join before restarting:
93   @code 93   @code
94   srv.start(); 94   srv.start();
95   ioc.run_for( 10s ); // Run for a while 95   ioc.run_for( 10s ); // Run for a while
96   srv.stop(); // Signal shutdown 96   srv.stop(); // Signal shutdown
97   ioc.run(); // REQUIRED: drain pending completions 97   ioc.run(); // REQUIRED: drain pending completions
98   srv.join(); // REQUIRED: wait for accept loops 98   srv.join(); // REQUIRED: wait for accept loops
99   99  
100   // Now safe to restart 100   // Now safe to restart
101   srv.start(); 101   srv.start();
102   ioc.run(); 102   ioc.run();
103   @endcode 103   @endcode
104   104  
105   @par WARNING: What NOT to Do 105   @par WARNING: What NOT to Do
106   - Do NOT call @ref join from inside a worker coroutine (deadlock). 106   - Do NOT call @ref join from inside a worker coroutine (deadlock).
107   - Do NOT call @ref join from a thread running `ioc.run()` (deadlock). 107   - Do NOT call @ref join from a thread running `ioc.run()` (deadlock).
108   - Do NOT call @ref start without completing @ref join after @ref stop. 108   - Do NOT call @ref start without completing @ref join after @ref stop.
109   - Do NOT call `ioc.stop()` for graceful shutdown; use @ref stop instead. 109   - Do NOT call `ioc.stop()` for graceful shutdown; use @ref stop instead.
110   110  
111   @par Example 111   @par Example
112   @code 112   @code
113   class my_worker : public tcp_server::worker_base 113   class my_worker : public tcp_server::worker_base
114   { 114   {
115   corosio::tcp_socket sock_; 115   corosio::tcp_socket sock_;
116   capy::any_executor ex_; 116   capy::any_executor ex_;
117   public: 117   public:
118   my_worker(io_context& ctx) 118   my_worker(io_context& ctx)
119   : sock_(ctx) 119   : sock_(ctx)
120   , ex_(ctx.get_executor()) 120   , ex_(ctx.get_executor())
121   { 121   {
122   } 122   }
123   123  
124   corosio::tcp_socket& socket() override { return sock_; } 124   corosio::tcp_socket& socket() override { return sock_; }
125   125  
126   void run(launcher launch) override 126   void run(launcher launch) override
127   { 127   {
128   launch(ex_, [](corosio::tcp_socket* sock) -> capy::task<> 128   launch(ex_, [](corosio::tcp_socket* sock) -> capy::task<>
129   { 129   {
130   // handle connection using sock 130   // handle connection using sock
131   co_return; 131   co_return;
132   }(&sock_)); 132   }(&sock_));
133   } 133   }
134   }; 134   };
135   135  
136   auto make_workers(io_context& ctx, int n) 136   auto make_workers(io_context& ctx, int n)
137   { 137   {
138   std::vector<std::unique_ptr<tcp_server::worker_base>> v; 138   std::vector<std::unique_ptr<tcp_server::worker_base>> v;
139   v.reserve(n); 139   v.reserve(n);
140   for(int i = 0; i < n; ++i) 140   for(int i = 0; i < n; ++i)
141   v.push_back(std::make_unique<my_worker>(ctx)); 141   v.push_back(std::make_unique<my_worker>(ctx));
142   return v; 142   return v;
143   } 143   }
144   144  
145   io_context ioc; 145   io_context ioc;
146   tcp_server srv(ioc, ioc.get_executor()); 146   tcp_server srv(ioc, ioc.get_executor());
147   srv.set_workers(make_workers(ioc, 100)); 147   srv.set_workers(make_workers(ioc, 100));
148   @endcode 148   @endcode
149   149  
150   @see worker_base, set_workers, launcher 150   @see worker_base, set_workers, launcher
151   */ 151   */
152   class BOOST_COROSIO_DECL tcp_server 152   class BOOST_COROSIO_DECL tcp_server
153   { 153   {
154   public: 154   public:
155   class worker_base; ///< Abstract base for connection handlers. 155   class worker_base; ///< Abstract base for connection handlers.
156   class launcher; ///< Move-only handle to launch worker coroutines. 156   class launcher; ///< Move-only handle to launch worker coroutines.
157   157  
158   private: 158   private:
159   struct waiter 159   struct waiter
160   { 160   {
161   waiter* next; 161   waiter* next;
162   std::coroutine_handle<> h; 162   std::coroutine_handle<> h;
163   capy::continuation cont; 163   capy::continuation cont;
164   worker_base* w; 164   worker_base* w;
165   }; 165   };
166   166  
167   struct impl; 167   struct impl;
168   168  
169   static impl* make_impl(capy::execution_context& ctx); 169   static impl* make_impl(capy::execution_context& ctx);
170   170  
171   impl* impl_; 171   impl* impl_;
172   capy::any_executor ex_; 172   capy::any_executor ex_;
173   waiter* waiters_ = nullptr; 173   waiter* waiters_ = nullptr;
174   worker_base* idle_head_ = nullptr; // Forward list: available workers 174   worker_base* idle_head_ = nullptr; // Forward list: available workers
175   worker_base* active_head_ = 175   worker_base* active_head_ =
176   nullptr; // Doubly linked: workers handling connections 176   nullptr; // Doubly linked: workers handling connections
177   worker_base* active_tail_ = nullptr; // Tail for O(1) push_back 177   worker_base* active_tail_ = nullptr; // Tail for O(1) push_back
178   std::size_t active_accepts_ = 0; // Number of active do_accept coroutines 178   std::size_t active_accepts_ = 0; // Number of active do_accept coroutines
179   std::shared_ptr<void> storage_; // Owns the worker container (type-erased) 179   std::shared_ptr<void> storage_; // Owns the worker container (type-erased)
180   bool running_ = false; 180   bool running_ = false;
181   181  
182   // Idle list (forward/singly linked) - push front, pop front 182   // Idle list (forward/singly linked) - push front, pop front
HITCBC 183   162 void idle_push(worker_base* w) noexcept 183   162 void idle_push(worker_base* w) noexcept
184   { 184   {
HITCBC 185   162 w->next_ = idle_head_; 185   162 w->next_ = idle_head_;
HITCBC 186   162 idle_head_ = w; 186   162 idle_head_ = w;
HITCBC 187   162 } 187   162 }
188   188  
HITCBC 189   36 worker_base* idle_pop() noexcept 189   36 worker_base* idle_pop() noexcept
190   { 190   {
HITCBC 191   36 auto* w = idle_head_; 191   36 auto* w = idle_head_;
HITCBC 192   36 if (w) 192   36 if (w)
HITCBC 193   36 idle_head_ = w->next_; 193   36 idle_head_ = w->next_;
HITCBC 194   36 return w; 194   36 return w;
195   } 195   }
196   196  
HITCBC 197   42 bool idle_empty() const noexcept 197   42 bool idle_empty() const noexcept
198   { 198   {
HITCBC 199   42 return idle_head_ == nullptr; 199   42 return idle_head_ == nullptr;
200   } 200   }
201   201  
202   // Active list (doubly linked) - push back, remove anywhere 202   // Active list (doubly linked) - push back, remove anywhere
HITCBC 203   18 void active_push(worker_base* w) noexcept 203   18 void active_push(worker_base* w) noexcept
204   { 204   {
HITCBC 205   18 w->next_ = nullptr; 205   18 w->next_ = nullptr;
HITCBC 206   18 w->prev_ = active_tail_; 206   18 w->prev_ = active_tail_;
HITCBC 207   18 if (active_tail_) 207   18 if (active_tail_)
HITCBC 208   4 active_tail_->next_ = w; 208   4 active_tail_->next_ = w;
209   else 209   else
HITCBC 210   14 active_head_ = w; 210   14 active_head_ = w;
HITCBC 211   18 active_tail_ = w; 211   18 active_tail_ = w;
HITCBC 212   18 } 212   18 }
213   213  
HITCBC 214   42 void active_remove(worker_base* w) noexcept 214   42 void active_remove(worker_base* w) noexcept
215   { 215   {
216   // Skip if not in active list (e.g., after failed accept) 216   // Skip if not in active list (e.g., after failed accept)
HITCBC 217   42 if (w != active_head_ && w->prev_ == nullptr) 217   42 if (w != active_head_ && w->prev_ == nullptr)
HITCBC 218   24 return; 218   24 return;
HITCBC 219   18 if (w->prev_) 219   18 if (w->prev_)
HITCBC 220   4 w->prev_->next_ = w->next_; 220   4 w->prev_->next_ = w->next_;
221   else 221   else
HITCBC 222   14 active_head_ = w->next_; 222   14 active_head_ = w->next_;
HITCBC 223   18 if (w->next_) 223   18 if (w->next_)
HITCBC 224   2 w->next_->prev_ = w->prev_; 224   2 w->next_->prev_ = w->prev_;
225   else 225   else
HITCBC 226   16 active_tail_ = w->prev_; 226   16 active_tail_ = w->prev_;
HITCBC 227   18 w->prev_ = nullptr; // Mark as not in active list 227   18 w->prev_ = nullptr; // Mark as not in active list
228   } 228   }
229   229  
230   template<capy::Executor Ex> 230   template<capy::Executor Ex>
231   struct launch_wrapper 231   struct launch_wrapper
232   { 232   {
233   struct promise_type 233   struct promise_type
234   { 234   {
235   Ex ex; // Executor stored directly in frame (outlives child tasks) 235   Ex ex; // Executor stored directly in frame (outlives child tasks)
236   capy::io_env env_; 236   capy::io_env env_;
237   237  
238   // For regular coroutines: first arg is executor, second is stop token 238   // For regular coroutines: first arg is executor, second is stop token
239   template<class E, class S, class... Args> 239   template<class E, class S, class... Args>
240   requires capy::Executor<std::decay_t<E>> 240   requires capy::Executor<std::decay_t<E>>
241   promise_type(E e, S s, Args&&...) 241   promise_type(E e, S s, Args&&...)
242   : ex(std::move(e)) 242   : ex(std::move(e))
243   , env_{ 243   , env_{
244   capy::executor_ref(ex), std::move(s), 244   capy::executor_ref(ex), std::move(s),
245   capy::get_current_frame_allocator()} 245   capy::get_current_frame_allocator()}
246   { 246   {
247   } 247   }
248   248  
249   // For lambda coroutines: first arg is closure, second is executor, third is stop token 249   // For lambda coroutines: first arg is closure, second is executor, third is stop token
250   template<class Closure, class E, class S, class... Args> 250   template<class Closure, class E, class S, class... Args>
251   requires(!capy::Executor<std::decay_t<Closure>> && 251   requires(!capy::Executor<std::decay_t<Closure>> &&
252   capy::Executor<std::decay_t<E>>) 252   capy::Executor<std::decay_t<E>>)
HITCBC 253   18 promise_type(Closure&&, E e, S s, Args&&...) 253   18 promise_type(Closure&&, E e, S s, Args&&...)
HITCBC 254   18 : ex(std::move(e)) 254   18 : ex(std::move(e))
HITCBC 255   18 , env_{ 255   18 , env_{
HITCBC 256   18 capy::executor_ref(ex), std::move(s), 256   18 capy::executor_ref(ex), std::move(s),
HITCBC 257   18 capy::get_current_frame_allocator()} 257   18 capy::get_current_frame_allocator()}
258   { 258   {
HITCBC 259   18 } 259   18 }
260   260  
HITCBC 261   18 launch_wrapper get_return_object() noexcept 261   18 launch_wrapper get_return_object() noexcept
262   { 262   {
263   return { 263   return {
HITCBC 264   18 std::coroutine_handle<promise_type>::from_promise(*this)}; 264   18 std::coroutine_handle<promise_type>::from_promise(*this)};
265   } 265   }
HITCBC 266   18 std::suspend_always initial_suspend() noexcept 266   18 std::suspend_always initial_suspend() noexcept
267   { 267   {
HITCBC 268   18 return {}; 268   18 return {};
269   } 269   }
HITCBC 270   18 std::suspend_never final_suspend() noexcept 270   18 std::suspend_never final_suspend() noexcept
271   { 271   {
HITCBC 272   18 return {}; 272   18 return {};
273   } 273   }
HITCBC 274   18 void return_void() noexcept {} 274   18 void return_void() noexcept {}
MISUBC 275   ✗ void unhandled_exception() 275   ✗ void unhandled_exception()
276   { 276   {
MISUBC 277   ✗ std::terminate(); 277   ✗ std::terminate();
278   } 278   }
279   279  
280   // Inject io_env for IoAwaitable 280   // Inject io_env for IoAwaitable
281   template<capy::IoAwaitable Awaitable> 281   template<capy::IoAwaitable Awaitable>
HITCBC 282   36 auto await_transform(Awaitable&& a) 282   36 auto await_transform(Awaitable&& a)
283   { 283   {
284   using AwaitableT = std::decay_t<Awaitable>; 284   using AwaitableT = std::decay_t<Awaitable>;
285   struct adapter 285   struct adapter
286   { 286   {
287   AwaitableT aw; 287   AwaitableT aw;
288   capy::io_env const* env; 288   capy::io_env const* env;
289   289  
HITCBC 290   36 bool await_ready() 290   36 bool await_ready()
291   { 291   {
HITCBC 292   36 return aw.await_ready(); 292   36 return aw.await_ready();
293   } 293   }
HITCBC 294   36 decltype(auto) await_resume() 294   36 decltype(auto) await_resume()
295   { 295   {
HITCBC 296   36 return aw.await_resume(); 296   36 return aw.await_resume();
297   } 297   }
298   298  
HITCBC 299   36 auto await_suspend(std::coroutine_handle<promise_type> h) 299   36 auto await_suspend(std::coroutine_handle<promise_type> h)
300   { 300   {
HITCBC 301   36 return aw.await_suspend(h, env); 301   36 return aw.await_suspend(h, env);
302   } 302   }
303   }; 303   };
HITCBC 304   54 return adapter{std::forward<Awaitable>(a), &env_}; 304   54 return adapter{std::forward<Awaitable>(a), &env_};
HITCBC 305   18 } 305   18 }
306   }; 306   };
307   307  
308   std::coroutine_handle<promise_type> h; 308   std::coroutine_handle<promise_type> h;
309   309  
HITCBC 310   18 launch_wrapper(std::coroutine_handle<promise_type> handle) noexcept 310   18 launch_wrapper(std::coroutine_handle<promise_type> handle) noexcept
HITCBC 311   18 : h(handle) 311   18 : h(handle)
312   { 312   {
HITCBC 313   18 } 313   18 }
314   314  
HITCBC 315   18 ~launch_wrapper() 315   18 ~launch_wrapper()
316   { 316   {
HITCBC 317   18 if (h) 317   18 if (h)
MISUBC 318   ✗ h.destroy(); 318   ✗ h.destroy();
HITCBC 319   18 } 319   18 }
320   320  
321   launch_wrapper(launch_wrapper&& o) noexcept 321   launch_wrapper(launch_wrapper&& o) noexcept
322   : h(std::exchange(o.h, nullptr)) 322   : h(std::exchange(o.h, nullptr))
323   { 323   {
324   } 324   }
325   325  
326   launch_wrapper(launch_wrapper const&) = delete; 326   launch_wrapper(launch_wrapper const&) = delete;
327   launch_wrapper& operator=(launch_wrapper const&) = delete; 327   launch_wrapper& operator=(launch_wrapper const&) = delete;
328   launch_wrapper& operator=(launch_wrapper&&) = delete; 328   launch_wrapper& operator=(launch_wrapper&&) = delete;
329   }; 329   };
330   330  
331   // Named functor to avoid incomplete lambda type in coroutine promise 331   // Named functor to avoid incomplete lambda type in coroutine promise
332   template<class Executor> 332   template<class Executor>
333   struct launch_coro 333   struct launch_coro
334   { 334   {
HITCBC 335   18 launch_wrapper<Executor> operator()( 335   18 launch_wrapper<Executor> operator()(
336   Executor, 336   Executor,
337   std::stop_token, 337   std::stop_token,
338   tcp_server* self, 338   tcp_server* self,
339   capy::task<void> t, 339   capy::task<void> t,
340   worker_base* wp) 340   worker_base* wp)
341   { 341   {
342   // Executor and stop token stored in promise via constructor 342   // Executor and stop token stored in promise via constructor
343   co_await std::move(t); 343   co_await std::move(t);
344   co_await self->push(*wp); // worker goes back to idle list 344   co_await self->push(*wp); // worker goes back to idle list
HITCBC 345   36 } 345   36 }
346   }; 346   };
347   347  
348   class push_awaitable 348   class push_awaitable
349   { 349   {
350   tcp_server& self_; 350   tcp_server& self_;
351   worker_base& w_; 351   worker_base& w_;
352   capy::continuation cont_; 352   capy::continuation cont_;
353   353  
354   public: 354   public:
HITCBC 355   38 push_awaitable(tcp_server& self, worker_base& w) noexcept 355   38 push_awaitable(tcp_server& self, worker_base& w) noexcept
HITCBC 356   38 : self_(self) 356   38 : self_(self)
HITCBC 357   38 , w_(w) 357   38 , w_(w)
358   { 358   {
HITCBC 359   38 } 359   38 }
360   360  
HITCBC 361   38 bool await_ready() const noexcept 361   38 bool await_ready() const noexcept
362   { 362   {
HITCBC 363   38 return false; 363   38 return false;
364   } 364   }
365   365  
366   std::coroutine_handle<> 366   std::coroutine_handle<>
HITCBC 367   38 await_suspend(std::coroutine_handle<> h, capy::io_env const*) noexcept 367   38 await_suspend(std::coroutine_handle<> h, capy::io_env const*) noexcept
368   { 368   {
369   // Symmetric transfer to server's executor 369   // Symmetric transfer to server's executor
HITCBC 370   38 cont_.h = h; 370   38 cont_.h = h;
HITCBC 371   38 return self_.ex_.dispatch(cont_); 371   38 return self_.ex_.dispatch(cont_);
372   } 372   }
373   373  
HITCBC 374   38 void await_resume() noexcept 374   38 void await_resume() noexcept
375   { 375   {
376   // Running on server executor - safe to modify lists 376   // Running on server executor - safe to modify lists
377   // Remove from active (if present), then wake waiter or add to idle 377   // Remove from active (if present), then wake waiter or add to idle
HITCBC 378   38 self_.active_remove(&w_); 378   38 self_.active_remove(&w_);
HITCBC 379   38 if (self_.waiters_) 379   38 if (self_.waiters_)
380   { 380   {
HITCBC 381   6 auto* wait = self_.waiters_; 381   6 auto* wait = self_.waiters_;
HITCBC 382   6 self_.waiters_ = wait->next; 382   6 self_.waiters_ = wait->next;
HITCBC 383   6 wait->w = &w_; 383   6 wait->w = &w_;
HITCBC 384   6 wait->cont.h = wait->h; 384   6 wait->cont.h = wait->h;
HITCBC 385   6 self_.ex_.post(wait->cont); 385   6 self_.ex_.post(wait->cont);
386   } 386   }
387   else 387   else
388   { 388   {
HITCBC 389   32 self_.idle_push(&w_); 389   32 self_.idle_push(&w_);
390   } 390   }
HITCBC 391   38 } 391   38 }
392   }; 392   };
393   393  
394   class pop_awaitable 394   class pop_awaitable
395   { 395   {
396   tcp_server& self_; 396   tcp_server& self_;
397   waiter wait_; 397   waiter wait_;
398   398  
399   public: 399   public:
HITCBC 400   42 pop_awaitable(tcp_server& self) noexcept : self_(self), wait_{} {} 400   42 pop_awaitable(tcp_server& self) noexcept : self_(self), wait_{} {}
401   401  
HITCBC 402   42 bool await_ready() const noexcept 402   42 bool await_ready() const noexcept
403   { 403   {
HITCBC 404   42 return !self_.idle_empty(); 404   42 return !self_.idle_empty();
405   } 405   }
406   406  
407   bool 407   bool
HITCBC 408   6 await_suspend(std::coroutine_handle<> h, capy::io_env const*) noexcept 408   6 await_suspend(std::coroutine_handle<> h, capy::io_env const*) noexcept
409   { 409   {
410   // Running on server executor (do_accept runs there) 410   // Running on server executor (do_accept runs there)
HITCBC 411   6 wait_.h = h; 411   6 wait_.h = h;
HITCBC 412   6 wait_.w = nullptr; 412   6 wait_.w = nullptr;
HITCBC 413   6 wait_.next = self_.waiters_; 413   6 wait_.next = self_.waiters_;
HITCBC 414   6 self_.waiters_ = &wait_; 414   6 self_.waiters_ = &wait_;
HITCBC 415   6 return true; 415   6 return true;
416   } 416   }
417   417  
HITCBC 418   42 worker_base& await_resume() noexcept 418   42 worker_base& await_resume() noexcept
419   { 419   {
420   // Running on server executor 420   // Running on server executor
HITCBC 421   42 if (wait_.w) 421   42 if (wait_.w)
HITCBC 422   6 return *wait_.w; // Woken by push_awaitable 422   6 return *wait_.w; // Woken by push_awaitable
HITCBC 423   36 return *self_.idle_pop(); 423   36 return *self_.idle_pop();
424   } 424   }
425   }; 425   };
426   426  
HITCBC 427   38 push_awaitable push(worker_base& w) 427   38 push_awaitable push(worker_base& w)
428   { 428   {
HITCBC 429   38 return push_awaitable{*this, w}; 429   38 return push_awaitable{*this, w};
430   } 430   }
431   431  
432   // Synchronous version for destructor/guard paths 432   // Synchronous version for destructor/guard paths
433   // Must be called from server executor context 433   // Must be called from server executor context
HITCBC 434   4 void push_sync(worker_base& w) noexcept 434   4 void push_sync(worker_base& w) noexcept
435   { 435   {
HITCBC 436   4 active_remove(&w); 436   4 active_remove(&w);
HITCBC 437   4 if (waiters_) 437   4 if (waiters_)
438   { 438   {
MISUBC 439   ✗ auto* wait = waiters_; 439   ✗ auto* wait = waiters_;
MISUBC 440   ✗ waiters_ = wait->next; 440   ✗ waiters_ = wait->next;
MISUBC 441   ✗ wait->w = &w; 441   ✗ wait->w = &w;
MISUBC 442   ✗ wait->cont.h = wait->h; 442   ✗ wait->cont.h = wait->h;
MISUBC 443   ✗ ex_.post(wait->cont); 443   ✗ ex_.post(wait->cont);
444   } 444   }
445   else 445   else
446   { 446   {
HITCBC 447   4 idle_push(&w); 447   4 idle_push(&w);
448   } 448   }
HITCBC 449   4 } 449   4 }
450   450  
HITCBC 451   42 pop_awaitable pop() 451   42 pop_awaitable pop()
452   { 452   {
HITCBC 453   42 return pop_awaitable{*this}; 453   42 return pop_awaitable{*this};
454   } 454   }
455   455  
456   capy::task<void> do_accept(tcp_acceptor& acc); 456   capy::task<void> do_accept(tcp_acceptor& acc);
457   457  
458   public: 458   public:
459   /** Abstract base class for connection handlers. 459   /** Abstract base class for connection handlers.
460   460  
461   Derive from this class to implement custom connection handling. 461   Derive from this class to implement custom connection handling.
462   Each worker owns a socket and is reused across multiple 462   Each worker owns a socket and is reused across multiple
463   connections to avoid per-connection allocation. 463   connections to avoid per-connection allocation.
464   464  
465   @see tcp_server, launcher 465   @see tcp_server, launcher
466   */ 466   */
467   class BOOST_COROSIO_DECL worker_base 467   class BOOST_COROSIO_DECL worker_base
468   { 468   {
469   // Ordered largest to smallest for optimal packing 469   // Ordered largest to smallest for optimal packing
470   std::stop_source stop_; // ~16 bytes 470   std::stop_source stop_; // ~16 bytes
471   worker_base* next_ = nullptr; // 8 bytes - used by idle and active lists 471   worker_base* next_ = nullptr; // 8 bytes - used by idle and active lists
472   worker_base* prev_ = nullptr; // 8 bytes - used only by active list 472   worker_base* prev_ = nullptr; // 8 bytes - used only by active list
473   473  
474   friend class tcp_server; 474   friend class tcp_server;
475   475  
476   public: 476   public:
477   /// Construct a worker. 477   /// Construct a worker.
478   worker_base(); 478   worker_base();
479   479  
480   /// Destroy the worker. 480   /// Destroy the worker.
481   virtual ~worker_base(); 481   virtual ~worker_base();
482   482  
483   /** Handle an accepted connection. 483   /** Handle an accepted connection.
484   484  
485   Called when this worker is dispatched to handle a new 485   Called when this worker is dispatched to handle a new
486   connection. The implementation must invoke the launcher 486   connection. The implementation must invoke the launcher
487   exactly once to start the handling coroutine. 487   exactly once to start the handling coroutine.
488   488  
489   @param launch Handle to launch the connection coroutine. 489   @param launch Handle to launch the connection coroutine.
490   */ 490   */
491   virtual void run(launcher launch) = 0; 491   virtual void run(launcher launch) = 0;
492   492  
493   /// Return the socket used for connections. 493   /// Return the socket used for connections.
494   virtual corosio::tcp_socket& socket() = 0; 494   virtual corosio::tcp_socket& socket() = 0;
495   }; 495   };
496   496  
497   /** Move-only handle to launch a worker coroutine. 497   /** Move-only handle to launch a worker coroutine.
498   498  
499   Passed to @ref worker_base::run to start the connection-handling 499   Passed to @ref worker_base::run to start the connection-handling
500   coroutine. The launcher ensures the worker returns to the idle 500   coroutine. The launcher ensures the worker returns to the idle
501   pool when the coroutine completes or if launching fails. 501   pool when the coroutine completes or if launching fails.
502   502  
503   The launcher must be invoked exactly once via `operator()`. 503   The launcher must be invoked exactly once via `operator()`.
504   If destroyed without invoking, the worker is returned to the 504   If destroyed without invoking, the worker is returned to the
505   idle pool automatically. 505   idle pool automatically.
506   506  
507   @see worker_base::run 507   @see worker_base::run
508   */ 508   */
509   class BOOST_COROSIO_DECL launcher 509   class BOOST_COROSIO_DECL launcher
510   { 510   {
511   tcp_server* srv_; 511   tcp_server* srv_;
512   worker_base* w_; 512   worker_base* w_;
513   513  
514   friend class tcp_server; 514   friend class tcp_server;
515   515  
HITCBC 516   22 launcher(tcp_server& srv, worker_base& w) noexcept : srv_(&srv), w_(&w) 516   22 launcher(tcp_server& srv, worker_base& w) noexcept : srv_(&srv), w_(&w)
517   { 517   {
HITCBC 518   22 } 518   22 }
519   519  
520   public: 520   public:
521   /// Return the worker to the pool if not launched. 521   /// Return the worker to the pool if not launched.
HITCBC 522   22 ~launcher() 522   22 ~launcher()
523   { 523   {
HITCBC 524   22 if (w_) 524   22 if (w_)
HITCBC 525   4 srv_->push_sync(*w_); 525   4 srv_->push_sync(*w_);
HITCBC 526   22 } 526   22 }
527   527  
528   launcher(launcher&& o) noexcept 528   launcher(launcher&& o) noexcept
529   : srv_(o.srv_) 529   : srv_(o.srv_)
530   , w_(std::exchange(o.w_, nullptr)) 530   , w_(std::exchange(o.w_, nullptr))
531   { 531   {
532   } 532   }
533   launcher(launcher const&) = delete; 533   launcher(launcher const&) = delete;
534   launcher& operator=(launcher const&) = delete; 534   launcher& operator=(launcher const&) = delete;
535   launcher& operator=(launcher&&) = delete; 535   launcher& operator=(launcher&&) = delete;
536   536  
537   /** Launch the connection-handling coroutine. 537   /** Launch the connection-handling coroutine.
538   538  
539   Starts the given coroutine on the specified executor. When 539   Starts the given coroutine on the specified executor. When
540   the coroutine completes, the worker is automatically returned 540   the coroutine completes, the worker is automatically returned
541   to the idle pool. 541   to the idle pool.
542   542  
543   @param ex The executor to run the coroutine on. 543   @param ex The executor to run the coroutine on.
544   @param task The coroutine to execute. 544   @param task The coroutine to execute.
545   545  
546   @throws std::logic_error If this launcher was already invoked. 546   @throws std::logic_error If this launcher was already invoked.
547   */ 547   */
548   template<class Executor> 548   template<class Executor>
HITCBC 549   20 void operator()(Executor const& ex, capy::task<void> task) 549   20 void operator()(Executor const& ex, capy::task<void> task)
550   { 550   {
HITCBC 551   20 if (!w_) 551   20 if (!w_)
HITCBC 552   2 detail::throw_logic_error(); // launcher already invoked 552   2 detail::throw_logic_error(); // launcher already invoked
553   553  
HITCBC 554   18 auto* w = std::exchange(w_, nullptr); 554   18 auto* w = std::exchange(w_, nullptr);
555   555  
556   // Worker is being dispatched - add to active list 556   // Worker is being dispatched - add to active list
HITCBC 557   18 srv_->active_push(w); 557   18 srv_->active_push(w);
558   558  
559   // Return worker to pool if coroutine setup throws 559   // Return worker to pool if coroutine setup throws
560   struct guard_t 560   struct guard_t
561   { 561   {
562   tcp_server* srv; 562   tcp_server* srv;
563   worker_base* w; 563   worker_base* w;
HITCBC 564   18 ~guard_t() 564   18 ~guard_t()
565   { 565   {
HITCBC 566   18 if (w) 566   18 if (w)
MISUBC 567   ✗ srv->push_sync(*w); 567   ✗ srv->push_sync(*w);
HITCBC 568   18 } 568   18 }
HITCBC 569   18 } guard{srv_, w}; 569   18 } guard{srv_, w};
570   570  
571   // Reset worker's stop source for this connection 571   // Reset worker's stop source for this connection
HITCBC 572   18 w->stop_ = {}; 572   18 w->stop_ = {};
HITCBC 573   18 auto st = w->stop_.get_token(); 573   18 auto st = w->stop_.get_token();
574   574  
HITCBC 575   18 auto wrapper = 575   18 auto wrapper =
HITCBC 576   18 launch_coro<Executor>{}(ex, st, srv_, std::move(task), w); 576   18 launch_coro<Executor>{}(ex, st, srv_, std::move(task), w);
577   577  
578   // Executor and stop token stored in promise via constructor 578   // Executor and stop token stored in promise via constructor
HITCBC 579   18 ex.post(std::exchange(wrapper.h, nullptr)); // Release before post 579   18 ex.post(std::exchange(wrapper.h, nullptr)); // Release before post
HITCBC 580   18 guard.w = nullptr; // Success - dismiss guard 580   18 guard.w = nullptr; // Success - dismiss guard
HITCBC 581   18 } 581   18 }
582   }; 582   };
583   583  
584   /** Construct a TCP server. 584   /** Construct a TCP server.
585   585  
586   @tparam Ctx Execution context type satisfying ExecutionContext. 586   @tparam Ctx Execution context type satisfying ExecutionContext.
587   @tparam Ex Executor type satisfying Executor. 587   @tparam Ex Executor type satisfying Executor.
588   588  
589   @param ctx The execution context for socket operations. 589   @param ctx The execution context for socket operations.
590   @param ex The executor for dispatching coroutines. 590   @param ex The executor for dispatching coroutines.
591   591  
592   @par Example 592   @par Example
593   @code 593   @code
594   tcp_server srv(ctx, ctx.get_executor()); 594   tcp_server srv(ctx, ctx.get_executor());
595   srv.set_workers(make_workers(ctx, 100)); 595   srv.set_workers(make_workers(ctx, 100));
596   if (auto ec = srv.bind(endpoint{...})) 596   if (auto ec = srv.bind(endpoint{...}))
597   return; 597   return;
598   srv.start(); 598   srv.start();
599   @endcode 599   @endcode
600   */ 600   */
601   template<capy::ExecutionContext Ctx, capy::Executor Ex> 601   template<capy::ExecutionContext Ctx, capy::Executor Ex>
HITCBC 602   36 tcp_server(Ctx& ctx, Ex ex) : impl_(make_impl(ctx)) 602   36 tcp_server(Ctx& ctx, Ex ex) : impl_(make_impl(ctx))
HITCBC 603   36 , ex_(std::move(ex)) 603   36 , ex_(std::move(ex))
604   { 604   {
HITCBC 605   36 } 605   36 }
606   606  
607   public: 607   public:
608   /// Destroy the server, stopping all accept loops. 608   /// Destroy the server, stopping all accept loops.
609   ~tcp_server(); 609   ~tcp_server();
610   610  
611   tcp_server(tcp_server const&) = delete; 611   tcp_server(tcp_server const&) = delete;
612   tcp_server& operator=(tcp_server const&) = delete; 612   tcp_server& operator=(tcp_server const&) = delete;
613   613  
614   /** Move construct from another server. 614   /** Move construct from another server.
615   615  
616   @param o The source server. After the move, @p o is 616   @param o The source server. After the move, @p o is
617   in a valid but unspecified state. 617   in a valid but unspecified state.
618   */ 618   */
619   tcp_server(tcp_server&& o) noexcept; 619   tcp_server(tcp_server&& o) noexcept;
620   620  
621   /** Move assign from another server. 621   /** Move assign from another server.
622   622  
623   @param o The source server. After the move, @p o is 623   @param o The source server. After the move, @p o is
624   in a valid but unspecified state. 624   in a valid but unspecified state.
625   625  
626   @return `*this`. 626   @return `*this`.
627   */ 627   */
628   tcp_server& operator=(tcp_server&& o) noexcept; 628   tcp_server& operator=(tcp_server&& o) noexcept;
629   629  
630   /** Bind to a local endpoint. 630   /** Bind to a local endpoint.
631   631  
632   Creates an acceptor listening on the specified endpoint. 632   Creates an acceptor listening on the specified endpoint.
633   Multiple endpoints can be bound by calling this method 633   Multiple endpoints can be bound by calling this method
634   multiple times before @ref start. 634   multiple times before @ref start.
635   635  
636   @param ep The local endpoint to bind to. 636   @param ep The local endpoint to bind to.
637   637  
638   @return The error code if binding fails. 638   @return The error code if binding fails.
639   */ 639   */
640   [[nodiscard]] std::error_code bind(endpoint ep); 640   [[nodiscard]] std::error_code bind(endpoint ep);
641   641  
642   /** Set the worker pool. 642   /** Set the worker pool.
643   643  
644   Replaces any existing workers with the given range. Any 644   Replaces any existing workers with the given range. Any
645   previous workers are released and the idle/active lists 645   previous workers are released and the idle/active lists
646   are cleared before populating with new workers. 646   are cleared before populating with new workers.
647   647  
648   @tparam Range Forward range of pointer-like objects to worker_base. 648   @tparam Range Forward range of pointer-like objects to worker_base.
649   649  
650   @param workers Range of workers to manage. Each element must 650   @param workers Range of workers to manage. Each element must
651   support `std::to_address()` yielding `worker_base*`. 651   support `std::to_address()` yielding `worker_base*`.
652   652  
653   @par Example 653   @par Example
654   @code 654   @code
655   std::vector<std::unique_ptr<my_worker>> workers; 655   std::vector<std::unique_ptr<my_worker>> workers;
656   for(int i = 0; i < 100; ++i) 656   for(int i = 0; i < 100; ++i)
657   workers.push_back(std::make_unique<my_worker>(ctx)); 657   workers.push_back(std::make_unique<my_worker>(ctx));
658   srv.set_workers(std::move(workers)); 658   srv.set_workers(std::move(workers));
659   @endcode 659   @endcode
660   */ 660   */
661   template<std::ranges::forward_range Range> 661   template<std::ranges::forward_range Range>
662   requires std::convertible_to< 662   requires std::convertible_to<
663   decltype(std::to_address( 663   decltype(std::to_address(
664   std::declval<std::ranges::range_value_t<Range>&>())), 664   std::declval<std::ranges::range_value_t<Range>&>())),
665   worker_base*> 665   worker_base*>
HITCBC 666   36 void set_workers(Range&& workers) 666   36 void set_workers(Range&& workers)
667   { 667   {
668   // Clear existing state 668   // Clear existing state
HITCBC 669   36 storage_.reset(); 669   36 storage_.reset();
HITCBC 670   36 idle_head_ = nullptr; 670   36 idle_head_ = nullptr;
HITCBC 671   36 active_head_ = nullptr; 671   36 active_head_ = nullptr;
HITCBC 672   36 active_tail_ = nullptr; 672   36 active_tail_ = nullptr;
673   673  
674   // Take ownership and populate idle list 674   // Take ownership and populate idle list
675   using StorageType = std::decay_t<Range>; 675   using StorageType = std::decay_t<Range>;
HITCBC 676   36 auto* p = new StorageType(std::forward<Range>(workers)); 676   36 auto* p = new StorageType(std::forward<Range>(workers));
HITCBC 677   36 storage_ = std::shared_ptr<void>( 677   36 storage_ = std::shared_ptr<void>(
HITCBC 678   36 p, [](void* ptr) { delete static_cast<StorageType*>(ptr); }); 678   36 p, [](void* ptr) { delete static_cast<StorageType*>(ptr); });
HITCBC 679   162 for (auto&& elem : *static_cast<StorageType*>(p)) 679   162 for (auto&& elem : *static_cast<StorageType*>(p))
HITCBC 680   126 idle_push(std::to_address(elem)); 680   126 idle_push(std::to_address(elem));
HITCBC 681   36 } 681   36 }
682   682  
683   /** Start accepting connections. 683   /** Start accepting connections.
684   684  
685   Launches accept loops for all bound endpoints. Incoming 685   Launches accept loops for all bound endpoints. Incoming
686   connections are dispatched to idle workers from the pool. 686   connections are dispatched to idle workers from the pool.
687   687  
688   Calling `start()` on an already-running server has no effect. 688   Calling `start()` on an already-running server has no effect.
689   689  
690   @par Preconditions 690   @par Preconditions
691   - At least one endpoint bound via @ref bind. 691   - At least one endpoint bound via @ref bind.
692   - Workers provided via @ref set_workers. 692   - Workers provided via @ref set_workers.
693   - If restarting, @ref join must have completed first. 693   - If restarting, @ref join must have completed first.
694   694  
695   @par Effects 695   @par Effects
696   Creates one accept coroutine per bound endpoint. Each coroutine 696   Creates one accept coroutine per bound endpoint. Each coroutine
697   runs on the server's executor, waiting for connections and 697   runs on the server's executor, waiting for connections and
698   dispatching them to idle workers. 698   dispatching them to idle workers.
699   699  
700   @par Restart Sequence 700   @par Restart Sequence
701   To restart after stopping, complete the full shutdown cycle: 701   To restart after stopping, complete the full shutdown cycle:
702   @code 702   @code
703   srv.start(); 703   srv.start();
704   ioc.run_for( 1s ); 704   ioc.run_for( 1s );
705   srv.stop(); // 1. Signal shutdown 705   srv.stop(); // 1. Signal shutdown
706   ioc.run(); // 2. Drain remaining completions 706   ioc.run(); // 2. Drain remaining completions
707   srv.join(); // 3. Wait for accept loops 707   srv.join(); // 3. Wait for accept loops
708   708  
709   // Now safe to restart 709   // Now safe to restart
710   srv.start(); 710   srv.start();
711   ioc.run(); 711   ioc.run();
712   @endcode 712   @endcode
713   713  
714   @par Thread Safety 714   @par Thread Safety
715   Not thread safe. 715   Not thread safe.
716   716  
717   @throws std::logic_error If a previous session has not been 717   @throws std::logic_error If a previous session has not been
718   joined (accept loops still active). 718   joined (accept loops still active).
719   */ 719   */
720   void start(); 720   void start();
721   721  
722   /** Return the local endpoint for the i-th bound port. 722   /** Return the local endpoint for the i-th bound port.
723   723  
724   @param index Zero-based index into the list of bound ports. 724   @param index Zero-based index into the list of bound ports.
725   725  
726   @return The local endpoint, or a default-constructed endpoint 726   @return The local endpoint, or a default-constructed endpoint
727   if @p index is out of range or the acceptor is not open. 727   if @p index is out of range or the acceptor is not open.
728   */ 728   */
729   endpoint local_endpoint(std::size_t index = 0) const noexcept; 729   endpoint local_endpoint(std::size_t index = 0) const noexcept;
730   730  
731   /** Stop accepting connections. 731   /** Stop accepting connections.
732   732  
733   Requests the accept loops' stop token and requests cancellation 733   Requests the accept loops' stop token and requests cancellation
734   of active workers via their stop tokens. The acceptors are not 734   of active workers via their stop tokens. The acceptors are not
735   closed; a suspended accept completes once more before its loop 735   closed; a suspended accept completes once more before its loop
736   observes the stop token and ends. 736   observes the stop token and ends.
737   737  
738   This function returns immediately; it does not wait for workers 738   This function returns immediately; it does not wait for workers
739   to finish. Pending I/O operations complete asynchronously. 739   to finish. Pending I/O operations complete asynchronously.
740   740  
741   Calling `stop()` on a non-running server has no effect. 741   Calling `stop()` on a non-running server has no effect.
742   742  
743   @par Effects 743   @par Effects
744   - Requests stop on the accept loops' stop token. The acceptors 744   - Requests stop on the accept loops' stop token. The acceptors
745   are not closed; a pending accept completes once more before 745   are not closed; a pending accept completes once more before
746   the accept loop ends. 746   the accept loop ends.
747   - Requests stop on each active worker's stop token. 747   - Requests stop on each active worker's stop token.
748   - Workers observing their stop token should exit promptly. 748   - Workers observing their stop token should exit promptly.
749   749  
750   @par Postconditions 750   @par Postconditions
751   No new connections will be accepted. Active workers continue 751   No new connections will be accepted. Active workers continue
752   until they observe their stop token or complete naturally. 752   until they observe their stop token or complete naturally.
753   753  
754   @par What Happens Next 754   @par What Happens Next
755   After calling `stop()`: 755   After calling `stop()`:
756   1. Let `ioc.run()` return (drains pending completions). 756   1. Let `ioc.run()` return (drains pending completions).
757   2. Call @ref join to wait for accept loops to finish. 757   2. Call @ref join to wait for accept loops to finish.
758   3. Only then is it safe to restart or destroy the server. 758   3. Only then is it safe to restart or destroy the server.
759   759  
760   @par Thread Safety 760   @par Thread Safety
761   Not thread safe. 761   Not thread safe.
762   762  
763   @see join, start 763   @see join, start
764   */ 764   */
765   void stop(); 765   void stop();
766   766  
767   /** Block until all accept loops complete. 767   /** Block until all accept loops complete.
768   768  
769   Blocks the calling thread until all accept coroutines launched 769   Blocks the calling thread until all accept coroutines launched
770   by @ref start have finished executing. This synchronizes the 770   by @ref start have finished executing. This synchronizes the
771   shutdown sequence, ensuring the server is fully stopped before 771   shutdown sequence, ensuring the server is fully stopped before
772   restarting or destroying it. 772   restarting or destroying it.
773   773  
774   @par Preconditions 774   @par Preconditions
775   @ref stop has been called and `ioc.run()` has returned. 775   @ref stop has been called and `ioc.run()` has returned.
776   776  
777   @par Postconditions 777   @par Postconditions
778   All accept loops have completed. The server is in the stopped 778   All accept loops have completed. The server is in the stopped
779   state and may be restarted via @ref start. 779   state and may be restarted via @ref start.
780   780  
781   @par Example (Correct Usage) 781   @par Example (Correct Usage)
782   @code 782   @code
783   // main thread 783   // main thread
784   srv.start(); 784   srv.start();
785   ioc.run(); // Blocks until work completes 785   ioc.run(); // Blocks until work completes
786   srv.join(); // Safe: called after ioc.run() returns 786   srv.join(); // Safe: called after ioc.run() returns
787   @endcode 787   @endcode
788   788  
789   @par WARNING: Deadlock Scenarios 789   @par WARNING: Deadlock Scenarios
790   Calling `join()` from the wrong context causes deadlock: 790   Calling `join()` from the wrong context causes deadlock:
791   791  
792   @code 792   @code
793   // WRONG: calling join() from inside a worker coroutine 793   // WRONG: calling join() from inside a worker coroutine
794   void run( launcher launch ) override 794   void run( launcher launch ) override
795   { 795   {
796   launch( ex, [this]() -> capy::task<> 796   launch( ex, [this]() -> capy::task<>
797   { 797   {
798   srv_.join(); // DEADLOCK: blocks the executor 798   srv_.join(); // DEADLOCK: blocks the executor
799   co_return; 799   co_return;
800   }()); 800   }());
801   } 801   }
802   802  
803   // WRONG: calling join() while ioc.run() is still active 803   // WRONG: calling join() while ioc.run() is still active
804   std::thread t( [&]{ ioc.run(); } ); 804   std::thread t( [&]{ ioc.run(); } );
805   srv.stop(); 805   srv.stop();
806   srv.join(); // DEADLOCK: ioc.run() still running in thread t 806   srv.join(); // DEADLOCK: ioc.run() still running in thread t
807   @endcode 807   @endcode
808   808  
809   @par Thread Safety 809   @par Thread Safety
810   May be called from any thread, but will deadlock if called 810   May be called from any thread, but will deadlock if called
811   from within the io_context event loop or from a worker coroutine. 811   from within the io_context event loop or from a worker coroutine.
812   812  
813   @see stop, start 813   @see stop, start
814   */ 814   */
815   void join(); 815   void join();
816   816  
817   private: 817   private:
818   capy::task<> do_stop(); 818   capy::task<> do_stop();
819   }; 819   };
820   820  
821   #ifdef _MSC_VER 821   #ifdef _MSC_VER
822   #pragma warning(pop) 822   #pragma warning(pop)
823   #endif 823   #endif
824   824  
825   } // namespace boost::corosio 825   } // namespace boost::corosio
826   826  
827   #endif 827   #endif