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