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

33.2% Lines (650/1958) 50.7% List of functions (72/142)

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

reactor_basic_socket.hpp

f(x) Functions (142)

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	21927x	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	21927x	~reactor_basic_socket() override = default;
78
79		/// Return the underlying file descriptor.
80	44689x	native_handle_type native_handle() const noexcept override
81		{
82	44689x	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	80x	std::error_code set_option(
99		int level,
100		int optname,
101		void const* data,
102		std::size_t size) noexcept override
103		{
104	80x	if (::setsockopt(
105	80x	fd_, level, optname, data, static_cast<socklen_t>(size)) != 0)
106	✗	return make_err(errno);
107	80x	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	7228x	void set_socket(int fd) noexcept
124		{
125	7228x	fd_ = fd;
126	7228x	}
127
128		/// Cache the local endpoint.
129		void set_local_endpoint(Endpoint ep) noexcept
130		{
131		local_endpoint_ = ep;
132		}
133
134		/** Bind the socket to a local endpoint.
135
136		Calls ::bind() and caches the resulting local endpoint
137		via getsockname().
138
139		@param ep The endpoint to bind to.
140		@return Error code on failure, empty on success.
141		*/
142	76x	std::error_code do_bind(Endpoint const& ep) noexcept
143		{
144	76x	sockaddr_storage storage{};
145	76x	socklen_t addrlen = to_sockaddr(ep, socket_family(fd_), storage);
146	76x	if (::bind(fd_, reinterpret_cast<sockaddr*>(&storage), addrlen) != 0)
147	10x	return make_err(errno);
148
149	66x	sockaddr_storage local_storage{};
150	66x	socklen_t local_len = sizeof(local_storage);
151	66x	if (::getsockname(
152	66x	fd_, reinterpret_cast<sockaddr*>(&local_storage), &local_len) ==
153		0)
154	52x	local_endpoint_ =
155	66x	from_sockaddr_as(local_storage, local_len, Endpoint{});
156
157	66x	return {};
158		}
159
160		/** Register an op with the reactor.
161
162		Handles cached edge events and deferred cancellation.
163		Called on the EAGAIN/EINPROGRESS path when speculative
164		I/O failed.
165		*/
166		template<class Op>
167		void register_op(
168		Op& op,
169		reactor_op_base*& desc_slot,
170		bool& ready_flag,
171		bool& cancel_flag) noexcept;
172
173		/** Cancel a single pending operation.
174
175		Claims the operation from its descriptor_state slot under
176		the mutex and posts it to the scheduler as cancelled.
177		Derived must implement:
178		op_to_desc_slot(Op&) -> reactor_op_base**
179		op_to_cancel_flag(Op&) -> bool*
180		*/
181		template<class Op>
182		void cancel_single_op(Op& op) noexcept;
183
184		/** Cancel all pending operations.
185
186		Invoked by the derived class's cancel() override.
187		Derived must implement:
188		for_each_op(auto fn)
189		for_each_desc_entry(auto fn)
190		*/
191		void do_cancel() noexcept;
192
193		/** Close the socket and cancel pending operations.
194
195		Invoked by the derived class's close_socket(). The
196		derived class may add backend-specific cleanup after
197		calling this method.
198		Derived must implement:
199		for_each_op(auto fn)
200		for_each_desc_entry(auto fn)
201		*/
202		void do_close_socket() noexcept;
203
204		/** Release the socket without closing the fd.
205
206		Like do_close_socket() but does not call ::close().
207		Returns the fd so the caller can take ownership.
208		*/
209		native_handle_type do_release_socket() noexcept;
210		};
211
212		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
213		template<class Op>
214		void
215	7642x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::register_op(
216		Op& op,
217		reactor_op_base*& desc_slot,
218		bool& ready_flag,
219		bool& cancel_flag) noexcept
220		{
221	7642x	svc_.work_started();
222
223	7642x	std::lock_guard lock(desc_state_.mutex);
224	7642x	bool io_done = false;
225	7642x	if (ready_flag)
226		{
227	193x	ready_flag = false;
228	193x	op.perform_io();
229	193x	io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
230	193x	if (!io_done)
231	193x	op.errn = 0;
232		}
233
234	7642x	if (cancel_flag)
235		{
236	✗	cancel_flag = false;
237	✗	op.cancelled.store(true, std::memory_order_relaxed);
238		}
239
240	7642x	if (io_done \|\| op.cancelled.load(std::memory_order_acquire))
241		{
242	✗	svc_.post(&op);
243	✗	svc_.work_finished();
244		}
245		else
246		{
247	7642x	desc_slot = &op;
248		}
249	7642x	}
250
251		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
252		template<class Op>
253		void
254	201x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::cancel_single_op(
255		Op& op) noexcept
256		{
257	201x	auto self = this->weak_from_this().lock();
258	201x	if (!self)
259	✗	return;
260
261	201x	op.request_cancel();
262
263	201x	auto* d = static_cast<Derived*>(this);
264	201x	reactor_op_base** desc_op_ptr = d->op_to_desc_slot(op);
265
266	201x	if (desc_op_ptr)
267		{
268	201x	reactor_op_base* claimed = nullptr;
269		{
270	201x	std::lock_guard lock(desc_state_.mutex);
271	201x	if (*desc_op_ptr == &op)
272	201x	claimed = std::exchange(*desc_op_ptr, nullptr);
273		else
274		{
275	✗	bool* cflag = d->op_to_cancel_flag(op);
276	✗	if (cflag)
277	✗	*cflag = true;
278		}
279	201x	}
280	201x	if (claimed)
281		{
282	201x	op.impl_ptr = self;
283	201x	svc_.post(&op);
284	201x	svc_.work_finished();
285		}
286		}
287	201x	}
288
289		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
290		void
291	198x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
292		do_cancel() noexcept
293		{
294	198x	auto self = this->weak_from_this().lock();
295	198x	if (!self)
296	✗	return;
297
298	198x	auto* d = static_cast<Derived*>(this);
299
300	198x	d->for_each_op([](auto& op) { op.request_cancel(); });
301
302		// Claim ops under a single lock acquisition
303		struct claimed_entry
304		{
305		reactor_op_base* op = nullptr;
306		reactor_op_base* base = nullptr;
307		};
308		// Max 3 ops (conn, rd, wr)
309	198x	claimed_entry claimed[3];
310	198x	int count = 0;
311
312		{
313	198x	std::lock_guard lock(desc_state_.mutex);
314	198x	d->for_each_desc_entry([&](auto& op, reactor_op_base*& desc_slot) {
315		if (desc_slot == &op)
316		{
317		claimed[count].op = std::exchange(desc_slot, nullptr);
318		claimed[count].base = &op;
319		++count;
320		}
321		});
322	198x	}
323
324	304x	for (int i = 0; i < count; ++i)
325		{
326	106x	claimed[i].base->impl_ptr = self;
327	106x	svc_.post(claimed[i].base);
328	106x	svc_.work_finished();
329		}
330	198x	}
331
332		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
333		void
334	65888x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
335		do_close_socket() noexcept
336		{
337	65888x	auto self = this->weak_from_this().lock();
338	65888x	if (self)
339		{
340	65888x	auto* d = static_cast<Derived*>(this);
341
342	65888x	d->for_each_op([](auto& op) { op.request_cancel(); });
343
344		struct claimed_entry
345		{
346		reactor_op_base* base = nullptr;
347		};
348	65888x	claimed_entry claimed[3];
349	65888x	int count = 0;
350
351		{
352	65888x	std::lock_guard lock(desc_state_.mutex);
353	65888x	d->for_each_desc_entry(
354		[&](auto& /op/, reactor_op_base*& desc_slot) {
355		auto* c = std::exchange(desc_slot, nullptr);
356		if (c)
357		{
358		claimed[count].base = c;
359		++count;
360		}
361		});
362	65888x	desc_state_.read_ready = false;
363	65888x	desc_state_.write_ready = false;
364	65888x	desc_state_.read_cancel_pending = false;
365	65888x	desc_state_.write_cancel_pending = false;
366	65888x	desc_state_.connect_cancel_pending = false;
367
368	65888x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
369	271x	desc_state_.impl_ref_ = self;
370	65888x	}
371
372	65892x	for (int i = 0; i < count; ++i)
373		{
374	4x	claimed[i].base->impl_ptr = self;
375	4x	svc_.post(claimed[i].base);
376	4x	svc_.work_finished();
377		}
378		}
379
380	65888x	if (fd_ >= 0)
381		{
382	14630x	if (desc_state_.registered_events != 0)
383	14630x	svc_.scheduler().deregister_descriptor(fd_);
384	14630x	::close(fd_);
385	14630x	fd_ = -1;
386		}
387
388	65888x	desc_state_.fd = -1;
389	65888x	desc_state_.registered_events = 0;
390
391	65888x	local_endpoint_ = Endpoint{};
392	65888x	}
393
394		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
395		native_handle_type
396	2x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
397		do_release_socket() noexcept
398		{
399		// Cancel pending ops (same as do_close_socket)
400	2x	auto self = this->weak_from_this().lock();
401	2x	if (self)
402		{
403	2x	auto* d = static_cast<Derived*>(this);
404
405	2x	d->for_each_op([](auto& op) { op.request_cancel(); });
406
407		struct claimed_entry
408		{
409		reactor_op_base* base = nullptr;
410		};
411	2x	claimed_entry claimed[3];
412	2x	int count = 0;
413
414		{
415	2x	std::lock_guard lock(desc_state_.mutex);
416	2x	d->for_each_desc_entry(
417		[&](auto& /op/, reactor_op_base*& desc_slot) {
418		auto* c = std::exchange(desc_slot, nullptr);
419		if (c)
420		{
421		claimed[count].base = c;
422		++count;
423		}
424		});
425	2x	desc_state_.read_ready = false;
426	2x	desc_state_.write_ready = false;
427	2x	desc_state_.read_cancel_pending = false;
428	2x	desc_state_.write_cancel_pending = false;
429	2x	desc_state_.connect_cancel_pending = false;
430
431	2x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
432	✗	desc_state_.impl_ref_ = self;
433	2x	}
434
435	2x	for (int i = 0; i < count; ++i)
436		{
437	✗	claimed[i].base->impl_ptr = self;
438	✗	svc_.post(claimed[i].base);
439	✗	svc_.work_finished();
440		}
441		}
442
443	2x	native_handle_type released = fd_;
444
445	2x	if (fd_ >= 0)
446		{
447	2x	if (desc_state_.registered_events != 0)
448	2x	svc_.scheduler().deregister_descriptor(fd_);
449		// Do NOT close -- caller takes ownership
450	2x	fd_ = -1;
451		}
452
453	2x	desc_state_.fd = -1;
454	2x	desc_state_.registered_events = 0;
455
456	2x	local_endpoint_ = Endpoint{};
457
458	4x	return released;
459	2x	}
460
461		} // namespace boost::corosio::detail
462
463		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BASIC_SOCKET_HPP
464