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

92.8% Lines (193/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	508x	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	508x	~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	4756x	Endpoint local_endpoint() const noexcept override
107		{
108	4756x	return local_endpoint_;
109		}
110
111		/// Return true if the acceptor has an open file descriptor.
112	7587x	bool is_open() const noexcept override
113		{
114	7587x	return fd_ >= 0;
115		}
116
117		/// Set a socket option.
118	400x	std::error_code set_option(
119		int level,
120		int optname,
121		void const* data,
122		std::size_t size) noexcept override
123		{
124	400x	if (::setsockopt(
125	400x	fd_, level, optname, data, static_cast<socklen_t>(size)) != 0)
126	✗	return make_err(errno);
127	400x	return {};
128		}
129
130		/// Get a socket option.
131		std::error_code
132	8x	get_option(int level, int optname, void* data, std::size_t* size)
133		const noexcept override
134		{
135	8x	socklen_t len = static_cast<socklen_t>(*size);
136	8x	if (::getsockopt(fd_, level, optname, data, &len) != 0)
137	✗	return make_err(errno);
138	8x	*size = static_cast<std::size_t>(len);
139	8x	return {};
140		}
141
142		/// Cache the local endpoint.
143	463x	void set_local_endpoint(Endpoint ep) noexcept
144		{
145	463x	local_endpoint_ = std::move(ep);
146	463x	}
147
148		/// Assign the fd and initialize descriptor state for the acceptor.
149	487x	void init_acceptor_fd(int fd) noexcept
150		{
151	487x	fd_ = fd;
152	487x	desc_state_.fd = fd;
153		{
154	487x	std::lock_guard lock(desc_state_.mutex);
155	487x	desc_state_.read_op = nullptr;
156	487x	desc_state_.wait_read_op = nullptr;
157	487x	desc_state_.wait_write_op = nullptr;
158	487x	desc_state_.wait_error_op = nullptr;
159	487x	}
160	487x	}
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	4387x	Service& service() noexcept
190		{
191	4387x	return svc_;
192		}
193
194	13x	void cancel() noexcept override { do_cancel(); }
195
196		/// Close the acceptor (non-virtual, called by the service).
197	1968x	void close_socket() noexcept { do_close_socket(); }
198
199	23x	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	23x	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	85x	reactor_acceptor<Derived, Service, Op, AcceptOp, WaitOp, DescState, ImplBase, Endpoint>::
283		cancel_single_op(Op& op) noexcept
284		{
285	85x	auto self = this->weak_from_this().lock();
286	85x	if (!self)
287	✗	return;
288
289	85x	op.request_cancel();
290
291	85x	reactor_op_base* claimed = nullptr;
292		{
293	85x	std::lock_guard lock(desc_state_.mutex);
294	85x	auto try_claim = [&](reactor_op_base*& slot) {
295	✗	if (!claimed && slot == &op)
296	✗	claimed = std::exchange(slot, nullptr);
297		};
298	85x	try_claim(desc_state_.read_op);
299	85x	try_claim(desc_state_.wait_read_op);
300	85x	try_claim(desc_state_.wait_write_op);
301	85x	try_claim(desc_state_.wait_error_op);
302	85x	}
303	85x	if (claimed)
304		{
305	36x	op.impl_ptr = self;
306	36x	svc_.post(&op);
307	36x	svc_.work_finished();
308		}
309	85x	}
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	13x	reactor_acceptor<Derived, Service, Op, AcceptOp, WaitOp, DescState, ImplBase, Endpoint>::
322		do_cancel() noexcept
323		{
324	13x	cancel_single_op(acc_);
325	13x	cancel_single_op(wait_rd_);
326	13x	cancel_single_op(wait_wr_);
327	13x	cancel_single_op(wait_er_);
328	13x	}
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	1968x	reactor_acceptor<Derived, Service, Op, AcceptOp, WaitOp, DescState, ImplBase, Endpoint>::
341		do_close_socket() noexcept
342		{
343	1968x	auto self = this->weak_from_this().lock();
344	1968x	if (self)
345		{
346	1968x	acc_.request_cancel();
347	1968x	wait_rd_.request_cancel();
348	1968x	wait_wr_.request_cancel();
349	1968x	wait_er_.request_cancel();
350
351	1968x	reactor_op_base* claimed_acc = nullptr;
352	1968x	reactor_op_base* claimed_wr = nullptr;
353	1968x	reactor_op_base* claimed_ww = nullptr;
354	1968x	reactor_op_base* claimed_we = nullptr;
355		{
356	1968x	std::lock_guard lock(desc_state_.mutex);
357	1968x	claimed_acc = std::exchange(desc_state_.read_op, nullptr);
358	1968x	claimed_wr = std::exchange(desc_state_.wait_read_op, nullptr);
359	1968x	claimed_ww = std::exchange(desc_state_.wait_write_op, nullptr);
360	1968x	claimed_we = std::exchange(desc_state_.wait_error_op, nullptr);
361	1968x	desc_state_.read_ready = false;
362	1968x	desc_state_.write_ready = false;
363
364	1968x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
365	✗	desc_state_.impl_ref_ = self;
366	1968x	}
367
368	1968x	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	1968x	repost(claimed_acc, acc_);
377	1968x	repost(claimed_wr, wait_rd_);
378	1968x	repost(claimed_ww, wait_wr_);
379	1968x	repost(claimed_we, wait_er_);
380		}
381
382	1968x	if (fd_ >= 0)
383		{
384	469x	if (desc_state_.registered_events != 0)
385	412x	svc_.scheduler().deregister_descriptor(fd_);
386	469x	::close(fd_);
387	469x	fd_ = -1;
388		}
389
390	1968x	desc_state_.fd = -1;
391	1968x	desc_state_.registered_events = 0;
392
393	1968x	local_endpoint_ = Endpoint{};
394	1968x	}
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	457x	reactor_acceptor<Derived, Service, Op, AcceptOp, WaitOp, DescState, ImplBase, Endpoint>::
476		do_bind(Endpoint const& ep)
477		{
478	457x	sockaddr_storage storage{};
479	457x	socklen_t addrlen = to_sockaddr(ep, storage);
480	457x	if (::bind(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen) < 0)
481	10x	return make_err(errno);
482
483		// Cache local endpoint (resolves ephemeral port / path)
484	447x	sockaddr_storage local{};
485	447x	socklen_t local_len = sizeof(local);
486	447x	if (::getsockname(fd_, reinterpret_cast<sockaddr*>(&local), &local_len) ==
487		0)
488	447x	set_local_endpoint(from_sockaddr_as(local, local_len, Endpoint{}));
489
490	447x	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	416x	reactor_acceptor<Derived, Service, Op, AcceptOp, WaitOp, DescState, ImplBase, Endpoint>::
504		do_listen(int backlog)
505		{
506	416x	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	416x	if (desc_state_.registered_events != 0)
512	2x	return {};
513
514	414x	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	23x	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	23x	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	17x	if (w == wait_type::read)
559		{
560	15x	op_ptr = &wait_rd_;
561	15x	desc_slot_ptr = &desc_state_.wait_read_op;
562	15x	event = reactor_event_read;
563		}
564		else // wait_type::error
565		{
566	2x	op_ptr = &wait_er_;
567	2x	desc_slot_ptr = &desc_state_.wait_error_op;
568	2x	event = reactor_event_error;
569		}
570
571	17x	auto& op = *op_ptr;
572	17x	op.reset();
573	17x	op.wait_event = event;
574	17x	op.h = h;
575	17x	op.ex = ex;
576	17x	op.ec_out = ec;
577	17x	op.fd = this->fd_;
578	17x	op.start(token, static_cast<Derived*>(this));
579	17x	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	17x	int perr = 0;
586	17x	if (WaitOp::probe(this->fd_, event, perr))
587		{
588	4x	op.complete(perr, 0);
589	4x	svc_.post(&op);
590	4x	return std::noop_coroutine();
591		}
592
593	13x	svc_.work_started();
594
595	13x	std::lock_guard lock(desc_state_.mutex);
596	13x	if (op.cancelled.load(std::memory_order_acquire))
597		{
598	2x	svc_.post(&op);
599	2x	svc_.work_finished();
600		}
601	11x	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	11x	*desc_slot_ptr = &op;
613		}
614	13x	return std::noop_coroutine();
615	13x	}
616
617		} // namespace boost::corosio::detail
618
619		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
620