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

46.1% Lines (717/1554) 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	24200x	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	24200x	~reactor_basic_socket() override = default;
78
79		/// Return the underlying file descriptor.
80	73283x	native_handle_type native_handle() const noexcept override
81		{
82	73283x	return fd_;
83		}
84
85		/// Return the cached local endpoint.
86	84x	Endpoint local_endpoint() const noexcept override
87		{
88	84x	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	7983x	void set_socket(int fd) noexcept
124		{
125	7983x	fd_ = fd;
126	7983x	}
127
128		/// Assign the fd, initialize descriptor state, and register with the reactor.
129	8167x	void init_and_register(int fd) noexcept
130		{
131	8167x	fd_ = fd;
132	8167x	desc_state_.fd = fd;
133		{
134	8167x	std::lock_guard lock(desc_state_.mutex);
135	8167x	desc_state_.read_op = nullptr;
136	8167x	desc_state_.write_op = nullptr;
137	8167x	desc_state_.connect_op = nullptr;
138	8167x	}
139	8167x	svc_.scheduler().register_descriptor(fd, &desc_state_);
140	8167x	}
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	80x	std::error_code do_bind(Endpoint const& ep) noexcept
157		{
158	80x	sockaddr_storage storage{};
159	80x	socklen_t addrlen = to_sockaddr(ep, socket_family(fd_), storage);
160	80x	if (::bind(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen) != 0)
161	10x	return make_err(errno);
162
163	70x	sockaddr_storage local_storage{};
164	70x	socklen_t local_len = sizeof(local_storage);
165	70x	if (::getsockname(
166	70x	fd_, reinterpret_cast<sockaddr*>(&local_storage), &local_len) ==
167		0)
168	56x	local_endpoint_ =
169	70x	from_sockaddr_as(local_storage, local_len, Endpoint{});
170
171	70x	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	8379x	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	8379x	svc_.work_started();
238
239	8379x	std::lock_guard lock(desc_state_.mutex);
240	8379x	bool io_done = false;
241	8379x	if (ready_flag)
242		{
243	185x	ready_flag = false;
244	185x	op.perform_io();
245	185x	io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
246	185x	if (!io_done)
247	185x	op.errn = 0;
248		}
249
250	8379x	if (cancel_flag)
251		{
252	✗	cancel_flag = false;
253	✗	op.cancelled.store(true, std::memory_order_relaxed);
254		}
255
256	8379x	if (io_done \|\| op.cancelled.load(std::memory_order_acquire))
257		{
258	✗	svc_.post(&op);
259	✗	svc_.work_finished();
260		}
261		else
262		{
263	8379x	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	3643x	if (is_write_direction)
271	3451x	svc_.scheduler().notify_reactor();
272		}
273		}
274	8379x	}
275
276		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
277		template<class Op>
278		void
279	193x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::cancel_single_op(
280		Op& op) noexcept
281		{
282	193x	auto self = this->weak_from_this().lock();
283	193x	if (!self)
284	✗	return;
285
286	193x	op.request_cancel();
287
288	193x	auto* d = static_cast<Derived*>(this);
289	193x	reactor_op_base** desc_op_ptr = d->op_to_desc_slot(op);
290
291	193x	if (desc_op_ptr)
292		{
293	193x	reactor_op_base* claimed = nullptr;
294		{
295	193x	std::lock_guard lock(desc_state_.mutex);
296	193x	if (*desc_op_ptr == &op)
297	193x	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	193x	}
305	193x	if (claimed)
306		{
307	193x	op.impl_ptr = self;
308	193x	svc_.post(&op);
309	193x	svc_.work_finished();
310		}
311		}
312	193x	}
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	72715x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
360		do_close_socket() noexcept
361		{
362	72715x	auto self = this->weak_from_this().lock();
363	72715x	if (self)
364		{
365	72715x	auto* d = static_cast<Derived*>(this);
366
367	72715x	d->for_each_op([](auto& op) { op.request_cancel(); });
368
369		struct claimed_entry
370		{
371		reactor_op_base* base = nullptr;
372		};
373	72715x	claimed_entry claimed[3];
374	72715x	int count = 0;
375
376		{
377	72715x	std::lock_guard lock(desc_state_.mutex);
378	72715x	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	72715x	desc_state_.read_ready = false;
388	72715x	desc_state_.write_ready = false;
389	72715x	desc_state_.read_cancel_pending = false;
390	72715x	desc_state_.write_cancel_pending = false;
391	72715x	desc_state_.connect_cancel_pending = false;
392
393	72715x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
394	280x	desc_state_.impl_ref_ = self;
395	72715x	}
396
397	72719x	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	72715x	if (fd_ >= 0)
406		{
407		// init_and_register always registers the descriptor, so any
408		// live fd is registered. Deregister unconditionally.
409	16148x	svc_.scheduler().deregister_descriptor(fd_);
410	16148x	::close(fd_);
411	16148x	fd_ = -1;
412		}
413
414	72715x	desc_state_.fd = -1;
415	72715x	desc_state_.registered_events = 0;
416
417	72715x	local_endpoint_ = Endpoint{};
418	72715x	}
419
420		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
421		native_handle_type
422	2x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
423		do_release_socket() noexcept
424		{
425		// Cancel pending ops (same as do_close_socket)
426	2x	auto self = this->weak_from_this().lock();
427	2x	if (self)
428		{
429	2x	auto* d = static_cast<Derived*>(this);
430
431	2x	d->for_each_op([](auto& op) { op.request_cancel(); });
432
433		struct claimed_entry
434		{
435		reactor_op_base* base = nullptr;
436		};
437	2x	claimed_entry claimed[3];
438	2x	int count = 0;
439
440		{
441	2x	std::lock_guard lock(desc_state_.mutex);
442	2x	d->for_each_desc_entry(
443		[&](auto& /op/, reactor_op_base*& desc_slot) {
444		auto* c = std::exchange(desc_slot, nullptr);
445		if (c)
446		{
447		claimed[count].base = c;
448		++count;
449		}
450		});
451	2x	desc_state_.read_ready = false;
452	2x	desc_state_.write_ready = false;
453	2x	desc_state_.read_cancel_pending = false;
454	2x	desc_state_.write_cancel_pending = false;
455	2x	desc_state_.connect_cancel_pending = false;
456
457	2x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
458	✗	desc_state_.impl_ref_ = self;
459	2x	}
460
461	2x	for (int i = 0; i < count; ++i)
462		{
463	✗	claimed[i].base->impl_ptr = self;
464	✗	svc_.post(claimed[i].base);
465	✗	svc_.work_finished();
466		}
467		}
468
469	2x	native_handle_type released = fd_;
470
471	2x	if (fd_ >= 0)
472		{
473		// init_and_register always registers the descriptor, so any
474		// live fd is registered. Deregister unconditionally.
475	2x	svc_.scheduler().deregister_descriptor(fd_);
476		// Do NOT close -- caller takes ownership
477	2x	fd_ = -1;
478		}
479
480	2x	desc_state_.fd = -1;
481	2x	desc_state_.registered_events = 0;
482
483	2x	local_endpoint_ = Endpoint{};
484
485	4x	return released;
486	2x	}
487
488		} // namespace boost::corosio::detail
489
490		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BASIC_SOCKET_HPP
491