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

46.4% Lines (727/1566) 61.9% List of functions (78/126)

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

reactor_basic_socket.hpp

f(x) Functions (126)

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		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
46		*/
47		template<
48		class Derived,
49		class ImplBase,
50		class Service,
51		class DescState,
52		class Endpoint = endpoint>
53		class reactor_basic_socket
54		: public ImplBase
55		, public std::enable_shared_from_this<Derived>
56		, public intrusive_list<Derived>::node
57		{
58		friend Derived;
59
60		template<class, class, class, class, class, class, class, class>
61		friend class reactor_stream_socket;
62
63		template<class, class, class, class, class, class, class, class, class, class>
64		friend class reactor_datagram_socket;
65
66	15367x	explicit reactor_basic_socket(Service& svc) noexcept : svc_(svc) {}
67
68		protected:
69		Service& svc_;
70		int fd_ = -1;
71		Endpoint local_endpoint_;
72
73		public:
74		/// Per-descriptor state for persistent reactor registration.
75		DescState desc_state_;
76
77	15367x	~reactor_basic_socket() override = default;
78
79		/// Return the underlying file descriptor.
80	46774x	native_handle_type native_handle() const noexcept override
81		{
82	46774x	return fd_;
83		}
84
85		/// Return the cached local endpoint.
86	80x	Endpoint local_endpoint() const noexcept override
87		{
88	80x	return local_endpoint_;
89		}
90
91		/// Return true if the socket has an open file descriptor.
92		bool is_open() const noexcept
93		{
94		return fd_ >= 0;
95		}
96
97		/// Set a socket option.
98	20x	std::error_code set_option(
99		int level,
100		int optname,
101		void const* data,
102		std::size_t size) noexcept override
103		{
104	20x	if (::setsockopt(
105	20x	fd_, level, optname, data, static_cast<socklen_t>(size)) != 0)
106	✗	return make_err(errno);
107	20x	return {};
108		}
109
110		/// Get a socket option.
111		std::error_code
112	78x	get_option(int level, int optname, void* data, std::size_t* size)
113		const noexcept override
114		{
115	78x	socklen_t len = static_cast<socklen_t>(*size);
116	78x	if (::getsockopt(fd_, level, optname, data, &len) != 0)
117	✗	return make_err(errno);
118	78x	*size = static_cast<std::size_t>(len);
119	78x	return {};
120		}
121
122		/// Assign the file descriptor.
123	5041x	void set_socket(int fd) noexcept
124		{
125	5041x	fd_ = fd;
126	5041x	}
127
128		/// Assign the fd, initialize descriptor state, and register with the reactor.
129	5218x	void init_and_register(int fd) noexcept
130		{
131	5218x	fd_ = fd;
132	5218x	desc_state_.fd = fd;
133		{
134	5218x	std::lock_guard lock(desc_state_.mutex);
135	5218x	desc_state_.read_op = nullptr;
136	5218x	desc_state_.write_op = nullptr;
137	5218x	desc_state_.connect_op = nullptr;
138	5218x	}
139	5218x	svc_.scheduler().register_descriptor(fd, &desc_state_);
140	5218x	}
141
142		/// Cache the local endpoint.
143		void set_local_endpoint(Endpoint ep) noexcept
144		{
145		local_endpoint_ = ep;
146		}
147
148		/** Bind the socket to a local endpoint.
149
150		Calls ::bind() and caches the resulting local endpoint
151		via getsockname().
152
153		@param ep The endpoint to bind to.
154		@return Error code on failure, empty on success.
155		*/
156	76x	std::error_code do_bind(Endpoint const& ep) noexcept
157		{
158	76x	sockaddr_storage storage{};
159	76x	socklen_t addrlen = to_sockaddr(ep, socket_family(fd_), storage);
160	76x	if (::bind(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen) != 0)
161	10x	return make_err(errno);
162
163	66x	sockaddr_storage local_storage{};
164	66x	socklen_t local_len = sizeof(local_storage);
165	66x	if (::getsockname(
166	66x	fd_, reinterpret_cast<sockaddr*>(&local_storage), &local_len) ==
167		0)
168	52x	local_endpoint_ =
169	66x	from_sockaddr_as(local_storage, local_len, Endpoint{});
170
171	66x	return {};
172		}
173
174		/** Register an op with the reactor.
175
176		Handles cached edge events and deferred cancellation.
177		Called on the EAGAIN/EINPROGRESS path when speculative
178		I/O failed.
179		*/
180		template<class Op>
181		void register_op(
182		Op& op,
183		reactor_op_base*& desc_slot,
184		bool& ready_flag,
185		bool& cancel_flag,
186		bool is_write_direction = false) noexcept;
187
188		/** Cancel a single pending operation.
189
190		Claims the operation from its descriptor_state slot under
191		the mutex and posts it to the scheduler as cancelled.
192		Derived must implement:
193		op_to_desc_slot(Op&) -> reactor_op_base**
194		op_to_cancel_flag(Op&) -> bool*
195		*/
196		template<class Op>
197		void cancel_single_op(Op& op) noexcept;
198
199		/** Cancel all pending operations.
200
201		Invoked by the derived class's cancel() override.
202		Derived must implement:
203		for_each_op(auto fn)
204		for_each_desc_entry(auto fn)
205		*/
206		void do_cancel() noexcept;
207
208		/** Close the socket and cancel pending operations.
209
210		Invoked by the derived class's close_socket(). The
211		derived class may add backend-specific cleanup after
212		calling this method.
213		Derived must implement:
214		for_each_op(auto fn)
215		for_each_desc_entry(auto fn)
216		*/
217		void do_close_socket() noexcept;
218
219		/** Release the socket without closing the fd.
220
221		Like do_close_socket() but does not call ::close().
222		Returns the fd so the caller can take ownership.
223		*/
224		native_handle_type do_release_socket() noexcept;
225		};
226
227		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
228		template<class Op>
229		void
230	5436x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::register_op(
231		Op& op,
232		reactor_op_base*& desc_slot,
233		bool& ready_flag,
234		bool& cancel_flag,
235		bool is_write_direction) noexcept
236		{
237	5436x	svc_.work_started();
238
239	5436x	std::lock_guard lock(desc_state_.mutex);
240	5436x	bool io_done = false;
241	5436x	if (ready_flag)
242		{
243	183x	ready_flag = false;
244	183x	op.perform_io();
245	183x	io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
246	183x	if (!io_done)
247	183x	op.errn = 0;
248		}
249
250	5436x	if (cancel_flag)
251		{
252	✗	cancel_flag = false;
253	✗	op.cancelled.store(true, std::memory_order_relaxed);
254		}
255
256	5436x	if (io_done \|\| op.cancelled.load(std::memory_order_acquire))
257		{
258	✗	svc_.post(&op);
259	✗	svc_.work_finished();
260		}
261		else
262		{
263	5436x	desc_slot = &op;
264
265		// Select must rebuild its fd_sets when a write-direction op
266		// is parked, so select() watches for writability. Compiled
267		// away to nothing for epoll and kqueue.
268		if constexpr (Service::needs_write_notification)
269		{
270	2265x	if (is_write_direction)
271	2074x	svc_.scheduler().notify_reactor();
272		}
273		}
274	5436x	}
275
276		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
277		template<class Op>
278		void
279	192x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::cancel_single_op(
280		Op& op) noexcept
281		{
282	192x	auto self = this->weak_from_this().lock();
283	192x	if (!self)
284	✗	return;
285
286	192x	op.request_cancel();
287
288	192x	auto* d = static_cast<Derived*>(this);
289	192x	reactor_op_base** desc_op_ptr = d->op_to_desc_slot(op);
290
291	192x	if (desc_op_ptr)
292		{
293	192x	reactor_op_base* claimed = nullptr;
294		{
295	192x	std::lock_guard lock(desc_state_.mutex);
296	192x	if (*desc_op_ptr == &op)
297	192x	claimed = std::exchange(*desc_op_ptr, nullptr);
298		else
299		{
300	✗	bool* cflag = d->op_to_cancel_flag(op);
301	✗	if (cflag)
302	✗	*cflag = true;
303		}
304	192x	}
305	192x	if (claimed)
306		{
307	192x	op.impl_ptr = self;
308	192x	svc_.post(&op);
309	192x	svc_.work_finished();
310		}
311		}
312	192x	}
313
314		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
315		void
316	188x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
317		do_cancel() noexcept
318		{
319	188x	auto self = this->weak_from_this().lock();
320	188x	if (!self)
321	✗	return;
322
323	188x	auto* d = static_cast<Derived*>(this);
324
325	188x	d->for_each_op([](auto& op) { op.request_cancel(); });
326
327		// Claim ops under a single lock acquisition
328		struct claimed_entry
329		{
330		reactor_op_base* op = nullptr;
331		reactor_op_base* base = nullptr;
332		};
333		// Max 3 ops (conn, rd, wr)
334	188x	claimed_entry claimed[3];
335	188x	int count = 0;
336
337		{
338	188x	std::lock_guard lock(desc_state_.mutex);
339	188x	d->for_each_desc_entry([&](auto& op, reactor_op_base*& desc_slot) {
340		if (desc_slot == &op)
341		{
342		claimed[count].op = std::exchange(desc_slot, nullptr);
343		claimed[count].base = &op;
344		++count;
345		}
346		});
347	188x	}
348
349	288x	for (int i = 0; i < count; ++i)
350		{
351	100x	claimed[i].base->impl_ptr = self;
352	100x	svc_.post(claimed[i].base);
353	100x	svc_.work_finished();
354		}
355	188x	}
356
357		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
358		void
359	46209x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
360		do_close_socket() noexcept
361		{
362	46209x	auto self = this->weak_from_this().lock();
363	46209x	if (self)
364		{
365	46209x	auto* d = static_cast<Derived*>(this);
366
367	46209x	d->for_each_op([](auto& op) { op.request_cancel(); });
368
369		struct claimed_entry
370		{
371		reactor_op_base* base = nullptr;
372		};
373	46209x	claimed_entry claimed[3];
374	46209x	int count = 0;
375
376		{
377	46209x	std::lock_guard lock(desc_state_.mutex);
378	46209x	d->for_each_desc_entry(
379		[&](auto& /op/, reactor_op_base*& desc_slot) {
380		auto* c = std::exchange(desc_slot, nullptr);
381		if (c)
382		{
383		claimed[count].base = c;
384		++count;
385		}
386		});
387	46209x	desc_state_.read_ready = false;
388	46209x	desc_state_.write_ready = false;
389	46209x	desc_state_.read_cancel_pending = false;
390	46209x	desc_state_.write_cancel_pending = false;
391	46209x	desc_state_.connect_cancel_pending = false;
392
393	46209x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
394	271x	desc_state_.impl_ref_ = self;
395	46209x	}
396
397	46213x	for (int i = 0; i < count; ++i)
398		{
399	4x	claimed[i].base->impl_ptr = self;
400	4x	svc_.post(claimed[i].base);
401	4x	svc_.work_finished();
402		}
403		}
404
405	46209x	if (fd_ >= 0)
406		{
407	10257x	if (desc_state_.registered_events != 0)
408	10257x	svc_.scheduler().deregister_descriptor(fd_);
409	10257x	::close(fd_);
410	10257x	fd_ = -1;
411		}
412
413	46209x	desc_state_.fd = -1;
414	46209x	desc_state_.registered_events = 0;
415
416	46209x	local_endpoint_ = Endpoint{};
417	46209x	}
418
419		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
420		native_handle_type
421	2x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
422		do_release_socket() noexcept
423		{
424		// Cancel pending ops (same as do_close_socket)
425	2x	auto self = this->weak_from_this().lock();
426	2x	if (self)
427		{
428	2x	auto* d = static_cast<Derived*>(this);
429
430	2x	d->for_each_op([](auto& op) { op.request_cancel(); });
431
432		struct claimed_entry
433		{
434		reactor_op_base* base = nullptr;
435		};
436	2x	claimed_entry claimed[3];
437	2x	int count = 0;
438
439		{
440	2x	std::lock_guard lock(desc_state_.mutex);
441	2x	d->for_each_desc_entry(
442		[&](auto& /op/, reactor_op_base*& desc_slot) {
443		auto* c = std::exchange(desc_slot, nullptr);
444		if (c)
445		{
446		claimed[count].base = c;
447		++count;
448		}
449		});
450	2x	desc_state_.read_ready = false;
451	2x	desc_state_.write_ready = false;
452	2x	desc_state_.read_cancel_pending = false;
453	2x	desc_state_.write_cancel_pending = false;
454	2x	desc_state_.connect_cancel_pending = false;
455
456	2x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
457	✗	desc_state_.impl_ref_ = self;
458	2x	}
459
460	2x	for (int i = 0; i < count; ++i)
461		{
462	✗	claimed[i].base->impl_ptr = self;
463	✗	svc_.post(claimed[i].base);
464	✗	svc_.work_finished();
465		}
466		}
467
468	2x	native_handle_type released = fd_;
469
470	2x	if (fd_ >= 0)
471		{
472	2x	if (desc_state_.registered_events != 0)
473	2x	svc_.scheduler().deregister_descriptor(fd_);
474		// Do NOT close -- caller takes ownership
475	2x	fd_ = -1;
476		}
477
478	2x	desc_state_.fd = -1;
479	2x	desc_state_.registered_events = 0;
480
481	2x	local_endpoint_ = Endpoint{};
482
483	4x	return released;
484	2x	}
485
486		} // namespace boost::corosio::detail
487
488		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BASIC_SOCKET_HPP
489