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

65.8% Lines (121/184) 75.7% List of functions (56/74)

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

reactor_acceptor.hpp

f(x) Functions (74)

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	195x	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	195x	~reactor_acceptor() override = default;
92
93		/// Return the underlying file descriptor.
94		int native_handle() const noexcept
95		{
96		return fd_;
97		}
98
99		/// Return the cached local endpoint.
100	8582x	Endpoint local_endpoint() const noexcept override
101		{
102	8582x	return local_endpoint_;
103		}
104
105		/// Return true if the acceptor has an open file descriptor.
106	9696x	bool is_open() const noexcept override
107		{
108	9696x	return fd_ >= 0;
109		}
110
111		/// Set a socket option.
112	171x	std::error_code set_option(
113		int level,
114		int optname,
115		void const* data,
116		std::size_t size) noexcept override
117		{
118	171x	if (::setsockopt(
119	171x	fd_, level, optname, data, static_cast<socklen_t>(size)) != 0)
120	✗	return make_err(errno);
121	171x	return {};
122		}
123
124		/// Get a socket option.
125		std::error_code
126	✗	get_option(int level, int optname, void* data, std::size_t* size)
127		const noexcept override
128		{
129	✗	socklen_t len = static_cast<socklen_t>(*size);
130	✗	if (::getsockopt(fd_, level, optname, data, &len) != 0)
131	✗	return make_err(errno);
132	✗	*size = static_cast<std::size_t>(len);
133	✗	return {};
134		}
135
136		/// Cache the local endpoint.
137	180x	void set_local_endpoint(Endpoint ep) noexcept
138		{
139	180x	local_endpoint_ = std::move(ep);
140	180x	}
141
142		/// Assign the fd and initialize descriptor state for the acceptor.
143	190x	void init_acceptor_fd(int fd) noexcept
144		{
145	190x	fd_ = fd;
146	190x	desc_state_.fd = fd;
147		{
148	190x	std::lock_guard lock(desc_state_.mutex);
149	190x	desc_state_.read_op = nullptr;
150	190x	desc_state_.wait_read_op = nullptr;
151	190x	desc_state_.wait_write_op = nullptr;
152	190x	desc_state_.wait_error_op = nullptr;
153	190x	}
154	190x	}
155
156		/// Return a reference to the owning service.
157	8420x	Service& service() noexcept
158		{
159	8420x	return svc_;
160		}
161
162	6x	void cancel() noexcept override { do_cancel(); }
163
164		/// Close the acceptor (non-virtual, called by the service).
165	770x	void close_socket() noexcept { do_close_socket(); }
166
167	2x	std::coroutine_handle<> wait(
168		std::coroutine_handle<> h,
169		capy::executor_ref ex,
170		wait_type w,
171		std::stop_token token,
172		std::error_code* ec) override
173		{
174	2x	return do_wait(h, ex, w, token, ec);
175		}
176
177		/** Wait for readiness on the listen socket.
178
179		Registers a wait op on the matching event slot. For
180		`wait_type::read`, completion signals that an incoming
181		connection is pending and a subsequent accept will
182		succeed without blocking.
183		*/
184		std::coroutine_handle<> do_wait(
185		std::coroutine_handle<>,
186		capy::executor_ref,
187		wait_type,
188		std::stop_token const&,
189		std::error_code*);
190
191		/** Cancel a single pending operation.
192
193		Claims the operation from the read_op descriptor slot
194		under the mutex and posts it to the scheduler as cancelled.
195
196		@param op The operation to cancel.
197		*/
198		void cancel_single_op(Op& op) noexcept;
199
200		/** Cancel the pending accept operation. */
201		void do_cancel() noexcept;
202
203		/** Close the acceptor and cancel pending operations.
204
205		Invoked by the derived class's close_socket(). The
206		derived class may add backend-specific cleanup after
207		calling this method.
208		*/
209		void do_close_socket() noexcept;
210
211		/** Release the acceptor without closing the fd. */
212		native_handle_type do_release_socket() noexcept;
213
214		/** Bind the acceptor socket to an endpoint.
215
216		Caches the resolved local endpoint (including ephemeral
217		port) after a successful bind.
218
219		@param ep The endpoint to bind to.
220		@return The error code from bind(), or success.
221		*/
222		std::error_code do_bind(Endpoint const& ep);
223
224		/** Start listening on the acceptor socket.
225
226		Registers the file descriptor with the reactor after
227		a successful listen() call.
228
229		@param backlog The listen backlog.
230		@return The error code from listen(), or success.
231		*/
232		std::error_code do_listen(int backlog);
233		};
234
235		template<
236		class Derived,
237		class Service,
238		class Op,
239		class AcceptOp,
240		class WaitOp,
241		class DescState,
242		class ImplBase,
243		class Endpoint>
244		void
245	30x	reactor_acceptor<Derived, Service, Op, AcceptOp, WaitOp, DescState, ImplBase, Endpoint>::
246		cancel_single_op(Op& op) noexcept
247		{
248	30x	auto self = this->weak_from_this().lock();
249	30x	if (!self)
250	✗	return;
251
252	30x	op.request_cancel();
253
254	30x	reactor_op_base* claimed = nullptr;
255		{
256	30x	std::lock_guard lock(desc_state_.mutex);
257	30x	auto try_claim = [&](reactor_op_base*& slot) {
258		if (!claimed && slot == &op)
259		claimed = std::exchange(slot, nullptr);
260		};
261	30x	try_claim(desc_state_.read_op);
262	30x	try_claim(desc_state_.wait_read_op);
263	30x	try_claim(desc_state_.wait_write_op);
264	30x	try_claim(desc_state_.wait_error_op);
265	30x	}
266	30x	if (claimed)
267		{
268	8x	op.impl_ptr = self;
269	8x	svc_.post(&op);
270	8x	svc_.work_finished();
271		}
272	30x	}
273
274		template<
275		class Derived,
276		class Service,
277		class Op,
278		class AcceptOp,
279		class WaitOp,
280		class DescState,
281		class ImplBase,
282		class Endpoint>
283		void
284	6x	reactor_acceptor<Derived, Service, Op, AcceptOp, WaitOp, DescState, ImplBase, Endpoint>::
285		do_cancel() noexcept
286		{
287	6x	cancel_single_op(acc_);
288	6x	cancel_single_op(wait_rd_);
289	6x	cancel_single_op(wait_wr_);
290	6x	cancel_single_op(wait_er_);
291	6x	}
292
293		template<
294		class Derived,
295		class Service,
296		class Op,
297		class AcceptOp,
298		class WaitOp,
299		class DescState,
300		class ImplBase,
301		class Endpoint>
302		void
303	770x	reactor_acceptor<Derived, Service, Op, AcceptOp, WaitOp, DescState, ImplBase, Endpoint>::
304		do_close_socket() noexcept
305		{
306	770x	auto self = this->weak_from_this().lock();
307	770x	if (self)
308		{
309	770x	acc_.request_cancel();
310	770x	wait_rd_.request_cancel();
311	770x	wait_wr_.request_cancel();
312	770x	wait_er_.request_cancel();
313
314	770x	reactor_op_base* claimed_acc = nullptr;
315	770x	reactor_op_base* claimed_wr = nullptr;
316	770x	reactor_op_base* claimed_ww = nullptr;
317	770x	reactor_op_base* claimed_we = nullptr;
318		{
319	770x	std::lock_guard lock(desc_state_.mutex);
320	770x	claimed_acc = std::exchange(desc_state_.read_op, nullptr);
321	770x	claimed_wr = std::exchange(desc_state_.wait_read_op, nullptr);
322	770x	claimed_ww = std::exchange(desc_state_.wait_write_op, nullptr);
323	770x	claimed_we = std::exchange(desc_state_.wait_error_op, nullptr);
324	770x	desc_state_.read_ready = false;
325	770x	desc_state_.write_ready = false;
326
327	770x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
328	✗	desc_state_.impl_ref_ = self;
329	770x	}
330
331	770x	auto repost = [&](reactor_op_base* claimed, reactor_op_base& op) {
332		if (claimed)
333		{
334		op.impl_ptr = self;
335		svc_.post(&op);
336		svc_.work_finished();
337		}
338		};
339	770x	repost(claimed_acc, acc_);
340	770x	repost(claimed_wr, wait_rd_);
341	770x	repost(claimed_ww, wait_wr_);
342	770x	repost(claimed_we, wait_er_);
343		}
344
345	770x	if (fd_ >= 0)
346		{
347	190x	if (desc_state_.registered_events != 0)
348	160x	svc_.scheduler().deregister_descriptor(fd_);
349	190x	::close(fd_);
350	190x	fd_ = -1;
351		}
352
353	770x	desc_state_.fd = -1;
354	770x	desc_state_.registered_events = 0;
355
356	770x	local_endpoint_ = Endpoint{};
357	770x	}
358
359		template<
360		class Derived,
361		class Service,
362		class Op,
363		class AcceptOp,
364		class WaitOp,
365		class DescState,
366		class ImplBase,
367		class Endpoint>
368		native_handle_type
369	✗	reactor_acceptor<Derived, Service, Op, AcceptOp, WaitOp, DescState, ImplBase, Endpoint>::
370		do_release_socket() noexcept
371		{
372	✗	auto self = this->weak_from_this().lock();
373	✗	if (self)
374		{
375	✗	acc_.request_cancel();
376	✗	wait_rd_.request_cancel();
377	✗	wait_wr_.request_cancel();
378	✗	wait_er_.request_cancel();
379
380	✗	reactor_op_base* claimed_acc = nullptr;
381	✗	reactor_op_base* claimed_wr = nullptr;
382	✗	reactor_op_base* claimed_ww = nullptr;
383	✗	reactor_op_base* claimed_we = nullptr;
384		{
385	✗	std::lock_guard lock(desc_state_.mutex);
386	✗	claimed_acc = std::exchange(desc_state_.read_op, nullptr);
387	✗	claimed_wr = std::exchange(desc_state_.wait_read_op, nullptr);
388	✗	claimed_ww = std::exchange(desc_state_.wait_write_op, nullptr);
389	✗	claimed_we = std::exchange(desc_state_.wait_error_op, nullptr);
390	✗	desc_state_.read_ready = false;
391	✗	desc_state_.write_ready = false;
392
393	✗	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
394	✗	desc_state_.impl_ref_ = self;
395	✗	}
396
397	✗	auto repost = [&](reactor_op_base* claimed, reactor_op_base& op) {
398		if (claimed)
399		{
400		op.impl_ptr = self;
401		svc_.post(&op);
402		svc_.work_finished();
403		}
404		};
405	✗	repost(claimed_acc, acc_);
406	✗	repost(claimed_wr, wait_rd_);
407	✗	repost(claimed_ww, wait_wr_);
408	✗	repost(claimed_we, wait_er_);
409		}
410
411	✗	native_handle_type released = fd_;
412
413	✗	if (fd_ >= 0)
414		{
415	✗	if (desc_state_.registered_events != 0)
416	✗	svc_.scheduler().deregister_descriptor(fd_);
417	✗	fd_ = -1;
418		}
419
420	✗	desc_state_.fd = -1;
421	✗	desc_state_.registered_events = 0;
422
423	✗	local_endpoint_ = Endpoint{};
424
425	✗	return released;
426	✗	}
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		std::error_code
438	188x	reactor_acceptor<Derived, Service, Op, AcceptOp, WaitOp, DescState, ImplBase, Endpoint>::
439		do_bind(Endpoint const& ep)
440		{
441	188x	sockaddr_storage storage{};
442	188x	socklen_t addrlen = to_sockaddr(ep, storage);
443	188x	if (::bind(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen) < 0)
444	8x	return make_err(errno);
445
446		// Cache local endpoint (resolves ephemeral port / path)
447	180x	sockaddr_storage local{};
448	180x	socklen_t local_len = sizeof(local);
449	180x	if (::getsockname(fd_, reinterpret_cast<sockaddr*>(&local), &local_len) ==
450		0)
451	180x	set_local_endpoint(from_sockaddr_as(local, local_len, Endpoint{}));
452
453	180x	return {};
454		}
455
456		template<
457		class Derived,
458		class Service,
459		class Op,
460		class AcceptOp,
461		class WaitOp,
462		class DescState,
463		class ImplBase,
464		class Endpoint>
465		std::error_code
466	160x	reactor_acceptor<Derived, Service, Op, AcceptOp, WaitOp, DescState, ImplBase, Endpoint>::
467		do_listen(int backlog)
468		{
469	160x	if (::listen(fd_, backlog) < 0)
470	✗	return make_err(errno);
471
472	160x	svc_.scheduler().register_descriptor(fd_, &desc_state_);
473	160x	return {};
474		}
475
476		template<
477		class Derived,
478		class Service,
479		class Op,
480		class AcceptOp,
481		class WaitOp,
482		class DescState,
483		class ImplBase,
484		class Endpoint>
485		std::coroutine_handle<>
486	2x	reactor_acceptor<Derived, Service, Op, AcceptOp, WaitOp, DescState, ImplBase, Endpoint>::
487		do_wait(
488		std::coroutine_handle<> h,
489		capy::executor_ref ex,
490		wait_type w,
491		std::stop_token const& token,
492		std::error_code* ec)
493		{
494		// wait_type::write completes immediately (see reactor_stream_socket::do_wait).
495	2x	if (w == wait_type::write)
496		{
497	✗	auto& op = wait_wr_;
498	✗	op.reset();
499	✗	op.wait_event = reactor_event_write;
500	✗	op.h = h;
501	✗	op.ex = ex;
502	✗	op.ec_out = ec;
503	✗	op.fd = this->fd_;
504	✗	op.start(token, static_cast<Derived*>(this));
505	✗	op.impl_ptr = this->shared_from_this();
506	✗	op.complete(0, 0);
507	✗	svc_.post(&op);
508	✗	return std::noop_coroutine();
509		}
510
511		WaitOp* op_ptr;
512		reactor_op_base** desc_slot_ptr;
513		std::uint32_t event;
514
515	2x	if (w == wait_type::read)
516		{
517	2x	op_ptr = &wait_rd_;
518	2x	desc_slot_ptr = &desc_state_.wait_read_op;
519	2x	event = reactor_event_read;
520		}
521		else // wait_type::error
522		{
523	✗	op_ptr = &wait_er_;
524	✗	desc_slot_ptr = &desc_state_.wait_error_op;
525	✗	event = reactor_event_error;
526		}
527
528	2x	auto& op = *op_ptr;
529	2x	op.reset();
530	2x	op.wait_event = event;
531	2x	op.h = h;
532	2x	op.ex = ex;
533	2x	op.ec_out = ec;
534	2x	op.fd = this->fd_;
535	2x	op.start(token, static_cast<Derived*>(this));
536	2x	op.impl_ptr = this->shared_from_this();
537
538	2x	svc_.work_started();
539
540	2x	std::lock_guard lock(desc_state_.mutex);
541	2x	if (op.cancelled.load(std::memory_order_acquire))
542		{
543	✗	svc_.post(&op);
544	✗	svc_.work_finished();
545		}
546		else
547		{
548	2x	*desc_slot_ptr = &op;
549		}
550	2x	return std::noop_coroutine();
551	2x	}
552
553		} // namespace boost::corosio::detail
554
555		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
556