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

61.6% Lines (220/357) 76.0% List of functions (38/50)

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

reactor_acceptor.hpp

f(x) Functions (50)

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/detail/intrusive.hpp>
15		#include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
16		#include <boost/corosio/native/detail/make_err.hpp>
17		#include <boost/corosio/native/detail/endpoint_convert.hpp>
18
19		#include <memory>
20		#include <mutex>
21		#include <utility>
22
23		#include <errno.h>
24		#include <netinet/in.h>
25		#include <sys/socket.h>
26		#include <unistd.h>
27
28		namespace boost::corosio::detail {
29
30		/** CRTP base for reactor-backed acceptor implementations.
31
32		Provides shared data members, trivial virtual overrides, and
33		non-virtual helper methods for cancellation and close. Concrete
34		backends inherit and add `cancel()`, `close_socket()`, and
35		`accept()` overrides that delegate to the `do_*` helpers.
36
37		@tparam Derived The concrete acceptor type (CRTP).
38		@tparam Service The backend's acceptor service type.
39		@tparam Op The backend's base op type.
40		@tparam AcceptOp The backend's accept op type.
41		@tparam DescState The backend's descriptor_state type.
42		@tparam ImplBase The public vtable base
43		(tcp_acceptor::implementation or
44		local_stream_acceptor::implementation).
45		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
46		*/
47		template<
48		class Derived,
49		class Service,
50		class Op,
51		class AcceptOp,
52		class DescState,
53		class ImplBase = tcp_acceptor::implementation,
54		class Endpoint = endpoint>
55		class reactor_acceptor
56		: public ImplBase
57		, public std::enable_shared_from_this<Derived>
58		, public intrusive_list<Derived>::node
59		{
60		friend Derived;
61
62	161x	explicit reactor_acceptor(Service& svc) noexcept : svc_(svc) {}
63
64		protected:
65		Service& svc_;
66		int fd_ = -1;
67		Endpoint local_endpoint_;
68
69		public:
70		/// Pending accept operation slot.
71		AcceptOp acc_;
72
73		/// Per-descriptor state for persistent reactor registration.
74		DescState desc_state_;
75
76	161x	~reactor_acceptor() override = default;
77
78		/// Return the underlying file descriptor.
79		int native_handle() const noexcept
80		{
81		return fd_;
82		}
83
84		/// Return the cached local endpoint.
85	7358x	Endpoint local_endpoint() const noexcept override
86		{
87	7358x	return local_endpoint_;
88		}
89
90		/// Return true if the acceptor has an open file descriptor.
91	8292x	bool is_open() const noexcept override
92		{
93	8292x	return fd_ >= 0;
94		}
95
96		/// Set a socket option.
97	139x	std::error_code set_option(
98		int level,
99		int optname,
100		void const* data,
101		std::size_t size) noexcept override
102		{
103	139x	if (::setsockopt(
104	139x	fd_, level, optname, data, static_cast<socklen_t>(size)) != 0)
105	✗	return make_err(errno);
106	139x	return {};
107		}
108
109		/// Get a socket option.
110		std::error_code
111	✗	get_option(int level, int optname, void* data, std::size_t* size)
112		const noexcept override
113		{
114	✗	socklen_t len = static_cast<socklen_t>(*size);
115	✗	if (::getsockopt(fd_, level, optname, data, &len) != 0)
116	✗	return make_err(errno);
117	✗	*size = static_cast<std::size_t>(len);
118	✗	return {};
119		}
120
121		/// Cache the local endpoint.
122	146x	void set_local_endpoint(Endpoint ep) noexcept
123		{
124	146x	local_endpoint_ = std::move(ep);
125	146x	}
126
127		/// Return a reference to the owning service.
128	7228x	Service& service() noexcept
129		{
130	7228x	return svc_;
131		}
132
133		/** Cancel a single pending operation.
134
135		Claims the operation from the read_op descriptor slot
136		under the mutex and posts it to the scheduler as cancelled.
137
138		@param op The operation to cancel.
139		*/
140		void cancel_single_op(Op& op) noexcept;
141
142		/** Cancel the pending accept operation.
143
144		Invoked by the derived class's cancel() override.
145		*/
146		void do_cancel() noexcept;
147
148		/** Close the acceptor and cancel pending operations.
149
150		Invoked by the derived class's close_socket(). The
151		derived class may add backend-specific cleanup after
152		calling this method.
153		*/
154		void do_close_socket() noexcept;
155
156		/** Release the acceptor without closing the fd. */
157		native_handle_type do_release_socket() noexcept;
158
159		/** Bind the acceptor socket to an endpoint.
160
161		Caches the resolved local endpoint (including ephemeral
162		port) after a successful bind.
163
164		@param ep The endpoint to bind to.
165		@return The error code from bind(), or success.
166		*/
167		std::error_code do_bind(Endpoint const& ep);
168
169		/** Start listening on the acceptor socket.
170
171		Registers the file descriptor with the reactor after
172		a successful listen() call.
173
174		@param backlog The listen backlog.
175		@return The error code from listen(), or success.
176		*/
177		std::error_code do_listen(int backlog);
178		};
179
180		template<
181		class Derived,
182		class Service,
183		class Op,
184		class AcceptOp,
185		class DescState,
186		class ImplBase,
187		class Endpoint>
188		void
189	10x	reactor_acceptor<Derived, Service, Op, AcceptOp, DescState, ImplBase, Endpoint>::
190		cancel_single_op(Op& op) noexcept
191		{
192	10x	auto self = this->weak_from_this().lock();
193	10x	if (!self)
194	✗	return;
195
196	10x	op.request_cancel();
197
198	10x	reactor_op_base* claimed = nullptr;
199		{
200	10x	std::lock_guard lock(desc_state_.mutex);
201	10x	if (desc_state_.read_op == &op)
202	8x	claimed = std::exchange(desc_state_.read_op, nullptr);
203	10x	}
204	10x	if (claimed)
205		{
206	8x	op.impl_ptr = self;
207	8x	svc_.post(&op);
208	8x	svc_.work_finished();
209		}
210	10x	}
211
212		template<
213		class Derived,
214		class Service,
215		class Op,
216		class AcceptOp,
217		class DescState,
218		class ImplBase,
219		class Endpoint>
220		void
221	4x	reactor_acceptor<Derived, Service, Op, AcceptOp, DescState, ImplBase, Endpoint>::
222		do_cancel() noexcept
223		{
224	4x	cancel_single_op(acc_);
225	4x	}
226
227		template<
228		class Derived,
229		class Service,
230		class Op,
231		class AcceptOp,
232		class DescState,
233		class ImplBase,
234		class Endpoint>
235		void
236	634x	reactor_acceptor<Derived, Service, Op, AcceptOp, DescState, ImplBase, Endpoint>::
237		do_close_socket() noexcept
238		{
239	634x	auto self = this->weak_from_this().lock();
240	634x	if (self)
241		{
242	634x	acc_.request_cancel();
243
244	634x	reactor_op_base* claimed = nullptr;
245		{
246	634x	std::lock_guard lock(desc_state_.mutex);
247	634x	claimed = std::exchange(desc_state_.read_op, nullptr);
248	634x	desc_state_.read_ready = false;
249	634x	desc_state_.write_ready = false;
250
251	634x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
252	✗	desc_state_.impl_ref_ = self;
253	634x	}
254
255	634x	if (claimed)
256		{
257	4x	acc_.impl_ptr = self;
258	4x	svc_.post(&acc_);
259	4x	svc_.work_finished();
260		}
261		}
262
263	634x	if (fd_ >= 0)
264		{
265	156x	if (desc_state_.registered_events != 0)
266	138x	svc_.scheduler().deregister_descriptor(fd_);
267	156x	::close(fd_);
268	156x	fd_ = -1;
269		}
270
271	634x	desc_state_.fd = -1;
272	634x	desc_state_.registered_events = 0;
273
274	634x	local_endpoint_ = Endpoint{};
275	634x	}
276
277		template<
278		class Derived,
279		class Service,
280		class Op,
281		class AcceptOp,
282		class DescState,
283		class ImplBase,
284		class Endpoint>
285		native_handle_type
286	✗	reactor_acceptor<Derived, Service, Op, AcceptOp, DescState, ImplBase, Endpoint>::
287		do_release_socket() noexcept
288		{
289	✗	auto self = this->weak_from_this().lock();
290	✗	if (self)
291		{
292	✗	acc_.request_cancel();
293
294	✗	reactor_op_base* claimed = nullptr;
295		{
296	✗	std::lock_guard lock(desc_state_.mutex);
297	✗	claimed = std::exchange(desc_state_.read_op, nullptr);
298	✗	desc_state_.read_ready = false;
299	✗	desc_state_.write_ready = false;
300
301	✗	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
302	✗	desc_state_.impl_ref_ = self;
303	✗	}
304
305	✗	if (claimed)
306		{
307	✗	acc_.impl_ptr = self;
308	✗	svc_.post(&acc_);
309	✗	svc_.work_finished();
310		}
311		}
312
313	✗	native_handle_type released = fd_;
314
315	✗	if (fd_ >= 0)
316		{
317	✗	if (desc_state_.registered_events != 0)
318	✗	svc_.scheduler().deregister_descriptor(fd_);
319	✗	fd_ = -1;
320		}
321
322	✗	desc_state_.fd = -1;
323	✗	desc_state_.registered_events = 0;
324
325	✗	local_endpoint_ = Endpoint{};
326
327	✗	return released;
328	✗	}
329
330		template<
331		class Derived,
332		class Service,
333		class Op,
334		class AcceptOp,
335		class DescState,
336		class ImplBase,
337		class Endpoint>
338		std::error_code
339	154x	reactor_acceptor<Derived, Service, Op, AcceptOp, DescState, ImplBase, Endpoint>::
340		do_bind(Endpoint const& ep)
341		{
342	154x	sockaddr_storage storage{};
343	154x	socklen_t addrlen = to_sockaddr(ep, storage);
344	154x	if (::bind(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen) < 0)
345	8x	return make_err(errno);
346
347		// Cache local endpoint (resolves ephemeral port / path)
348	146x	sockaddr_storage local{};
349	146x	socklen_t local_len = sizeof(local);
350	146x	if (::getsockname(fd_, reinterpret_cast<sockaddr*>(&local), &local_len) ==
351		0)
352	146x	set_local_endpoint(from_sockaddr_as(local, local_len, Endpoint{}));
353
354	146x	return {};
355		}
356
357		template<
358		class Derived,
359		class Service,
360		class Op,
361		class AcceptOp,
362		class DescState,
363		class ImplBase,
364		class Endpoint>
365		std::error_code
366	138x	reactor_acceptor<Derived, Service, Op, AcceptOp, DescState, ImplBase, Endpoint>::
367		do_listen(int backlog)
368		{
369	138x	if (::listen(fd_, backlog) < 0)
370	✗	return make_err(errno);
371
372	138x	svc_.scheduler().register_descriptor(fd_, &desc_state_);
373	138x	return {};
374		}
375
376		} // namespace boost::corosio::detail
377
378		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_ACCEPTOR_HPP
379