include/boost/corosio/tcp_server.hpp

98.4% Lines (248 / 252, 1 excl) 42.5% Branches (136 / 320) 78.5% Functions (84 / 107)

[ ] prev/next file n p prev/next uncovered line

tcp_server.hpp

f(x) Functions (107)

Function Calls Lines Branches Blocks

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