src/corosio/src/tcp_server.cpp

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