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