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

88.1% Lines (133/0/151) 71.7% List of functions (86/0/120)

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

reactor_basic_socket.hpp

f(x) Functions (120)

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	14045x	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	14045x	~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	5002x	std::error_code init_and_register(int fd) noexcept
97		{
98	5002x	fd_ = fd;
99	5002x	desc_state_.fd = fd;
100		{
101	5002x	std::lock_guard lock(desc_state_.mutex);
102	5002x	desc_state_.read_op = nullptr;
103	5002x	desc_state_.write_op = nullptr;
104	5002x	desc_state_.connect_op = nullptr;
105	5002x	}
106	5002x	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	1x	fd_ = -1;
111	1x	desc_state_.fd = -1;
112	1x	desc_state_.registered_events = 0;
113	1x	return ec;
114		}
115	5001x	return {};
116		}
117
118		/** Register an op with the reactor.
119
120		Handles cached edge events and deferred cancellation.
121		Called on the EAGAIN/EINPROGRESS path when speculative
122		I/O failed.
123		*/
124		template<class Op>
125		void register_op(
126		Op& op,
127		reactor_op_base*& desc_slot,
128		bool& ready_flag,
129		bool& cancel_flag,
130		bool is_write_direction = false) noexcept;
131
132		/** Cancel a single pending operation.
133
134		Claims the operation from its descriptor_state slot under
135		the mutex and posts it to the scheduler as cancelled.
136		Derived must implement:
137		op_to_desc_slot(Op&) -> reactor_op_base**
138		op_to_cancel_flag(Op&) -> bool*
139		*/
140		template<class Op>
141		void cancel_single_op(Op& op) noexcept;
142
143		/** Cancel all pending operations.
144
145		Invoked by the derived class's cancel() override.
146		Derived must implement:
147		for_each_op(auto fn)
148		for_each_desc_entry(auto fn)
149		*/
150		void do_cancel() noexcept;
151
152		/** Close the socket and cancel pending operations.
153
154		Invoked by the derived class's close_socket(). The
155		derived class may add backend-specific cleanup after
156		calling this method.
157		Derived must implement:
158		for_each_op(auto fn)
159		for_each_desc_entry(auto fn)
160		*/
161		void do_close_socket() noexcept;
162
163		/** Release the socket without closing the fd.
164
165		Like do_close_socket() but does not call ::close().
166		Returns the fd so the caller can take ownership.
167		*/
168		native_handle_type do_release_socket() noexcept;
169		};
170
171		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
172		template<class Op>
173		void
174	5499x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::register_op(
175		Op& op,
176		reactor_op_base*& desc_slot,
177		bool& ready_flag,
178		bool& cancel_flag,
179		bool is_write_direction) noexcept
180		{
181	5499x	svc_.work_started();
182
183	5499x	std::lock_guard lock(desc_state_.mutex);
184	5499x	bool io_done = false;
185	5499x	if (ready_flag)
186		{
187	259x	ready_flag = false;
188	259x	op.perform_io();
189	259x	io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
190	259x	if (!io_done)
191	259x	op.errn = 0;
192		}
193
194	5499x	if (cancel_flag)
195		{
196	✗	cancel_flag = false;
197	✗	op.cancelled.store(true, std::memory_order_relaxed);
198		}
199
200	5499x	if (io_done \|\| op.cancelled.load(std::memory_order_acquire))
201		{
202	26x	svc_.post(&op);
203	26x	svc_.work_finished();
204		}
205		else
206		{
207	5473x	desc_slot = &op;
208
209		// Select must rebuild its fd_sets when a write-direction op
210		// is parked, so select() watches for writability. Compiled
211		// away to nothing for epoll and kqueue.
212		if constexpr (requires { Service::needs_write_notification; })
213		{
214		if constexpr (Service::needs_write_notification)
215		{
216	2677x	if (is_write_direction)
217	2242x	svc_.scheduler().notify_reactor();
218		}
219		}
220		}
221	5499x	}
222
223		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
224		template<class Op>
225		void
226	269x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::cancel_single_op(
227		Op& op) noexcept
228		{
229	269x	auto self = this->weak_from_this().lock();
230	269x	if (!self)
231	✗	return;
232
233	269x	op.request_cancel();
234
235	269x	auto* d = static_cast<Derived*>(this);
236	269x	reactor_op_base** desc_op_ptr = d->op_to_desc_slot(op);
237
238	269x	if (desc_op_ptr)
239		{
240	269x	reactor_op_base* claimed = nullptr;
241		{
242	269x	std::lock_guard lock(desc_state_.mutex);
243	269x	if (*desc_op_ptr == &op)
244	225x	claimed = std::exchange(*desc_op_ptr, nullptr);
245		// Not in the slot: request_cancel() above already set
246		// op.cancelled, which register_op consults before parking
247		// and the completion decode consults on delivery. Latching
248		// a descriptor flag here instead would outlive this op and
249		// cancel the next wait in the same direction.
250	269x	}
251	269x	if (claimed)
252		{
253	225x	op.impl_ptr = self;
254	225x	svc_.post(&op);
255	225x	svc_.work_finished();
256		}
257		}
258	269x	}
259
260		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
261		void
262	224x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
263		do_cancel() noexcept
264		{
265	224x	auto self = this->weak_from_this().lock();
266	224x	if (!self)
267	✗	return;
268
269	224x	auto* d = static_cast<Derived*>(this);
270
271	224x	d->for_each_op([](auto& op) { op.request_cancel(); });
272
273		// Claim ops under a single lock acquisition
274		struct claimed_entry
275		{
276		reactor_op_base* op = nullptr;
277		reactor_op_base* base = nullptr;
278		};
279		// Max 8 ops: conn, rd, wr, wait_rd, wait_wr, wait_er, recv_rd, send_wr
280	224x	claimed_entry claimed[8];
281	224x	int count = 0;
282
283		{
284	224x	std::lock_guard lock(desc_state_.mutex);
285	224x	d->for_each_desc_entry([&](auto& op, reactor_op_base*& desc_slot) {
286	✗	if (desc_slot == &op)
287		{
288	✗	claimed[count].op = std::exchange(desc_slot, nullptr);
289	✗	claimed[count].base = &op;
290	✗	++count;
291		}
292		});
293	224x	}
294
295	357x	for (int i = 0; i < count; ++i)
296		{
297	133x	claimed[i].base->impl_ptr = self;
298	133x	svc_.post(claimed[i].base);
299	133x	svc_.work_finished();
300		}
301	224x	}
302
303		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
304		void
305	42468x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
306		do_close_socket() noexcept
307		{
308	42468x	auto self = this->weak_from_this().lock();
309	42468x	if (self)
310		{
311	42468x	auto* d = static_cast<Derived*>(this);
312
313	42468x	d->for_each_op([](auto& op) { op.request_cancel(); });
314
315		struct claimed_entry
316		{
317		reactor_op_base* base = nullptr;
318		};
319	42468x	claimed_entry claimed[8];
320	42468x	int count = 0;
321
322		{
323	42468x	std::lock_guard lock(desc_state_.mutex);
324	42468x	d->for_each_desc_entry(
325	✗	[&](auto& /op/, reactor_op_base*& desc_slot) {
326	✗	auto* c = std::exchange(desc_slot, nullptr);
327	✗	if (c)
328		{
329	✗	claimed[count].base = c;
330	✗	++count;
331		}
332		});
333	42468x	desc_state_.read_ready = false;
334	42468x	desc_state_.write_ready = false;
335	42468x	desc_state_.read_cancel_pending = false;
336	42468x	desc_state_.write_cancel_pending = false;
337	42468x	desc_state_.connect_cancel_pending = false;
338	42468x	desc_state_.wait_read_cancel_pending = false;
339	42468x	desc_state_.wait_write_cancel_pending = false;
340	42468x	desc_state_.wait_error_cancel_pending = false;
341
342	42468x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
343	691x	desc_state_.impl_ref_ = self;
344	42468x	}
345
346	42496x	for (int i = 0; i < count; ++i)
347		{
348	28x	claimed[i].base->impl_ptr = self;
349	28x	svc_.post(claimed[i].base);
350	28x	svc_.work_finished();
351		}
352		}
353
354	42468x	if (fd_ >= 0)
355		{
356	9384x	if (desc_state_.registered_events != 0)
357	9384x	svc_.scheduler().deregister_descriptor(fd_);
358	9384x	::close(fd_);
359	9384x	fd_ = -1;
360		}
361
362	42468x	desc_state_.fd = -1;
363	42468x	desc_state_.registered_events = 0;
364
365	42468x	local_endpoint_ = Endpoint{};
366	42468x	}
367
368		template<class Derived, class ImplBase, class Service, class DescState, class Endpoint>
369		native_handle_type
370	10x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
371		do_release_socket() noexcept
372		{
373		// Cancel pending ops (same as do_close_socket)
374	10x	auto self = this->weak_from_this().lock();
375	10x	if (self)
376		{
377	10x	auto* d = static_cast<Derived*>(this);
378
379	10x	d->for_each_op([](auto& op) { op.request_cancel(); });
380
381		struct claimed_entry
382		{
383		reactor_op_base* base = nullptr;
384		};
385	10x	claimed_entry claimed[8];
386	10x	int count = 0;
387
388		{
389	10x	std::lock_guard lock(desc_state_.mutex);
390	10x	d->for_each_desc_entry(
391	✗	[&](auto& /op/, reactor_op_base*& desc_slot) {
392	✗	auto* c = std::exchange(desc_slot, nullptr);
393	✗	if (c)
394		{
395	✗	claimed[count].base = c;
396	✗	++count;
397		}
398		});
399	10x	desc_state_.read_ready = false;
400	10x	desc_state_.write_ready = false;
401	10x	desc_state_.read_cancel_pending = false;
402	10x	desc_state_.write_cancel_pending = false;
403	10x	desc_state_.connect_cancel_pending = false;
404	10x	desc_state_.wait_read_cancel_pending = false;
405	10x	desc_state_.wait_write_cancel_pending = false;
406	10x	desc_state_.wait_error_cancel_pending = false;
407
408	10x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
409	3x	desc_state_.impl_ref_ = self;
410	10x	}
411
412	16x	for (int i = 0; i < count; ++i)
413		{
414	6x	claimed[i].base->impl_ptr = self;
415	6x	svc_.post(claimed[i].base);
416	6x	svc_.work_finished();
417		}
418		}
419
420	10x	native_handle_type released = fd_;
421
422	10x	if (fd_ >= 0)
423		{
424	10x	if (desc_state_.registered_events != 0)
425	10x	svc_.scheduler().deregister_descriptor(fd_);
426		// Do NOT close -- caller takes ownership
427	10x	fd_ = -1;
428		}
429
430	10x	desc_state_.fd = -1;
431	10x	desc_state_.registered_events = 0;
432
433	10x	local_endpoint_ = Endpoint{};
434
435	20x	return released;
436	10x	}
437
438		} // namespace boost::corosio::detail
439
440		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BASIC_SOCKET_HPP
441