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

97.2% Lines (206/0/212) 98.9% List of functions (88/0/89)

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

reactor_acceptor.hpp

f(x) Functions (89)

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