src/corosio/src/tcp_server.cpp

95.9% Lines (94 / 98) 100.0% Functions (18 / 18)

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

tcp_server.cpp

f(x) Functions (18)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Vinnie Falco (vinnie.falco@gmail.com)
3		// Copyright (c) 2026 Steve Gerbino
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		#include <boost/corosio/tcp_server.hpp>
12		#include <boost/corosio/detail/except.hpp>
13		#include <condition_variable>
14		#include <mutex>
15		#include <utility>
16
17		namespace boost::corosio {
18
19	419x	tcp_server::worker_base::worker_base() = default;
20	419x	tcp_server::worker_base::~worker_base() = default;
21
22		struct tcp_server::impl
23		{
24		std::mutex join_mutex;
25		std::condition_variable join_cv;
26		capy::execution_context& ctx;
27		std::vector<tcp_acceptor> ports;
28		std::stop_source stop;
29		// Guarded by join_mutex. Each listener is closed exactly once after
30		// the server is gone: by its loop's completion if the loop was still
31		// running when the server went, otherwise by the departing server.
32		std::vector<bool> loop_running;
33		bool discarded = false;
34
35	169x	explicit impl(capy::execution_context& c) noexcept : ctx(c) {}
36		};
37
38		tcp_server::state*
39	169x	tcp_server::make_state(
40		capy::execution_context& ctx, capy::any_executor const& ex)
41		{
42	169x	auto* i = new impl(ctx);
43		try
44		{
45	169x	return new state(i, ex);
46		}
47	✗	catch (...)
48		{
49	✗	delete i;
50	✗	throw;
51	✗	}
52		}
53
54		void
55	2658x	tcp_server::release(state* s) noexcept
56		{
57	5316x	if (s->refs.fetch_sub(1, std::memory_order_acq_rel) == 1)
58		{
59	169x	delete s->impl_;
60	169x	delete s;
61		}
62	2658x	}
63
64		void
65	175x	tcp_server::discard() noexcept
66		{
67	175x	if (!st_)
68	6x	return;
69	169x	auto& im = *st_->impl_;
70	169x	std::vector<std::size_t> idle;
71		{
72	169x	std::lock_guard lock(im.join_mutex);
73	169x	im.discarded = true;
74	336x	for (std::size_t i = 0; i < im.ports.size(); ++i)
75	167x	if (i >= im.loop_running.size() \|\| !im.loop_running[i])
76	113x	idle.push_back(i);
77	169x	}
78	282x	for (auto i : idle)
79	113x	im.ports[i].close();
80	169x	stop();
81	169x	release(std::exchange(st_, nullptr));
82	169x	}
83
84	173x	tcp_server::~tcp_server()
85		{
86	173x	discard();
87	173x	}
88
89	4x	tcp_server::tcp_server(tcp_server&& o) noexcept
90	4x	: st_(std::exchange(o.st_, nullptr))
91		{
92	4x	}
93
94		tcp_server&
95	2x	tcp_server::operator=(tcp_server&& o) noexcept
96		{
97	2x	if (this != &o)
98		{
99	2x	discard();
100	2x	st_ = std::exchange(o.st_, nullptr);
101		}
102	2x	return *this;
103		}
104
105		// Accept loop: wait for idle worker, accept connection, dispatch
106		capy::task<void>
107	159x	tcp_server::do_accept(state_ref st, tcp_acceptor& acc)
108		{
109		// Analyzer can't trace value through coroutine await_transform
110		// NOLINTNEXTLINE(clang-analyzer-core.uninitialized.UndefReturn)
111		auto env = co_await capy::this_coro::environment;
112		while (!env->stop_token.stop_requested())
113		{
114		// Wait for an idle worker before blocking on accept
115		auto* w = co_await pop_awaitable{*st};
116		if (!w)
117		break;
118		// Not dropped if this frame is destroyed while accepting: that
119		// happens only when the listener's context is torn down, which
120		// the server cannot outlive.
121		w->holds_.store(1, std::memory_order_relaxed);
122		auto [ec] = co_await acc.accept(w->socket());
123		if (ec)
124		{
125		co_await hop_awaitable{*st};
126		st->drop_hold(*w);
127		continue;
128		}
129		w->holds_.fetch_add(1, std::memory_order_relaxed);
130		w->run(launcher{st, w});
131		st->drop_hold(*w);
132		}
133	318x	}
134
135		std::error_code
136	169x	tcp_server::bind(endpoint ep)
137		{
138		try
139		{
140	169x	st_->impl_->ports.emplace_back(st_->impl_->ctx, ep);
141	167x	return {};
142		}
143	2x	catch (std::system_error const& e)
144		{
145	2x	return e.code();
146	2x	}
147		}
148
149		endpoint
150	159x	tcp_server::local_endpoint(std::size_t index) const noexcept
151		{
152	159x	if (!st_ \|\| index >= st_->impl_->ports.size())
153	4x	return endpoint{};
154	155x	return st_->impl_->ports[index].local_endpoint();
155		}
156
157		void
158	161x	tcp_server::start()
159		{
160		// Idempotent - only start if not already running
161	161x	if (st_->running)
162	2x	return;
163
164	159x	auto& im = *st_->impl_;
165		{
166		// Previous session must be fully stopped before restart
167	159x	std::lock_guard lock(im.join_mutex);
168	159x	if (st_->active_accepts != 0)
169	2x	detail::throw_logic_error(
170		"tcp_server::start: previous session not joined");
171	157x	st_->active_accepts = im.ports.size();
172	157x	im.loop_running.assign(im.ports.size(), true);
173	159x	}
174		{
175	157x	std::lock_guard<std::mutex> lock(st_->mutex);
176	157x	st_->stopping = false;
177	157x	}
178
179	157x	st_->running = true;
180	157x	im.stop = {}; // Fresh stop source
181	157x	auto token = im.stop.get_token();
182
183		// The completion runs once the loop's frame is done with its
184		// listener, so a close from here cannot race the loop.
185	316x	for (std::size_t i = 0; i < im.ports.size(); ++i)
186	318x	capy::run_async(st_->ex, token, [st = state_ref(st_), i]() {
187	153x	auto& im = *st->impl_;
188	153x	bool close = false;
189		{
190	153x	std::lock_guard lock(im.join_mutex);
191	153x	im.loop_running[i] = false;
192	153x	close = im.discarded;
193	153x	if (--st->active_accepts == 0)
194	151x	im.join_cv.notify_all();
195	153x	}
196	153x	if (close)
197	48x	im.ports[i].close();
198	312x	})(do_accept(state_ref(st_), im.ports[i]));
199	157x	}
200
201		void
202	280x	tcp_server::stop()
203		{
204		// Idempotent - only stop if running
205	280x	if (!st_ \|\| !st_->running)
206	123x	return;
207	157x	st_->running = false;
208
209		// Stop accept loops, including those waiting for a worker
210	157x	st_->impl_->stop.request_stop();
211	157x	st_->stop_waiters();
212
213		// Launch cancellation coroutine on server executor
214	157x	capy::run_async(st_->ex, std::stop_token{})(do_stop(state_ref(st_)));
215		}
216
217		void
218	89x	tcp_server::join()
219		{
220	89x	auto& im = *st_->impl_;
221	89x	std::unique_lock lock(im.join_mutex);
222	178x	im.join_cv.wait(lock, [this] { return st_->active_accepts == 0; });
223	89x	}
224
225		capy::task<>
226	157x	tcp_server::do_stop(state_ref st)
227		{
228		// Requested outside the lock: a stop callback may return its worker,
229		// which takes the lock.
230		std::vector<std::stop_source> active;
231		{
232		std::lock_guard<std::mutex> lock(st->mutex);
233		for (auto* w = st->active_head; w; w = w->next_)
234		active.push_back(w->stop_);
235		}
236		for (auto& s : active)
237		s.request_stop();
238		co_return;
239	314x	}
240
241		} // namespace boost::corosio
242