include/boost/corosio/tcp_server.hpp

98.1% Lines (203 / 207, 1 excl) 84.9% Functions (62 / 73)

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

tcp_server.hpp

f(x) Functions (73)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Vinnie Falco (vinnie.falco@gmail.com)
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 worker_base, set_workers, 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	169x	state(impl* i, capy::any_executor const& e) noexcept : impl_(i), ex(e)
145		{
146	169x	}
147
148		// The list operations below require `mutex`.
149
150	1198x	void idle_push(worker_base* w) noexcept
151		{
152	1198x	w->next_ = idle_head;
153	1198x	idle_head = w;
154	1198x	}
155
156	1158x	worker_base* idle_pop() noexcept
157		{
158	1158x	auto* w = idle_head;
159	1158x	if (w)
160		{
161	787x	idle_head = w->next_;
162	787x	++lent;
163		}
164	1158x	return w;
165		}
166
167	1000x	void active_push(worker_base* w) noexcept
168		{
169	1000x	w->next_ = nullptr;
170	1000x	w->prev_ = active_tail;
171	1000x	if (active_tail)
172	703x	active_tail->next_ = w;
173		else
174	297x	active_head = w;
175	1000x	active_tail = w;
176	1000x	}
177
178	1130x	void active_remove(worker_base* w) noexcept
179		{
180		// Skip if not in active list (e.g., after failed accept)
181	1130x	if (w != active_head && w->prev_ == nullptr)
182	130x	return;
183	1000x	if (w->prev_)
184	268x	w->prev_->next_ = w->next_;
185		else
186	732x	active_head = w->next_;
187	1000x	if (w->next_)
188	608x	w->next_->prev_ = w->prev_;
189		else
190	392x	active_tail = w->prev_;
191	1000x	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	1130x	void push(worker_base& w) noexcept
197		{
198	1130x	waiter* wake = nullptr;
199		{
200	1130x	std::lock_guard<std::mutex> lock(mutex);
201	1130x	active_remove(&w);
202	1130x	if (waiters)
203		{
204	345x	wake = waiters;
205	345x	waiters = wake->next;
206	345x	wake->w = &w;
207		}
208		else
209		{
210	785x	idle_push(&w);
211	785x	--lent;
212		}
213	1130x	}
214	1130x	if (wake)
215	345x	ex.post(wake->cont);
216	1130x	}
217
218		/// Drop one hold on @p w, returning it to the pool with the last.
219	2144x	void drop_hold(worker_base& w) noexcept
220		{
221	4288x	if (w.holds_.fetch_sub(1, std::memory_order_acq_rel) == 1)
222	1130x	push(w);
223	2144x	}
224
225		/// Mark the pool as stopping and release every accept loop
226		/// waiting for a worker, with none.
227	157x	void stop_waiters() noexcept
228		{
229	157x	waiter* list = nullptr;
230		{
231	157x	std::lock_guard<std::mutex> lock(mutex);
232	157x	stopping = true;
233	157x	list = std::exchange(waiters, nullptr);
234	157x	}
235	183x	while (list)
236		{
237	26x	auto* next = list->next;
238	26x	ex.post(list->cont);
239	26x	list = next;
240		}
241	157x	}
242		};
243
244		/// Add a reference to @p s.
245	2489x	static void add_ref(state* s) noexcept
246		{
247	2489x	s->refs.fetch_add(1, std::memory_order_relaxed);
248	2489x	}
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	1475x	explicit state_ref(state* p) noexcept : p_(p)
260		{
261	1475x	add_ref(p_);
262	1475x	}
263
264		state_ref(state_ref const& o) noexcept : p_(o.p_)
265		{
266		if (p_)
267		add_ref(p_);
268		}
269
270	3944x	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	5419x	~state_ref()
276		{
277	5419x	if (p_)
278	1475x	release(p_);
279	5419x	}
280
281	2742x	state* operator->() const noexcept
282		{
283	2742x	return p_;
284		}
285
286	3276x	state& operator*() const noexcept
287		{
288	3276x	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	1000x	promise_type(Closure&&, E e, S s, Args&&...)
330	1000x	: ex(std::move(e))
331	1000x	, env_{
332	1000x	capy::executor_ref(ex), std::move(s),
333	2000x	capy::get_current_frame_allocator(), &owner_}
334		{
335	1000x	}
336
337	1000x	launch_wrapper get_return_object() noexcept
338		{
339		return {
340	1000x	std::coroutine_handle<promise_type>::from_promise(*this)};
341		}
342	1000x	std::suspend_always initial_suspend() noexcept
343		{
344	1000x	return {};
345		}
346	983x	std::suspend_never final_suspend() noexcept
347		{
348	983x	return {};
349		}
350	983x	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	1978x	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		bool await_ready()
370		{
371		return aw.await_ready();
372		}
373		decltype(auto) await_resume()
374		{
375		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		auto await_suspend(std::coroutine_handle<promise_type> h)
382		{
383		return capy::detail::symmetric_transfer(
384		capy::detail::call_await_suspend(&aw, h, env));
385		}
386		};
387	1978x	return adapter{std::forward<Awaitable>(a), &env_};
388		}
389		};
390
391		std::coroutine_handle<promise_type> h;
392
393	1000x	launch_wrapper(std::coroutine_handle<promise_type> handle) noexcept
394	1000x	: h(handle)
395		{
396	1000x	}
397
398	1000x	~launch_wrapper()
399		{
400		// An unstarted root owns no chain yet: destroying it is
401		// the whole teardown.
402	1000x	if (h)
403	4x	h.destroy();
404	1000x	}
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	1000x	worker_hold(state& st, worker_base& w) noexcept : st_(&st), w_(&w) {}
424
425	2992x	worker_hold(worker_hold&& o) noexcept
426	2992x	: st_(std::move(o.st_))
427	2992x	, w_(std::exchange(o.w_, nullptr))
428		{
429	2992x	}
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	3992x	~worker_hold()
436		{
437	3992x	if (w_)
438	1000x	st_->drop_hold(*w_);
439	3992x	}
440
441	986x	state& server() const noexcept
442		{
443	986x	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	1000x	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	2000x	}
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	1104x	explicit hop_awaitable(state& st, bool root = false) noexcept
494	1104x	: st_(st)
495	1104x	, root_(root)
496		{
497	1104x	}
498
499	1104x	bool await_ready() const noexcept
500		{
501	1104x	return false;
502		}
503
504		std::coroutine_handle<>
505	1104x	await_suspend(
506		std::coroutine_handle<> h, capy::await_context const* env) noexcept
507		{
508	1104x	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	1104x	if (!root_)
513	118x	cont_.owner = env->owner;
514	1104x	return st_.ex.dispatch(cont_);
515		}
516
517	1099x	void await_resume() noexcept {}
518		};
519
520		class pop_awaitable
521		{
522		state& st_;
523		waiter wait_;
524
525		public:
526	1158x	pop_awaitable(state& st) noexcept : st_(st), wait_{} {}
527
528	1158x	bool await_ready() const noexcept
529		{
530	1158x	return false;
531		}
532
533		bool
534	1158x	await_suspend(std::coroutine_handle<> h, capy::await_context const* env) noexcept
535		{
536	1158x	std::lock_guard<std::mutex> lock(st_.mutex);
537	1158x	if (st_.stopping)
538		{
539	✗	wait_.w = nullptr;
540	✗	return false;
541		}
542	1158x	if (auto* w = st_.idle_pop())
543		{
544	787x	wait_.w = w;
545	787x	return false;
546		}
547	371x	wait_.cont.h = h;
548	371x	wait_.cont.owner = env->owner;
549	371x	wait_.w = nullptr;
550	371x	wait_.next = st_.waiters;
551	371x	st_.waiters = &wait_;
552	371x	return true;
553	1158x	}
554
555		/// Return the worker, or null once the server is stopping.
556	1154x	worker_base* await_resume() noexcept
557		{
558		// Set by await_suspend, or by the push that woke us.
559	1154x	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 tcp_server, 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 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	1014x	launcher(state& st, worker_base& w) noexcept : st_(&st), w_(&w)
643		{
644	1014x	add_ref(st_);
645	1014x	}
646
647		public:
648		/// Return the worker to the pool if not started.
649	1026x	~launcher()
650		{
651	1026x	if (w_)
652	14x	st_->drop_hold(*w_);
653	1026x	if (st_)
654	1014x	release(st_);
655	1026x	}
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	12x	launcher(launcher&& o) noexcept
663	12x	: st_(std::exchange(o.st_, nullptr))
664	12x	, w_(std::exchange(o.w_, nullptr))
665		{
666	12x	}
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	1002x	void operator()(Executor const& ex, capy::task<void> task)
712		{
713	1002x	if (!w_)
714	2x	detail::throw_logic_error(); // launcher already invoked
715
716	1000x	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	1000x	std::stop_token st;
725	1000x	bool stopped_before = false;
726		{
727	1000x	std::lock_guard<std::mutex> lock(st_->mutex);
728	1000x	if (w->stop_.stop_requested())
729	✗	w->stop_ = {};
730	1000x	st = w->stop_.get_token();
731	1000x	st_->active_push(w);
732	1000x	stopped_before = st_->stopping;
733	1000x	}
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	1000x	if (stopped_before)
738	10x	w->stop_.request_stop();
739
740	1000x	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	134x	allocator_scope() noexcept
750	134x	{
751	134x	capy::set_current_frame_allocator(
752		capy::get_recycling_memory_resource());
753	134x	}
754	134x	~allocator_scope()
755		{
756	134x	capy::set_current_frame_allocator(prev);
757	134x	}
758		};
759	1268x	launch_wrapper<Executor> wrapper = [&] {
760	134x	allocator_scope scope;
761		return launch_coro<Executor>{}(
762		ex, st,
763		session<Executor>{
764	134x	std::move(hold), capy::work_guard<Executor>(ex),
765	402x	std::move(task)});
766	134x	}();
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	1000x	auto h = wrapper.h;
774	1000x	h.promise().owner_.bind(h, h.promise().ex);
775	1000x	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	1000x	ex.post(h.promise().cont_);
780	996x	wrapper.h = nullptr;
781	1008x	}
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	169x	tcp_server(Ctx& ctx, Ex ex)
797	169x	: st_(make_state(ctx, capy::any_executor(std::move(ex))))
798		{
799	169x	}
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	183x	void set_workers(Range&& workers)
881		{
882		// The previous workers are released after the lock.
883	183x	std::shared_ptr<void> previous;
884	183x	std::lock_guard<std::mutex> lock(st_->mutex);
885	183x	if (st_->lent != 0)
886	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	177x	auto* p = new StorageType(std::forward<Range>(workers));
891	177x	std::shared_ptr<void> storage(
892	177x	p, [](void* ptr) { delete static_cast<StorageType*>(ptr); });
893
894	177x	st_->idle_head = nullptr;
895	177x	st_->active_head = nullptr;
896	177x	st_->active_tail = nullptr;
897	177x	previous = std::exchange(st_->storage, std::move(storage));
898	590x	for (auto&& elem : *p)
899	413x	st_->idle_push(std::to_address(elem));
900	189x	}
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 join, 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 stop, 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