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

96.7% Lines (118 / 122) 98.1% Functions (206 / 210)

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

reactor_io_core.hpp

f(x) Functions (210)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Michael Vandeberg
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_IO_CORE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_IO_CORE_HPP
12
13		#include <boost/corosio/detail/config.hpp>
14		#include <boost/corosio/detail/object_ref.hpp>
15		#include <boost/corosio/native/detail/reactor/reactor_op_base.hpp>
16
17		#include <atomic>
18		#include <cstring>
19		#include <mutex>
20		#include <system_error>
21		#include <utility>
22
23		#include <errno.h>
24
25		/* Shared reactor I/O protocol.
26
27		One implementation of the register/park/cancel/teardown protocol
28		for every reactor-backed object -- stream and datagram sockets,
29		acceptors, and posix descriptors -- over a backend's
30		descriptor_state. Asio keeps the same protocol in its reactor
31		(start_op, cancel_ops, deregister_descriptor); services and the
32		objects' own verbs stay per type.
33
34		Derived supplies its op slots through for_each_op,
35		for_each_desc_entry and op_to_desc_slot, and is an
36		io_object::implementation whose callers hold a reference on it.
37		*/
38
39		namespace boost::corosio::detail {
40
41		/** CRTP base holding the reactor parking and cancel protocol.
42
43		@tparam Derived The concrete object type (CRTP).
44		@tparam Service The backend service that owns Derived.
45		@tparam DescState The backend's descriptor_state type.
46		*/
47		template<class Derived, class Service, class DescState>
48		class reactor_io_core
49		{
50	229287x	Derived* self_ptr() noexcept
51		{
52	229287x	return static_cast<Derived*>(this);
53		}
54
55		protected:
56		// NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
57	7572x	explicit reactor_io_core(Service& svc) noexcept : svc_(svc) {}
58
59		Service& svc_;
60
61		public:
62		/// Per-descriptor state for persistent reactor registration.
63		DescState desc_state_;
64
65		/** Cancel a single pending operation.
66
67		Claims the operation from its descriptor_state slot under
68		the mutex and posts it to the scheduler as cancelled.
69		*/
70		template<class Op>
71	676x	void cancel_single_op(Op& op) noexcept
72		{
73	676x	op.request_cancel();
74
75	676x	reactor_op_base** desc_op_ptr = self_ptr()->op_to_desc_slot(op);
76	676x	if (!desc_op_ptr)
77	✗	return;
78
79	676x	reactor_op_base* claimed = nullptr;
80		{
81	676x	std::lock_guard lock(desc_state_.mutex);
82	676x	if (*desc_op_ptr == &op)
83		{
84	544x	claimed = std::exchange(*desc_op_ptr, nullptr);
85	544x	svc_.scheduler().cancel_posting_begin();
86		}
87		// Not in the slot: request_cancel() above already set
88		// op.cancelled, which register_op consults before parking
89		// and the completion decode consults on delivery. Latching
90		// a descriptor flag here instead would outlive this op and
91		// cancel the next wait in the same direction.
92	676x	}
93	676x	if (claimed)
94		{
95	544x	op.object_ref_ = detail::object_ref(self_ptr());
96	544x	svc_.scheduler().post_counted(&op);
97	544x	svc_.scheduler().cancel_posting_end();
98		}
99		}
100
101		protected:
102		/** Clear every slot and register @a fd with the reactor.
103
104		@return The reactor's refusal, in which case the state is left
105		unregistered.
106		*/
107	14253x	std::error_code register_fd(int fd) noexcept
108		{
109		{
110		// A completion from an earlier registration may still read
111		// these under the mutex.
112	14253x	std::lock_guard lock(desc_state_.mutex);
113	14253x	desc_state_.fd = fd;
114	14253x	self_ptr()->for_each_desc_entry(
115		[](auto&, reactor_op_base*& slot) { slot = nullptr; });
116	14253x	}
117	14253x	if (auto ec = svc_.scheduler().register_descriptor(fd, &desc_state_))
118		{
119	4x	std::lock_guard lock(desc_state_.mutex);
120	4x	desc_state_.fd = -1;
121	4x	desc_state_.registered_events = 0;
122	4x	return ec;
123	4x	}
124	14249x	return {};
125		}
126
127		/** Register an op with the reactor.
128
129		Handles cached edge events. Called on the EAGAIN/EINPROGRESS
130		path when speculative I/O failed.
131		*/
132		template<class Op>
133	10346x	void register_op(
134		Op& op,
135		reactor_op_base*& desc_slot,
136		bool& ready_flag,
137		bool is_write_direction = false) noexcept
138		{
139	10346x	svc_.work_started();
140	10346x	if (park_op(op, desc_slot, ready_flag, is_write_direction))
141	10339x	return;
142		// Posted after the descriptor lock is released: the scheduler
143		// takes its own lock before descriptor locks.
144	7x	svc_.scheduler().post_counted(&op);
145		}
146
147		/// Cancel every pending operation.
148	367x	void cancel_all() noexcept
149		{
150	367x	self_ptr()->for_each_op([](auto& op) { op.request_cancel(); });
151
152		reactor_op_base* claimed[max_claimed];
153	367x	int count = 0;
154		{
155	367x	std::lock_guard lock(desc_state_.mutex);
156	367x	self_ptr()->for_each_desc_entry(
157		[&](auto& op, reactor_op_base*& desc_slot) {
158		if (desc_slot == &op)
159		{
160		BOOST_COROSIO_ASSERT(count < max_claimed);
161		claimed[count++] = std::exchange(desc_slot, nullptr);
162		}
163		});
164	367x	}
165	367x	post_claimed(claimed, count);
166	367x	}
167
168		/** Cancel every operation, drop the reactor registration, and
169		claim every parked op for teardown.
170
171		Deregistration hands a reference to the scheduler, which keeps
172		the object alive while a reactor pass may still deliver an
173		event for the descriptor. Afterwards the descriptor state reads
174		as closed, so the caller may close or release the fd.
175		*/
176	94331x	void abandon_all() noexcept
177		{
178	94331x	self_ptr()->for_each_op([](auto& op) { op.request_cancel(); });
179
180	94331x	int fd = -1;
181	94331x	bool registered = false;
182		{
183	94331x	std::lock_guard lock(desc_state_.mutex);
184	94331x	fd = desc_state_.fd;
185	94331x	registered = desc_state_.registered_events != 0;
186	94331x	}
187	94331x	if (fd >= 0 && registered)
188		{
189	22997x	svc_.scheduler().deregister_descriptor(fd);
190	45994x	svc_.scheduler().retire_descriptor(
191	45994x	desc_state_, detail::object_ref(self_ptr()));
192		}
193
194		reactor_op_base* claimed[max_claimed];
195	94331x	int count = 0;
196		{
197	94331x	std::lock_guard lock(desc_state_.mutex);
198	94331x	self_ptr()->for_each_desc_entry(
199		[&](auto& /op/, reactor_op_base*& desc_slot) {
200		if (auto* c = std::exchange(desc_slot, nullptr))
201		{
202		BOOST_COROSIO_ASSERT(count < max_claimed);
203		claimed[count++] = c;
204		}
205		});
206	94331x	desc_state_.read_ready = false;
207	94331x	desc_state_.write_ready = false;
208		// Cleared before the fd is closed: a queued completion reads
209		// it under this mutex, and the number can be reused once
210		// closed.
211	94331x	desc_state_.fd = -1;
212	94331x	desc_state_.registered_events = 0;
213	94331x	desc_state_.unpollable = false;
214	94331x	}
215	94331x	post_claimed(claimed, count);
216	94331x	}
217
218		/** Reset the descriptor state a recycled object starts from.
219
220		Every field is reset rather than asserted so a debug build may
221		poison them on retirement. References and the queued flag are
222		asserted: nothing may hold one once the count reached zero.
223		*/
224	21977x	void reset_desc_state() noexcept
225		{
226	21977x	desc_state_.fd = -1;
227	21977x	desc_state_.registered_events = 0;
228	21977x	desc_state_.unpollable = false;
229	21977x	desc_state_.read_op = nullptr;
230	21977x	desc_state_.write_op = nullptr;
231	21977x	desc_state_.connect_op = nullptr;
232	21977x	desc_state_.wait_read_op = nullptr;
233	21977x	desc_state_.wait_write_op = nullptr;
234	21977x	desc_state_.wait_error_op = nullptr;
235	21977x	desc_state_.read_ready = false;
236	21977x	desc_state_.write_ready = false;
237	21977x	BOOST_COROSIO_ASSERT(!desc_state_.object_ref_);
238	21977x	BOOST_COROSIO_ASSERT(!desc_state_.retired_ref_);
239	21977x	BOOST_COROSIO_ASSERT(!desc_state_.dropped_ref_);
240	21977x	BOOST_COROSIO_ASSERT(
241		!desc_state_.is_enqueued_.load(std::memory_order_relaxed));
242	21977x	}
243
244		#if !defined(NDEBUG)
245		/** Poison the descriptor fields `reset_desc_state()` re-initializes.
246
247		Leaves the mutex, the atomics, `scheduler_`, the references and
248		the retirement links alone: the scheduler and `recycle()` may
249		still read them.
250		*/
251	29549x	void poison_desc_state() noexcept
252		{
253		auto smash = [](auto& field) {
254		std::memset(static_cast<void*>(&field), 0xDB, sizeof(field));
255		};
256	29549x	smash(desc_state_.fd);
257	29549x	smash(desc_state_.registered_events);
258	29549x	smash(desc_state_.unpollable);
259	29549x	smash(desc_state_.read_op);
260	29549x	smash(desc_state_.write_op);
261	29549x	smash(desc_state_.connect_op);
262	29549x	smash(desc_state_.wait_read_op);
263	29549x	smash(desc_state_.wait_write_op);
264	29549x	smash(desc_state_.wait_error_op);
265	29549x	smash(desc_state_.read_ready);
266	29549x	smash(desc_state_.write_ready);
267	29549x	}
268		#endif
269
270		private:
271		// A claim empties its slot, so no more ops are claimed than
272		// descriptor_state has slots (read, write, connect, wait_read,
273		// wait_write, wait_error), however many ops share them.
274		static constexpr int max_claimed = 6;
275
276		/** Park @a op unless it can complete now.
277
278		@return True if parked; false if the caller must post it.
279		*/
280		template<class Op>
281	10346x	bool park_op(
282		Op& op,
283		reactor_op_base*& desc_slot,
284		bool& ready_flag,
285		bool is_write_direction) noexcept
286		{
287	10346x	std::lock_guard lock(desc_state_.mutex);
288	10346x	if (ready_flag)
289		{
290	228x	ready_flag = false;
291	228x	op.perform_io();
292	228x	if (op.errn != EAGAIN && op.errn != EWOULDBLOCK)
293	6x	return false;
294	222x	op.errn = 0;
295		}
296
297	10340x	if (op.cancelled.load(std::memory_order_acquire))
298	✗	return false;
299
300	10340x	if (desc_state_.unpollable)
301		{
302		// Nothing will ever report readiness for this fd.
303	1x	op.complete(EOPNOTSUPP, 0);
304	1x	return false;
305		}
306
307	10339x	if (is_write_direction)
308		{
309	8362x	if (auto ec = svc_.scheduler().ensure_write_registered(
310		desc_state_.fd, &desc_state_))
311		{
312	✗	op.complete(ec.value(), 0);
313	✗	return false;
314		}
315		}
316
317	10339x	desc_slot = &op;
318		// Select rebuilds its fd_sets from parked ops only, so parking
319		// must wake it, under the lock: once released, the op can
320		// complete and its context be destroyed. Compiled away for
321		// epoll and kqueue.
322		if constexpr (Service::needs_park_notification)
323	3778x	svc_.scheduler().notify_reactor();
324	10339x	return true;
325	10346x	}
326
327	94698x	void post_claimed(reactor_op_base** claimed, int count) noexcept
328		{
329	96119x	for (int i = 0; i < count; ++i)
330		{
331	1421x	claimed[i]->object_ref_ = detail::object_ref(self_ptr());
332	1421x	svc_.scheduler().post_counted(claimed[i]);
333		}
334	94698x	}
335		};
336
337		} // namespace boost::corosio::detail
338
339		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_IO_CORE_HPP
340