. /include/boost/corosio /tcp_server.hpp
97.97% Lines (145/148) 97.30% Functions (36/37)
[ ] 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   238 void idle_push(worker_base* w) noexcept
184   { 184   {
HITCBC 185   162 w->next_ = idle_head_; 185   238 w->next_ = idle_head_;
HITCBC 186   162 idle_head_ = w; 186   238 idle_head_ = w;
HITCBC 187   162 } 187   238 }
188   188  
HITCBC 189   36 worker_base* idle_pop() noexcept 189   75 worker_base* idle_pop() noexcept
190   { 190   {
HITCBC 191   36 auto* w = idle_head_; 191   75 auto* w = idle_head_;
HITCBC 192   36 if (w) 192   75 if (w)
HITCBC 193   36 idle_head_ = w->next_; 193   75 idle_head_ = w->next_;
HITCBC 194   36 return w; 194   75 return w;
195   } 195   }
196   196  
HITCBC 197   42 bool idle_empty() const noexcept 197   153 bool idle_empty() const noexcept
198   { 198   {
HITCBC 199   42 return idle_head_ == nullptr; 199   153 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   88 void active_push(worker_base* w) noexcept
204   { 204   {
HITCBC 205   18 w->next_ = nullptr; 205   88 w->next_ = nullptr;
HITCBC 206   18 w->prev_ = active_tail_; 206   88 w->prev_ = active_tail_;
HITCBC 207   18 if (active_tail_) 207   88 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   84 active_head_ = w;
HITCBC 211   18 active_tail_ = w; 211   88 active_tail_ = w;
HITCBC 212   18 } 212   88 }
213   213  
HITCBC 214   42 void active_remove(worker_base* w) noexcept 214   153 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   153 if (w != active_head_ && w->prev_ == nullptr)
HITCBC 218   24 return; 218   65 return;
HITCBC 219   18 if (w->prev_) 219   88 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   84 active_head_ = w->next_;
HITCBC 223   18 if (w->next_) 223   88 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   86 active_tail_ = w->prev_;
HITCBC 227   18 w->prev_ = nullptr; // Mark as not in active list 227   88 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   88 promise_type(Closure&&, E e, S s, Args&&...)
HITCBC 254   18 : ex(std::move(e)) 254   88 : ex(std::move(e))
HITCBC 255   18 , env_{ 255   88 , env_{
HITCBC 256   18 capy::executor_ref(ex), std::move(s), 256   88 capy::executor_ref(ex), std::move(s),
HITCBC 257   18 capy::get_current_frame_allocator()} 257   88 capy::get_current_frame_allocator()}
258   { 258   {
HITCBC 259   18 } 259   88 }
260   260  
HITCBC 261   18 launch_wrapper get_return_object() noexcept 261   88 launch_wrapper get_return_object() noexcept
262   { 262   {
263   return { 263   return {
HITCBC 264   18 std::coroutine_handle<promise_type>::from_promise(*this)}; 264   88 std::coroutine_handle<promise_type>::from_promise(*this)};
265   } 265   }
HITCBC 266   18 std::suspend_always initial_suspend() noexcept 266   88 std::suspend_always initial_suspend() noexcept
267   { 267   {
HITCBC 268   18 return {}; 268   88 return {};
269   } 269   }
HITCBC 270   18 std::suspend_never final_suspend() noexcept 270   88 std::suspend_never final_suspend() noexcept
271   { 271   {
HITCBC 272   18 return {}; 272   88 return {};
273   } 273   }
HITCBC 274   18 void return_void() noexcept {} 274   88 void return_void() noexcept {}
MISUBC 275   ✗ void unhandled_exception() 275   ✗ void unhandled_exception()
276   { 276   {
  277 + // LCOV_EXCL_START: terminating by contract is not a
  278 + // coverable outcome.
EUB 277   ✗ std::terminate(); 279   std::terminate();
  280 + // LCOV_EXCL_STOP
278   } 281   }
279   282  
280   // Inject io_env for IoAwaitable 283   // Inject io_env for IoAwaitable
281   template<capy::IoAwaitable Awaitable> 284   template<capy::IoAwaitable Awaitable>
HITCBC 282   36 auto await_transform(Awaitable&& a) 285   176 auto await_transform(Awaitable&& a)
283   { 286   {
284   using AwaitableT = std::decay_t<Awaitable>; 287   using AwaitableT = std::decay_t<Awaitable>;
285   struct adapter 288   struct adapter
286   { 289   {
287   AwaitableT aw; 290   AwaitableT aw;
288   capy::io_env const* env; 291   capy::io_env const* env;
289   292  
HITCBC 290   36 bool await_ready() 293   176 bool await_ready()
291   { 294   {
HITCBC 292   36 return aw.await_ready(); 295   176 return aw.await_ready();
293   } 296   }
HITCBC 294   36 decltype(auto) await_resume() 297   176 decltype(auto) await_resume()
295   { 298   {
HITCBC 296   36 return aw.await_resume(); 299   176 return aw.await_resume();
297   } 300   }
298   301  
HITCBC 299   36 auto await_suspend(std::coroutine_handle<promise_type> h) 302   176 auto await_suspend(std::coroutine_handle<promise_type> h)
300   { 303   {
HITCBC 301   36 return aw.await_suspend(h, env); 304   176 return aw.await_suspend(h, env);
302   } 305   }
303   }; 306   };
HITCBC 304   54 return adapter{std::forward<Awaitable>(a), &env_}; 307   264 return adapter{std::forward<Awaitable>(a), &env_};
HITCBC 305   18 } 308   88 }
306   }; 309   };
307   310  
308   std::coroutine_handle<promise_type> h; 311   std::coroutine_handle<promise_type> h;
309   312  
HITCBC 310   18 launch_wrapper(std::coroutine_handle<promise_type> handle) noexcept 313   88 launch_wrapper(std::coroutine_handle<promise_type> handle) noexcept
HITCBC 311   18 : h(handle) 314   88 : h(handle)
312   { 315   {
HITCBC 313   18 } 316   88 }
314   317  
HITCBC 315   18 ~launch_wrapper() 318   88 ~launch_wrapper()
316   { 319   {
HITCBC 317   18 if (h) 320   88 if (h)
MISUBC 318   ✗ h.destroy(); 321   ✗ h.destroy();
HITCBC 319   18 } 322   88 }
320   323  
321   launch_wrapper(launch_wrapper&& o) noexcept 324   launch_wrapper(launch_wrapper&& o) noexcept
322   : h(std::exchange(o.h, nullptr)) 325   : h(std::exchange(o.h, nullptr))
323   { 326   {
324   } 327   }
325   328  
326   launch_wrapper(launch_wrapper const&) = delete; 329   launch_wrapper(launch_wrapper const&) = delete;
327   launch_wrapper& operator=(launch_wrapper const&) = delete; 330   launch_wrapper& operator=(launch_wrapper const&) = delete;
328   launch_wrapper& operator=(launch_wrapper&&) = delete; 331   launch_wrapper& operator=(launch_wrapper&&) = delete;
329   }; 332   };
330   333  
331   // Named functor to avoid incomplete lambda type in coroutine promise 334   // Named functor to avoid incomplete lambda type in coroutine promise
332   template<class Executor> 335   template<class Executor>
333   struct launch_coro 336   struct launch_coro
334   { 337   {
HITCBC 335   18 launch_wrapper<Executor> operator()( 338   88 launch_wrapper<Executor> operator()(
336   Executor, 339   Executor,
337   std::stop_token, 340   std::stop_token,
338   tcp_server* self, 341   tcp_server* self,
339   capy::task<void> t, 342   capy::task<void> t,
340   worker_base* wp) 343   worker_base* wp)
341   { 344   {
342   // Executor and stop token stored in promise via constructor 345   // Executor and stop token stored in promise via constructor
343   co_await std::move(t); 346   co_await std::move(t);
344   co_await self->push(*wp); // worker goes back to idle list 347   co_await self->push(*wp); // worker goes back to idle list
HITCBC 345   36 } 348   176 }
346   }; 349   };
347   350  
348   class push_awaitable 351   class push_awaitable
349   { 352   {
350   tcp_server& self_; 353   tcp_server& self_;
351   worker_base& w_; 354   worker_base& w_;
352   capy::continuation cont_; 355   capy::continuation cont_;
353   356  
354   public: 357   public:
HITCBC 355   38 push_awaitable(tcp_server& self, worker_base& w) noexcept 358   145 push_awaitable(tcp_server& self, worker_base& w) noexcept
HITCBC 356   38 : self_(self) 359   145 : self_(self)
HITCBC 357   38 , w_(w) 360   145 , w_(w)
358   { 361   {
HITCBC 359   38 } 362   145 }
360   363  
HITCBC 361   38 bool await_ready() const noexcept 364   145 bool await_ready() const noexcept
362   { 365   {
HITCBC 363   38 return false; 366   145 return false;
364   } 367   }
365   368  
366   std::coroutine_handle<> 369   std::coroutine_handle<>
HITCBC 367   38 await_suspend(std::coroutine_handle<> h, capy::io_env const*) noexcept 370   145 await_suspend(std::coroutine_handle<> h, capy::io_env const*) noexcept
368   { 371   {
369   // Symmetric transfer to server's executor 372   // Symmetric transfer to server's executor
HITCBC 370   38 cont_.h = h; 373   145 cont_.h = h;
HITCBC 371   38 return self_.ex_.dispatch(cont_); 374   145 return self_.ex_.dispatch(cont_);
372   } 375   }
373   376  
HITCBC 374   38 void await_resume() noexcept 377   145 void await_resume() noexcept
375   { 378   {
376   // Running on server executor - safe to modify lists 379   // Running on server executor - safe to modify lists
377   // Remove from active (if present), then wake waiter or add to idle 380   // Remove from active (if present), then wake waiter or add to idle
HITCBC 378   38 self_.active_remove(&w_); 381   145 self_.active_remove(&w_);
HITCBC 379   38 if (self_.waiters_) 382   145 if (self_.waiters_)
380   { 383   {
HITCBC 381   6 auto* wait = self_.waiters_; 384   76 auto* wait = self_.waiters_;
HITCBC 382   6 self_.waiters_ = wait->next; 385   76 self_.waiters_ = wait->next;
HITCBC 383   6 wait->w = &w_; 386   76 wait->w = &w_;
HITCBC 384   6 wait->cont.h = wait->h; 387   76 wait->cont.h = wait->h;
HITCBC 385   6 self_.ex_.post(wait->cont); 388   76 self_.ex_.post(wait->cont);
386   } 389   }
387   else 390   else
388   { 391   {
HITCBC 389   32 self_.idle_push(&w_); 392   69 self_.idle_push(&w_);
390   } 393   }
HITCBC 391   38 } 394   145 }
392   }; 395   };
393   396  
394   class pop_awaitable 397   class pop_awaitable
395   { 398   {
396   tcp_server& self_; 399   tcp_server& self_;
397   waiter wait_; 400   waiter wait_;
398   401  
399   public: 402   public:
HITCBC 400   42 pop_awaitable(tcp_server& self) noexcept : self_(self), wait_{} {} 403   153 pop_awaitable(tcp_server& self) noexcept : self_(self), wait_{} {}
401   404  
HITCBC 402   42 bool await_ready() const noexcept 405   153 bool await_ready() const noexcept
403   { 406   {
HITCBC 404   42 return !self_.idle_empty(); 407   153 return !self_.idle_empty();
405   } 408   }
406   409  
407   bool 410   bool
HITCBC 408   6 await_suspend(std::coroutine_handle<> h, capy::io_env const*) noexcept 411   78 await_suspend(std::coroutine_handle<> h, capy::io_env const*) noexcept
409   { 412   {
410   // Running on server executor (do_accept runs there) 413   // Running on server executor (do_accept runs there)
HITCBC 411   6 wait_.h = h; 414   78 wait_.h = h;
HITCBC 412   6 wait_.w = nullptr; 415   78 wait_.w = nullptr;
HITCBC 413   6 wait_.next = self_.waiters_; 416   78 wait_.next = self_.waiters_;
HITCBC 414   6 self_.waiters_ = &wait_; 417   78 self_.waiters_ = &wait_;
HITCBC 415   6 return true; 418   78 return true;
416   } 419   }
417   420  
HITCBC 418   42 worker_base& await_resume() noexcept 421   153 worker_base& await_resume() noexcept
419   { 422   {
420   // Running on server executor 423   // Running on server executor
HITCBC 421   42 if (wait_.w) 424   153 if (wait_.w)
HITCBC 422   6 return *wait_.w; // Woken by push_awaitable 425   78 return *wait_.w; // Woken by push_awaitable
HITCBC 423   36 return *self_.idle_pop(); 426   75 return *self_.idle_pop();
424   } 427   }
425   }; 428   };
426   429  
HITCBC 427   38 push_awaitable push(worker_base& w) 430   145 push_awaitable push(worker_base& w)
428   { 431   {
HITCBC 429   38 return push_awaitable{*this, w}; 432   145 return push_awaitable{*this, w};
430   } 433   }
431   434  
432   // Synchronous version for destructor/guard paths 435   // Synchronous version for destructor/guard paths
433   // Must be called from server executor context 436   // Must be called from server executor context
HITCBC 434   4 void push_sync(worker_base& w) noexcept 437   8 void push_sync(worker_base& w) noexcept
435   { 438   {
HITCBC 436   4 active_remove(&w); 439   8 active_remove(&w);
HITCBC 437   4 if (waiters_) 440   8 if (waiters_)
438   { 441   {
HITGBC 439   ✗ auto* wait = waiters_; 442   2 auto* wait = waiters_;
HITGBC 440   ✗ waiters_ = wait->next; 443   2 waiters_ = wait->next;
HITGBC 441   ✗ wait->w = &w; 444   2 wait->w = &w;
HITGBC 442   ✗ wait->cont.h = wait->h; 445   2 wait->cont.h = wait->h;
HITGBC 443   ✗ ex_.post(wait->cont); 446   2 ex_.post(wait->cont);
444   } 447   }
445   else 448   else
446   { 449   {
HITCBC 447   4 idle_push(&w); 450   6 idle_push(&w);
448   } 451   }
HITCBC 449   4 } 452   8 }
450   453  
HITCBC 451   42 pop_awaitable pop() 454   153 pop_awaitable pop()
452   { 455   {
HITCBC 453   42 return pop_awaitable{*this}; 456   153 return pop_awaitable{*this};
454   } 457   }
455   458  
456   capy::task<void> do_accept(tcp_acceptor& acc); 459   capy::task<void> do_accept(tcp_acceptor& acc);
457   460  
458   public: 461   public:
459   /** Abstract base class for connection handlers. 462   /** Abstract base class for connection handlers.
460   463  
461   Derive from this class to implement custom connection handling. 464   Derive from this class to implement custom connection handling.
462   Each worker owns a socket and is reused across multiple 465   Each worker owns a socket and is reused across multiple
463   connections to avoid per-connection allocation. 466   connections to avoid per-connection allocation.
464   467  
465   @see tcp_server, launcher 468   @see tcp_server, launcher
466   */ 469   */
467   class BOOST_COROSIO_DECL worker_base 470   class BOOST_COROSIO_DECL worker_base
468   { 471   {
469   // Ordered largest to smallest for optimal packing 472   // Ordered largest to smallest for optimal packing
470   std::stop_source stop_; // ~16 bytes 473   std::stop_source stop_; // ~16 bytes
471   worker_base* next_ = nullptr; // 8 bytes - used by idle and active lists 474   worker_base* next_ = nullptr; // 8 bytes - used by idle and active lists
472   worker_base* prev_ = nullptr; // 8 bytes - used only by active list 475   worker_base* prev_ = nullptr; // 8 bytes - used only by active list
473   476  
474   friend class tcp_server; 477   friend class tcp_server;
475   478  
476   public: 479   public:
477   /// Construct a worker. 480   /// Construct a worker.
478   worker_base(); 481   worker_base();
479   482  
480   /// Destroy the worker. 483   /// Destroy the worker.
481   virtual ~worker_base(); 484   virtual ~worker_base();
482   485  
483   /** Handle an accepted connection. 486   /** Handle an accepted connection.
484   487  
485   Called when this worker is dispatched to handle a new 488   Called when this worker is dispatched to handle a new
486   connection. The implementation must invoke the launcher 489   connection. The implementation must invoke the launcher
487   exactly once to start the handling coroutine. 490   exactly once to start the handling coroutine.
488   491  
489   @param launch Handle to launch the connection coroutine. 492   @param launch Handle to launch the connection coroutine.
490   */ 493   */
491   virtual void run(launcher launch) = 0; 494   virtual void run(launcher launch) = 0;
492   495  
493   /// Return the socket used for connections. 496   /// Return the socket used for connections.
494   virtual corosio::tcp_socket& socket() = 0; 497   virtual corosio::tcp_socket& socket() = 0;
495   }; 498   };
496   499  
497   /** Move-only handle to launch a worker coroutine. 500   /** Move-only handle to launch a worker coroutine.
498   501  
499   Passed to @ref worker_base::run to start the connection-handling 502   Passed to @ref worker_base::run to start the connection-handling
500   coroutine. The launcher ensures the worker returns to the idle 503   coroutine. The launcher ensures the worker returns to the idle
501   pool when the coroutine completes or if launching fails. 504   pool when the coroutine completes or if launching fails.
502   505  
503   The launcher must be invoked exactly once via `operator()`. 506   The launcher must be invoked exactly once via `operator()`.
504   If destroyed without invoking, the worker is returned to the 507   If destroyed without invoking, the worker is returned to the
505   idle pool automatically. 508   idle pool automatically.
506   509  
507   @see worker_base::run 510   @see worker_base::run
508   */ 511   */
509   class BOOST_COROSIO_DECL launcher 512   class BOOST_COROSIO_DECL launcher
510   { 513   {
511   tcp_server* srv_; 514   tcp_server* srv_;
512   worker_base* w_; 515   worker_base* w_;
513   516  
514   friend class tcp_server; 517   friend class tcp_server;
515   518  
HITCBC 516   22 launcher(tcp_server& srv, worker_base& w) noexcept : srv_(&srv), w_(&w) 519   96 launcher(tcp_server& srv, worker_base& w) noexcept : srv_(&srv), w_(&w)
517   { 520   {
HITCBC 518   22 } 521   96 }
519   522  
520   public: 523   public:
521   /// Return the worker to the pool if not launched. 524   /// Return the worker to the pool if not launched.
HITCBC 522   22 ~launcher() 525   98 ~launcher()
523   { 526   {
HITCBC 524   22 if (w_) 527   98 if (w_)
HITCBC 525   4 srv_->push_sync(*w_); 528   8 srv_->push_sync(*w_);
HITCBC 526   22 } 529   98 }
527   530  
HITGIC 528   launcher(launcher&& o) noexcept 531   2 launcher(launcher&& o) noexcept
HITGIC 529   : srv_(o.srv_) 532   2 : srv_(o.srv_)
HITGIC 530   , w_(std::exchange(o.w_, nullptr)) 533   2 , w_(std::exchange(o.w_, nullptr))
531   { 534   {
HITGIC 532   } 535   2 }
533   launcher(launcher const&) = delete; 536   launcher(launcher const&) = delete;
534   launcher& operator=(launcher const&) = delete; 537   launcher& operator=(launcher const&) = delete;
535   launcher& operator=(launcher&&) = delete; 538   launcher& operator=(launcher&&) = delete;
536   539  
537   /** Launch the connection-handling coroutine. 540   /** Launch the connection-handling coroutine.
538   541  
539   Starts the given coroutine on the specified executor. When 542   Starts the given coroutine on the specified executor. When
540   the coroutine completes, the worker is automatically returned 543   the coroutine completes, the worker is automatically returned
541   to the idle pool. 544   to the idle pool.
542   545  
543   @param ex The executor to run the coroutine on. 546   @param ex The executor to run the coroutine on.
544   @param task The coroutine to execute. 547   @param task The coroutine to execute.
545   548  
546   @throws std::logic_error If this launcher was already invoked. 549   @throws std::logic_error If this launcher was already invoked.
547   */ 550   */
548   template<class Executor> 551   template<class Executor>
HITCBC 549   20 void operator()(Executor const& ex, capy::task<void> task) 552   90 void operator()(Executor const& ex, capy::task<void> task)
550   { 553   {
HITCBC 551   20 if (!w_) 554   90 if (!w_)
HITCBC 552   2 detail::throw_logic_error(); // launcher already invoked 555   2 detail::throw_logic_error(); // launcher already invoked
553   556  
HITCBC 554   18 auto* w = std::exchange(w_, nullptr); 557   88 auto* w = std::exchange(w_, nullptr);
555   558  
556   // Worker is being dispatched - add to active list 559   // Worker is being dispatched - add to active list
HITCBC 557   18 srv_->active_push(w); 560   88 srv_->active_push(w);
558   561  
559   // Return worker to pool if coroutine setup throws 562   // Return worker to pool if coroutine setup throws
560   struct guard_t 563   struct guard_t
561   { 564   {
562   tcp_server* srv; 565   tcp_server* srv;
563   worker_base* w; 566   worker_base* w;
HITCBC 564   18 ~guard_t() 567   88 ~guard_t()
565   { 568   {
HITCBC 566   18 if (w) 569   88 if (w)
MISUBC 567   ✗ srv->push_sync(*w); 570   ✗ srv->push_sync(*w);
HITCBC 568   18 } 571   88 }
HITCBC 569   18 } guard{srv_, w}; 572   88 } guard{srv_, w};
570   573  
571   // Reset worker's stop source for this connection 574   // Reset worker's stop source for this connection
HITCBC 572   18 w->stop_ = {}; 575   88 w->stop_ = {};
HITCBC 573   18 auto st = w->stop_.get_token(); 576   88 auto st = w->stop_.get_token();
574   577  
HITCBC 575   18 auto wrapper = 578   88 auto wrapper =
HITCBC 576   18 launch_coro<Executor>{}(ex, st, srv_, std::move(task), w); 579   88 launch_coro<Executor>{}(ex, st, srv_, std::move(task), w);
577   580  
578   // Executor and stop token stored in promise via constructor 581   // Executor and stop token stored in promise via constructor
HITCBC 579   18 ex.post(std::exchange(wrapper.h, nullptr)); // Release before post 582   88 ex.post(std::exchange(wrapper.h, nullptr)); // Release before post
HITCBC 580   18 guard.w = nullptr; // Success - dismiss guard 583   88 guard.w = nullptr; // Success - dismiss guard
HITCBC 581   18 } 584   88 }
582   }; 585   };
583   586  
584   /** Construct a TCP server. 587   /** Construct a TCP server.
585   588  
586   @tparam Ctx Execution context type satisfying ExecutionContext. 589   @tparam Ctx Execution context type satisfying ExecutionContext.
587   @tparam Ex Executor type satisfying Executor. 590   @tparam Ex Executor type satisfying Executor.
588   591  
589   @param ctx The execution context for socket operations. 592   @param ctx The execution context for socket operations.
590   @param ex The executor for dispatching coroutines. 593   @param ex The executor for dispatching coroutines.
591   594  
592   @par Example 595   @par Example
593   @code 596   @code
594   tcp_server srv(ctx, ctx.get_executor()); 597   tcp_server srv(ctx, ctx.get_executor());
595   srv.set_workers(make_workers(ctx, 100)); 598   srv.set_workers(make_workers(ctx, 100));
596   if (auto ec = srv.bind(endpoint{...})) 599   if (auto ec = srv.bind(endpoint{...}))
597   return; 600   return;
598   srv.start(); 601   srv.start();
599   @endcode 602   @endcode
600   */ 603   */
601   template<capy::ExecutionContext Ctx, capy::Executor Ex> 604   template<capy::ExecutionContext Ctx, capy::Executor Ex>
HITCBC 602   36 tcp_server(Ctx& ctx, Ex ex) : impl_(make_impl(ctx)) 605   73 tcp_server(Ctx& ctx, Ex ex) : impl_(make_impl(ctx))
HITCBC 603   36 , ex_(std::move(ex)) 606   73 , ex_(std::move(ex))
604   { 607   {
HITCBC 605   36 } 608   73 }
606   609  
607   public: 610   public:
608   /// Destroy the server, stopping all accept loops. 611   /// Destroy the server, stopping all accept loops.
609   ~tcp_server(); 612   ~tcp_server();
610   613  
611   tcp_server(tcp_server const&) = delete; 614   tcp_server(tcp_server const&) = delete;
612   tcp_server& operator=(tcp_server const&) = delete; 615   tcp_server& operator=(tcp_server const&) = delete;
613   616  
614   /** Move construct from another server. 617   /** Move construct from another server.
615   618  
616   @param o The source server. After the move, @p o is 619   @param o The source server. After the move, @p o is
617   in a valid but unspecified state. 620   in a valid but unspecified state.
618   */ 621   */
619   tcp_server(tcp_server&& o) noexcept; 622   tcp_server(tcp_server&& o) noexcept;
620   623  
621   /** Move assign from another server. 624   /** Move assign from another server.
622   625  
623   @param o The source server. After the move, @p o is 626   @param o The source server. After the move, @p o is
624   in a valid but unspecified state. 627   in a valid but unspecified state.
625   628  
626   @return `*this`. 629   @return `*this`.
627   */ 630   */
628   tcp_server& operator=(tcp_server&& o) noexcept; 631   tcp_server& operator=(tcp_server&& o) noexcept;
629   632  
630   /** Bind to a local endpoint. 633   /** Bind to a local endpoint.
631   634  
632   Creates an acceptor listening on the specified endpoint. 635   Creates an acceptor listening on the specified endpoint.
633   Multiple endpoints can be bound by calling this method 636   Multiple endpoints can be bound by calling this method
634   multiple times before @ref start. 637   multiple times before @ref start.
635   638  
636   @param ep The local endpoint to bind to. 639   @param ep The local endpoint to bind to.
637   640  
638   @return The error code if binding fails. 641   @return The error code if binding fails.
639   */ 642   */
640   [[nodiscard]] std::error_code bind(endpoint ep); 643   [[nodiscard]] std::error_code bind(endpoint ep);
641   644  
642   /** Set the worker pool. 645   /** Set the worker pool.
643   646  
644   Replaces any existing workers with the given range. Any 647   Replaces any existing workers with the given range. Any
645   previous workers are released and the idle/active lists 648   previous workers are released and the idle/active lists
646   are cleared before populating with new workers. 649   are cleared before populating with new workers.
647   650  
648   @tparam Range Forward range of pointer-like objects to worker_base. 651   @tparam Range Forward range of pointer-like objects to worker_base.
649   652  
650   @param workers Range of workers to manage. Each element must 653   @param workers Range of workers to manage. Each element must
651   support `std::to_address()` yielding `worker_base*`. 654   support `std::to_address()` yielding `worker_base*`.
652   655  
653   @par Example 656   @par Example
654   @code 657   @code
655   std::vector<std::unique_ptr<my_worker>> workers; 658   std::vector<std::unique_ptr<my_worker>> workers;
656   for(int i = 0; i < 100; ++i) 659   for(int i = 0; i < 100; ++i)
657   workers.push_back(std::make_unique<my_worker>(ctx)); 660   workers.push_back(std::make_unique<my_worker>(ctx));
658   srv.set_workers(std::move(workers)); 661   srv.set_workers(std::move(workers));
659   @endcode 662   @endcode
660   */ 663   */
661   template<std::ranges::forward_range Range> 664   template<std::ranges::forward_range Range>
662   requires std::convertible_to< 665   requires std::convertible_to<
663   decltype(std::to_address( 666   decltype(std::to_address(
664   std::declval<std::ranges::range_value_t<Range>&>())), 667   std::declval<std::ranges::range_value_t<Range>&>())),
665   worker_base*> 668   worker_base*>
HITCBC 666   36 void set_workers(Range&& workers) 669   73 void set_workers(Range&& workers)
667   { 670   {
668   // Clear existing state 671   // Clear existing state
HITCBC 669   36 storage_.reset(); 672   73 storage_.reset();
HITCBC 670   36 idle_head_ = nullptr; 673   73 idle_head_ = nullptr;
HITCBC 671   36 active_head_ = nullptr; 674   73 active_head_ = nullptr;
HITCBC 672   36 active_tail_ = nullptr; 675   73 active_tail_ = nullptr;
673   676  
674   // Take ownership and populate idle list 677   // Take ownership and populate idle list
675   using StorageType = std::decay_t<Range>; 678   using StorageType = std::decay_t<Range>;
HITCBC 676   36 auto* p = new StorageType(std::forward<Range>(workers)); 679   73 auto* p = new StorageType(std::forward<Range>(workers));
HITCBC 677   36 storage_ = std::shared_ptr<void>( 680   73 storage_ = std::shared_ptr<void>(
HITCBC 678   36 p, [](void* ptr) { delete static_cast<StorageType*>(ptr); }); 681   73 p, [](void* ptr) { delete static_cast<StorageType*>(ptr); });
HITCBC 679   162 for (auto&& elem : *static_cast<StorageType*>(p)) 682   236 for (auto&& elem : *static_cast<StorageType*>(p))
HITCBC 680   126 idle_push(std::to_address(elem)); 683   163 idle_push(std::to_address(elem));
HITCBC 681   36 } 684   73 }
682   685  
683   /** Start accepting connections. 686   /** Start accepting connections.
684   687  
685   Launches accept loops for all bound endpoints. Incoming 688   Launches accept loops for all bound endpoints. Incoming
686   connections are dispatched to idle workers from the pool. 689   connections are dispatched to idle workers from the pool.
687   690  
688   Calling `start()` on an already-running server has no effect. 691   Calling `start()` on an already-running server has no effect.
689   692  
690   @par Preconditions 693   @par Preconditions
691   - At least one endpoint bound via @ref bind. 694   - At least one endpoint bound via @ref bind.
692   - Workers provided via @ref set_workers. 695   - Workers provided via @ref set_workers.
693   - If restarting, @ref join must have completed first. 696   - If restarting, @ref join must have completed first.
694   697  
695   @par Effects 698   @par Effects
696   Creates one accept coroutine per bound endpoint. Each coroutine 699   Creates one accept coroutine per bound endpoint. Each coroutine
697   runs on the server's executor, waiting for connections and 700   runs on the server's executor, waiting for connections and
698   dispatching them to idle workers. 701   dispatching them to idle workers.
699   702  
700   @par Restart Sequence 703   @par Restart Sequence
701   To restart after stopping, complete the full shutdown cycle: 704   To restart after stopping, complete the full shutdown cycle:
702   @code 705   @code
703   srv.start(); 706   srv.start();
704   ioc.run_for( 1s ); 707   ioc.run_for( 1s );
705   srv.stop(); // 1. Signal shutdown 708   srv.stop(); // 1. Signal shutdown
706   ioc.run(); // 2. Drain remaining completions 709   ioc.run(); // 2. Drain remaining completions
707   srv.join(); // 3. Wait for accept loops 710   srv.join(); // 3. Wait for accept loops
708   711  
709   // Now safe to restart 712   // Now safe to restart
710   srv.start(); 713   srv.start();
711   ioc.run(); 714   ioc.run();
712   @endcode 715   @endcode
713   716  
714   @par Thread Safety 717   @par Thread Safety
715   Not thread safe. 718   Not thread safe.
716   719  
717   @throws std::logic_error If a previous session has not been 720   @throws std::logic_error If a previous session has not been
718   joined (accept loops still active). 721   joined (accept loops still active).
719   */ 722   */
720   void start(); 723   void start();
721   724  
722   /** Return the local endpoint for the i-th bound port. 725   /** Return the local endpoint for the i-th bound port.
723   726  
724   @param index Zero-based index into the list of bound ports. 727   @param index Zero-based index into the list of bound ports.
725   728  
726   @return The local endpoint, or a default-constructed endpoint 729   @return The local endpoint, or a default-constructed endpoint
727   if @p index is out of range or the acceptor is not open. 730   if @p index is out of range or the acceptor is not open.
728   */ 731   */
729   endpoint local_endpoint(std::size_t index = 0) const noexcept; 732   endpoint local_endpoint(std::size_t index = 0) const noexcept;
730   733  
731   /** Stop accepting connections. 734   /** Stop accepting connections.
732   735  
733   Requests the accept loops' stop token and requests cancellation 736   Requests the accept loops' stop token and requests cancellation
734   of active workers via their stop tokens. The acceptors are not 737   of active workers via their stop tokens. The acceptors are not
735   closed; a suspended accept completes once more before its loop 738   closed; a suspended accept completes once more before its loop
736   observes the stop token and ends. 739   observes the stop token and ends.
737   740  
738   This function returns immediately; it does not wait for workers 741   This function returns immediately; it does not wait for workers
739   to finish. Pending I/O operations complete asynchronously. 742   to finish. Pending I/O operations complete asynchronously.
740   743  
741   Calling `stop()` on a non-running server has no effect. 744   Calling `stop()` on a non-running server has no effect.
742   745  
743   @par Effects 746   @par Effects
744   - Requests stop on the accept loops' stop token. The acceptors 747   - Requests stop on the accept loops' stop token. The acceptors
745   are not closed; a pending accept completes once more before 748   are not closed; a pending accept completes once more before
746   the accept loop ends. 749   the accept loop ends.
747   - Requests stop on each active worker's stop token. 750   - Requests stop on each active worker's stop token.
748   - Workers observing their stop token should exit promptly. 751   - Workers observing their stop token should exit promptly.
749   752  
750   @par Postconditions 753   @par Postconditions
751   No new connections will be accepted. Active workers continue 754   No new connections will be accepted. Active workers continue
752   until they observe their stop token or complete naturally. 755   until they observe their stop token or complete naturally.
753   756  
754   @par What Happens Next 757   @par What Happens Next
755   After calling `stop()`: 758   After calling `stop()`:
756   1. Let `ioc.run()` return (drains pending completions). 759   1. Let `ioc.run()` return (drains pending completions).
757   2. Call @ref join to wait for accept loops to finish. 760   2. Call @ref join to wait for accept loops to finish.
758   3. Only then is it safe to restart or destroy the server. 761   3. Only then is it safe to restart or destroy the server.
759   762  
760   @par Thread Safety 763   @par Thread Safety
761   Not thread safe. 764   Not thread safe.
762   765  
763   @see join, start 766   @see join, start
764   */ 767   */
765   void stop(); 768   void stop();
766   769  
767   /** Block until all accept loops complete. 770   /** Block until all accept loops complete.
768   771  
769   Blocks the calling thread until all accept coroutines launched 772   Blocks the calling thread until all accept coroutines launched
770   by @ref start have finished executing. This synchronizes the 773   by @ref start have finished executing. This synchronizes the
771   shutdown sequence, ensuring the server is fully stopped before 774   shutdown sequence, ensuring the server is fully stopped before
772   restarting or destroying it. 775   restarting or destroying it.
773   776  
774   @par Preconditions 777   @par Preconditions
775   @ref stop has been called and `ioc.run()` has returned. 778   @ref stop has been called and `ioc.run()` has returned.
776   779  
777   @par Postconditions 780   @par Postconditions
778   All accept loops have completed. The server is in the stopped 781   All accept loops have completed. The server is in the stopped
779   state and may be restarted via @ref start. 782   state and may be restarted via @ref start.
780   783  
781   @par Example (Correct Usage) 784   @par Example (Correct Usage)
782   @code 785   @code
783   // main thread 786   // main thread
784   srv.start(); 787   srv.start();
785   ioc.run(); // Blocks until work completes 788   ioc.run(); // Blocks until work completes
786   srv.join(); // Safe: called after ioc.run() returns 789   srv.join(); // Safe: called after ioc.run() returns
787   @endcode 790   @endcode
788   791  
789   @par WARNING: Deadlock Scenarios 792   @par WARNING: Deadlock Scenarios
790   Calling `join()` from the wrong context causes deadlock: 793   Calling `join()` from the wrong context causes deadlock:
791   794  
792   @code 795   @code
793   // WRONG: calling join() from inside a worker coroutine 796   // WRONG: calling join() from inside a worker coroutine
794   void run( launcher launch ) override 797   void run( launcher launch ) override
795   { 798   {
796   launch( ex, [this]() -> capy::task<> 799   launch( ex, [this]() -> capy::task<>
797   { 800   {
798   srv_.join(); // DEADLOCK: blocks the executor 801   srv_.join(); // DEADLOCK: blocks the executor
799   co_return; 802   co_return;
800   }()); 803   }());
801   } 804   }
802   805  
803   // WRONG: calling join() while ioc.run() is still active 806   // WRONG: calling join() while ioc.run() is still active
804   std::thread t( [&]{ ioc.run(); } ); 807   std::thread t( [&]{ ioc.run(); } );
805   srv.stop(); 808   srv.stop();
806   srv.join(); // DEADLOCK: ioc.run() still running in thread t 809   srv.join(); // DEADLOCK: ioc.run() still running in thread t
807   @endcode 810   @endcode
808   811  
809   @par Thread Safety 812   @par Thread Safety
810   May be called from any thread, but will deadlock if called 813   May be called from any thread, but will deadlock if called
811   from within the io_context event loop or from a worker coroutine. 814   from within the io_context event loop or from a worker coroutine.
812   815  
813   @see stop, start 816   @see stop, start
814   */ 817   */
815   void join(); 818   void join();
816   819  
817   private: 820   private:
818   capy::task<> do_stop(); 821   capy::task<> do_stop();
819   }; 822   };
820   823  
821   #ifdef _MSC_VER 824   #ifdef _MSC_VER
822   #pragma warning(pop) 825   #pragma warning(pop)
823   #endif 826   #endif
824   827  
825   } // namespace boost::corosio 828   } // namespace boost::corosio
826   829  
827   #endif 830   #endif