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

98.7% Lines (155 / 157) 91.5% Functions (119 / 130)

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

reactor_basic_socket.hpp

f(x) Functions (130)

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/config.hpp>
14		#include <boost/corosio/detail/intrusive.hpp>
15		#include <boost/corosio/detail/native_handle.hpp>
16		#include <boost/corosio/endpoint.hpp>
17		#include <boost/corosio/native/detail/native_socket_base.hpp>
18		#include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
19		#include <boost/corosio/native/detail/make_err.hpp>
20		#include <boost/corosio/native/detail/endpoint_convert.hpp>
21
22		#include <cstring>
23		#include <memory>
24		#include <mutex>
25		#include <utility>
26
27		#include <errno.h>
28		#include <netinet/in.h>
29		#include <sys/socket.h>
30		#include <unistd.h>
31
32		namespace boost::corosio::detail {
33
34		/** CRTP base for reactor-backed socket implementations.
35
36		Extracts the shared data members, virtual overrides, and
37		cancel/close/register logic that is identical across TCP
38		(reactor_stream_socket) and UDP (reactor_datagram_socket).
39
40		Derived classes provide CRTP callbacks that enumerate their
41		specific op slots so cancel/close can iterate them generically.
42
43		@tparam Derived The concrete socket type (CRTP).
44		@tparam ImplBase The public vtable base (tcp_socket::implementation
45		or udp_socket::implementation).
46		@tparam Service The backend's service type.
47		@tparam DescState The backend's descriptor_state type.
48		@tparam Endpoint The endpoint type (endpoint or local_endpoint).
49		*/
50		template<
51		class Derived,
52		class ImplBase,
53		class Service,
54		class DescState,
55		class Endpoint = endpoint>
56		class reactor_basic_socket
57		: public native_socket_base<Derived, ImplBase, Endpoint>
58		, public intrusive_list<Derived>::node
59		{
60		friend Derived;
61
62		template<class, class, class, class, class, class, class, class, class>
63		friend class reactor_stream_socket;
64
65		template<
66		class,
67		class,
68		class,
69		class,
70		class,
71		class,
72		class,
73		class,
74		class,
75		class,
76		class>
77		friend class reactor_datagram_socket;
78
79	2765x	explicit reactor_basic_socket(Service& svc) noexcept : svc_(svc) {}
80
81		protected:
82		// fd_ / local_endpoint_ and the synchronous accessors (native_handle,
83		// is_open, set_option/get_option, set_socket/set_local_endpoint, do_bind)
84		// live in native_socket_base — the readiness/completion-agnostic base
85		// shared with io_uring's sockets. The using-declarations make the
86		// inherited members visible to this template's own unqualified
87		// references below (two-phase lookup).
88		using native_socket_base<Derived, ImplBase, Endpoint>::fd_;
89		using native_socket_base<Derived, ImplBase, Endpoint>::local_endpoint_;
90
91		Service& svc_;
92
93		public:
94		/// Per-descriptor state for persistent reactor registration.
95		DescState desc_state_;
96
97	2765x	~reactor_basic_socket() override = default;
98
99		/** Assign the fd, initialize descriptor state, and register with
100		the reactor.
101
102		@param fd The descriptor to adopt.
103
104		@return The error if the reactor rejects the descriptor, in
105		which case the implementation is left closed and the caller
106		retains ownership of @a fd; otherwise a default constructed
107		error code.
108		*/
109	5753x	std::error_code init_and_register(int fd) noexcept
110		{
111	5753x	fd_ = fd;
112	5753x	desc_state_.fd = fd;
113		{
114	5753x	std::lock_guard lock(desc_state_.mutex);
115	5753x	desc_state_.read_op = nullptr;
116	5753x	desc_state_.write_op = nullptr;
117	5753x	desc_state_.connect_op = nullptr;
118	5753x	}
119	5753x	if (auto ec = svc_.scheduler().register_descriptor(fd, &desc_state_))
120		{
121		// Undo the partial state so a failed adopt is
122		// indistinguishable from a closed implementation.
123	3x	fd_ = -1;
124	3x	desc_state_.fd = -1;
125	3x	desc_state_.registered_events = 0;
126	3x	return ec;
127		}
128	5750x	return {};
129		}
130
131		/** Register an op with the reactor.
132
133		Handles cached edge events. Called on the EAGAIN/EINPROGRESS
134		path when speculative I/O failed.
135		*/
136		template<class Op>
137		void register_op(
138		Op& op,
139		reactor_op_base*& desc_slot,
140		bool& ready_flag,
141		bool is_write_direction = false) noexcept;
142
143		/** Cancel a single pending operation.
144
145		Claims the operation from its descriptor_state slot under
146		the mutex and posts it to the scheduler as cancelled.
147		Derived must implement:
148		op_to_desc_slot(Op&) -> reactor_op_base**
149		*/
150		template<class Op>
151		void cancel_single_op(Op& op) noexcept;
152
153		/** Cancel all pending operations.
154
155		Invoked by the derived class's cancel() override.
156		Derived must implement:
157		for_each_op(auto fn)
158		for_each_desc_entry(auto fn)
159		*/
160		void do_cancel() noexcept;
161
162		/** Close the socket and cancel pending operations.
163
164		Invoked by the derived class's close_socket(). The
165		derived class may add backend-specific cleanup after
166		calling this method.
167		Derived must implement:
168		for_each_op(auto fn)
169		for_each_desc_entry(auto fn)
170		*/
171		void do_close_socket() noexcept;
172
173		/** Release the socket without closing the fd.
174
175		Like do_close_socket() but does not call ::close().
176		Returns the fd so the caller can take ownership.
177		*/
178		native_handle_type do_release_socket() noexcept;
179
180		/** Reset descriptor state for recycling.
181
182		Called by the owning service's `construct()` when popping this
183		impl back off the free list. Actively re-initializes every
184		field it owns rather than trusting `close_socket()`'s prior
185		writes to still be there — `poison()` below depends on that,
186		and a future field added to either side without the other
187		would otherwise go unnoticed until this one re-sets it.
188		`object_ref_` and `is_enqueued_` are asserted instead: `poison()`
189		never touches them, so they must already hold their
190		zero-action-complete state.
191
192		@pre refs_ == 0, fd closed and deregistered, no op in flight.
193		*/
194	13287x	void reuse() noexcept
195		{
196	13287x	fd_ = -1;
197	13287x	local_endpoint_ = Endpoint{};
198	13287x	desc_state_.fd = -1;
199	13287x	desc_state_.registered_events = 0;
200	13287x	desc_state_.read_op = nullptr;
201	13287x	desc_state_.write_op = nullptr;
202	13287x	desc_state_.connect_op = nullptr;
203	13287x	desc_state_.wait_read_op = nullptr;
204	13287x	desc_state_.wait_write_op = nullptr;
205	13287x	desc_state_.wait_error_op = nullptr;
206	13287x	desc_state_.read_ready = false;
207	13287x	desc_state_.write_ready = false;
208	13287x	BOOST_COROSIO_ASSERT(!desc_state_.object_ref_);
209	13287x	BOOST_COROSIO_ASSERT(!desc_state_.is_enqueued_.load(
210		std::memory_order_relaxed));
211	13287x	}
212
213		#if !defined(NDEBUG)
214		/** Poison the fields `reuse()` re-initializes, before the pool
215		parks this impl on the free list.
216
217		A field `reuse()` forgets to reset then reads back as this
218		unmistakable byte pattern instead of silently matching
219		whatever `close_socket()` happened to leave, turning a missed
220		reset into an immediate, loud failure rather than a latent
221		stale-state bug. Does not touch the mutex, `ready_events_` /
222		`is_enqueued_` (atomics), `scheduler_`, `object_ref_`, the
223		`intrusive_list` hook, or `refs_`: `recycle()` still reads
224		some of those after this call returns, and smashing the rest
225		would be undefined rather than diagnostic.
226
227		@pre refs_ == 0, fd closed and deregistered, no op in flight.
228		*/
229	16047x	void poison() noexcept
230		{
231	2712x	auto smash = [](auto& field) {
232	2712x	std::memset(static_cast<void*>(&field), 0xDB, sizeof(field));
233		};
234	16047x	smash(fd_);
235	16047x	smash(local_endpoint_);
236	16047x	smash(desc_state_.fd);
237	16047x	smash(desc_state_.registered_events);
238	16047x	smash(desc_state_.read_op);
239	16047x	smash(desc_state_.write_op);
240	16047x	smash(desc_state_.connect_op);
241	16047x	smash(desc_state_.wait_read_op);
242	16047x	smash(desc_state_.wait_write_op);
243	16047x	smash(desc_state_.wait_error_op);
244	16047x	smash(desc_state_.read_ready);
245	16047x	smash(desc_state_.write_ready);
246	16047x	}
247		#endif
248		};
249
250		template<
251		class Derived,
252		class ImplBase,
253		class Service,
254		class DescState,
255		class Endpoint>
256		template<class Op>
257		void
258	6140x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
259		register_op(
260		Op& op,
261		reactor_op_base*& desc_slot,
262		bool& ready_flag,
263		bool is_write_direction) noexcept
264		{
265	6140x	svc_.work_started();
266
267	6140x	std::lock_guard lock(desc_state_.mutex);
268	6140x	bool io_done = false;
269	6140x	if (ready_flag)
270		{
271	285x	ready_flag = false;
272	285x	op.perform_io();
273	285x	io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
274	285x	if (!io_done)
275	279x	op.errn = 0;
276		}
277
278	6140x	if (io_done \|\| op.cancelled.load(std::memory_order_acquire))
279		{
280	6x	svc_.post(&op);
281	6x	svc_.work_finished();
282		}
283		else
284		{
285	6134x	desc_slot = &op;
286
287		// Select must rebuild its fd_sets when a write-direction op
288		// is parked, so select() watches for writability. Compiled
289		// away to nothing for epoll and kqueue.
290		if constexpr (requires { Service::needs_write_notification; })
291		{
292		if constexpr (Service::needs_write_notification)
293		{
294	2817x	if (is_write_direction)
295	2282x	svc_.scheduler().notify_reactor();
296		}
297		}
298		}
299	6140x	}
300
301		template<
302		class Derived,
303		class ImplBase,
304		class Service,
305		class DescState,
306		class Endpoint>
307		template<class Op>
308		void
309	239x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
310		cancel_single_op(Op& op) noexcept
311		{
312	239x	op.request_cancel();
313
314	239x	auto* d = static_cast<Derived*>(this);
315	239x	reactor_op_base** desc_op_ptr = d->op_to_desc_slot(op);
316
317	239x	if (desc_op_ptr)
318		{
319	239x	reactor_op_base* claimed = nullptr;
320		{
321	239x	std::lock_guard lock(desc_state_.mutex);
322	239x	if (*desc_op_ptr == &op)
323	229x	claimed = std::exchange(*desc_op_ptr, nullptr);
324		// Not in the slot: request_cancel() above already set
325		// op.cancelled, which register_op consults before parking
326		// and the completion decode consults on delivery. Latching
327		// a descriptor flag here instead would outlive this op and
328		// cancel the next wait in the same direction.
329	239x	}
330	239x	if (claimed)
331		{
332	229x	op.object_ref_ = detail::object_ref(this);
333	229x	svc_.post(&op);
334	229x	svc_.work_finished();
335		}
336		}
337	239x	}
338
339		template<
340		class Derived,
341		class ImplBase,
342		class Service,
343		class DescState,
344		class Endpoint>
345		void
346	283x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
347		do_cancel() noexcept
348		{
349	283x	auto* d = static_cast<Derived*>(this);
350
351	293x	d->for_each_op([](auto& op) { op.request_cancel(); });
352
353		// Claim ops under a single lock acquisition
354		struct claimed_entry
355		{
356		reactor_op_base* op = nullptr;
357		reactor_op_base* base = nullptr;
358		};
359		// Max 8 ops: conn, rd, wr, wait_rd, wait_wr, wait_er, recv_rd, send_wr
360	283x	claimed_entry claimed[8];
361	283x	int count = 0;
362
363		{
364	283x	std::lock_guard lock(desc_state_.mutex);
365	305x	d->for_each_desc_entry([&](auto& op, reactor_op_base*& desc_slot) {
366	12x	if (desc_slot == &op)
367		{
368	1x	claimed[count].op = std::exchange(desc_slot, nullptr);
369	1x	claimed[count].base = &op;
370	1x	++count;
371		}
372		});
373	283x	}
374
375	479x	for (int i = 0; i < count; ++i)
376		{
377	196x	claimed[i].base->object_ref_ = detail::object_ref(this);
378	196x	svc_.post(claimed[i].base);
379	196x	svc_.work_finished();
380		}
381	283x	}
382
383		template<
384		class Derived,
385		class ImplBase,
386		class Service,
387		class DescState,
388		class Endpoint>
389		void
390	48764x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
391		do_close_socket() noexcept
392		{
393	48764x	auto* d = static_cast<Derived*>(this);
394
395	52249x	d->for_each_op([](auto& op) { op.request_cancel(); });
396
397		struct claimed_entry
398		{
399		reactor_op_base* base = nullptr;
400		};
401	48764x	claimed_entry claimed[8];
402	48764x	int count = 0;
403
404		{
405	48764x	std::lock_guard lock(desc_state_.mutex);
406	48764x	d->for_each_desc_entry(
407	8364x	[&](auto& /op/, reactor_op_base*& desc_slot) {
408	4182x	auto* c = std::exchange(desc_slot, nullptr);
409	4182x	if (c)
410		{
411	✗	claimed[count].base = c;
412	✗	++count;
413		}
414		});
415	48764x	desc_state_.read_ready = false;
416	48764x	desc_state_.write_ready = false;
417
418	48764x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
419	1333x	desc_state_.object_ref_ = detail::object_ref(this);
420	48764x	}
421
422	48822x	for (int i = 0; i < count; ++i)
423		{
424	58x	claimed[i].base->object_ref_ = detail::object_ref(this);
425	58x	svc_.post(claimed[i].base);
426	58x	svc_.work_finished();
427		}
428
429	48764x	if (fd_ >= 0)
430		{
431	10602x	if (desc_state_.registered_events != 0)
432	10600x	svc_.scheduler().deregister_descriptor(fd_);
433	10602x	::close(fd_);
434	10602x	fd_ = -1;
435		}
436
437	48764x	desc_state_.fd = -1;
438	48764x	desc_state_.registered_events = 0;
439
440	48764x	local_endpoint_ = Endpoint{};
441	48764x	}
442
443		template<
444		class Derived,
445		class ImplBase,
446		class Service,
447		class DescState,
448		class Endpoint>
449		native_handle_type
450	16x	reactor_basic_socket<Derived, ImplBase, Service, DescState, Endpoint>::
451		do_release_socket() noexcept
452		{
453		// Cancel pending ops (same as do_close_socket)
454	16x	auto* d = static_cast<Derived*>(this);
455
456	128x	d->for_each_op([](auto& op) { op.request_cancel(); });
457
458		struct claimed_entry
459		{
460		reactor_op_base* base = nullptr;
461		};
462	16x	claimed_entry claimed[8];
463	16x	int count = 0;
464
465		{
466	16x	std::lock_guard lock(desc_state_.mutex);
467	16x	d->for_each_desc_entry(
468	224x	[&](auto& /op/, reactor_op_base*& desc_slot) {
469	112x	auto* c = std::exchange(desc_slot, nullptr);
470	112x	if (c)
471		{
472	12x	claimed[count].base = c;
473	12x	++count;
474		}
475		});
476	16x	desc_state_.read_ready = false;
477	16x	desc_state_.write_ready = false;
478
479	16x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
480	3x	desc_state_.object_ref_ = detail::object_ref(this);
481	16x	}
482
483	28x	for (int i = 0; i < count; ++i)
484		{
485	12x	claimed[i].base->object_ref_ = detail::object_ref(this);
486	12x	svc_.post(claimed[i].base);
487	12x	svc_.work_finished();
488		}
489
490	16x	native_handle_type released = fd_;
491
492	16x	if (fd_ >= 0)
493		{
494	16x	if (desc_state_.registered_events != 0)
495	16x	svc_.scheduler().deregister_descriptor(fd_);
496		// Do NOT close -- caller takes ownership
497	16x	fd_ = -1;
498		}
499
500	16x	desc_state_.fd = -1;
501	16x	desc_state_.registered_events = 0;
502
503	16x	local_endpoint_ = Endpoint{};
504
505	16x	return released;
506		}
507
508		} // namespace boost::corosio::detail
509
510		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_BASIC_SOCKET_HPP
511