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

87.2% Lines (123 / 141) 78.7% Branches (74 / 94) 100.0% Functions (7 / 7)

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

reactor_descriptor_state.hpp

f(x) Functions (7)

Function Calls Lines Branches Blocks

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2026 Steve Gerbino
3			// Copyright (c) 2026 Michael Vandeberg
4			//
5			// Distributed under the Boost Software License, Version 1.0. (See accompanying
6			// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7			//
8			// Official repository: https://github.com/cppalliance/corosio
9			//
10
11			#ifndef BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
12			#define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
13
14			#include <boost/corosio/native/detail/reactor/reactor_events.hpp>
15			#include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
16			#include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
17			#include <boost/corosio/detail/ready_queue.hpp>
18
19			#include <boost/corosio/detail/conditionally_enabled_mutex.hpp>
20
21			#include <atomic>
22			#include <cstdint>
23			#include <memory>
24
25			#include <errno.h>
26			#include <sys/socket.h>
27
28			namespace boost::corosio::detail {
29
30			/** Per-descriptor state shared across reactor backends.
31
32			Tracks pending operations for a file descriptor. The fd is registered
33			once with the reactor and stays registered until closed. Uses deferred
34			I/O: the reactor sets ready_events atomically, then enqueues this state.
35			When popped by the scheduler, invoke_deferred_io() performs I/O under
36			the mutex and queues completed ops.
37
38			Non-template: uses reactor_op_base pointers so the scheduler and
39			descriptor_state code exist as a single copy in the binary regardless
40			of how many backends are compiled in.
41
42			@par Thread Safety
43			The mutex protects operation pointers and ready flags. ready_events_
44			and is_enqueued_ are atomic for lock-free reactor access.
45			*/
46		14331x	struct reactor_descriptor_state
47			: scheduler_op
48			, reactor_retirable
49			{
50			/// Protects operation pointers and ready/cancel flags.
51			/// Becomes a no-op in single-threaded mode.
52		14331x	conditionally_enabled_mutex mutex{true};
53
54			/// Pending read operation (guarded by `mutex`).
55		14331x	reactor_op_base* read_op = nullptr;
56
57			/// Pending write operation (guarded by `mutex`).
58		14331x	reactor_op_base* write_op = nullptr;
59
60			/// Pending connect operation (guarded by `mutex`).
61		14331x	reactor_op_base* connect_op = nullptr;
62
63			/// Pending wait-for-read operation (guarded by `mutex`).
64		14331x	reactor_op_base* wait_read_op = nullptr;
65
66			/// Pending wait-for-write operation (guarded by `mutex`).
67		14331x	reactor_op_base* wait_write_op = nullptr;
68
69			/// Pending wait-for-error operation (guarded by `mutex`).
70		14331x	reactor_op_base* wait_error_op = nullptr;
71
72			/// True if a read edge event arrived before an op was registered.
73		14331x	bool read_ready = false;
74
75			/// True if a write edge event arrived before an op was registered.
76		14331x	bool write_ready = false;
77
78			/// Event mask set during registration (no mutex needed).
79		14331x	std::uint32_t registered_events = 0;
80
81			/// The reactor refused to watch this fd (e.g. /dev/null on epoll);
82			/// its I/O never blocks, and an op that would park must not.
83		14331x	bool unpollable = false;
84
85			/// File descriptor this state tracks.
86		14331x	int fd = -1;
87
88			/// Accumulated ready events (set by reactor, read by scheduler).
89		14331x	std::atomic<std::uint32_t> ready_events_{0};
90
91			/// True while this state is queued in the scheduler's completed_ops.
92		14331x	std::atomic<bool> is_enqueued_{false};
93
94			/// Owning scheduler for posting completions.
95		14331x	reactor_scheduler const* scheduler_ = nullptr;
96
97			/// Prevents impl destruction while queued in the scheduler.
98			detail::object_ref object_ref_;
99
100			/// Add ready events atomically.
101			/// Release pairs with the consumer's acquire exchange on
102			/// ready_events_ so the consumer sees all flags. Acquire pairs with
103			/// registration's release store, so a reactor pass on another
104			/// thread sees the state registration initialized before handing
105			/// it to the kernel. On x86 (TSO) this compiles to the same LOCK OR
106			/// as relaxed.
107		123337x	void add_ready_events(std::uint32_t ev) noexcept
108			{
109		123337x	ready_events_.fetch_or(ev, std::memory_order_acq_rel);
110		123337x	}
111
112			/// Invoke deferred I/O and dispatch completions.
113		122894x	void operator()() override
114			{
115		122894x	invoke_deferred_io();
116		122894x	}
117
118			/// Move @p keep into this state if it is queued without one.
119		35328x	void hand_off(object_ref& keep) noexcept override
120			{
121			// Locked even to read the flag: a popped invocation clears it at
122			// the start of its critical section and still uses this state
123			// until it unlocks, so an unlocked false does not mean idle.
124	1/2 ✓ Branch 0 taken 35328 times. ✗ Branch 1 not taken.	35328x	conditionally_enabled_mutex::scoped_lock lock(mutex);
125	3/4 ✓ Branch 0 taken 551 times. ✓ Branch 1 taken 34777 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 551 times.	35328x	if (is_enqueued_.load(std::memory_order_acquire) && !object_ref_)
126		551x	object_ref_ = std::move(keep);
127		35328x	}
128
129			/// Destroy without invoking.
130			/// Called during scheduler::shutdown() drain. Clear object_ref_ to break
131			/// the self-referential cycle set by close_socket().
132		156x	void destroy() override
133			{
134		156x	object_ref_.reset();
135		156x	}
136
137			/** Perform deferred I/O and queue completions.
138
139			Performs I/O under the mutex and queues completed ops. EAGAIN
140			ops stay parked in their slot for re-delivery on the next
141			edge event.
142			*/
143			void invoke_deferred_io();
144			};
145
146			inline void
147		122897x	reactor_descriptor_state::invoke_deferred_io()
148			{
149		122897x	detail::object_ref prevent_impl_destruction;
150		122897x	ready_queue local_ops;
151			// Read under the lock: once it drops, a retired owner may be
152			// recycled, so nothing below touches this state.
153		122897x	reactor_scheduler const* sched = nullptr;
154
155			{
156	2/2 ✓ Branch 0 taken 122893 times. ✓ Branch 1 taken 4 times.	122897x	conditionally_enabled_mutex::scoped_lock lock(mutex);
157		122893x	sched = scheduler_;
158
159			// Must clear is_enqueued_ and move object_ref_ under the same
160			// lock that processes I/O. close_socket() checks is_enqueued_
161			// under this mutex — without atomicity between the flag store
162			// and the ref move, close_socket() could see is_enqueued_==false,
163			// skip setting object_ref_, and destroy the impl under us.
164		122893x	prevent_impl_destruction = std::move(object_ref_);
165		122893x	is_enqueued_.store(false, std::memory_order_release);
166
167		122893x	std::uint32_t ev = ready_events_.exchange(0, std::memory_order_acquire);
168	1/2 ✓ Branch 0 taken 122893 times. ✗ Branch 1 not taken.	122893x	if (ev == 0)
169			{
170			// Mutex unlocks here; compensate for work_cleanup's decrement
171		✗	scheduler_->compensating_work_started();
172		✗	return;
173			}
174
175		122893x	int err = 0;
176	2/2 ✓ Branch 0 taken 121796 times. ✓ Branch 1 taken 1097 times.	122893x	if (ev & reactor_event_error)
177			{
178			// Force the read/write dispatch below to run: an
179			// edge-triggered EPOLLERR can arrive alone, and without this
180			// a parked op never calls perform_io() and, the edge being
181			// one-shot, never gets another chance -- a permanent hang.
182			// Every parked op then re-runs its own syscall or probe.
183			//
184			// Assumes at least one parked op's own syscall makes
185			// non-EAGAIN progress; if every op re-parks with EAGAIN this
186			// sticky error is never redelivered and they hang. No such
187			// case is known -- a future descriptor type that hits one
188			// should be handled here.
189		1097x	ev \|= reactor_event_read \| reactor_event_write;
190
191			// SO_ERROR clears on read, so take it only for a parked op
192			// that reports it. A readiness wait reports readiness and
193			// leaves the error for the next read or write to name, as
194			// asio does; reading it here with nothing to report it to
195			// would turn a reset into a clean EOF.
196		1097x	bool const reports_error =
197	6/6 ✓ Branch 0 taken 221 times. ✓ Branch 1 taken 876 times. ✓ Branch 2 taken 220 times. ✓ Branch 3 taken 1 time. ✓ Branch 4 taken 69 times. ✓ Branch 5 taken 151 times.	1097x	read_op \|\| write_op \|\| connect_op \|\| wait_error_op;
198		1097x	socklen_t len = sizeof(err);
199	4/4 ✓ Branch 0 taken 148 times. ✓ Branch 1 taken 949 times. ✓ Branch 2 taken 4 times. ✓ Branch 3 taken 945 times.	2046x	if (reports_error &&
200	1/2 ✓ Branch 0 taken 949 times. ✗ Branch 1 not taken.	949x	::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
201			{
202			// Non-socket fd (pipe, chardev, ...): no SO_ERROR, so
203			// let the op's own syscall name the real failure.
204	3/6 ✓ Branch 0 taken 4 times. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✓ Branch 3 taken 4 times. ✓ Branch 4 taken 4 times. ✗ Branch 5 not taken.	4x	err = (errno == ENOTSOCK) ? 0 : errno;
205		4x	}
206			// select raises its exceptional set for out-of-band/urgent
207			// data as well as for genuine faults; on a healthy socket the
208			// probe then reads SO_ERROR == 0. Faulting a pending read or
209			// write on that is wrong, so an I/O operation completes only
210			// on a real (non-zero) error. wait(error) still names a code
211			// below.
212		1097x	}
213
214	2/2 ✓ Branch 0 taken 16477 times. ✓ Branch 1 taken 106416 times.	122893x	if (ev & reactor_event_read)
215			{
216	2/2 ✓ Branch 0 taken 6612 times. ✓ Branch 1 taken 99804 times.	106416x	if (read_op)
217			{
218		99804x	auto* rd = read_op;
219	2/2 ✓ Branch 0 taken 874 times. ✓ Branch 1 taken 98930 times.	99804x	if (err)
220		874x	rd->complete(err, 0);
221			else
222		98930x	rd->perform_io();
223
224	3/4 ✓ Branch 0 taken 99604 times. ✓ Branch 1 taken 200 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 99604 times.	99804x	if (rd->errn == EAGAIN \|\| rd->errn == EWOULDBLOCK)
225			{
226		200x	rd->errn = 0;
227		200x	}
228			else
229			{
230		99604x	read_op = nullptr;
231		99604x	local_ops.push(rd);
232			}
233		99804x	}
234			else
235			{
236		6612x	read_ready = true;
237			}
238
239			// The event does not prove the socket is still readable: a
240			// parked read op above may have drained it, or a speculative
241			// read consumed the data before this dispatch ran. The wait
242			// op's perform_io() re-probes and reports EAGAIN to stay
243			// parked.
244	2/2 ✓ Branch 0 taken 106378 times. ✓ Branch 1 taken 38 times.	106416x	if (wait_read_op)
245			{
246		38x	auto* wo = wait_read_op;
247		38x	wo->perform_io();
248
249	2/4 ✓ Branch 0 taken 38 times. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✓ Branch 3 taken 38 times.	38x	if (wo->errn == EAGAIN \|\| wo->errn == EWOULDBLOCK)
250			{
251		✗	wo->errn = 0;
252		✗	}
253			else
254			{
255		38x	wait_read_op = nullptr;
256		38x	local_ops.push(wo);
257			}
258		38x	}
259		106416x	}
260	2/2 ✓ Branch 0 taken 105155 times. ✓ Branch 1 taken 17738 times.	122893x	if (ev & reactor_event_write)
261			{
262	2/2 ✓ Branch 0 taken 3182 times. ✓ Branch 1 taken 14556 times.	17738x	bool had_write_op = (connect_op \|\| write_op);
263			// A writable event on a socket still in SYN_SENT (e.g. the
264			// spurious pre-connect readiness of a fresh socket) must
265			// not complete the connect; perform_io() reports EAGAIN
266			// until a peer is actually established.
267	2/2 ✓ Branch 0 taken 3178 times. ✓ Branch 1 taken 14560 times.	17738x	if (connect_op)
268			{
269		14560x	auto* cn = connect_op;
270	2/2 ✓ Branch 0 taken 69 times. ✓ Branch 1 taken 14491 times.	14560x	if (err)
271		69x	cn->complete(err, 0);
272			else
273		14491x	cn->perform_io();
274
275	3/4 ✓ Branch 0 taken 14564 times. ✓ Branch 1 taken 4 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 14564 times.	14560x	if (cn->errn == EAGAIN \|\| cn->errn == EWOULDBLOCK)
276			{
277		4x	cn->errn = 0;
278		4x	}
279			else
280			{
281		14564x	connect_op = nullptr;
282		14564x	local_ops.push(cn);
283			}
284		14568x	}
285	2/2 ✓ Branch 0 taken 16718 times. ✓ Branch 1 taken 1028 times.	17746x	if (write_op)
286			{
287		1028x	auto* wr = write_op;
288	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 1028 times.	1028x	if (err)
289		✗	wr->complete(err, 0);
290			else
291		1028x	wr->perform_io();
292
293	2/4 ✓ Branch 0 taken 1028 times. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✓ Branch 3 taken 1028 times.	1028x	if (wr->errn == EAGAIN \|\| wr->errn == EWOULDBLOCK)
294			{
295		✗	wr->errn = 0;
296		✗	}
297			else
298			{
299		1028x	write_op = nullptr;
300		1028x	local_ops.push(wr);
301			}
302		1028x	}
303	2/2 ✓ Branch 0 taken 2154 times. ✓ Branch 1 taken 15592 times.	17746x	if (!had_write_op)
304		2154x	write_ready = true;
305
306			// Same re-probe discipline as the wait-for-read dispatch.
307	2/2 ✓ Branch 0 taken 17739 times. ✓ Branch 1 taken 7 times.	17746x	if (wait_write_op)
308			{
309		7x	auto* wo = wait_write_op;
310		7x	wo->perform_io();
311
312	2/4 ✓ Branch 0 taken 7 times. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✓ Branch 3 taken 7 times.	7x	if (wo->errn == EAGAIN \|\| wo->errn == EWOULDBLOCK)
313			{
314		✗	wo->errn = 0;
315		✗	}
316			else
317			{
318		7x	wait_write_op = nullptr;
319		7x	local_ops.push(wo);
320			}
321		7x	}
322		17746x	}
323			// Complete a parked wait-for-error on any error condition.
324	2/2 ✓ Branch 0 taken 121804 times. ✓ Branch 1 taken 1097 times.	122901x	if (ev & reactor_event_error)
325			{
326	2/2 ✓ Branch 0 taken 1094 times. ✓ Branch 1 taken 3 times.	1097x	if (wait_error_op)
327			{
328			// wait(error) fired on the exceptional condition; name a
329			// code even when the kernel exposed none (e.g. urgent
330			// data leaves SO_ERROR == 0).
331	1/2 ✓ Branch 0 taken 3 times. ✗ Branch 1 not taken.	3x	int const werr = err ? err : EIO;
332		3x	wait_error_op->complete(werr, 0);
333		3x	local_ops.push(std::exchange(wait_error_op, nullptr));
334		3x	}
335		1097x	}
336	2/2 ✓ Branch 0 taken 121955 times. ✓ Branch 1 taken 946 times.	122901x	if (err)
337			{
338	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 946 times.	946x	if (read_op)
339			{
340		✗	read_op->complete(err, 0);
341		✗	local_ops.push(std::exchange(read_op, nullptr));
342		✗	}
343	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 946 times.	946x	if (write_op)
344			{
345		✗	write_op->complete(err, 0);
346		✗	local_ops.push(std::exchange(write_op, nullptr));
347		✗	}
348	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 946 times.	946x	if (connect_op)
349			{
350		✗	connect_op->complete(err, 0);
351		✗	local_ops.push(std::exchange(connect_op, nullptr));
352		✗	}
353		946x	}
354		122901x	}
355
356			// Execute first handler inline — the scheduler's work_cleanup
357			// accounts for this as the "consumed" work item. local_ops holds
358			// only ops, so the popped entry decodes directly.
359		122897x	scheduler_op* first = ready_as_op(local_ops.pop());
360	2/2 ✓ Branch 0 taken 115238 times. ✓ Branch 1 taken 7659 times.	122897x	if (first)
361			{
362	2/2 ✓ Branch 0 taken 115239 times. ✓ Branch 1 taken 1 time.	115238x	sched->post_deferred_completions(local_ops);
363	2/2 ✓ Branch 0 taken 115236 times. ✓ Branch 1 taken 3 times.	115239x	(*first)();
364		115236x	}
365			else
366			{
367		7659x	sched->compensating_work_started();
368			}
369		122907x	}
370
371			} // namespace boost::corosio::detail
372
373			#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_STATE_HPP
374