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 (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		#include "src/detail/open_listener.hpp"
18
19		namespace boost::corosio {
20
21	427x	tcp_server::worker_base::worker_base() = default;
22	427x	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	171x	explicit impl(capy::execution_context& c) noexcept : ctx(c) {}
38		};
39
40		tcp_server::state*
41	171x	tcp_server::make_state(
42		capy::execution_context& ctx, capy::any_executor const& ex)
43		{
44	171x	auto* i = new impl(ctx);
45		try
46		{
47	171x	return new state(i, ex);
48		}
49	✗	catch (...)
50		{
51	✗	delete i;
52	✗	throw;
53	✗	}
54		}
55
56		void
57	2694x	tcp_server::release(state* s) noexcept
58		{
59	5388x	if (s->refs.fetch_sub(1, std::memory_order_acq_rel) == 1)
60		{
61	171x	delete s->impl_;
62	171x	delete s;
63		}
64	2694x	}
65
66		void
67	177x	tcp_server::discard() noexcept
68		{
69	177x	if (!st_)
70	6x	return;
71	171x	auto& im = *st_->impl_;
72	171x	std::vector<std::size_t> idle;
73		{
74	171x	std::lock_guard lock(im.join_mutex);
75	171x	im.discarded = true;
76	338x	for (std::size_t i = 0; i < im.ports.size(); ++i)
77	167x	if (i >= im.loop_running.size() \|\| !im.loop_running[i])
78	113x	idle.push_back(i);
79	171x	}
80	284x	for (auto i : idle)
81	113x	im.ports[i].close();
82	171x	stop();
83	171x	release(std::exchange(st_, nullptr));
84	171x	}
85
86	175x	tcp_server::~tcp_server()
87		{
88	175x	discard();
89	175x	}
90
91	4x	tcp_server::tcp_server(tcp_server&& o) noexcept
92	4x	: st_(std::exchange(o.st_, nullptr))
93		{
94	4x	}
95
96		tcp_server&
97	2x	tcp_server::operator=(tcp_server&& o) noexcept
98		{
99	2x	if (this != &o)
100		{
101	2x	discard();
102	2x	st_ = std::exchange(o.st_, nullptr);
103		}
104	2x	return *this;
105		}
106
107		// Accept loop: wait for idle worker, accept connection, dispatch
108		capy::task<void>
109	159x	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	318x	}
136
137		std::error_code
138	171x	tcp_server::bind(endpoint ep)
139		{
140	171x	tcp_acceptor acc(st_->impl_->ctx);
141	171x	if (auto ec = detail::open_listener(acc, ep, 128))
142	4x	return ec;
143	167x	st_->impl_->ports.push_back(std::move(acc));
144	167x	return {};
145	171x	}
146
147		endpoint
148	159x	tcp_server::local_endpoint(std::size_t index) const noexcept
149		{
150	159x	if (!st_ \|\| index >= st_->impl_->ports.size())
151	4x	return endpoint{};
152	155x	return st_->impl_->ports[index].local_endpoint();
153		}
154
155		void
156	161x	tcp_server::start()
157		{
158		// Idempotent - only start if not already running
159	161x	if (st_->running)
160	2x	return;
161
162	159x	auto& im = *st_->impl_;
163		{
164		// Previous session must be fully stopped before restart
165	159x	std::lock_guard lock(im.join_mutex);
166	159x	if (st_->active_accepts != 0)
167	2x	detail::throw_logic_error(
168		"tcp_server::start: previous session not joined");
169	157x	st_->active_accepts = im.ports.size();
170	157x	im.loop_running.assign(im.ports.size(), true);
171	159x	}
172		{
173	157x	std::lock_guard<std::mutex> lock(st_->mutex);
174	157x	st_->stopping = false;
175	157x	}
176
177	157x	st_->running = true;
178	157x	im.stop = {}; // Fresh stop source
179	157x	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	316x	for (std::size_t i = 0; i < im.ports.size(); ++i)
184	318x	capy::run_async(st_->ex, token, [st = state_ref(st_), i]() {
185	153x	auto& im = *st->impl_;
186	153x	bool close = false;
187		{
188	153x	std::lock_guard lock(im.join_mutex);
189	153x	im.loop_running[i] = false;
190	153x	close = im.discarded;
191	153x	if (--st->active_accepts == 0)
192	151x	im.join_cv.notify_all();
193	153x	}
194	153x	if (close)
195	48x	im.ports[i].close();
196	312x	})(do_accept(state_ref(st_), im.ports[i]));
197	157x	}
198
199		void
200	282x	tcp_server::stop()
201		{
202		// Idempotent - only stop if running
203	282x	if (!st_ \|\| !st_->running)
204	125x	return;
205	157x	st_->running = false;
206
207		// Stop accept loops, including those waiting for a worker
208	157x	st_->impl_->stop.request_stop();
209	157x	st_->stop_waiters();
210
211		// Launch cancellation coroutine on server executor
212	157x	capy::run_async(st_->ex, std::stop_token{})(do_stop(state_ref(st_)));
213		}
214
215		void
216	89x	tcp_server::join()
217		{
218	89x	auto& im = *st_->impl_;
219	89x	std::unique_lock lock(im.join_mutex);
220	178x	im.join_cv.wait(lock, [this] { return st_->active_accepts == 0; });
221	89x	}
222
223		capy::task<>
224	157x	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	314x	}
238
239		} // namespace boost::corosio
240