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

96.3% Lines (206 / 214) 98.9% Functions (92 / 93)

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

reactor_acceptor.hpp

f(x) Functions (93)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Steve Gerbino
3		//
4		// Distributed under the Boost Software License, Version 1.0. (See accompanying
5		// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6		//
7		// Official repository: https://github.com/cppalliance/corosio
8		//
9
10		#ifndef BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
12
13		#include <boost/corosio/tcp_acceptor.hpp>
14		#include <boost/corosio/wait_type.hpp>
15		#include <boost/corosio/detail/intrusive.hpp>
16		#include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
17		#include <boost/corosio/native/detail/reactor/reactor_descriptor_state.hpp>
18		#include <boost/corosio/native/detail/make_err.hpp>
19		#include <boost/corosio/native/detail/endpoint_convert.hpp>
20
21		#include <memory>
22		#include <mutex>
23		#include <utility>
24
25		#include <errno.h>
26		#include <netinet/in.h>
27		#include <sys/socket.h>
28		#include <unistd.h>
29
30		namespace boost::corosio::detail {
31
32		/** CRTP base for reactor-backed acceptor implementations.
33
34		Provides shared data members, trivial virtual overrides, and
35		non-virtual helper methods for cancellation and close. Concrete
36		backends inherit and add `cancel()`, `close_socket()`, and
37		`accept()` overrides that delegate to the `do_*` helpers.
38
39		@tparam Derived The concrete acceptor type (CRTP).
40		@tparam Service The backend's acceptor service type.
41		@tparam Op The backend's base op type.
42		@tparam AcceptOp The backend's accept op type.
43		@tparam WaitOp The backend's wait op type.
44		@tparam DescState The backend's descriptor_state type.
45		@tparam ImplBase The public vtable base
46		(tcp_acceptor::implementation or
47		local_stream_acceptor::implementation).
48		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
49		*/
50		template<
51		class Derived,
52		class Service,
53		class Op,
54		class AcceptOp,
55		class WaitOp,
56		class DescState,
57		class ImplBase = tcp_acceptor::implementation,
58		class Endpoint = endpoint>
59		class reactor_acceptor
60		: public ImplBase
61		, public std::enable_shared_from_this<Derived>
62		, public intrusive_list<Derived>::node
63		{
64		friend Derived;
65
66		protected:
67		// NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
68	772x	explicit reactor_acceptor(Service& svc) noexcept : svc_(svc) {}
69
70		protected:
71		Service& svc_;
72		int fd_ = -1;
73		Endpoint local_endpoint_;
74
75		public:
76		/// Pending accept operation slot.
77		AcceptOp acc_;
78
79		/// Pending wait-for-read operation slot.
80		WaitOp wait_rd_;
81
82		/// Pending wait-for-write operation slot.
83		WaitOp wait_wr_;
84
85		/// Pending wait-for-error operation slot.
86		WaitOp wait_er_;
87
88		/// Per-descriptor state for persistent reactor registration.
89		DescState desc_state_;
90
91	772x	~reactor_acceptor() override = default;
92
93		/// Return the underlying file descriptor.
94	59x	native_handle_type native_handle() const noexcept override
95		{
96	59x	return fd_;
97		}
98
99	636x	corosio::family family() const noexcept override
100		{
101	636x	return to_family(socket_family(fd_));
102		}
103
104		/// Release and return the native handle without closing it.
105	22x	native_handle_type release_socket() noexcept override
106		{
107	22x	return do_release_socket();
108		}
109
110		/// Return the cached local endpoint.
111	5031x	Endpoint local_endpoint() const noexcept override
112		{
113	5031x	return local_endpoint_;
114		}
115
116		/// Return true if the acceptor has an open file descriptor.
117	9310x	bool is_open() const noexcept override
118		{
119	9310x	return fd_ >= 0;
120		}
121
122		/// Set a socket option.
123	611x	std::error_code set_option(
124		int level,
125		int optname,
126		void const* data,
127		std::size_t size) noexcept override
128		{
129	611x	if (::setsockopt(
130	611x	fd_, level, optname, data, static_cast<socklen_t>(size)) != 0)
131	10x	return make_err(errno);
132	601x	return {};
133		}
134
135		/// Get a socket option.
136		std::error_code
137	25x	get_option(int level, int optname, void* data, std::size_t* size)
138		const noexcept override
139		{
140	25x	socklen_t len = static_cast<socklen_t>(*size);
141	25x	if (::getsockopt(fd_, level, optname, data, &len) != 0)
142	10x	return make_err(errno);
143	15x	*size = static_cast<std::size_t>(len);
144	15x	return {};
145		}
146
147		/// Cache the local endpoint.
148	685x	void set_local_endpoint(Endpoint ep) noexcept
149		{
150	685x	local_endpoint_ = std::move(ep);
151	685x	}
152
153		/// Assign the fd and initialize descriptor state for the acceptor.
154	726x	void init_acceptor_fd(int fd) noexcept
155		{
156	726x	fd_ = fd;
157	726x	desc_state_.fd = fd;
158		{
159	726x	std::lock_guard lock(desc_state_.mutex);
160	726x	desc_state_.read_op = nullptr;
161	726x	desc_state_.wait_read_op = nullptr;
162	726x	desc_state_.wait_write_op = nullptr;
163	726x	desc_state_.wait_error_op = nullptr;
164	726x	}
165	726x	}
166
167		/** Assign the fd, initialize descriptor state, and register with
168		the reactor.
169
170		Adoption skips `do_listen`, so the registration it performs
171		has to happen here instead.
172
173		@param fd The already-listening descriptor to adopt.
174
175		@return The error if the reactor rejects the descriptor, in
176		which case the implementation is left closed and the caller
177		retains ownership of @a fd; otherwise a default constructed
178		error code.
179		*/
180	18x	std::error_code init_and_register(int fd) noexcept
181		{
182	18x	init_acceptor_fd(fd);
183	18x	if (auto ec = svc_.scheduler().register_descriptor(fd, &desc_state_))
184		{
185	1x	fd_ = -1;
186	1x	desc_state_.fd = -1;
187	1x	desc_state_.registered_events = 0;
188	1x	return ec;
189		}
190	17x	return {};
191		}
192
193		/// Return a reference to the owning service.
194	4473x	Service& service() noexcept
195		{
196	4473x	return svc_;
197		}
198
199	19x	void cancel() noexcept override
200		{
201	19x	do_cancel();
202	19x	}
203
204		/// Close the acceptor (non-virtual, called by the service).
205	2972x	void close_socket() noexcept
206		{
207	2972x	do_close_socket();
208	2972x	}
209
210	38x	std::coroutine_handle<> wait(
211		std::coroutine_handle<> h,
212		capy::executor_ref ex,
213		wait_type w,
214		std::stop_token token,
215		std::error_code* ec) override
216		{
217	38x	return do_wait(h, ex, w, token, ec);
218		}
219
220		/** Wait for readiness on the listen socket.
221
222		For `wait_type::read`, completion signals that an incoming
223		connection is pending and a subsequent accept will succeed
224		without blocking; a connection already queued when the wait
225		begins completes it immediately via an initiation probe.
226
227		`wait_type::write` fails with `operation_not_supported` on
228		every backend: writability carries no meaning for a
229		listening socket.
230		*/
231		std::coroutine_handle<> do_wait(
232		std::coroutine_handle<>,
233		capy::executor_ref,
234		wait_type,
235		std::stop_token const&,
236		std::error_code*);
237
238		/** Cancel a single pending operation.
239
240		Claims the operation from the read_op descriptor slot
241		under the mutex and posts it to the scheduler as cancelled.
242
243		@param op The operation to cancel.
244		*/
245		void cancel_single_op(Op& op) noexcept;
246
247		/** Cancel the pending accept operation. */
248		void do_cancel() noexcept;
249
250		/** Close the acceptor and cancel pending operations.
251
252		Invoked by the derived class's close_socket(). The
253		derived class may add backend-specific cleanup after
254		calling this method.
255		*/
256		void do_close_socket() noexcept;
257
258		/** Release the acceptor without closing the fd. */
259		native_handle_type do_release_socket() noexcept;
260
261		/** Bind the acceptor socket to an endpoint.
262
263		Caches the resolved local endpoint (including ephemeral
264		port) after a successful bind.
265
266		@param ep The endpoint to bind to.
267		@return The error code from bind(), or success.
268		*/
269		std::error_code do_bind(Endpoint const& ep);
270
271		/** Start listening on the acceptor socket.
272
273		Registers the file descriptor with the reactor after
274		a successful listen() call.
275
276		@param backlog The listen backlog.
277		@return The error code from listen() or from reactor
278		registration, or success.
279		*/
280		std::error_code do_listen(int backlog);
281		};
282
283		template<
284		class Derived,
285		class Service,
286		class Op,
287		class AcceptOp,
288		class WaitOp,
289		class DescState,
290		class ImplBase,
291		class Endpoint>
292		void
293	146x	reactor_acceptor<
294		Derived,
295		Service,
296		Op,
297		AcceptOp,
298		WaitOp,
299		DescState,
300		ImplBase,
301		Endpoint>::cancel_single_op(Op& op) noexcept
302		{
303	146x	auto self = this->weak_from_this().lock();
304	146x	if (!self)
305	✗	return;
306
307	146x	op.request_cancel();
308
309	146x	reactor_op_base* claimed = nullptr;
310		{
311	146x	std::lock_guard lock(desc_state_.mutex);
312	146x	auto try_claim = [&](reactor_op_base*& slot) {
313	✗	if (!claimed && slot == &op)
314	✗	claimed = std::exchange(slot, nullptr);
315		};
316	146x	try_claim(desc_state_.read_op);
317	146x	try_claim(desc_state_.wait_read_op);
318	146x	try_claim(desc_state_.wait_write_op);
319	146x	try_claim(desc_state_.wait_error_op);
320	146x	}
321	146x	if (claimed)
322		{
323	87x	op.impl_ptr = self;
324	87x	svc_.post(&op);
325	87x	svc_.work_finished();
326		}
327	146x	}
328
329		template<
330		class Derived,
331		class Service,
332		class Op,
333		class AcceptOp,
334		class WaitOp,
335		class DescState,
336		class ImplBase,
337		class Endpoint>
338		void
339	19x	reactor_acceptor<
340		Derived,
341		Service,
342		Op,
343		AcceptOp,
344		WaitOp,
345		DescState,
346		ImplBase,
347		Endpoint>::do_cancel() noexcept
348		{
349	19x	cancel_single_op(acc_);
350	19x	cancel_single_op(wait_rd_);
351	19x	cancel_single_op(wait_wr_);
352	19x	cancel_single_op(wait_er_);
353	19x	}
354
355		template<
356		class Derived,
357		class Service,
358		class Op,
359		class AcceptOp,
360		class WaitOp,
361		class DescState,
362		class ImplBase,
363		class Endpoint>
364		void
365	2972x	reactor_acceptor<
366		Derived,
367		Service,
368		Op,
369		AcceptOp,
370		WaitOp,
371		DescState,
372		ImplBase,
373		Endpoint>::do_close_socket() noexcept
374		{
375	2972x	auto self = this->weak_from_this().lock();
376	2972x	if (self)
377		{
378	2972x	acc_.request_cancel();
379	2972x	wait_rd_.request_cancel();
380	2972x	wait_wr_.request_cancel();
381	2972x	wait_er_.request_cancel();
382
383	2972x	reactor_op_base* claimed_acc = nullptr;
384	2972x	reactor_op_base* claimed_wr = nullptr;
385	2972x	reactor_op_base* claimed_ww = nullptr;
386	2972x	reactor_op_base* claimed_we = nullptr;
387		{
388	2972x	std::lock_guard lock(desc_state_.mutex);
389	2972x	claimed_acc = std::exchange(desc_state_.read_op, nullptr);
390	2972x	claimed_wr = std::exchange(desc_state_.wait_read_op, nullptr);
391	2972x	claimed_ww = std::exchange(desc_state_.wait_write_op, nullptr);
392	2972x	claimed_we = std::exchange(desc_state_.wait_error_op, nullptr);
393	2972x	desc_state_.read_ready = false;
394	2972x	desc_state_.write_ready = false;
395
396	2972x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
397	32x	desc_state_.impl_ref_ = self;
398	2972x	}
399
400	2972x	auto repost = [&](reactor_op_base* claimed, reactor_op_base& op) {
401		if (claimed)
402		{
403		op.impl_ptr = self;
404		svc_.post(&op);
405		svc_.work_finished();
406		}
407		};
408	2972x	repost(claimed_acc, acc_);
409	2972x	repost(claimed_wr, wait_rd_);
410	2972x	repost(claimed_ww, wait_wr_);
411	2972x	repost(claimed_we, wait_er_);
412		}
413
414	2972x	if (fd_ >= 0)
415		{
416	703x	if (desc_state_.registered_events != 0)
417	620x	svc_.scheduler().deregister_descriptor(fd_);
418	703x	::close(fd_);
419	703x	fd_ = -1;
420		}
421
422	2972x	desc_state_.fd = -1;
423	2972x	desc_state_.registered_events = 0;
424
425	2972x	local_endpoint_ = Endpoint{};
426	2972x	}
427
428		template<
429		class Derived,
430		class Service,
431		class Op,
432		class AcceptOp,
433		class WaitOp,
434		class DescState,
435		class ImplBase,
436		class Endpoint>
437		native_handle_type
438	22x	reactor_acceptor<
439		Derived,
440		Service,
441		Op,
442		AcceptOp,
443		WaitOp,
444		DescState,
445		ImplBase,
446		Endpoint>::do_release_socket() noexcept
447		{
448	22x	auto self = this->weak_from_this().lock();
449	22x	if (self)
450		{
451	22x	acc_.request_cancel();
452	22x	wait_rd_.request_cancel();
453	22x	wait_wr_.request_cancel();
454	22x	wait_er_.request_cancel();
455
456	22x	reactor_op_base* claimed_acc = nullptr;
457	22x	reactor_op_base* claimed_wr = nullptr;
458	22x	reactor_op_base* claimed_ww = nullptr;
459	22x	reactor_op_base* claimed_we = nullptr;
460		{
461	22x	std::lock_guard lock(desc_state_.mutex);
462	22x	claimed_acc = std::exchange(desc_state_.read_op, nullptr);
463	22x	claimed_wr = std::exchange(desc_state_.wait_read_op, nullptr);
464	22x	claimed_ww = std::exchange(desc_state_.wait_write_op, nullptr);
465	22x	claimed_we = std::exchange(desc_state_.wait_error_op, nullptr);
466	22x	desc_state_.read_ready = false;
467	22x	desc_state_.write_ready = false;
468
469	22x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
470	1x	desc_state_.impl_ref_ = self;
471	22x	}
472
473	22x	auto repost = [&](reactor_op_base* claimed, reactor_op_base& op) {
474		if (claimed)
475		{
476		op.impl_ptr = self;
477		svc_.post(&op);
478		svc_.work_finished();
479		}
480		};
481	22x	repost(claimed_acc, acc_);
482	22x	repost(claimed_wr, wait_rd_);
483	22x	repost(claimed_ww, wait_wr_);
484	22x	repost(claimed_we, wait_er_);
485		}
486
487	22x	native_handle_type released = fd_;
488
489	22x	if (fd_ >= 0)
490		{
491	22x	if (desc_state_.registered_events != 0)
492	22x	svc_.scheduler().deregister_descriptor(fd_);
493	22x	fd_ = -1;
494		}
495
496	22x	desc_state_.fd = -1;
497	22x	desc_state_.registered_events = 0;
498
499	22x	local_endpoint_ = Endpoint{};
500
501	44x	return released;
502	22x	}
503
504		template<
505		class Derived,
506		class Service,
507		class Op,
508		class AcceptOp,
509		class WaitOp,
510		class DescState,
511		class ImplBase,
512		class Endpoint>
513		std::error_code
514	684x	reactor_acceptor<
515		Derived,
516		Service,
517		Op,
518		AcceptOp,
519		WaitOp,
520		DescState,
521		ImplBase,
522		Endpoint>::do_bind(Endpoint const& ep)
523		{
524	684x	sockaddr_storage storage{};
525	684x	socklen_t addrlen = to_sockaddr(ep, storage);
526	684x	if (::bind(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen) < 0)
527	16x	return make_err(errno);
528
529		// Cache local endpoint (resolves ephemeral port / path)
530	668x	sockaddr_storage local{};
531	668x	socklen_t local_len = sizeof(local);
532	668x	if (::getsockname(fd_, reinterpret_cast<sockaddr*>(&local), &local_len) ==
533		0)
534	668x	set_local_endpoint(from_sockaddr_as(local, local_len, Endpoint{}));
535
536	668x	return {};
537		}
538
539		template<
540		class Derived,
541		class Service,
542		class Op,
543		class AcceptOp,
544		class WaitOp,
545		class DescState,
546		class ImplBase,
547		class Endpoint>
548		std::error_code
549	640x	reactor_acceptor<
550		Derived,
551		Service,
552		Op,
553		AcceptOp,
554		WaitOp,
555		DescState,
556		ImplBase,
557		Endpoint>::do_listen(int backlog)
558		{
559	640x	if (::listen(fd_, backlog) < 0)
560	12x	return make_err(errno);
561
562		// A re-listen only changes the backlog; the descriptor is already
563		// registered and re-adding it would fail on epoll.
564	628x	if (desc_state_.registered_events != 0)
565	2x	return {};
566
567	626x	return svc_.scheduler().register_descriptor(fd_, &desc_state_);
568		}
569
570		template<
571		class Derived,
572		class Service,
573		class Op,
574		class AcceptOp,
575		class WaitOp,
576		class DescState,
577		class ImplBase,
578		class Endpoint>
579		std::coroutine_handle<>
580	38x	reactor_acceptor<
581		Derived,
582		Service,
583		Op,
584		AcceptOp,
585		WaitOp,
586		DescState,
587		ImplBase,
588		Endpoint>::
589		do_wait(
590		std::coroutine_handle<> h,
591		capy::executor_ref ex,
592		wait_type w,
593		std::stop_token const& token,
594		std::error_code* ec)
595		{
596		// Writability carries no meaning for a listening socket; some
597		// backends could only lie about it and others could never report
598		// it, so the wait fails the same way everywhere instead.
599	38x	if (w == wait_type::write)
600		{
601	6x	auto& op = wait_wr_;
602	6x	op.reset();
603	6x	op.wait_event = reactor_event_write;
604	6x	op.h = h;
605	6x	op.ex = ex;
606	6x	op.ec_out = ec;
607	6x	op.fd = this->fd_;
608	6x	op.start(token, static_cast<Derived*>(this));
609	6x	op.impl_ptr = this->shared_from_this();
610	6x	op.complete(ENOTSUP, 0);
611	6x	svc_.post(&op);
612	6x	return std::noop_coroutine();
613		}
614
615		WaitOp* op_ptr;
616		reactor_op_base** desc_slot_ptr;
617		std::uint32_t event;
618
619	32x	if (w == wait_type::read)
620		{
621	26x	op_ptr = &wait_rd_;
622	26x	desc_slot_ptr = &desc_state_.wait_read_op;
623	26x	event = reactor_event_read;
624		}
625		else // wait_type::error
626		{
627	6x	op_ptr = &wait_er_;
628	6x	desc_slot_ptr = &desc_state_.wait_error_op;
629	6x	event = reactor_event_error;
630		}
631
632	32x	auto& op = *op_ptr;
633	32x	op.reset();
634	32x	op.wait_event = event;
635	32x	op.h = h;
636	32x	op.ex = ex;
637	32x	op.ec_out = ec;
638	32x	op.fd = this->fd_;
639	32x	op.start(token, static_cast<Derived*>(this));
640	32x	op.impl_ptr = this->shared_from_this();
641
642		// A listener's readiness can predate the wait: an adopted or
643		// shared descriptor has history the reactor never saw, and an
644		// edge already dispatched will not be re-announced. Probe before
645		// parking.
646	32x	int perr = 0;
647	32x	if (WaitOp::probe(this->fd_, event, perr))
648		{
649	11x	op.complete(perr, 0);
650	11x	svc_.post(&op);
651	11x	return std::noop_coroutine();
652		}
653
654	21x	svc_.work_started();
655
656	21x	std::lock_guard lock(desc_state_.mutex);
657	21x	if (op.cancelled.load(std::memory_order_acquire))
658		{
659	✗	svc_.post(&op);
660	✗	svc_.work_finished();
661		}
662	21x	else if (WaitOp::probe(this->fd_, event, perr))
663		{
664		// Close the probe-to-park window: an edge that landed after
665		// the first probe was consumed, so re-check under the mutex
666		// the dispatch path holds.
667	✗	op.complete(perr, 0);
668	✗	svc_.post(&op);
669	✗	svc_.work_finished();
670		}
671		else
672		{
673	21x	*desc_slot_ptr = &op;
674		}
675	21x	return std::noop_coroutine();
676	21x	}
677
678		} // namespace boost::corosio::detail
679
680		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
681