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

91.8% Lines (90 / 98) 98.7% Functions (152 / 154)

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

reactor_io_core.hpp

f(x) Functions (154)

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/native/detail/reactor/reactor_op_base.hpp>
15
16		#include <atomic>
17		#include <memory>
18		#include <mutex>
19		#include <system_error>
20		#include <utility>
21
22		#include <errno.h>
23
24		/* Shared reactor I/O protocol.
25
26		One implementation of the register/park/cancel/teardown protocol
27		for every reactor-backed object -- stream and datagram sockets,
28		acceptors, and posix descriptors -- over a backend's
29		descriptor_state. Asio keeps the same protocol in its reactor
30		(start_op, cancel_ops, deregister_descriptor); services and the
31		objects' own verbs stay per type.
32
33		Derived supplies its op slots through for_each_op,
34		for_each_desc_entry and op_to_desc_slot, and must derive from
35		std::enable_shared_from_this<Derived>.
36		*/
37
38		namespace boost::corosio::detail {
39
40		/** CRTP base holding the reactor parking and cancel protocol.
41
42		@tparam Derived The concrete object type (CRTP).
43		@tparam Service The backend service that owns Derived.
44		@tparam DescState The backend's descriptor_state type.
45		*/
46		template<class Derived, class Service, class DescState>
47		class reactor_io_core
48		{
49	223971x	Derived* self_ptr() noexcept
50		{
51	223971x	return static_cast<Derived*>(this);
52		}
53
54		protected:
55		// NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
56	23113x	explicit reactor_io_core(Service& svc) noexcept : svc_(svc) {}
57
58		Service& svc_;
59
60		public:
61		/// Per-descriptor state for persistent reactor registration.
62		DescState desc_state_;
63
64		/** Cancel a single pending operation.
65
66		Claims the operation from its descriptor_state slot under
67		the mutex and posts it to the scheduler as cancelled.
68		*/
69		template<class Op>
70	1067x	void cancel_single_op(Op& op) noexcept
71		{
72	1067x	auto self = self_ptr()->weak_from_this().lock();
73	1067x	if (!self)
74	✗	return;
75
76	1067x	op.request_cancel();
77
78	1067x	reactor_op_base** desc_op_ptr = self_ptr()->op_to_desc_slot(op);
79	1067x	if (!desc_op_ptr)
80	✗	return;
81
82	1067x	reactor_op_base* claimed = nullptr;
83		{
84	1067x	std::lock_guard lock(desc_state_.mutex);
85	1067x	if (*desc_op_ptr == &op)
86	1053x	claimed = std::exchange(*desc_op_ptr, nullptr);
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	1067x	}
93	1067x	if (claimed)
94		{
95	1053x	op.impl_ptr = self;
96	1053x	svc_.post(&op);
97	1053x	svc_.work_finished();
98		}
99	1067x	}
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	7448x	std::error_code register_fd(int fd) noexcept
108		{
109	7448x	desc_state_.fd = fd;
110		{
111	7448x	std::lock_guard lock(desc_state_.mutex);
112	7448x	self_ptr()->for_each_desc_entry(
113		[](auto&, reactor_op_base*& slot) { slot = nullptr; });
114	7448x	}
115	7448x	if (auto ec = svc_.scheduler().register_descriptor(fd, &desc_state_))
116		{
117	4x	desc_state_.fd = -1;
118	4x	desc_state_.registered_events = 0;
119	4x	return ec;
120		}
121	7444x	return {};
122		}
123
124		/** Register an op with the reactor.
125
126		Handles cached edge events. Called on the EAGAIN/EINPROGRESS
127		path when speculative I/O failed.
128		*/
129		template<class Op>
130	9921x	void register_op(
131		Op& op,
132		reactor_op_base*& desc_slot,
133		bool& ready_flag,
134		bool is_write_direction = false) noexcept
135		{
136	9921x	svc_.work_started();
137
138	9921x	std::lock_guard lock(desc_state_.mutex);
139	9921x	bool io_done = false;
140	9921x	if (ready_flag)
141		{
142	691x	ready_flag = false;
143	691x	op.perform_io();
144	691x	io_done = (op.errn != EAGAIN && op.errn != EWOULDBLOCK);
145	691x	if (!io_done)
146	685x	op.errn = 0;
147		}
148
149	9921x	if (io_done \|\| op.cancelled.load(std::memory_order_acquire))
150		{
151	6x	svc_.post(&op);
152	6x	svc_.work_finished();
153	6x	return;
154		}
155
156	9915x	if (desc_state_.unpollable)
157		{
158		// Nothing will ever report readiness for this fd.
159	1x	op.complete(EOPNOTSUPP, 0);
160	1x	svc_.post(&op);
161	1x	svc_.work_finished();
162	1x	return;
163		}
164
165	9914x	if (is_write_direction)
166		{
167	6686x	if (auto ec = svc_.scheduler().ensure_write_registered(
168		desc_state_.fd, &desc_state_))
169		{
170	✗	op.complete(ec.value(), 0);
171	✗	svc_.post(&op);
172	✗	svc_.work_finished();
173	✗	return;
174		}
175		}
176
177	9914x	desc_slot = &op;
178
179		// Select rebuilds its fd_sets from parked ops only, so parking
180		// must wake it. Compiled away for epoll and kqueue.
181		if constexpr (Service::needs_park_notification)
182	2863x	svc_.scheduler().notify_reactor();
183	9921x	}
184
185		/// Cancel every pending operation.
186	364x	void cancel_all() noexcept
187		{
188	364x	auto self = self_ptr()->weak_from_this().lock();
189	364x	if (!self)
190	✗	return;
191
192	364x	self_ptr()->for_each_op([](auto& op) { op.request_cancel(); });
193
194		reactor_op_base* claimed[max_claimed];
195	364x	int count = 0;
196		{
197	364x	std::lock_guard lock(desc_state_.mutex);
198	364x	self_ptr()->for_each_desc_entry(
199		[&](auto& op, reactor_op_base*& desc_slot) {
200		if (desc_slot == &op)
201		{
202		BOOST_COROSIO_ASSERT(count < max_claimed);
203		claimed[count++] = std::exchange(desc_slot, nullptr);
204		}
205		});
206	364x	}
207	364x	post_claimed(claimed, count, self);
208	364x	}
209
210		/** Cancel every operation and claim every parked one for teardown.
211
212		Also clears the cached edge flags and, if the state is queued
213		in the scheduler, pins the object alive until it is drained.
214		*/
215	71099x	void abandon_all() noexcept
216		{
217	71099x	auto self = self_ptr()->weak_from_this().lock();
218	71099x	if (!self)
219	✗	return;
220
221	71099x	self_ptr()->for_each_op([](auto& op) { op.request_cancel(); });
222
223		reactor_op_base* claimed[max_claimed];
224	71099x	int count = 0;
225		{
226	71099x	std::lock_guard lock(desc_state_.mutex);
227	71099x	self_ptr()->for_each_desc_entry(
228		[&](auto& /op/, reactor_op_base*& desc_slot) {
229		if (auto* c = std::exchange(desc_slot, nullptr))
230		{
231		BOOST_COROSIO_ASSERT(count < max_claimed);
232		claimed[count++] = c;
233		}
234		});
235	71099x	desc_state_.read_ready = false;
236	71099x	desc_state_.write_ready = false;
237
238		// Must be set under the same lock that invoke_deferred_io
239		// clears is_enqueued_ under, or the object could be destroyed
240		// while the scheduler still holds the queued descriptor_state.
241	71099x	if (desc_state_.is_enqueued_.load(std::memory_order_acquire))
242	1551x	desc_state_.impl_ref_ = self;
243	71099x	}
244	71099x	post_claimed(claimed, count, self);
245	71099x	}
246
247		/// Drop the reactor registration of @a fd and reset the state.
248	71099x	void unregister_fd(int fd) noexcept
249		{
250	71099x	if (fd >= 0 && desc_state_.registered_events != 0)
251	15789x	svc_.scheduler().deregister_descriptor(fd);
252	71099x	desc_state_.fd = -1;
253	71099x	desc_state_.registered_events = 0;
254	71099x	desc_state_.unpollable = false;
255	71099x	}
256
257		private:
258		// A claim empties its slot, so no more ops are claimed than
259		// descriptor_state has slots (read, write, connect, wait_read,
260		// wait_write, wait_error), however many ops share them.
261		static constexpr int max_claimed = 6;
262
263	71463x	void post_claimed(
264		reactor_op_base** claimed,
265		int count,
266		std::shared_ptr<Derived> const& self) noexcept
267		{
268	71889x	for (int i = 0; i < count; ++i)
269		{
270	426x	claimed[i]->impl_ptr = self;
271	426x	svc_.post(claimed[i]);
272	426x	svc_.work_finished();
273		}
274	71463x	}
275		};
276
277		} // namespace boost::corosio::detail
278
279		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_IO_CORE_HPP
280