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

90.4% Lines (123/136) 76.1% List of functions (70/92)

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

reactor_basic_socket.hpp

f(x) Functions (92)

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