TLA Line data 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 HIT 229287 : Derived* self_ptr() noexcept
51 : {
52 229287 : return static_cast<Derived*>(this);
53 : }
54 :
55 : protected:
56 : // NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
57 7572 : 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 676 : void cancel_single_op(Op& op) noexcept
72 : {
73 676 : op.request_cancel();
74 :
75 676 : reactor_op_base** desc_op_ptr = self_ptr()->op_to_desc_slot(op);
76 676 : if (!desc_op_ptr)
77 MIS 0 : return;
78 :
79 HIT 676 : reactor_op_base* claimed = nullptr;
80 : {
81 676 : std::lock_guard lock(desc_state_.mutex);
82 676 : if (*desc_op_ptr == &op)
83 : {
84 544 : claimed = std::exchange(*desc_op_ptr, nullptr);
85 544 : 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 676 : }
93 676 : if (claimed)
94 : {
95 544 : op.object_ref_ = detail::object_ref(self_ptr());
96 544 : svc_.scheduler().post_counted(&op);
97 544 : 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 14253 : 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 14253 : std::lock_guard lock(desc_state_.mutex);
113 14253 : desc_state_.fd = fd;
114 14253 : self_ptr()->for_each_desc_entry(
115 86460 : [](auto&, reactor_op_base*& slot) { slot = nullptr; });
116 14253 : }
117 14253 : if (auto ec = svc_.scheduler().register_descriptor(fd, &desc_state_))
118 : {
119 4 : std::lock_guard lock(desc_state_.mutex);
120 4 : desc_state_.fd = -1;
121 4 : desc_state_.registered_events = 0;
122 4 : return ec;
123 4 : }
124 14249 : 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 10346 : 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 10346 : svc_.work_started();
140 10346 : if (park_op(op, desc_slot, ready_flag, is_write_direction))
141 10339 : return;
142 : // Posted after the descriptor lock is released: the scheduler
143 : // takes its own lock before descriptor locks.
144 7 : svc_.scheduler().post_counted(&op);
145 : }
146 :
147 : /// Cancel every pending operation.
148 367 : void cancel_all() noexcept
149 : {
150 2588 : self_ptr()->for_each_op([](auto& op) { op.request_cancel(); });
151 :
152 : reactor_op_base* claimed[max_claimed];
153 367 : int count = 0;
154 : {
155 367 : std::lock_guard lock(desc_state_.mutex);
156 367 : self_ptr()->for_each_desc_entry(
157 4442 : [&](auto& op, reactor_op_base*& desc_slot) {
158 2221 : if (desc_slot == &op)
159 : {
160 269 : BOOST_COROSIO_ASSERT(count < max_claimed);
161 269 : claimed[count++] = std::exchange(desc_slot, nullptr);
162 : }
163 : });
164 367 : }
165 367 : post_claimed(claimed, count);
166 367 : }
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 94331 : void abandon_all() noexcept
177 : {
178 645613 : self_ptr()->for_each_op([](auto& op) { op.request_cancel(); });
179 :
180 94331 : int fd = -1;
181 94331 : bool registered = false;
182 : {
183 94331 : std::lock_guard lock(desc_state_.mutex);
184 94331 : fd = desc_state_.fd;
185 94331 : registered = desc_state_.registered_events != 0;
186 94331 : }
187 94331 : if (fd >= 0 && registered)
188 : {
189 22997 : svc_.scheduler().deregister_descriptor(fd);
190 45994 : svc_.scheduler().retire_descriptor(
191 45994 : desc_state_, detail::object_ref(self_ptr()));
192 : }
193 :
194 : reactor_op_base* claimed[max_claimed];
195 94331 : int count = 0;
196 : {
197 94331 : std::lock_guard lock(desc_state_.mutex);
198 94331 : self_ptr()->for_each_desc_entry(
199 1102564 : [&](auto& /*op*/, reactor_op_base*& desc_slot) {
200 551282 : if (auto* c = std::exchange(desc_slot, nullptr))
201 : {
202 1152 : BOOST_COROSIO_ASSERT(count < max_claimed);
203 1152 : claimed[count++] = c;
204 : }
205 : });
206 94331 : desc_state_.read_ready = false;
207 94331 : 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 94331 : desc_state_.fd = -1;
212 94331 : desc_state_.registered_events = 0;
213 94331 : desc_state_.unpollable = false;
214 94331 : }
215 94331 : post_claimed(claimed, count);
216 94331 : }
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 21977 : void reset_desc_state() noexcept
225 : {
226 21977 : desc_state_.fd = -1;
227 21977 : desc_state_.registered_events = 0;
228 21977 : desc_state_.unpollable = false;
229 21977 : desc_state_.read_op = nullptr;
230 21977 : desc_state_.write_op = nullptr;
231 21977 : desc_state_.connect_op = nullptr;
232 21977 : desc_state_.wait_read_op = nullptr;
233 21977 : desc_state_.wait_write_op = nullptr;
234 21977 : desc_state_.wait_error_op = nullptr;
235 21977 : desc_state_.read_ready = false;
236 21977 : desc_state_.write_ready = false;
237 21977 : BOOST_COROSIO_ASSERT(!desc_state_.object_ref_);
238 21977 : BOOST_COROSIO_ASSERT(!desc_state_.retired_ref_);
239 21977 : BOOST_COROSIO_ASSERT(!desc_state_.dropped_ref_);
240 21977 : BOOST_COROSIO_ASSERT(
241 : !desc_state_.is_enqueued_.load(std::memory_order_relaxed));
242 21977 : }
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 29549 : void poison_desc_state() noexcept
252 : {
253 325039 : auto smash = [](auto& field) {
254 325039 : std::memset(static_cast<void*>(&field), 0xDB, sizeof(field));
255 : };
256 29549 : smash(desc_state_.fd);
257 29549 : smash(desc_state_.registered_events);
258 29549 : smash(desc_state_.unpollable);
259 29549 : smash(desc_state_.read_op);
260 29549 : smash(desc_state_.write_op);
261 29549 : smash(desc_state_.connect_op);
262 29549 : smash(desc_state_.wait_read_op);
263 29549 : smash(desc_state_.wait_write_op);
264 29549 : smash(desc_state_.wait_error_op);
265 29549 : smash(desc_state_.read_ready);
266 29549 : smash(desc_state_.write_ready);
267 29549 : }
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 10346 : bool park_op(
282 : Op& op,
283 : reactor_op_base*& desc_slot,
284 : bool& ready_flag,
285 : bool is_write_direction) noexcept
286 : {
287 10346 : std::lock_guard lock(desc_state_.mutex);
288 10346 : if (ready_flag)
289 : {
290 228 : ready_flag = false;
291 228 : op.perform_io();
292 228 : if (op.errn != EAGAIN && op.errn != EWOULDBLOCK)
293 6 : return false;
294 222 : op.errn = 0;
295 : }
296 :
297 10340 : if (op.cancelled.load(std::memory_order_acquire))
298 MIS 0 : return false;
299 :
300 HIT 10340 : if (desc_state_.unpollable)
301 : {
302 : // Nothing will ever report readiness for this fd.
303 1 : op.complete(EOPNOTSUPP, 0);
304 1 : return false;
305 : }
306 :
307 10339 : if (is_write_direction)
308 : {
309 8362 : if (auto ec = svc_.scheduler().ensure_write_registered(
310 : desc_state_.fd, &desc_state_))
311 : {
312 MIS 0 : op.complete(ec.value(), 0);
313 0 : return false;
314 : }
315 : }
316 :
317 HIT 10339 : 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 3778 : svc_.scheduler().notify_reactor();
324 10339 : return true;
325 10346 : }
326 :
327 94698 : void post_claimed(reactor_op_base** claimed, int count) noexcept
328 : {
329 96119 : for (int i = 0; i < count; ++i)
330 : {
331 1421 : claimed[i]->object_ref_ = detail::object_ref(self_ptr());
332 1421 : svc_.scheduler().post_counted(claimed[i]);
333 : }
334 94698 : }
335 : };
336 :
337 : } // namespace boost::corosio::detail
338 :
339 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_IO_CORE_HPP
|
