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_DESCRIPTOR_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_HPP
12 :
13 : #include <boost/corosio/detail/platform.hpp>
14 :
15 : #if BOOST_COROSIO_POSIX
16 :
17 : #include <boost/corosio/posix_stream_descriptor.hpp>
18 : #include <boost/corosio/wait_type.hpp>
19 : #include <boost/corosio/detail/dispatch_coro.hpp>
20 : #include <boost/corosio/detail/intrusive.hpp>
21 : #include <boost/corosio/detail/native_handle.hpp>
22 : #include <boost/corosio/detail/object_ref.hpp>
23 : #include <boost/corosio/detail/config.hpp>
24 : #include <boost/corosio/native/detail/make_err.hpp>
25 : #include <boost/corosio/native/detail/validate_fd.hpp>
26 : #include <boost/corosio/native/detail/reactor/reactor_descriptor_state.hpp>
27 : #include <boost/corosio/native/detail/reactor/reactor_io_core.hpp>
28 : #include <boost/corosio/native/detail/reactor/reactor_op.hpp>
29 : #include <boost/corosio/native/detail/reactor/reactor_op_complete.hpp>
30 : #include <boost/capy/continuation.hpp>
31 : #include <boost/capy/buffers.hpp>
32 :
33 : #include <atomic>
34 : #include <coroutine>
35 : #include <cstring>
36 : #include <memory>
37 : #include <mutex>
38 : #include <utility>
39 :
40 : #include <errno.h>
41 : #include <sys/stat.h>
42 : #include <sys/uio.h>
43 : #include <unistd.h>
44 :
45 : /* Reactor-backed implementation of posix_stream_descriptor.
46 :
47 : The register/park/cancel/teardown protocol lives in reactor_io_core.
48 :
49 : The one behavior that is genuinely new: O_NONBLOCK is armed lazily,
50 : on the first read_some/write_some and never from assign() or wait().
51 : The flag lives on the shared open file description, so arming it is
52 : visible to every other holder of that description -- which is why a
53 : wait()-only user must never trigger it.
54 : */
55 :
56 : namespace boost::corosio::detail {
57 :
58 : // ============================================================
59 : // Op types
60 : // ============================================================
61 :
62 : /* Descriptor op family.
63 :
64 : Mirrors reactor_stream_ops.hpp. The Acceptor parameter is a
65 : placeholder: reactor_op is parameterized on both a socket and an
66 : acceptor impl type, and the shared completion helpers name
67 : acceptor_impl_ in a branch that a descriptor op never takes but the
68 : compiler still instantiates. Passing the backend's acceptor type (as
69 : reactor_dgram_socket_impl already does) keeps those helpers shared
70 : rather than duplicated here.
71 :
72 : @tparam Traits Backend traits (epoll_traits, kqueue_traits, ...).
73 : @tparam Descriptor The concrete descriptor type (forward-declared).
74 : @tparam Acceptor Placeholder acceptor type for the op base.
75 : */
76 :
77 : template<class Traits, class Descriptor, class Acceptor>
78 : struct reactor_descriptor_base_op : reactor_op<Descriptor, Acceptor>
79 : {
80 : void operator()() override;
81 : void cancel() noexcept override;
82 : };
83 :
84 : template<class Traits, class Descriptor, class Acceptor>
85 : struct reactor_descriptor_read_op final
86 : : reactor_read_op<reactor_descriptor_base_op<Traits, Descriptor, Acceptor>>
87 : {};
88 :
89 : template<class Traits, class Descriptor, class Acceptor>
90 : struct reactor_descriptor_write_op final
91 : : reactor_write_op<
92 : reactor_descriptor_base_op<Traits, Descriptor, Acceptor>,
93 : typename Traits::descriptor_write_policy>
94 : {};
95 :
96 : template<class Traits, class Descriptor, class Acceptor>
97 : struct reactor_descriptor_wait_op final
98 : : reactor_wait_op<reactor_descriptor_base_op<Traits, Descriptor, Acceptor>>
99 : {
100 : using base_type = reactor_wait_op<
101 : reactor_descriptor_base_op<Traits, Descriptor, Acceptor>>;
102 :
103 : /** Probe like reactor_wait_op::probe, but name a code for an error
104 : wait that sees POLLERR or POLLHUP. A non-socket has no SO_ERROR
105 : to consult, and success would claim no error condition exists.
106 : */
107 HIT 72 : static bool probe(int fd, std::uint32_t event, int& err) noexcept
108 : {
109 72 : if (event != reactor_event_error || fd < 0)
110 48 : return base_type::probe(fd, event, err);
111 :
112 24 : pollfd pfd{};
113 24 : pfd.fd = fd;
114 24 : pfd.events = POLLPRI;
115 : int r;
116 : do
117 : {
118 24 : r = ::poll(&pfd, 1, 0);
119 : }
120 24 : while (r < 0 && errno == EINTR);
121 24 : if (r < 0)
122 : {
123 MIS 0 : err = (errno == EAGAIN || errno == EWOULDBLOCK) ? ENOMEM : errno;
124 0 : return true;
125 : }
126 HIT 24 : if (r == 0)
127 22 : return false;
128 2 : if (pfd.revents & POLLNVAL)
129 MIS 0 : err = EBADF;
130 HIT 2 : else if (pfd.revents & (POLLERR | POLLHUP))
131 2 : err = EIO;
132 MIS 0 : else if (is_fifo(fd))
133 : // Darwin reports POLLPRI on a pipe that merely holds data;
134 : // a FIFO has no exceptional condition to signal.
135 0 : return false;
136 HIT 2 : return true;
137 : }
138 :
139 MIS 0 : static bool is_fifo(int fd) noexcept
140 : {
141 : struct stat st;
142 0 : return ::fstat(fd, &st) == 0 && S_ISFIFO(st.st_mode);
143 : }
144 :
145 HIT 35 : void perform_io() noexcept override
146 : {
147 35 : int err = 0;
148 35 : if (probe(this->fd, this->wait_event, err))
149 10 : this->complete(err, 0);
150 : else
151 25 : this->complete(EAGAIN, 0);
152 35 : }
153 :
154 : void operator()() override;
155 : };
156 :
157 : // --- Deferred implementations (instantiated when Descriptor is complete) ---
158 :
159 : template<class Traits, class Descriptor, class Acceptor>
160 : void
161 127 : reactor_descriptor_base_op<Traits, Descriptor, Acceptor>::operator()()
162 : {
163 127 : complete_io_op(*this);
164 127 : }
165 :
166 : template<class Traits, class Descriptor, class Acceptor>
167 : void
168 6 : reactor_descriptor_base_op<Traits, Descriptor, Acceptor>::cancel() noexcept
169 : {
170 : // A descriptor op is only ever started against a descriptor impl, so
171 : // the acceptor arm of the stream op's cancel() has no counterpart.
172 6 : if (this->socket_impl_)
173 6 : this->socket_impl_->cancel_single_op(*this);
174 : else
175 MIS 0 : this->request_cancel();
176 HIT 6 : }
177 :
178 : template<class Traits, class Descriptor, class Acceptor>
179 : void
180 27 : reactor_descriptor_wait_op<Traits, Descriptor, Acceptor>::operator()()
181 : {
182 27 : complete_wait_op(*this);
183 27 : }
184 :
185 : // ============================================================
186 : // Descriptor implementation
187 : // ============================================================
188 :
189 : /** CRTP base for reactor-backed posix_stream_descriptor implementations.
190 :
191 : Holds the adopted descriptor, its reactor registration state, and
192 : the five op slots (read, write, and one wait per direction).
193 :
194 : @tparam Derived The named final class (CRTP self).
195 : @tparam Traits Backend traits (epoll_traits, kqueue_traits, ...).
196 : @tparam Service The backend's descriptor service type.
197 : @tparam Acceptor Placeholder acceptor type for the op base.
198 : */
199 : template<class Derived, class Traits, class Service, class Acceptor>
200 : class reactor_descriptor
201 : : public posix_stream_descriptor::implementation
202 : , public reactor_io_core<Derived, Service, typename Traits::desc_state_type>
203 : , public intrusive_list<Derived>::node
204 : {
205 : using base_op = reactor_descriptor_base_op<Traits, Derived, Acceptor>;
206 : using read_op = reactor_descriptor_read_op<Traits, Derived, Acceptor>;
207 : using write_op = reactor_descriptor_write_op<Traits, Derived, Acceptor>;
208 : using wait_op = reactor_descriptor_wait_op<Traits, Derived, Acceptor>;
209 :
210 : using core_type =
211 : reactor_io_core<Derived, Service, typename Traits::desc_state_type>;
212 : friend core_type;
213 :
214 : protected:
215 : // NOLINTNEXTLINE(bugprone-crtp-constructor-accessibility)
216 140 : explicit reactor_descriptor(Service& svc) noexcept : core_type(svc) {}
217 :
218 : using core_type::svc_;
219 :
220 : public:
221 140 : ~reactor_descriptor() override = default;
222 :
223 : using core_type::desc_state_;
224 :
225 : // --- Virtual method overrides ---
226 :
227 121 : std::coroutine_handle<> read_some(
228 : capy::continuation& cont,
229 : capy::executor_ref ex,
230 : buffer_param param,
231 : std::stop_token token,
232 : std::error_code* ec,
233 : std::size_t* bytes_out) override
234 : {
235 121 : return do_read_some(cont, ex, param, token, ec, bytes_out);
236 : }
237 :
238 20 : std::coroutine_handle<> write_some(
239 : capy::continuation& cont,
240 : capy::executor_ref ex,
241 : buffer_param param,
242 : std::stop_token token,
243 : std::error_code* ec,
244 : std::size_t* bytes_out) override
245 : {
246 20 : return do_write_some(cont, ex, param, token, ec, bytes_out);
247 : }
248 :
249 37 : std::coroutine_handle<> wait(
250 : capy::continuation& cont,
251 : capy::executor_ref ex,
252 : wait_type w,
253 : std::stop_token token,
254 : std::error_code* ec) override
255 : {
256 37 : return do_wait(cont, ex, w, token, ec);
257 : }
258 :
259 484 : native_handle_type native_handle() const noexcept override
260 : {
261 484 : return fd_;
262 : }
263 :
264 : native_handle_type release_descriptor() noexcept override;
265 :
266 41 : void cancel() noexcept override
267 : {
268 41 : this->cancel_all();
269 41 : }
270 :
271 : /// Recycle into the owning service's pool at zero references.
272 205 : void retire() noexcept override
273 : {
274 : #if !defined(NDEBUG)
275 205 : poison();
276 : #endif
277 205 : svc_.state_->pool_.recycle(static_cast<Derived*>(this));
278 205 : }
279 :
280 : // --- Service-facing (non-virtual) ---
281 :
282 : /** Reset descriptor state for recycling.
283 :
284 : Re-initializes every field it owns rather than trusting
285 : `close_descriptor()`'s prior writes; see
286 : reactor_basic_socket::reuse().
287 :
288 : @pre refs_ == 0, fd closed and deregistered, no op in flight.
289 : */
290 65 : void reuse() noexcept
291 : {
292 65 : fd_ = -1;
293 65 : nonblocking_.store(false, std::memory_order_relaxed);
294 65 : this->reset_desc_state();
295 65 : }
296 :
297 : #if !defined(NDEBUG)
298 : /// Poison the fields `reuse()` re-initializes; see
299 : /// reactor_basic_socket::poison().
300 205 : void poison() noexcept
301 : {
302 205 : std::memset(static_cast<void*>(&fd_), 0xDB, sizeof(fd_));
303 205 : this->poison_desc_state();
304 205 : }
305 : #endif
306 :
307 : /** Adopt the fd, initialize descriptor state, and register it.
308 :
309 : @param fd The descriptor to adopt.
310 :
311 : @return The error if the reactor rejects the descriptor, in
312 : which case the implementation is left closed and the caller
313 : retains ownership of @a fd; otherwise a default constructed
314 : error code.
315 : */
316 : std::error_code init_and_register(int fd) noexcept;
317 :
318 : /// Close the descriptor and cancel pending operations.
319 : void close_descriptor() noexcept;
320 :
321 : private:
322 : /** Arm O_NONBLOCK, once, before the first speculative syscall.
323 :
324 : Reports an errno rather than an error_code because that is what
325 : the op result model records; the round trip is lossless here
326 : because fcntl only fails with codes make_err passes through.
327 : */
328 139 : int arm_nonblocking() noexcept
329 : {
330 : // Relaxed: ensure_nonblocking is idempotent, so a duplicate
331 : // fcntl from a racing first read and write is harmless.
332 139 : if (nonblocking_.load(std::memory_order_relaxed))
333 6 : return 0;
334 133 : if (auto ec = ensure_nonblocking(fd_))
335 MIS 0 : return ec.value();
336 HIT 133 : nonblocking_.store(true, std::memory_order_relaxed);
337 133 : return 0;
338 : }
339 :
340 : std::coroutine_handle<> do_read_some(
341 : capy::continuation&,
342 : capy::executor_ref,
343 : buffer_param,
344 : std::stop_token const&,
345 : std::error_code*,
346 : std::size_t*);
347 :
348 : std::coroutine_handle<> do_write_some(
349 : capy::continuation&,
350 : capy::executor_ref,
351 : buffer_param,
352 : std::stop_token const&,
353 : std::error_code*,
354 : std::size_t*);
355 :
356 : std::coroutine_handle<> do_wait(
357 : capy::continuation&,
358 : capy::executor_ref,
359 : wait_type,
360 : std::stop_token const&,
361 : std::error_code*);
362 :
363 : /// Apply @a fn to each of the five op slots.
364 : template<class Fn>
365 643 : void for_each_op(Fn fn) noexcept
366 : {
367 643 : fn(rd_);
368 643 : fn(wr_);
369 643 : fn(wait_rd_);
370 643 : fn(wait_wr_);
371 643 : fn(wait_er_);
372 643 : }
373 :
374 : /// Apply @a fn to each op and the descriptor_state slot it parks in.
375 : template<class Fn>
376 831 : void for_each_desc_entry(Fn fn) noexcept
377 : {
378 831 : fn(rd_, desc_state_.read_op);
379 831 : fn(wr_, desc_state_.write_op);
380 831 : fn(wait_rd_, desc_state_.wait_read_op);
381 831 : fn(wait_wr_, desc_state_.wait_write_op);
382 831 : fn(wait_er_, desc_state_.wait_error_op);
383 831 : }
384 :
385 : /// Sweep every op slot, then drop the reactor registration.
386 : void quiesce() noexcept;
387 :
388 : reactor_op_base** op_to_desc_slot(base_op& op) noexcept;
389 :
390 : int fd_ = -1;
391 : std::atomic<bool> nonblocking_{false};
392 :
393 : read_op rd_;
394 : write_op wr_;
395 : wait_op wait_rd_;
396 : wait_op wait_wr_;
397 : wait_op wait_er_;
398 : };
399 :
400 : // ============================================================
401 : // Registration and teardown
402 : // ============================================================
403 :
404 : template<class Derived, class Traits, class Service, class Acceptor>
405 : std::error_code
406 188 : reactor_descriptor<Derived, Traits, Service, Acceptor>::init_and_register(
407 : int fd) noexcept
408 : {
409 188 : fd_ = fd;
410 188 : if (auto ec = this->register_fd(fd))
411 : {
412 MIS 0 : fd_ = -1;
413 0 : return ec;
414 : }
415 HIT 188 : return {};
416 : }
417 :
418 : template<class Derived, class Traits, class Service, class Acceptor>
419 : void
420 602 : reactor_descriptor<Derived, Traits, Service, Acceptor>::quiesce() noexcept
421 : {
422 602 : this->abandon_all();
423 : // The next adopted fd starts from an unknown flag state.
424 602 : nonblocking_.store(false, std::memory_order_relaxed);
425 602 : }
426 :
427 : template<class Derived, class Traits, class Service, class Acceptor>
428 : void
429 596 : reactor_descriptor<Derived, Traits, Service, Acceptor>::
430 : close_descriptor() noexcept
431 : {
432 596 : quiesce();
433 :
434 596 : if (fd_ >= 0)
435 : {
436 182 : ::close(fd_);
437 182 : fd_ = -1;
438 : }
439 596 : }
440 :
441 : template<class Derived, class Traits, class Service, class Acceptor>
442 : native_handle_type
443 6 : reactor_descriptor<Derived, Traits, Service, Acceptor>::
444 : release_descriptor() noexcept
445 : {
446 6 : quiesce();
447 :
448 : // Do NOT close -- the caller takes ownership.
449 6 : native_handle_type released = fd_;
450 6 : fd_ = -1;
451 6 : return released;
452 : }
453 :
454 : // ============================================================
455 : // Op slot lookup
456 : // ============================================================
457 :
458 : template<class Derived, class Traits, class Service, class Acceptor>
459 : reactor_op_base**
460 6 : reactor_descriptor<Derived, Traits, Service, Acceptor>::op_to_desc_slot(
461 : base_op& op) noexcept
462 : {
463 6 : if (&op == static_cast<void*>(&rd_))
464 6 : return &desc_state_.read_op;
465 MIS 0 : if (&op == static_cast<void*>(&wr_))
466 0 : return &desc_state_.write_op;
467 0 : if (&op == static_cast<void*>(&wait_rd_))
468 0 : return &desc_state_.wait_read_op;
469 0 : if (&op == static_cast<void*>(&wait_wr_))
470 0 : return &desc_state_.wait_write_op;
471 0 : if (&op == static_cast<void*>(&wait_er_))
472 0 : return &desc_state_.wait_error_op;
473 0 : return nullptr;
474 : }
475 :
476 : // ============================================================
477 : // I/O dispatch
478 : // ============================================================
479 :
480 : template<class Derived, class Traits, class Service, class Acceptor>
481 : std::coroutine_handle<>
482 HIT 121 : reactor_descriptor<Derived, Traits, Service, Acceptor>::do_read_some(
483 : capy::continuation& cont,
484 : capy::executor_ref ex,
485 : buffer_param param,
486 : std::stop_token const& token,
487 : std::error_code* ec,
488 : std::size_t* bytes_out)
489 : {
490 121 : auto& op = rd_;
491 121 : op.reset();
492 121 : op.cont = &cont;
493 121 : op.ex = ex;
494 121 : op.ec_out = ec;
495 121 : op.bytes_out = bytes_out;
496 :
497 : // Closed-object contract: complete with bad_file_descriptor without
498 : // touching the kernel or the unregistered descriptor state.
499 121 : if (fd_ < 0)
500 : {
501 MIS 0 : op.start(token, static_cast<Derived*>(this));
502 0 : op.object_ref_ = detail::object_ref(this);
503 0 : op.complete(EBADF, 0);
504 0 : svc_.post(&op);
505 0 : return std::noop_coroutine();
506 : }
507 :
508 HIT 121 : capy::mutable_buffer bufs[read_op::max_buffers];
509 121 : op.iovec_count =
510 121 : static_cast<int>(param.copy_to(bufs, read_op::max_buffers));
511 :
512 121 : if (op.iovec_count == 0 || (op.iovec_count == 1 && bufs[0].size() == 0))
513 : {
514 2 : op.empty_buffer_read = true;
515 2 : op.start(token, static_cast<Derived*>(this));
516 2 : op.object_ref_ = detail::object_ref(this);
517 2 : op.complete(0, 0);
518 2 : svc_.post(&op);
519 2 : return std::noop_coroutine();
520 : }
521 :
522 : // The first transferring operation is what arms O_NONBLOCK; assign()
523 : // and wait() never do.
524 119 : if (int const nerr = arm_nonblocking())
525 : {
526 MIS 0 : op.start(token, static_cast<Derived*>(this));
527 0 : op.object_ref_ = detail::object_ref(this);
528 0 : op.complete(nerr, 0);
529 0 : svc_.post(&op);
530 0 : return std::noop_coroutine();
531 : }
532 :
533 HIT 240 : for (int i = 0; i < op.iovec_count; ++i)
534 : {
535 121 : op.iovecs[i].iov_base = bufs[i].data();
536 121 : op.iovecs[i].iov_len = bufs[i].size();
537 : }
538 :
539 : // Speculative read; the single-buffer case uses read() so the kernel
540 : // skips the readv iov_iter setup.
541 : ssize_t n;
542 119 : if (op.iovec_count == 1)
543 : {
544 : do
545 : {
546 117 : n = ::read(fd_, bufs[0].data(), bufs[0].size());
547 : }
548 117 : while (n < 0 && errno == EINTR);
549 : }
550 : else
551 : {
552 : do
553 : {
554 2 : n = ::readv(fd_, op.iovecs, op.iovec_count);
555 : }
556 2 : while (n < 0 && errno == EINTR);
557 : }
558 :
559 119 : if (n >= 0 || (errno != EAGAIN && errno != EWOULDBLOCK))
560 : {
561 57 : int err = (n < 0) ? errno : 0;
562 57 : auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
563 :
564 57 : if (svc_.scheduler().try_consume_inline_budget())
565 : {
566 6 : if (err)
567 MIS 0 : *ec = make_err(err);
568 HIT 6 : else if (n == 0)
569 2 : *ec = capy::error::eof;
570 : else
571 4 : *ec = {};
572 6 : *bytes_out = bytes;
573 6 : return dispatch_coro(ex, cont);
574 : }
575 51 : op.start(token, static_cast<Derived*>(this));
576 51 : op.object_ref_ = detail::object_ref(this);
577 51 : op.complete(err, bytes);
578 51 : svc_.post(&op);
579 51 : return std::noop_coroutine();
580 : }
581 :
582 : // EAGAIN — register with reactor
583 62 : op.fd = fd_;
584 62 : op.start(token, static_cast<Derived*>(this));
585 62 : op.object_ref_ = detail::object_ref(this);
586 :
587 62 : this->register_op(op, desc_state_.read_op, desc_state_.read_ready);
588 62 : return std::noop_coroutine();
589 : }
590 :
591 : template<class Derived, class Traits, class Service, class Acceptor>
592 : std::coroutine_handle<>
593 20 : reactor_descriptor<Derived, Traits, Service, Acceptor>::do_write_some(
594 : capy::continuation& cont,
595 : capy::executor_ref ex,
596 : buffer_param param,
597 : std::stop_token const& token,
598 : std::error_code* ec,
599 : std::size_t* bytes_out)
600 : {
601 20 : auto& op = wr_;
602 20 : op.reset();
603 20 : op.cont = &cont;
604 20 : op.ex = ex;
605 20 : op.ec_out = ec;
606 20 : op.bytes_out = bytes_out;
607 :
608 20 : if (fd_ < 0)
609 : {
610 MIS 0 : op.start(token, static_cast<Derived*>(this));
611 0 : op.object_ref_ = detail::object_ref(this);
612 0 : op.complete(EBADF, 0);
613 0 : svc_.post(&op);
614 0 : return std::noop_coroutine();
615 : }
616 :
617 HIT 20 : capy::mutable_buffer bufs[write_op::max_buffers];
618 20 : op.iovec_count =
619 20 : static_cast<int>(param.copy_to(bufs, write_op::max_buffers));
620 :
621 20 : if (op.iovec_count == 0 || (op.iovec_count == 1 && bufs[0].size() == 0))
622 : {
623 MIS 0 : op.start(token, static_cast<Derived*>(this));
624 0 : op.object_ref_ = detail::object_ref(this);
625 0 : op.complete(0, 0);
626 0 : svc_.post(&op);
627 0 : return std::noop_coroutine();
628 : }
629 :
630 HIT 20 : if (int const nerr = arm_nonblocking())
631 : {
632 MIS 0 : op.start(token, static_cast<Derived*>(this));
633 0 : op.object_ref_ = detail::object_ref(this);
634 0 : op.complete(nerr, 0);
635 0 : svc_.post(&op);
636 0 : return std::noop_coroutine();
637 : }
638 :
639 HIT 42 : for (int i = 0; i < op.iovec_count; ++i)
640 : {
641 22 : op.iovecs[i].iov_base = bufs[i].data();
642 22 : op.iovecs[i].iov_len = bufs[i].size();
643 : }
644 :
645 : // Speculative write; the single-buffer case skips the iov_iter setup.
646 : ssize_t n;
647 20 : if (op.iovec_count == 1)
648 : {
649 36 : n = write_op::write_policy::write_one(
650 18 : fd_, bufs[0].data(), bufs[0].size());
651 : }
652 : else
653 : {
654 2 : n = write_op::write_policy::write(fd_, op.iovecs, op.iovec_count);
655 : }
656 :
657 20 : if (n >= 0 || (errno != EAGAIN && errno != EWOULDBLOCK))
658 : {
659 14 : int err = (n < 0) ? errno : 0;
660 14 : auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
661 :
662 14 : if (svc_.scheduler().try_consume_inline_budget())
663 : {
664 2 : *ec = err ? make_err(err) : std::error_code{};
665 2 : *bytes_out = bytes;
666 2 : return dispatch_coro(ex, cont);
667 : }
668 12 : op.start(token, static_cast<Derived*>(this));
669 12 : op.object_ref_ = detail::object_ref(this);
670 12 : op.complete(err, bytes);
671 12 : svc_.post(&op);
672 12 : return std::noop_coroutine();
673 : }
674 :
675 : // EAGAIN — register with reactor
676 6 : op.fd = fd_;
677 6 : op.start(token, static_cast<Derived*>(this));
678 6 : op.object_ref_ = detail::object_ref(this);
679 :
680 6 : this->register_op(op, desc_state_.write_op, desc_state_.write_ready, true);
681 6 : return std::noop_coroutine();
682 : }
683 :
684 : template<class Derived, class Traits, class Service, class Acceptor>
685 : std::coroutine_handle<>
686 37 : reactor_descriptor<Derived, Traits, Service, Acceptor>::do_wait(
687 : capy::continuation& cont,
688 : capy::executor_ref ex,
689 : wait_type w,
690 : std::stop_token const& token,
691 : std::error_code* ec)
692 : {
693 : // Pick refs up-front to avoid duplicating the register_op call.
694 : wait_op* op_ptr;
695 : reactor_op_base** desc_slot_ptr;
696 : std::uint32_t event;
697 :
698 37 : if (w == wait_type::read)
699 : {
700 16 : op_ptr = &wait_rd_;
701 16 : desc_slot_ptr = &desc_state_.wait_read_op;
702 16 : event = reactor_event_read;
703 : }
704 21 : else if (w == wait_type::write)
705 : {
706 8 : op_ptr = &wait_wr_;
707 8 : desc_slot_ptr = &desc_state_.wait_write_op;
708 8 : event = reactor_event_write;
709 : }
710 : else // wait_type::error
711 : {
712 13 : op_ptr = &wait_er_;
713 13 : desc_slot_ptr = &desc_state_.wait_error_op;
714 13 : event = reactor_event_error;
715 : }
716 :
717 37 : auto& op = *op_ptr;
718 :
719 : // Speculative probe: an edge-triggered reactor cannot report a
720 : // condition that already holds, so a wait initiated on an already
721 : // ready descriptor would otherwise park forever. No syscall here
722 : // modifies the descriptor -- in particular O_NONBLOCK is untouched.
723 37 : int perr = 0;
724 37 : if (wait_op::probe(fd_, event, perr))
725 : {
726 12 : if (svc_.scheduler().try_consume_inline_budget())
727 : {
728 2 : *ec = perr ? make_err(perr) : std::error_code{};
729 2 : return dispatch_coro(ex, cont);
730 : }
731 10 : op.reset();
732 10 : op.wait_event = event;
733 10 : op.cont = &cont;
734 10 : op.ex = ex;
735 10 : op.ec_out = ec;
736 10 : op.fd = fd_;
737 10 : op.start(token, static_cast<Derived*>(this));
738 10 : op.object_ref_ = detail::object_ref(this);
739 10 : op.complete(perr, 0);
740 10 : svc_.post(&op);
741 10 : return std::noop_coroutine();
742 : }
743 :
744 25 : op.reset();
745 25 : op.wait_event = event;
746 25 : op.cont = &cont;
747 25 : op.ex = ex;
748 25 : op.ec_out = ec;
749 25 : op.fd = fd_;
750 25 : op.start(token, static_cast<Derived*>(this));
751 25 : op.object_ref_ = detail::object_ref(this);
752 :
753 : // Force register_op's ready path so the wait op re-probes under the
754 : // descriptor mutex before parking. A stale write_ready latched at
755 : // registration would otherwise report a full pipe as writable.
756 25 : bool force_probe = true;
757 25 : this->register_op(
758 : op, *desc_slot_ptr, force_probe, event == reactor_event_write);
759 25 : return std::noop_coroutine();
760 : }
761 :
762 : } // namespace boost::corosio::detail
763 :
764 : #endif // BOOST_COROSIO_POSIX
765 :
766 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_DESCRIPTOR_HPP
|
