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

88.2% Lines (120/0/136) 77.6% List of functions (90/0/116)

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

reactor_basic_socket.hpp

f(x) Functions (116)

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/native_socket_base.hpp>
17		#include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
18		#include <boost/corosio/native/detail/make_err.hpp>
19		#include <boost/corosio/native/detail/endpoint_convert.hpp>
20
21		#include <memory>
22		#include <mutex>
23		#include <utility>
24
25		#include <errno.h>
26		#include <netinet/in.h>
27		#include <sys/socket.h>
28		#include <unistd.h>
29
30		namespace boost::corosio::detail {
31
32		/** CRTP base for reactor-backed socket implementations.
33
34		Extracts the shared data members, virtual overrides, and
35		cancel/close/register logic that is identical across TCP
36		(reactor_stream_socket) and UDP (reactor_datagram_socket).
37
38		Derived classes provide CRTP callbacks that enumerate their
39		specific op slots so cancel/close can iterate them generically.
40
41		@tparam Derived The concrete socket type (CRTP).
42		@tparam ImplBase The public vtable base (tcp_socket::implementation
43		or udp_socket::implementation).
44		@tparam Service The backend's service type.
45		@tparam DescState The backend's descriptor_state type.
46		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
47		*/
48		template<
49		class Derived,
50		class ImplBase,
51		class Service,
52		class DescState,
53		class Endpoint = endpoint>
54		class reactor_basic_socket
55		: public native_socket_base<Derived, ImplBase, Endpoint>
56		, public intrusive_list<Derived>::node
57		{
58		friend Derived;
59
60		template<class, 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, class>
64		friend class reactor_datagram_socket;
65
66	20873x	explicit reactor_basic_socket(Service& svc) noexcept : svc_(svc) {}
67
68		protected:
69		// fd_ / local_endpoint_ and the synchronous accessors (native_handle,
70		// is_open, set_option/get_option, set_socket/set_local_endpoint, do_bind)
71		// live in native_socket_base — the readiness/completion-agnostic base
72		// shared with io_uring's sockets. The using-declarations make the
73		// inherited members visible to this template's own unqualified
74		// references below (two-phase lookup).
75		using native_socket_base<Derived, ImplBase, Endpoint>::fd_;
76		using native_socket_base<Derived, ImplBase, Endpoint>::local_endpoint_;
77
78		Service& svc_;
79
80		public:
81		/// Per-descriptor state for persistent reactor registration.
82		DescState desc_state_;
83
84	20873x	~reactor_basic_socket() override = default;
85
86		/** Assign the fd, initialize descriptor state, and register with
87		the reactor.
88
89		@param fd The descriptor to adopt.
90
91		@return The error if the reactor rejects the descriptor, in
92		which case the implementation is left closed and the caller
93		retains ownership of @a fd; otherwise a default constructed
94		error code.
95		*/
96	7410x	std::error_code init_and_register(int fd) noexcept
97		{
98	7410x	fd_ = fd;
99	7410x	desc_state_.fd = fd;
100		{
101	7410x	std::lock_guard lock(desc_state_.mutex);
102	7410x	desc_state_.read_op = nullptr;
103	7410x	desc_state_.write_op = nullptr;
104	7410x	desc_state_.connect_op = nullptr;
105	7410x	}
106	7410x	if (auto ec = svc_.scheduler().register_descriptor(fd, &desc_state_))
107		{
108		// Undo the partial state so a failed adopt is
109		// indistinguishable from a closed implementation.
110	3x	fd_ = -1;
111	3x	desc_state_.fd = -1;
112	3x	desc_state_.registered_events = 0;
113	3x	return ec;
114		}
115	7407x	return {};
116		}
117
118		/** Register an op with the reactor.
119
120		Handles cached edge events. Called on the EAGAIN/EINPROGRESS
121		path when speculative I/O failed.
122		*/
123		template<class Op>
124		void register_op(
125		Op& op,
126		reactor_op_base*& desc_slot,
127		bool& ready_flag,
128		bool is_write_direction = false) noexcept;
129
130		/** Cancel a single pending operation.
131
132		Claims the operation from its descriptor_state slot under
133		the mutex and posts it to the scheduler as cancelled.
134		Derived must implement:
135		op_to_desc_slot(Op&) -> reactor_op_base**
136		*/
137		template<class Op>
138		void cancel_single_op(Op& op) noexcept;
139
140		/** Cancel all pending operations.
141
142		Invoked by the derived class's cancel() override.
143		Derived must implement:
144		for_each_op(auto fn)
145		for_each_desc_entry(auto fn)
146		*/
147		void do_cancel() noexcept;
148
149		/** Close the socket and cancel pending operations.
150
151		Invoked by the derived class's close_socket(). The
152		derived class may add backend-specific cleanup after
153		calling this method.
154		Derived must implement:
155		for_each_op(auto fn)
156		for_each_desc_entry(auto fn)
157		*/
158		void do_close_socket() noexcept;
159
160		/** Release the socket without closing the fd.
161
162		Like do_close_socket() but does not call ::close().
163		Returns the fd so the caller can take ownership.
164		*/
165		native_handle_type do_release_socket() noexcept;
166		};
167
168		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
169		template<class Op>
170		void
171	7832x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::register_op(
172		Op& op,
173		reactor_op_base*& desc_slot,
174		bool& ready_flag,
175		bool is_write_direction) noexcept
176		{
177	7832x	svc_.work_started();
178
179	7832x	std::lock_guard lock(desc_state_.mutex);
180	7832x	bool io_done = false;
181	7832x	if (ready_flag)
182		{
183	301x	ready_flag = false;
184	301x	op.perform_io();
185	301x	io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
186	301x	if (!io_done)
187	295x	op.errn = 0;
188		}
189
190
191	7832x	if (io_done \|\| op.cancelled.load(std::memory_order_acquire))
192		{
193	52x	svc_.post(&op);
194	52x	svc_.work_finished();
195		}
196		else
197		{
198	7780x	desc_slot = &op;
199
200		// Select must rebuild its fd_sets when a write-direction op
201		// is parked, so select() watches for writability. Compiled
202		// away to nothing for epoll and kqueue.
203		if constexpr (requires { Service::needs_write_notification; })
204		{
205		if constexpr (Service::needs_write_notification)
206		{
207	2859x	if (is_write_direction)
208	2351x	svc_.scheduler().notify_reactor();
209		}
210		}
211		}
212	7832x	}
213
214		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
215		template<class Op>
216		void
217	297x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::cancel_single_op(
218		Op& op) noexcept
219		{
220	297x	auto self = this->weak_from_this().lock();
221	297x	if (!self)
222	✗	return;
223
224	297x	op.request_cancel();
225
226	297x	auto* d = static_cast<Derived*>(this);
227	297x	reactor_op_base** desc_op_ptr = d->op_to_desc_slot(op);
228
229	297x	if (desc_op_ptr)
230		{
231	297x	reactor_op_base* claimed = nullptr;
232		{
233	297x	std::lock_guard lock(desc_state_.mutex);
234	297x	if (*desc_op_ptr == &op)
235	227x	claimed = std::exchange(*desc_op_ptr, nullptr);
236		// Not in the slot: request_cancel() above already set
237		// op.cancelled, which register_op consults before parking
238		// and the completion decode consults on delivery. Latching
239		// a descriptor flag here instead would outlive this op and
240		// cancel the next wait in the same direction.
241	297x	}
242	297x	if (claimed)
243		{
244	227x	op.impl_ptr = self;
245	227x	svc_.post(&op);
246	227x	svc_.work_finished();
247		}
248		}
249	297x	}
250
251		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
252		void
253	277x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
254		do_cancel() noexcept
255		{
256	277x	auto self = this->weak_from_this().lock();
257	277x	if (!self)
258	✗	return;
259
260	277x	auto* d = static_cast<Derived*>(this);
261
262	277x	d->for_each_op([](auto& op) { op.request_cancel(); });
263
264		// Claim ops under a single lock acquisition
265		struct claimed_entry
266		{
267		reactor_op_base* op = nullptr;
268		reactor_op_base* base = nullptr;
269		};
270		// Max 8 ops: conn, rd, wr, wait_rd, wait_wr, wait_er, recv_rd, send_wr
271	277x	claimed_entry claimed[8];
272	277x	int count = 0;
273
274		{
275	277x	std::lock_guard lock(desc_state_.mutex);
276	277x	d->for_each_desc_entry([&](auto& op, reactor_op_base*& desc_slot) {
277	✗	if (desc_slot == &op)
278		{
279	✗	claimed[count].op = std::exchange(desc_slot, nullptr);
280	✗	claimed[count].base = &op;
281	✗	++count;
282		}
283		});
284	277x	}
285
286	465x	for (int i = 0; i < count; ++i)
287		{
288	188x	claimed[i].base->impl_ptr = self;
289	188x	svc_.post(claimed[i].base);
290	188x	svc_.work_finished();
291		}
292	277x	}
293
294		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
295		void
296	63099x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
297		do_close_socket() noexcept
298		{
299	63099x	auto self = this->weak_from_this().lock();
300	63099x	if (self)
301		{
302	63099x	auto* d = static_cast<Derived*>(this);
303
304	63099x	d->for_each_op([](auto& op) { op.request_cancel(); });
305
306		struct claimed_entry
307		{
308		reactor_op_base* base = nullptr;
309		};
310	63099x	claimed_entry claimed[8];
311	63099x	int count = 0;
312
313		{
314	63099x	std::lock_guard lock(desc_state_.mutex);
315	63099x	d->for_each_desc_entry(
316	✗	[&](auto& /op/, reactor_op_base*& desc_slot) {
317	✗	auto* c = std::exchange(desc_slot, nullptr);
318	✗	if (c)
319		{
320	✗	claimed[count].base = c;
321	✗	++count;
322		}
323		});
324	63099x	desc_state_.read_ready = false;
325	63099x	desc_state_.write_ready = false;
326
327	63099x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
328	958x	desc_state_.impl_ref_ = self;
329	63099x	}
330
331	63155x	for (int i = 0; i < count; ++i)
332		{
333	56x	claimed[i].base->impl_ptr = self;
334	56x	svc_.post(claimed[i].base);
335	56x	svc_.work_finished();
336		}
337		}
338
339	63099x	if (fd_ >= 0)
340		{
341	13950x	if (desc_state_.registered_events != 0)
342	13948x	svc_.scheduler().deregister_descriptor(fd_);
343	13950x	::close(fd_);
344	13950x	fd_ = -1;
345		}
346
347	63099x	desc_state_.fd = -1;
348	63099x	desc_state_.registered_events = 0;
349
350	63099x	local_endpoint_ = Endpoint{};
351	63099x	}
352
353		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
354		native_handle_type
355	16x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
356		do_release_socket() noexcept
357		{
358		// Cancel pending ops (same as do_close_socket)
359	16x	auto self = this->weak_from_this().lock();
360	16x	if (self)
361		{
362	16x	auto* d = static_cast<Derived*>(this);
363
364	16x	d->for_each_op([](auto& op) { op.request_cancel(); });
365
366		struct claimed_entry
367		{
368		reactor_op_base* base = nullptr;
369		};
370	16x	claimed_entry claimed[8];
371	16x	int count = 0;
372
373		{
374	16x	std::lock_guard lock(desc_state_.mutex);
375	16x	d->for_each_desc_entry(
376	✗	[&](auto& /op/, reactor_op_base*& desc_slot) {
377	✗	auto* c = std::exchange(desc_slot, nullptr);
378	✗	if (c)
379		{
380	✗	claimed[count].base = c;
381	✗	++count;
382		}
383		});
384	16x	desc_state_.read_ready = false;
385	16x	desc_state_.write_ready = false;
386
387	16x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
388	3x	desc_state_.impl_ref_ = self;
389	16x	}
390
391	28x	for (int i = 0; i < count; ++i)
392		{
393	12x	claimed[i].base->impl_ptr = self;
394	12x	svc_.post(claimed[i].base);
395	12x	svc_.work_finished();
396		}
397		}
398
399	16x	native_handle_type released = fd_;
400
401	16x	if (fd_ >= 0)
402		{
403	16x	if (desc_state_.registered_events != 0)
404	16x	svc_.scheduler().deregister_descriptor(fd_);
405		// Do NOT close -- caller takes ownership
406	16x	fd_ = -1;
407		}
408
409	16x	desc_state_.fd = -1;
410	16x	desc_state_.registered_events = 0;
411
412	16x	local_endpoint_ = Endpoint{};
413
414	32x	return released;
415	16x	}
416
417		} // namespace boost::corosio::detail
418
419		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BASIC_SOCKET_HPP
420