include/boost/corosio/native/detail/reactor/reactor_acceptor.hpp

92.3% Lines (144 / 156) 97.3% Functions (109 / 112, 1 excl)

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

reactor_acceptor.hpp

f(x) Functions (113)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Steve Gerbino
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_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
12		#define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
13
14		#include <boost/corosio/tcp_acceptor.hpp>
15		#include <boost/corosio/wait_type.hpp>
16		#include <boost/corosio/detail/intrusive.hpp>
17		#include <boost/corosio/native/detail/reactor/reactor_io_core.hpp>
18		#include <boost/corosio/native/detail/reactor/reactor_descriptor_state.hpp>
19		#include <boost/corosio/native/detail/make_err.hpp>
20		#include <boost/corosio/native/detail/endpoint_convert.hpp>
21
22		#include <memory>
23		#include <mutex>
24		#include <utility>
25
26		#include <errno.h>
27		#include <netinet/in.h>
28		#include <sys/socket.h>
29		#include <unistd.h>
30
31		namespace boost::corosio::detail {
32
33		/** CRTP base for reactor-backed acceptor implementations.
34
35		Provides shared data members, trivial virtual overrides, and
36		non-virtual helper methods for cancellation and close. Concrete
37		backends inherit and add `cancel()`, `close_socket()`, and
38		`accept()` overrides that delegate to the `do_*` helpers.
39
40		@tparam Derived The concrete acceptor type (CRTP).
41		@tparam Service The backend's acceptor service type.
42		@tparam Op The backend's base op type.
43		@tparam AcceptOp The backend's accept op type.
44		@tparam WaitOp The backend's wait op type.
45		@tparam DescState The backend's descriptor_state type.
46		@tparam ImplBase The public vtable base
47		(tcp_acceptor::implementation or
48		local_stream_acceptor::implementation).
49		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
50		*/
51		template<
52		class Derived,
53		class Service,
54		class Op,
55		class AcceptOp,
56		class WaitOp,
57		class DescState,
58		class ImplBase = tcp_acceptor::implementation,
59		class Endpoint = endpoint>
60		class reactor_acceptor
61		: public ImplBase
62		, public std::enable_shared_from_this<Derived>
63		, public reactor_io_core<Derived, Service, DescState>
64		, public intrusive_list<Derived>::node
65		{
66		friend Derived;
67
68		using core_type = reactor_io_core<Derived, Service, DescState>;
69		friend core_type;
70
71		protected:
72		// NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
73	2068x	explicit reactor_acceptor(Service& svc) noexcept : core_type(svc) {}
74
75		protected:
76		using core_type::svc_;
77		int fd_ = -1;
78		Endpoint local_endpoint_;
79
80		public:
81		/// Pending accept operation slot.
82		AcceptOp acc_;
83
84		/// Pending wait-for-read operation slot.
85		WaitOp wait_rd_;
86
87		/// Pending wait-for-write operation slot.
88		WaitOp wait_wr_;
89
90		/// Pending wait-for-error operation slot.
91		WaitOp wait_er_;
92
93		using core_type::desc_state_;
94
95	2068x	~reactor_acceptor() override = default;
96
97		/// Return the underlying file descriptor.
98	44x	native_handle_type native_handle() const noexcept override
99		{
100	44x	return fd_;
101		}
102
103	1920x	corosio::family family() const noexcept override
104		{
105	1920x	return to_family(socket_family(fd_));
106		}
107
108		/// Release and return the native handle without closing it.
109	20x	native_handle_type release_socket() noexcept override
110		{
111	20x	return do_release_socket();
112		}
113
114		/// Return the cached local endpoint.
115	8269x	Endpoint local_endpoint() const noexcept override
116		{
117	8269x	return local_endpoint_;
118		}
119
120		/// Return true if the acceptor has an open file descriptor.
121	20324x	bool is_open() const noexcept override
122		{
123	20324x	return fd_ >= 0;
124		}
125
126		/// Set a socket option.
127	1893x	std::error_code set_option(
128		int level,
129		int optname,
130		void const* data,
131		std::size_t size) noexcept override
132		{
133	1893x	if (::setsockopt(
134	1893x	fd_, level, optname, data, static_cast<socklen_t>(size)) != 0)
135	12x	return make_err(errno);
136	1881x	return {};
137		}
138
139		/// Get a socket option.
140		std::error_code
141	27x	get_option(int level, int optname, void* data, std::size_t* size)
142		const noexcept override
143		{
144	27x	socklen_t len = static_cast<socklen_t>(*size);
145	27x	if (::getsockopt(fd_, level, optname, data, &len) != 0)
146	12x	return make_err(errno);
147	15x	*size = static_cast<std::size_t>(len);
148	15x	return {};
149		}
150
151		/// Cache the local endpoint.
152	1965x	void set_local_endpoint(Endpoint ep) noexcept
153		{
154	1965x	local_endpoint_ = std::move(ep);
155	1965x	}
156
157		/// Assign the fd and initialize descriptor state for the acceptor.
158	1998x	void init_acceptor_fd(int fd) noexcept
159		{
160	1998x	fd_ = fd;
161	1998x	desc_state_.fd = fd;
162		{
163	1998x	std::lock_guard lock(desc_state_.mutex);
164	1998x	desc_state_.read_op = nullptr;
165	1998x	desc_state_.wait_read_op = nullptr;
166	1998x	desc_state_.wait_write_op = nullptr;
167	1998x	desc_state_.wait_error_op = nullptr;
168	1998x	}
169	1998x	}
170
171		/** Assign the fd, initialize descriptor state, and register with
172		the reactor.
173
174		Adoption skips `do_listen`, so the registration it performs
175		has to happen here instead.
176
177		@param fd The already-listening descriptor to adopt.
178
179		@return The error if the reactor rejects the descriptor, in
180		which case the implementation is left closed and the caller
181		retains ownership of @a fd; otherwise a default constructed
182		error code.
183		*/
184	14x	std::error_code init_and_register(int fd) noexcept
185		{
186	14x	fd_ = fd;
187	14x	if (auto ec = this->register_fd(fd))
188		{
189	1x	fd_ = -1;
190	1x	return ec;
191		}
192	13x	return {};
193		}
194
195		/// Return a reference to the owning service.
196	6431x	Service& service() noexcept
197		{
198	6431x	return svc_;
199		}
200
201	21x	void cancel() noexcept override
202		{
203	21x	do_cancel();
204	21x	}
205
206		/// Close the acceptor (non-virtual, called by the service).
207	8130x	void close_socket() noexcept
208		{
209	8130x	do_close_socket();
210	8130x	}
211
212	41x	std::coroutine_handle<> wait(
213		std::coroutine_handle<> h,
214		capy::executor_ref ex,
215		wait_type w,
216		std::stop_token token,
217		std::error_code* ec) override
218		{
219	41x	return do_wait(h, ex, w, token, ec);
220		}
221
222		/** Wait for readiness on the listen socket.
223
224		For `wait_type::read`, completion signals that an incoming
225		connection is pending and a subsequent accept succeeds
226		without blocking; a connection already queued when the wait
227		begins completes it immediately via an initiation probe.
228
229		`wait_type::write` fails with `operation_not_supported` on
230		every backend: writability carries no meaning for a
231		listening socket.
232		*/
233		std::coroutine_handle<> do_wait(
234		std::coroutine_handle<>,
235		capy::executor_ref,
236		wait_type,
237		std::stop_token const&,
238		std::error_code*);
239
240		/** Cancel the pending accept operation. */
241	21x	void do_cancel() noexcept
242		{
243	21x	this->cancel_all();
244	21x	}
245
246		/** Close the acceptor and cancel pending operations.
247
248		Invoked by the derived class's close_socket(). The
249		derived class may add backend-specific cleanup after
250		calling this method.
251		*/
252	8130x	void do_close_socket() noexcept
253		{
254	8130x	this->abandon_all();
255	8130x	this->unregister_fd(fd_);
256	8130x	if (fd_ >= 0)
257		{
258	1991x	::close(fd_);
259	1991x	fd_ = -1;
260		}
261	8130x	local_endpoint_ = Endpoint{};
262	8130x	}
263
264		/** Release the acceptor without closing the fd. */
265	20x	native_handle_type do_release_socket() noexcept
266		{
267	20x	this->abandon_all();
268	20x	native_handle_type released = fd_;
269	20x	this->unregister_fd(fd_);
270	20x	fd_ = -1;
271	20x	local_endpoint_ = Endpoint{};
272	20x	return released;
273		}
274
275		/** Bind the acceptor socket to an endpoint.
276
277		Caches the resolved local endpoint (including ephemeral
278		port) after a successful bind.
279
280		@param ep The endpoint to bind to.
281		@return The error code from bind(), or success.
282		*/
283		std::error_code do_bind(Endpoint const& ep);
284
285		/** Start listening on the acceptor socket.
286
287		Registers the file descriptor with the reactor after
288		a successful listen() call.
289
290		@param backlog The listen backlog.
291		@return The error code from listen() or from reactor
292		registration, or success.
293		*/
294		std::error_code do_listen(int backlog);
295
296		private:
297		// CRTP callbacks for reactor_io_core cancel/close
298
299		template<class AnyOp>
300	70x	reactor_op_base** op_to_desc_slot(AnyOp& op) noexcept
301		{
302	70x	if (&op == static_cast<void*>(&acc_))
303	70x	return &desc_state_.read_op;
304	✗	if (&op == static_cast<void*>(&wait_rd_))
305	✗	return &desc_state_.wait_read_op;
306	✗	if (&op == static_cast<void*>(&wait_wr_))
307	✗	return &desc_state_.wait_write_op;
308	✗	if (&op == static_cast<void*>(&wait_er_))
309	✗	return &desc_state_.wait_error_op;
310	✗	return nullptr;
311		}
312
313		template<class Fn>
314	8171x	void for_each_op(Fn fn) noexcept
315		{
316	8171x	fn(acc_);
317	8171x	fn(wait_rd_);
318	8171x	fn(wait_wr_);
319	8171x	fn(wait_er_);
320	8171x	}
321
322		template<class Fn>
323	8185x	void for_each_desc_entry(Fn fn) noexcept
324		{
325	8185x	fn(acc_, desc_state_.read_op);
326	8185x	fn(wait_rd_, desc_state_.wait_read_op);
327	8185x	fn(wait_wr_, desc_state_.wait_write_op);
328	8185x	fn(wait_er_, desc_state_.wait_error_op);
329	8185x	}
330		};
331
332		template<
333		class Derived,
334		class Service,
335		class Op,
336		class AcceptOp,
337		class WaitOp,
338		class DescState,
339		class ImplBase,
340		class Endpoint>
341		std::error_code
342	1968x	reactor_acceptor<
343		Derived,
344		Service,
345		Op,
346		AcceptOp,
347		WaitOp,
348		DescState,
349		ImplBase,
350		Endpoint>::do_bind(Endpoint const& ep)
351		{
352	1968x	sockaddr_storage storage{};
353	1968x	socklen_t addrlen = to_sockaddr(ep, storage);
354	1968x	if (::bind(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen) < 0)
355	16x	return make_err(errno);
356
357		// Cache local endpoint (resolves ephemeral port / path)
358	1952x	sockaddr_storage local{};
359	1952x	socklen_t local_len = sizeof(local);
360	1952x	if (::getsockname(fd_, reinterpret_cast<sockaddr*>(&local), &local_len) ==
361		0)
362	1952x	set_local_endpoint(from_sockaddr_as(local, local_len, Endpoint{}));
363
364	1952x	return {};
365		}
366
367		template<
368		class Derived,
369		class Service,
370		class Op,
371		class AcceptOp,
372		class WaitOp,
373		class DescState,
374		class ImplBase,
375		class Endpoint>
376		std::error_code
377	1924x	reactor_acceptor<
378		Derived,
379		Service,
380		Op,
381		AcceptOp,
382		WaitOp,
383		DescState,
384		ImplBase,
385		Endpoint>::do_listen(int backlog)
386		{
387	1924x	if (::listen(fd_, backlog) < 0)
388	16x	return make_err(errno);
389
390		// A re-listen only changes the backlog; the descriptor is already
391		// registered and re-adding it would fail on epoll.
392	1908x	if (desc_state_.registered_events != 0)
393	2x	return {};
394
395	1906x	return svc_.scheduler().register_descriptor(fd_, &desc_state_);
396		}
397
398		template<
399		class Derived,
400		class Service,
401		class Op,
402		class AcceptOp,
403		class WaitOp,
404		class DescState,
405		class ImplBase,
406		class Endpoint>
407		std::coroutine_handle<>
408	41x	reactor_acceptor<
409		Derived,
410		Service,
411		Op,
412		AcceptOp,
413		WaitOp,
414		DescState,
415		ImplBase,
416		Endpoint>::
417		do_wait(
418		std::coroutine_handle<> h,
419		capy::executor_ref ex,
420		wait_type w,
421		std::stop_token const& token,
422		std::error_code* ec)
423		{
424		// Writability carries no meaning for a listening socket; some
425		// backends could only lie about it and others could never report
426		// it, so the wait fails the same way everywhere instead.
427	41x	if (w == wait_type::write)
428		{
429	6x	auto& op = wait_wr_;
430	6x	op.reset();
431	6x	op.wait_event = reactor_event_write;
432	6x	op.h = h;
433	6x	op.ex = ex;
434	6x	op.ec_out = ec;
435	6x	op.fd = this->fd_;
436	6x	op.start(token, static_cast<Derived*>(this));
437	6x	op.impl_ptr = this->shared_from_this();
438	6x	op.complete(ENOTSUP, 0);
439	6x	svc_.post(&op);
440	6x	return std::noop_coroutine();
441		}
442
443		WaitOp* op_ptr;
444		reactor_op_base** desc_slot_ptr;
445		std::uint32_t event;
446
447	35x	if (w == wait_type::read)
448		{
449	29x	op_ptr = &wait_rd_;
450	29x	desc_slot_ptr = &desc_state_.wait_read_op;
451	29x	event = reactor_event_read;
452		}
453		else // wait_type::error
454		{
455	6x	op_ptr = &wait_er_;
456	6x	desc_slot_ptr = &desc_state_.wait_error_op;
457	6x	event = reactor_event_error;
458		}
459
460	35x	auto& op = *op_ptr;
461	35x	op.reset();
462	35x	op.wait_event = event;
463	35x	op.h = h;
464	35x	op.ex = ex;
465	35x	op.ec_out = ec;
466	35x	op.fd = this->fd_;
467	35x	op.start(token, static_cast<Derived*>(this));
468	35x	op.impl_ptr = this->shared_from_this();
469
470		// A listener's readiness can predate the wait: an adopted or
471		// shared descriptor has history the reactor never saw, and an
472		// edge already dispatched will not be re-announced. Probe before
473		// parking.
474	35x	int perr = 0;
475	35x	if (WaitOp::probe(this->fd_, event, perr))
476		{
477	12x	op.complete(perr, 0);
478	12x	svc_.post(&op);
479	12x	return std::noop_coroutine();
480		}
481
482	23x	svc_.work_started();
483
484	23x	std::lock_guard lock(desc_state_.mutex);
485	23x	if (op.cancelled.load(std::memory_order_acquire))
486		{
487	✗	svc_.post(&op);
488	✗	svc_.work_finished();
489		}
490	23x	else if (WaitOp::probe(this->fd_, event, perr))
491		{
492		// Close the probe-to-park window: an edge that landed after
493		// the first probe was consumed, so re-check under the mutex
494		// the dispatch path holds.
495	✗	op.complete(perr, 0);
496	✗	svc_.post(&op);
497	✗	svc_.work_finished();
498		}
499		else
500		{
501	23x	*desc_slot_ptr = &op;
502
503		// Select watches an fd only while an op is parked; see
504		// register_op.
505		if constexpr (Service::needs_park_notification)
506	11x	svc_.scheduler().notify_reactor();
507		}
508	23x	return std::noop_coroutine();
509	23x	}
510
511		} // namespace boost::corosio::detail
512
513		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
514