include/boost/corosio/tcp_server.hpp

98.1% Lines (211 / 215, 1 excl) 87.3% Branches (55 / 63) 84.0% Functions (63 / 75)

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

tcp_server.hpp

f(x) Functions (75)

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			std::atomic<std::size_t> refs{1};
130			impl* impl_;
131			capy::any_executor ex;
132			std::mutex mutex;
133			waiter* waiters = nullptr;
134			worker_base* idle_head = nullptr; // forward list
135			worker_base* active_head = nullptr; // doubly linked
136			worker_base* active_tail = nullptr;
137			std::size_t active_accepts = 0; // guarded by the join mutex
138			std::size_t lent = 0; // borrowed workers, guarded by mutex
139			std::shared_ptr<void> storage; // owns the worker container
140			bool running = false;
141			// Set by stop(); accept loops waiting for a worker give up.
142			bool stopping = false; // guarded by mutex
143
144		59x	state(impl* i, capy::any_executor const& e) noexcept : impl_(i), ex(e)
145			{
146		59x	}
147
148			// The list operations below require `mutex`.
149
150		837x	void idle_push(worker_base* w) noexcept
151			{
152		837x	w->next_ = idle_head;
153		837x	idle_head = w;
154		837x	}
155
156		671x	worker_base* idle_pop() noexcept
157			{
158		671x	auto* w = idle_head;
159	2/2 ✓ Branch 2 → 3 taken 652 times. ✓ Branch 2 → 4 taken 19 times.	671x	if (w)
160			{
161		652x	idle_head = w->next_;
162		652x	++lent;
163			}
164		671x	return w;
165			}
166
167		268x	void active_push(worker_base* w) noexcept
168			{
169		268x	w->next_ = nullptr;
170		268x	w->prev_ = active_tail;
171	2/2 ✓ Branch 2 → 3 taken 117 times. ✓ Branch 2 → 4 taken 151 times.	268x	if (active_tail)
172		117x	active_tail->next_ = w;
173			else
174		151x	active_head = w;
175		268x	active_tail = w;
176		268x	}
177
178		659x	void active_remove(worker_base* w) noexcept
179			{
180			// Skip if not in active list (e.g., after failed accept)
181	4/4 ✓ Branch 2 → 3 taken 437 times. ✓ Branch 2 → 5 taken 222 times. ✓ Branch 3 → 4 taken 391 times. ✓ Branch 3 → 5 taken 46 times.	659x	if (w != active_head && w->prev_ == nullptr)
182		391x	return;
183	2/2 ✓ Branch 5 → 6 taken 46 times. ✓ Branch 5 → 7 taken 222 times.	268x	if (w->prev_)
184		46x	w->prev_->next_ = w->next_;
185			else
186		222x	active_head = w->next_;
187	2/2 ✓ Branch 8 → 9 taken 79 times. ✓ Branch 8 → 10 taken 189 times.	268x	if (w->next_)
188		79x	w->next_->prev_ = w->prev_;
189			else
190		189x	active_tail = w->prev_;
191		268x	w->prev_ = nullptr; // Mark as not in active list
192			}
193
194			/// Return @p w to the pool, handing it to a waiting accept loop
195			/// if there is one.
196		659x	void push(worker_base& w) noexcept
197			{
198		659x	waiter* wake = nullptr;
199			{
200		659x	std::lock_guard<std::mutex> lock(mutex);
201		659x	active_remove(&w);
202	2/2 ✓ Branch 4 → 5 taken 8 times. ✓ Branch 4 → 6 taken 651 times.	659x	if (waiters)
203			{
204		8x	wake = waiters;
205		8x	waiters = wake->next;
206		8x	wake->w = &w;
207			}
208			else
209			{
210		651x	idle_push(&w);
211		651x	--lent;
212			}
213		659x	}
214	2/2 ✓ Branch 9 → 10 taken 8 times. ✓ Branch 9 → 11 taken 651 times.	659x	if (wake)
215		8x	ex.post(wake->cont);
216		659x	}
217
218			/// Drop one hold on @p w, returning it to the pool with the last.
219		934x	void drop_hold(worker_base& w) noexcept
220			{
221	2/2 ✓ Branch 4 → 5 taken 659 times. ✓ Branch 4 → 6 taken 275 times.	1868x	if (w.holds_.fetch_sub(1, std::memory_order_acq_rel) == 1)
222		659x	push(w);
223		934x	}
224
225			/// Mark the pool as stopping and release every accept loop
226			/// waiting for a worker, with none.
227		51x	void stop_waiters() noexcept
228			{
229		51x	waiter* list = nullptr;
230			{
231		51x	std::lock_guard<std::mutex> lock(mutex);
232		51x	stopping = true;
233		51x	list = std::exchange(waiters, nullptr);
234		51x	}
235	2/2 ✓ Branch 8 → 6 taken 11 times. ✓ Branch 8 → 9 taken 51 times.	62x	while (list)
236			{
237		11x	auto* next = list->next;
238		11x	ex.post(list->cont);
239		11x	list = next;
240			}
241		51x	}
242			};
243
244			/// Add a reference to @p s.
245		698x	static void add_ref(state* s) noexcept
246			{
247		698x	s->refs.fetch_add(1, std::memory_order_relaxed);
248		698x	}
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		423x	explicit state_ref(state* p) noexcept : p_(p)
260			{
261		423x	add_ref(p_);
262		423x	}
263
264			state_ref(state_ref const& o) noexcept : p_(o.p_)
265			{
266			if (p_)
267			add_ref(p_);
268			}
269
270		1111x	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		1534x	~state_ref()
276			{
277	2/2 ✓ Branch 2 → 3 taken 423 times. ✓ Branch 2 → 4 taken 1111 times.	1534x	if (p_)
278		423x	release(p_);
279		1534x	}
280
281		1123x	state* operator->() const noexcept
282			{
283		1123x	return p_;
284			}
285
286		1592x	state& operator*() const noexcept
287			{
288		1592x	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		268x	promise_type(Closure&&, E e, S s, Args&&...)
330		268x	: ex(std::move(e))
331		268x	, env_{
332		268x	capy::executor_ref(ex), std::move(s),
333		536x	capy::get_current_frame_allocator(), &owner_}
334			{
335		268x	}
336
337		268x	launch_wrapper get_return_object() noexcept
338			{
339			return {
340		268x	std::coroutine_handle<promise_type>::from_promise(*this)};
341			}
342		268x	std::suspend_always initial_suspend() noexcept
343			{
344		268x	return {};
345			}
346		259x	std::suspend_never final_suspend() noexcept
347			{
348		259x	return {};
349			}
350		259x	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		525x	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		525x	bool await_ready()
370			{
371		525x	return aw.await_ready();
372			}
373		520x	decltype(auto) await_resume()
374			{
375		520x	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		525x	auto await_suspend(std::coroutine_handle<promise_type> h)
382			{
383		525x	return capy::detail::symmetric_transfer(
384		525x	capy::detail::call_await_suspend(&aw, h, env));
385			}
386			};
387		525x	return adapter{std::forward<Awaitable>(a), &env_};
388			}
389			};
390
391			std::coroutine_handle<promise_type> h;
392
393		268x	launch_wrapper(std::coroutine_handle<promise_type> handle) noexcept
394		268x	: h(handle)
395			{
396		268x	}
397
398		268x	~launch_wrapper()
399			{
400			// An unstarted root owns no chain yet: destroying it is
401			// the whole teardown.
402	2/2 ✓ Branch 3 → 4 taken 2 times. ✓ Branch 3 → 5 taken 266 times.	268x	if (h)
403		2x	h.destroy();
404		268x	}
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		268x	worker_hold(state& st, worker_base& w) noexcept : st_(&st), w_(&w) {}
424
425		800x	worker_hold(worker_hold&& o) noexcept
426		800x	: st_(std::move(o.st_))
427		800x	, w_(std::exchange(o.w_, nullptr))
428			{
429		800x	}
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		1068x	~worker_hold()
436			{
437	2/2 ✓ Branch 2 → 3 taken 268 times. ✓ Branch 2 → 5 taken 800 times.	1068x	if (w_)
438		268x	st_->drop_hold(*w_);
439		1068x	}
440
441		261x	state& server() const noexcept
442			{
443		261x	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	1/1 ✓ Branch 9 → 10 taken 268 times.	268x	operator()(Executor, std::stop_token, session<Executor> in)
467			{
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			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			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			s.work.reset();
479			// The hold returns the worker from the server's executor.
480			co_await hop_awaitable{s.hold.server(), true};
481		536x	}
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		646x	explicit hop_awaitable(state& st, bool root = false) noexcept
494		646x	: st_(st)
495		646x	, root_(root)
496			{
497		646x	}
498
499		646x	bool await_ready() const noexcept
500			{
501		646x	return false;
502			}
503
504			std::coroutine_handle<>
505		646x	await_suspend(
506			std::coroutine_handle<> h, capy::await_context const* env) noexcept
507			{
508		646x	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 2 → 3 taken 385 times. ✓ Branch 2 → 4 taken 261 times.	646x	if (!root_)
513		385x	cont_.owner = env->owner;
514		646x	return st_.ex.dispatch(cont_);
515			}
516
517		643x	void await_resume() noexcept {}
518			};
519
520			class pop_awaitable
521			{
522			state& st_;
523			waiter wait_;
524
525			public:
526		671x	pop_awaitable(state& st) noexcept : st_(st), wait_{} {}
527
528		671x	bool await_ready() const noexcept
529			{
530		671x	return false;
531			}
532
533			bool
534		671x	await_suspend(std::coroutine_handle<> h, capy::await_context const* env) noexcept
535			{
536		671x	std::lock_guard<std::mutex> lock(st_.mutex);
537	1/2 ✗ Branch 3 → 4 not taken. ✓ Branch 3 → 5 taken 671 times.	671x	if (st_.stopping)
538			{
539		✗	wait_.w = nullptr;
540		✗	return false;
541			}
542	2/2 ✓ Branch 6 → 7 taken 652 times. ✓ Branch 6 → 8 taken 19 times.	671x	if (auto* w = st_.idle_pop())
543			{
544		652x	wait_.w = w;
545		652x	return false;
546			}
547		19x	wait_.cont.h = h;
548		19x	wait_.cont.owner = env->owner;
549		19x	wait_.w = nullptr;
550		19x	wait_.next = st_.waiters;
551		19x	st_.waiters = &wait_;
552		19x	return true;
553		671x	}
554
555			/// Return the worker, or null once the server is stopping.
556		669x	worker_base* await_resume() noexcept
557			{
558			// Set by await_suspend, or by the push that woke us.
559		669x	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		275x	launcher(state& st, worker_base& w) noexcept : st_(&st), w_(&w)
643			{
644		275x	add_ref(st_);
645		275x	}
646
647			public:
648			/// Return the worker to the pool if not started.
649		281x	~launcher()
650			{
651	2/2 ✓ Branch 2 → 3 taken 7 times. ✓ Branch 2 → 4 taken 274 times.	281x	if (w_)
652		7x	st_->drop_hold(*w_);
653	2/2 ✓ Branch 4 → 5 taken 275 times. ✓ Branch 4 → 6 taken 6 times.	281x	if (st_)
654		275x	release(st_);
655		281x	}
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		6x	launcher(launcher&& o) noexcept
663		6x	: st_(std::exchange(o.st_, nullptr))
664		6x	, w_(std::exchange(o.w_, nullptr))
665			{
666		6x	}
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		269x	void operator()(Executor const& ex, capy::task<void> task)
712			{
713	2/2 ✓ Branch 2 → 3 taken 1 time. ✓ Branch 2 → 4 taken 268 times.	269x	if (!w_)
714		1x	detail::throw_logic_error(); // launcher already invoked
715
716		268x	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		268x	std::stop_token st;
725		268x	bool stopped_before = false;
726			{
727	1/1 ✓ Branch 5 → 6 taken 268 times.	268x	std::lock_guard<std::mutex> lock(st_->mutex);
728	1/2 ✗ Branch 7 → 8 not taken. ✓ Branch 7 → 12 taken 268 times.	268x	if (w->stop_.stop_requested())
729	0/1 ✗ Branch 8 → 9 not taken.	✗	w->stop_ = {};
730		268x	st = w->stop_.get_token();
731		268x	st_->active_push(w);
732		268x	stopped_before = st_->stopping;
733		268x	}
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	2/2 ✓ Branch 17 → 18 taken 2 times. ✓ Branch 17 → 19 taken 266 times.	268x	if (stopped_before)
738		2x	w->stop_.request_stop();
739
740		268x	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			std::pmr::memory_resource* prev =
748			capy::get_current_frame_allocator();
749		268x	allocator_scope() noexcept
750		268x	{
751		268x	capy::set_current_frame_allocator(
752			capy::get_recycling_memory_resource());
753		268x	}
754		268x	~allocator_scope()
755			{
756		268x	capy::set_current_frame_allocator(prev);
757		268x	}
758			};
759	1/5 ✓ Branch 20 → 21 taken 268 times. ✗ Branch 21 → 22 not taken. ✗ Branch 21 → 23 not taken. ✗ Branch 24 → 25 not taken. ✗ Branch 24 → 26 not taken.	804x	launch_wrapper<Executor> wrapper = [&] {
760		268x	allocator_scope scope;
761		536x	return launch_coro<Executor>{}(
762			ex, st,
763			session<Executor>{
764		268x	std::move(hold), capy::work_guard<Executor>(ex),
765	1/1 ✓ Branch 9 → 10 taken 268 times.	1072x	std::move(task)});
766		268x	}();
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		268x	auto h = wrapper.h;
774		268x	h.promise().owner_.bind(h, h.promise().ex);
775		268x	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	1/1 ✓ Branch 29 → 30 taken 266 times.	268x	ex.post(h.promise().cont_);
780		266x	wrapper.h = nullptr;
781		272x	}
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		59x	tcp_server(Ctx& ctx, Ex ex)
797	2/2 ✓ Branch 3 → 4 taken 59 times. ✓ Branch 4 → 5 taken 59 times.	59x	: st_(make_state(ctx, capy::any_executor(std::move(ex))))
798			{
799		59x	}
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		66x	void set_workers(Range&& workers)
881			{
882			// The previous workers are released after the lock.
883		66x	std::shared_ptr<void> previous;
884	1/1 ✓ Branch 2 → 3 taken 66 times.	66x	std::lock_guard<std::mutex> lock(st_->mutex);
885	2/2 ✓ Branch 3 → 4 taken 3 times. ✓ Branch 3 → 5 taken 63 times.	66x	if (st_->lent != 0)
886		3x	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/1 ✓ Branch 5 → 6 taken 63 times.	63x	auto* p = new StorageType(std::forward<Range>(workers));
891	1/1 ✓ Branch 8 → 9 taken 63 times.	63x	std::shared_ptr<void> storage(
892	1/2 ✓ Branch 2 → 3 taken 63 times. ✗ Branch 2 → 5 not taken.	63x	p, [](void* ptr) { delete static_cast<StorageType*>(ptr); });
893
894		63x	st_->idle_head = nullptr;
895		63x	st_->active_head = nullptr;
896		63x	st_->active_tail = nullptr;
897		63x	previous = std::exchange(st_->storage, std::move(storage));
898	2/2 ✓ Branch 20 → 15 taken 186 times. ✓ Branch 20 → 21 taken 63 times.	249x	for (auto&& elem : *p)
899		186x	st_->idle_push(std::to_address(elem));
900		69x	}
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