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

82.9% Lines (223 / 269) 96.6% Functions (56 / 58)

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

reactor_descriptor.hpp

f(x) Functions (58)

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_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	72x	static bool probe(int fd, std::uint32_t event, int& err) noexcept
108		{
109	72x	if (event != reactor_event_error \|\| fd < 0)
110	48x	return base_type::probe(fd, event, err);
111
112	24x	pollfd pfd{};
113	24x	pfd.fd = fd;
114	24x	pfd.events = POLLPRI;
115		int r;
116		do
117		{
118	24x	r = ::poll(&pfd, 1, 0);
119		}
120	24x	while (r < 0 && errno == EINTR);
121	24x	if (r < 0)
122		{
123	✗	err = (errno == EAGAIN \|\| errno == EWOULDBLOCK) ? ENOMEM : errno;
124	✗	return true;
125		}
126	24x	if (r == 0)
127	22x	return false;
128	2x	if (pfd.revents & POLLNVAL)
129	✗	err = EBADF;
130	2x	else if (pfd.revents & (POLLERR \| POLLHUP))
131	2x	err = EIO;
132	✗	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	✗	return false;
136	2x	return true;
137		}
138
139	✗	static bool is_fifo(int fd) noexcept
140		{
141		struct stat st;
142	✗	return ::fstat(fd, &st) == 0 && S_ISFIFO(st.st_mode);
143		}
144
145	35x	void perform_io() noexcept override
146		{
147	35x	int err = 0;
148	35x	if (probe(this->fd, this->wait_event, err))
149	10x	this->complete(err, 0);
150		else
151	25x	this->complete(EAGAIN, 0);
152	35x	}
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	127x	reactor_descriptor_base_op<Traits, Descriptor, Acceptor>::operator()()
162		{
163	127x	complete_io_op(*this);
164	127x	}
165
166		template<class Traits, class Descriptor, class Acceptor>
167		void
168	6x	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	6x	if (this->socket_impl_)
173	6x	this->socket_impl_->cancel_single_op(*this);
174		else
175	✗	this->request_cancel();
176	6x	}
177
178		template<class Traits, class Descriptor, class Acceptor>
179		void
180	27x	reactor_descriptor_wait_op<Traits, Descriptor, Acceptor>::operator()()
181		{
182	27x	complete_wait_op(*this);
183	27x	}
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	140x	explicit reactor_descriptor(Service& svc) noexcept : core_type(svc) {}
217
218		using core_type::svc_;
219
220		public:
221	140x	~reactor_descriptor() override = default;
222
223		using core_type::desc_state_;
224
225		// --- Virtual method overrides ---
226
227	121x	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	121x	return do_read_some(cont, ex, param, token, ec, bytes_out);
236		}
237
238	20x	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	20x	return do_write_some(cont, ex, param, token, ec, bytes_out);
247		}
248
249	37x	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	37x	return do_wait(cont, ex, w, token, ec);
257		}
258
259	484x	native_handle_type native_handle() const noexcept override
260		{
261	484x	return fd_;
262		}
263
264		native_handle_type release_descriptor() noexcept override;
265
266	41x	void cancel() noexcept override
267		{
268	41x	this->cancel_all();
269	41x	}
270
271		/// Recycle into the owning service's pool at zero references.
272	205x	void retire() noexcept override
273		{
274		#if !defined(NDEBUG)
275	205x	poison();
276		#endif
277	205x	svc_.state_->pool_.recycle(static_cast<Derived*>(this));
278	205x	}
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	65x	void reuse() noexcept
291		{
292	65x	fd_ = -1;
293	65x	nonblocking_.store(false, std::memory_order_relaxed);
294	65x	this->reset_desc_state();
295	65x	}
296
297		#if !defined(NDEBUG)
298		/// Poison the fields `reuse()` re-initializes; see
299		/// reactor_basic_socket::poison().
300	205x	void poison() noexcept
301		{
302	205x	std::memset(static_cast<void*>(&fd_), 0xDB, sizeof(fd_));
303	205x	this->poison_desc_state();
304	205x	}
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	139x	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	139x	if (nonblocking_.load(std::memory_order_relaxed))
333	6x	return 0;
334	133x	if (auto ec = ensure_nonblocking(fd_))
335	✗	return ec.value();
336	133x	nonblocking_.store(true, std::memory_order_relaxed);
337	133x	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	643x	void for_each_op(Fn fn) noexcept
366		{
367	643x	fn(rd_);
368	643x	fn(wr_);
369	643x	fn(wait_rd_);
370	643x	fn(wait_wr_);
371	643x	fn(wait_er_);
372	643x	}
373
374		/// Apply @a fn to each op and the descriptor_state slot it parks in.
375		template<class Fn>
376	831x	void for_each_desc_entry(Fn fn) noexcept
377		{
378	831x	fn(rd_, desc_state_.read_op);
379	831x	fn(wr_, desc_state_.write_op);
380	831x	fn(wait_rd_, desc_state_.wait_read_op);
381	831x	fn(wait_wr_, desc_state_.wait_write_op);
382	831x	fn(wait_er_, desc_state_.wait_error_op);
383	831x	}
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	188x	reactor_descriptor<Derived, Traits, Service, Acceptor>::init_and_register(
407		int fd) noexcept
408		{
409	188x	fd_ = fd;
410	188x	if (auto ec = this->register_fd(fd))
411		{
412	✗	fd_ = -1;
413	✗	return ec;
414		}
415	188x	return {};
416		}
417
418		template<class Derived, class Traits, class Service, class Acceptor>
419		void
420	602x	reactor_descriptor<Derived, Traits, Service, Acceptor>::quiesce() noexcept
421		{
422	602x	this->abandon_all();
423		// The next adopted fd starts from an unknown flag state.
424	602x	nonblocking_.store(false, std::memory_order_relaxed);
425	602x	}
426
427		template<class Derived, class Traits, class Service, class Acceptor>
428		void
429	596x	reactor_descriptor<Derived, Traits, Service, Acceptor>::
430		close_descriptor() noexcept
431		{
432	596x	quiesce();
433
434	596x	if (fd_ >= 0)
435		{
436	182x	::close(fd_);
437	182x	fd_ = -1;
438		}
439	596x	}
440
441		template<class Derived, class Traits, class Service, class Acceptor>
442		native_handle_type
443	6x	reactor_descriptor<Derived, Traits, Service, Acceptor>::
444		release_descriptor() noexcept
445		{
446	6x	quiesce();
447
448		// Do NOT close -- the caller takes ownership.
449	6x	native_handle_type released = fd_;
450	6x	fd_ = -1;
451	6x	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	6x	reactor_descriptor<Derived, Traits, Service, Acceptor>::op_to_desc_slot(
461		base_op& op) noexcept
462		{
463	6x	if (&op == static_cast<void*>(&rd_))
464	6x	return &desc_state_.read_op;
465	✗	if (&op == static_cast<void*>(&wr_))
466	✗	return &desc_state_.write_op;
467	✗	if (&op == static_cast<void*>(&wait_rd_))
468	✗	return &desc_state_.wait_read_op;
469	✗	if (&op == static_cast<void*>(&wait_wr_))
470	✗	return &desc_state_.wait_write_op;
471	✗	if (&op == static_cast<void*>(&wait_er_))
472	✗	return &desc_state_.wait_error_op;
473	✗	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	121x	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	121x	auto& op = rd_;
491	121x	op.reset();
492	121x	op.cont = &cont;
493	121x	op.ex = ex;
494	121x	op.ec_out = ec;
495	121x	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	121x	if (fd_ < 0)
500		{
501	✗	op.start(token, static_cast<Derived*>(this));
502	✗	op.object_ref_ = detail::object_ref(this);
503	✗	op.complete(EBADF, 0);
504	✗	svc_.post(&op);
505	✗	return std::noop_coroutine();
506		}
507
508	121x	capy::mutable_buffer bufs[read_op::max_buffers];
509	121x	op.iovec_count =
510	121x	static_cast<int>(param.copy_to(bufs, read_op::max_buffers));
511
512	121x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
513		{
514	2x	op.empty_buffer_read = true;
515	2x	op.start(token, static_cast<Derived*>(this));
516	2x	op.object_ref_ = detail::object_ref(this);
517	2x	op.complete(0, 0);
518	2x	svc_.post(&op);
519	2x	return std::noop_coroutine();
520		}
521
522		// The first transferring operation is what arms O_NONBLOCK; assign()
523		// and wait() never do.
524	119x	if (int const nerr = arm_nonblocking())
525		{
526	✗	op.start(token, static_cast<Derived*>(this));
527	✗	op.object_ref_ = detail::object_ref(this);
528	✗	op.complete(nerr, 0);
529	✗	svc_.post(&op);
530	✗	return std::noop_coroutine();
531		}
532
533	240x	for (int i = 0; i < op.iovec_count; ++i)
534		{
535	121x	op.iovecs[i].iov_base = bufs[i].data();
536	121x	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	119x	if (op.iovec_count == 1)
543		{
544		do
545		{
546	117x	n = ::read(fd_, bufs[0].data(), bufs[0].size());
547		}
548	117x	while (n < 0 && errno == EINTR);
549		}
550		else
551		{
552		do
553		{
554	2x	n = ::readv(fd_, op.iovecs, op.iovec_count);
555		}
556	2x	while (n < 0 && errno == EINTR);
557		}
558
559	119x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
560		{
561	57x	int err = (n < 0) ? errno : 0;
562	57x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
563
564	57x	if (svc_.scheduler().try_consume_inline_budget())
565		{
566	6x	if (err)
567	✗	*ec = make_err(err);
568	6x	else if (n == 0)
569	2x	*ec = capy::error::eof;
570		else
571	4x	*ec = {};
572	6x	*bytes_out = bytes;
573	6x	return dispatch_coro(ex, cont);
574		}
575	51x	op.start(token, static_cast<Derived*>(this));
576	51x	op.object_ref_ = detail::object_ref(this);
577	51x	op.complete(err, bytes);
578	51x	svc_.post(&op);
579	51x	return std::noop_coroutine();
580		}
581
582		// EAGAIN — register with reactor
583	62x	op.fd = fd_;
584	62x	op.start(token, static_cast<Derived*>(this));
585	62x	op.object_ref_ = detail::object_ref(this);
586
587	62x	this->register_op(op, desc_state_.read_op, desc_state_.read_ready);
588	62x	return std::noop_coroutine();
589		}
590
591		template<class Derived, class Traits, class Service, class Acceptor>
592		std::coroutine_handle<>
593	20x	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	20x	auto& op = wr_;
602	20x	op.reset();
603	20x	op.cont = &cont;
604	20x	op.ex = ex;
605	20x	op.ec_out = ec;
606	20x	op.bytes_out = bytes_out;
607
608	20x	if (fd_ < 0)
609		{
610	✗	op.start(token, static_cast<Derived*>(this));
611	✗	op.object_ref_ = detail::object_ref(this);
612	✗	op.complete(EBADF, 0);
613	✗	svc_.post(&op);
614	✗	return std::noop_coroutine();
615		}
616
617	20x	capy::mutable_buffer bufs[write_op::max_buffers];
618	20x	op.iovec_count =
619	20x	static_cast<int>(param.copy_to(bufs, write_op::max_buffers));
620
621	20x	if (op.iovec_count == 0 \|\| (op.iovec_count == 1 && bufs[0].size() == 0))
622		{
623	✗	op.start(token, static_cast<Derived*>(this));
624	✗	op.object_ref_ = detail::object_ref(this);
625	✗	op.complete(0, 0);
626	✗	svc_.post(&op);
627	✗	return std::noop_coroutine();
628		}
629
630	20x	if (int const nerr = arm_nonblocking())
631		{
632	✗	op.start(token, static_cast<Derived*>(this));
633	✗	op.object_ref_ = detail::object_ref(this);
634	✗	op.complete(nerr, 0);
635	✗	svc_.post(&op);
636	✗	return std::noop_coroutine();
637		}
638
639	42x	for (int i = 0; i < op.iovec_count; ++i)
640		{
641	22x	op.iovecs[i].iov_base = bufs[i].data();
642	22x	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	20x	if (op.iovec_count == 1)
648		{
649	36x	n = write_op::write_policy::write_one(
650	18x	fd_, bufs[0].data(), bufs[0].size());
651		}
652		else
653		{
654	2x	n = write_op::write_policy::write(fd_, op.iovecs, op.iovec_count);
655		}
656
657	20x	if (n >= 0 \|\| (errno != EAGAIN && errno != EWOULDBLOCK))
658		{
659	14x	int err = (n < 0) ? errno : 0;
660	14x	auto bytes = (n > 0) ? static_cast<std::size_t>(n) : std::size_t(0);
661
662	14x	if (svc_.scheduler().try_consume_inline_budget())
663		{
664	2x	*ec = err ? make_err(err) : std::error_code{};
665	2x	*bytes_out = bytes;
666	2x	return dispatch_coro(ex, cont);
667		}
668	12x	op.start(token, static_cast<Derived*>(this));
669	12x	op.object_ref_ = detail::object_ref(this);
670	12x	op.complete(err, bytes);
671	12x	svc_.post(&op);
672	12x	return std::noop_coroutine();
673		}
674
675		// EAGAIN — register with reactor
676	6x	op.fd = fd_;
677	6x	op.start(token, static_cast<Derived*>(this));
678	6x	op.object_ref_ = detail::object_ref(this);
679
680	6x	this->register_op(op, desc_state_.write_op, desc_state_.write_ready, true);
681	6x	return std::noop_coroutine();
682		}
683
684		template<class Derived, class Traits, class Service, class Acceptor>
685		std::coroutine_handle<>
686	37x	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	37x	if (w == wait_type::read)
699		{
700	16x	op_ptr = &wait_rd_;
701	16x	desc_slot_ptr = &desc_state_.wait_read_op;
702	16x	event = reactor_event_read;
703		}
704	21x	else if (w == wait_type::write)
705		{
706	8x	op_ptr = &wait_wr_;
707	8x	desc_slot_ptr = &desc_state_.wait_write_op;
708	8x	event = reactor_event_write;
709		}
710		else // wait_type::error
711		{
712	13x	op_ptr = &wait_er_;
713	13x	desc_slot_ptr = &desc_state_.wait_error_op;
714	13x	event = reactor_event_error;
715		}
716
717	37x	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	37x	int perr = 0;
724	37x	if (wait_op::probe(fd_, event, perr))
725		{
726	12x	if (svc_.scheduler().try_consume_inline_budget())
727		{
728	2x	*ec = perr ? make_err(perr) : std::error_code{};
729	2x	return dispatch_coro(ex, cont);
730		}
731	10x	op.reset();
732	10x	op.wait_event = event;
733	10x	op.cont = &cont;
734	10x	op.ex = ex;
735	10x	op.ec_out = ec;
736	10x	op.fd = fd_;
737	10x	op.start(token, static_cast<Derived*>(this));
738	10x	op.object_ref_ = detail::object_ref(this);
739	10x	op.complete(perr, 0);
740	10x	svc_.post(&op);
741	10x	return std::noop_coroutine();
742		}
743
744	25x	op.reset();
745	25x	op.wait_event = event;
746	25x	op.cont = &cont;
747	25x	op.ex = ex;
748	25x	op.ec_out = ec;
749	25x	op.fd = fd_;
750	25x	op.start(token, static_cast<Derived*>(this));
751	25x	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	25x	bool force_probe = true;
757	25x	this->register_op(
758		op, *desc_slot_ptr, force_probe, event == reactor_event_write);
759	25x	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
767