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

93.6% Lines (160 / 171) 97.5% Functions (116 / 119, 1 excl)

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

reactor_acceptor.hpp

f(x) Functions (120)

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