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

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