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

48.1% Lines (435/904) 72.1% List of functions (49/68)

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

f(x) Functions (68)

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_BASIC_SOCKET_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BASIC_SOCKET_HPP
12
13		#include <boost/corosio/detail/intrusive.hpp>
14		#include <boost/corosio/detail/native_handle.hpp>
15		#include <boost/corosio/endpoint.hpp>
16		#include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
17		#include <boost/corosio/native/detail/make_err.hpp>
18		#include <boost/corosio/native/detail/endpoint_convert.hpp>
19
20		#include <memory>
21		#include <mutex>
22		#include <utility>
23
24		#include <errno.h>
25		#include <netinet/in.h>
26		#include <sys/socket.h>
27		#include <unistd.h>
28
29		namespace boost::corosio::detail {
30
31		/** CRTP base for reactor-backed socket implementations.
32
33		Extracts the shared data members, virtual overrides, and
34		cancel/close/register logic that is identical across TCP
35		(reactor_stream_socket) and UDP (reactor_datagram_socket).
36
37		Derived classes provide CRTP callbacks that enumerate their
38		specific op slots so cancel/close can iterate them generically.
39
40		@tparam Derived The concrete socket type (CRTP).
41		@tparam ImplBase The public vtable base (tcp_socket::implementation
42		or udp_socket::implementation).
43		@tparam Service The backend's service type.
44		@tparam DescState The backend's descriptor_state type.
45		*/
46		template<class Derived, class ImplBase, class Service, class DescState>
47		class reactor_basic_socket
48		: public ImplBase
49		, public std::enable_shared_from_this<Derived>
50		, public intrusive_list<Derived>::node
51		{
52		friend Derived;
53
54		template<class, class, class, class, class, class>
55		friend class reactor_stream_socket;
56
57		template<class, class, class, class, class, class, class, class>
58		friend class reactor_datagram_socket;
59
60	30385x	explicit reactor_basic_socket(Service& svc) noexcept : svc_(svc) {}
61
62		protected:
63		Service& svc_;
64		int fd_ = -1;
65		endpoint local_endpoint_;
66
67		public:
68		/// Per-descriptor state for persistent reactor registration.
69		DescState desc_state_;
70
71	30385x	~reactor_basic_socket() override = default;
72
73		/// Return the underlying file descriptor.
74	65009x	native_handle_type native_handle() const noexcept override
75		{
76	65009x	return fd_;
77		}
78
79		/// Return the cached local endpoint.
80	68x	endpoint local_endpoint() const noexcept override
81		{
82	68x	return local_endpoint_;
83		}
84
85		/// Return true if the socket has an open file descriptor.
86		bool is_open() const noexcept
87		{
88		return fd_ >= 0;
89		}
90
91		/// Set a socket option.
92	2132x	std::error_code set_option(
93		int level,
94		int optname,
95		void const* data,
96		std::size_t size) noexcept override
97		{
98	2132x	if (::setsockopt(
99	2132x	fd_, level, optname, data, static_cast<socklen_t>(size)) != 0)
100	✗	return make_err(errno);
101	2132x	return {};
102		}
103
104		/// Get a socket option.
105		std::error_code
106	78x	get_option(int level, int optname, void* data, std::size_t* size)
107		const noexcept override
108		{
109	78x	socklen_t len = static_cast<socklen_t>(*size);
110	78x	if (::getsockopt(fd_, level, optname, data, &len) != 0)
111	✗	return make_err(errno);
112	78x	*size = static_cast<std::size_t>(len);
113	78x	return {};
114		}
115
116		/// Assign the file descriptor.
117	10059x	void set_socket(int fd) noexcept
118		{
119	10059x	fd_ = fd;
120	10059x	}
121
122		/// Cache the local endpoint.
123		void set_local_endpoint(endpoint ep) noexcept
124		{
125		local_endpoint_ = ep;
126		}
127
128		/** Bind the socket to a local endpoint.
129
130		Calls ::bind() and caches the resulting local endpoint
131		via getsockname().
132
133		@param ep The endpoint to bind to.
134		@return Error code on failure, empty on success.
135		*/
136	42x	std::error_code do_bind(endpoint ep) noexcept
137		{
138	42x	sockaddr_storage storage{};
139	42x	socklen_t addrlen = to_sockaddr(ep, socket_family(fd_), storage);
140	42x	if (::bind(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen) != 0)
141	✗	return make_err(errno);
142
143	42x	sockaddr_storage local_storage{};
144	42x	socklen_t local_len = sizeof(local_storage);
145	42x	if (::getsockname(
146	42x	fd_, reinterpret_cast<sockaddr*>(&local_storage), &local_len) ==
147		0)
148	42x	local_endpoint_ = from_sockaddr(local_storage);
149
150	42x	return {};
151		}
152
153		/** Register an op with the reactor.
154
155		Handles cached edge events and deferred cancellation.
156		Called on the EAGAIN/EINPROGRESS path when speculative
157		I/O failed.
158		*/
159		template<class Op>
160		void register_op(
161		Op& op,
162		reactor_op_base*& desc_slot,
163		bool& ready_flag,
164		bool& cancel_flag) noexcept;
165
166		/** Cancel a single pending operation.
167
168		Claims the operation from its descriptor_state slot under
169		the mutex and posts it to the scheduler as cancelled.
170		Derived must implement:
171		op_to_desc_slot(Op&) -> reactor_op_base**
172		op_to_cancel_flag(Op&) -> bool*
173		*/
174		template<class Op>
175		void cancel_single_op(Op& op) noexcept;
176
177		/** Cancel all pending operations.
178
179		Invoked by the derived class's cancel() override.
180		Derived must implement:
181		for_each_op(auto fn)
182		for_each_desc_entry(auto fn)
183		*/
184		void do_cancel() noexcept;
185
186		/** Close the socket and cancel pending operations.
187
188		Invoked by the derived class's close_socket(). The
189		derived class may add backend-specific cleanup after
190		calling this method.
191		Derived must implement:
192		for_each_op(auto fn)
193		for_each_desc_entry(auto fn)
194		*/
195		void do_close_socket() noexcept;
196		};
197
198		template<class Derived, class ImplBase, class Service, class DescState>
199		template<class Op>
200		void
201	12792x	reactor_basic_socket<Derived, ImplBase, Service, DescState>::register_op(
202		Op& op,
203		reactor_op_base*& desc_slot,
204		bool& ready_flag,
205		bool& cancel_flag) noexcept
206		{
207	12792x	svc_.work_started();
208
209	12792x	std::lock_guard lock(desc_state_.mutex);
210	12792x	bool io_done = false;
211	12792x	if (ready_flag)
212		{
213	213x	ready_flag = false;
214	213x	op.perform_io();
215	213x	io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
216	213x	if (!io_done)
217	213x	op.errn = 0;
218		}
219
220	12792x	if (cancel_flag)
221		{
222	✗	cancel_flag = false;
223	✗	op.cancelled.store(true, std::memory_order_relaxed);
224		}
225
226	12792x	if (io_done \|\| op.cancelled.load(std::memory_order_acquire))
227		{
228	✗	svc_.post(&op);
229	✗	svc_.work_finished();
230		}
231		else
232		{
233	12792x	desc_slot = &op;
234		}
235	12792x	}
236
237		template<class Derived, class ImplBase, class Service, class DescState>
238		template<class Op>
239		void
240	896x	reactor_basic_socket<Derived, ImplBase, Service, DescState>::cancel_single_op(
241		Op& op) noexcept
242		{
243	896x	auto self = this->weak_from_this().lock();
244	896x	if (!self)
245	✗	return;
246
247	896x	op.request_cancel();
248
249	896x	auto* d = static_cast<Derived*>(this);
250	896x	reactor_op_base** desc_op_ptr = d->op_to_desc_slot(op);
251
252	896x	if (desc_op_ptr)
253		{
254	896x	reactor_op_base* claimed = nullptr;
255		{
256	896x	std::lock_guard lock(desc_state_.mutex);
257	896x	if (*desc_op_ptr == &op)
258	894x	claimed = std::exchange(*desc_op_ptr, nullptr);
259		else
260		{
261	2x	bool* cflag = d->op_to_cancel_flag(op);
262	2x	if (cflag)
263	2x	*cflag = true;
264		}
265	896x	}
266	896x	if (claimed)
267		{
268	894x	op.impl_ptr = self;
269	894x	svc_.post(&op);
270	894x	svc_.work_finished();
271		}
272		}
273	896x	}
274
275		template<class Derived, class ImplBase, class Service, class DescState>
276		void
277	221x	reactor_basic_socket<Derived, ImplBase, Service, DescState>::
278		do_cancel() noexcept
279		{
280	221x	auto self = this->weak_from_this().lock();
281	221x	if (!self)
282	✗	return;
283
284	221x	auto* d = static_cast<Derived*>(this);
285
286	221x	d->for_each_op([](auto& op) { op.request_cancel(); });
287
288		// Claim ops under a single lock acquisition
289		struct claimed_entry
290		{
291		reactor_op_base* op = nullptr;
292		reactor_op_base* base = nullptr;
293		};
294		// Max 3 ops (conn, rd, wr)
295	221x	claimed_entry claimed[3];
296	221x	int count = 0;
297
298		{
299	221x	std::lock_guard lock(desc_state_.mutex);
300	221x	d->for_each_desc_entry([&](auto& op, reactor_op_base*& desc_slot) {
301		if (desc_slot == &op)
302		{
303		claimed[count].op = std::exchange(desc_slot, nullptr);
304		claimed[count].base = &op;
305		++count;
306		}
307		});
308	221x	}
309
310	333x	for (int i = 0; i < count; ++i)
311		{
312	112x	claimed[i].base->impl_ptr = self;
313	112x	svc_.post(claimed[i].base);
314	112x	svc_.work_finished();
315		}
316	221x	}
317
318		template<class Derived, class ImplBase, class Service, class DescState>
319		void
320	91163x	reactor_basic_socket<Derived, ImplBase, Service, DescState>::
321		do_close_socket() noexcept
322		{
323	91163x	auto self = this->weak_from_this().lock();
324	91163x	if (self)
325		{
326	91163x	auto* d = static_cast<Derived*>(this);
327
328	91163x	d->for_each_op([](auto& op) { op.request_cancel(); });
329
330		struct claimed_entry
331		{
332		reactor_op_base* base = nullptr;
333		};
334	91163x	claimed_entry claimed[3];
335	91163x	int count = 0;
336
337		{
338	91163x	std::lock_guard lock(desc_state_.mutex);
339	91163x	d->for_each_desc_entry(
340		[&](auto& /op/, reactor_op_base*& desc_slot) {
341		auto* c = std::exchange(desc_slot, nullptr);
342		if (c)
343		{
344		claimed[count].base = c;
345		++count;
346		}
347		});
348	91163x	desc_state_.read_ready = false;
349	91163x	desc_state_.write_ready = false;
350	91163x	desc_state_.read_cancel_pending = false;
351	91163x	desc_state_.write_cancel_pending = false;
352	91163x	desc_state_.connect_cancel_pending = false;
353
354	91163x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
355	1183x	desc_state_.impl_ref_ = self;
356	91163x	}
357
358	91169x	for (int i = 0; i < count; ++i)
359		{
360	6x	claimed[i].base->impl_ptr = self;
361	6x	svc_.post(claimed[i].base);
362	6x	svc_.work_finished();
363		}
364		}
365
366	91163x	if (fd_ >= 0)
367		{
368	20226x	if (desc_state_.registered_events != 0)
369	20226x	svc_.scheduler().deregister_descriptor(fd_);
370	20226x	::close(fd_);
371	20226x	fd_ = -1;
372		}
373
374	91163x	desc_state_.fd = -1;
375	91163x	desc_state_.registered_events = 0;
376
377	91163x	local_endpoint_ = endpoint{};
378	91163x	}
379
380		} // namespace boost::corosio::detail
381
382		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BASIC_SOCKET_HPP
383