include/boost/corosio/native/detail/kqueue/kqueue_scheduler.hpp

98.3% Lines (169 / 172) 80.7% Branches (92 / 114) 100.0% Functions (15 / 15)

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

kqueue_scheduler.hpp

f(x) Functions (15)

Function Calls Lines Branches Blocks

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2026 Steve Gerbino
3			// Copyright (c) 2026 Michael Vandeberg
4			//
5			// Distributed under the Boost Software License, Version 1.0. (See accompanying
6			// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7			//
8			// Official repository: https://github.com/cppalliance/corosio
9			//
10
11			#ifndef BOOST_COROSIO_NATIVE_DETAIL_KQUEUE_KQUEUE_SCHEDULER_HPP
12			#define BOOST_COROSIO_NATIVE_DETAIL_KQUEUE_KQUEUE_SCHEDULER_HPP
13
14			#include <boost/corosio/detail/platform.hpp>
15
16			#if BOOST_COROSIO_HAS_KQUEUE
17
18			#include <boost/corosio/detail/config.hpp>
19			#include <boost/capy/ex/execution_context.hpp>
20
21			#include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
22			#include <boost/corosio/native/detail/reactor/reactor_signal_pipe.hpp>
23
24			#include <boost/corosio/native/detail/kqueue/kqueue_traits.hpp>
25			#include <boost/corosio/detail/timer_service.hpp>
26			#include <boost/corosio/native/detail/make_err.hpp>
27
28			#include <boost/corosio/detail/except.hpp>
29
30			#include <atomic>
31			#include <chrono>
32			#include <cstdint>
33			#include <limits>
34			#include <mutex>
35			#include <tuple>
36			#include <vector>
37
38			#include <errno.h>
39			#include <fcntl.h>
40			#include <sys/event.h>
41			#include <sys/time.h>
42			#include <unistd.h>
43
44			namespace boost::corosio::detail {
45
46			struct kqueue_op;
47
48			/* Every call on a kqueue goes through kqueue_call, because Darwin
49			needs kevent64 for one case: a poll with a zero timeout that finds
50			nothing ready. kevent with a zero timespec takes ~12us there; only
51			KEVENT_FLAG_IMMEDIATE returns at once. A kqueue accepts a single
52			flavor of call, so the registrations use kevent64 as well.
53			*/
54			#if defined(__APPLE__)
55			using kqueue_event = struct kevent64_s;
56			#else
57			using kqueue_event = struct kevent;
58			#endif
59
60			/// Fill @p ev, as EV_SET does.
61			inline void
62		110475x	kqueue_set(
63			kqueue_event& ev,
64			int ident,
65			short filter,
66			unsigned short flags,
67			unsigned fflags,
68			void* udata) noexcept
69			{
70			#if defined(__APPLE__)
71		110475x	EV_SET64(
72			&ev, static_cast<std::uint64_t>(ident), filter, flags, fflags, 0,
73			reinterpret_cast<std::uint64_t>(udata), 0, 0);
74			#else
75			EV_SET(&ev, static_cast<uintptr_t>(ident), filter, flags, fflags, 0, udata);
76			#endif
77		110475x	}
78
79			/// The udata stored by kqueue_set.
80			inline void*
81		81985x	kqueue_udata(kqueue_event const& ev) noexcept
82			{
83			#if defined(__APPLE__)
84		81985x	return reinterpret_cast<void*>(static_cast<std::uintptr_t>(ev.udata));
85			#else
86			return ev.udata;
87			#endif
88			}
89
90			/// Call kevent on @p kq; a zero @p ts polls without waiting.
91			inline int
92		300362x	kqueue_call(
93			int kq,
94			kqueue_event const* changes,
95			int nchanges,
96			kqueue_event* events,
97			int nevents,
98			struct timespec const* ts) noexcept
99			{
100			#if defined(__APPLE__)
101		300362x	unsigned flags = 0;
102	6/6 ✓ Branch 0 taken 201729 times. ✓ Branch 1 taken 98633 times. ✓ Branch 2 taken 200830 times. ✓ Branch 3 taken 899 times. ✓ Branch 4 taken 52347 times. ✓ Branch 5 taken 148483 times.	300362x	if (ts && ts->tv_sec == 0 && ts->tv_nsec == 0)
103			{
104		148483x	flags = KEVENT_FLAG_IMMEDIATE;
105		148483x	ts = nullptr;
106		148483x	}
107	1/2 ✓ Branch 0 taken 300362 times. ✗ Branch 1 not taken.	300362x	return ::kevent64(kq, changes, nchanges, events, nevents, flags, ts);
108			#else
109			return ::kevent(kq, changes, nchanges, events, nevents, ts);
110			#endif
111			}
112
113			/** macOS/BSD scheduler using kqueue for I/O multiplexing.
114
115			This scheduler implements the scheduler interface using the BSD kqueue
116			API for efficient I/O event notification. It uses a single reactor model
117			where one thread runs kevent() while other threads
118			wait on a condition variable for handler work. This design provides:
119
120			- Handler parallelism: N posted handlers can execute on N threads
121			- No thundering herd: condition_variable wakes exactly one thread
122			- IOCP parity: Behavior matches Windows I/O completion port semantics
123
124			When threads call run(), they first try to execute queued handlers.
125			If the queue is empty and no reactor is running, one thread becomes
126			the reactor and runs kevent(). Other threads wait on a condition
127			variable until handlers are available.
128
129			kqueue uses EV_CLEAR for edge-triggered semantics (equivalent to
130			epoll's EPOLLET). A descriptor gets EVFILT_READ at registration and
131			EVFILT_WRITE when a write-direction operation first parks, as asio
132			does; both stay registered until the descriptor is closed.
133
134			@par Thread Safety
135			All public member functions are thread-safe.
136			*/
137			class BOOST_COROSIO_DECL kqueue_scheduler final : public reactor_scheduler
138			{
139			public:
140			/** Construct the scheduler.
141
142			Creates a kqueue file descriptor via kqueue(), sets
143			close-on-exec, and registers EVFILT_USER for reactor
144			interruption. On failure the kqueue fd is closed before
145			throwing.
146
147			@param ctx Reference to the owning execution_context.
148			@param concurrency_hint Hint for expected thread count (unused).
149
150			@throws std::system_error if kqueue() fails, if setting
151			FD_CLOEXEC on the kqueue fd fails, or if registering
152			the EVFILT_USER event fails. The error code contains
153			the errno from the failed syscall.
154			*/
155			kqueue_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
156
157			/** Destructor.
158
159			Closes the kqueue file descriptor if valid. Does not throw.
160			*/
161			~kqueue_scheduler();
162
163			kqueue_scheduler(kqueue_scheduler const&) = delete;
164			kqueue_scheduler& operator=(kqueue_scheduler const&) = delete;
165
166			/// Shut down the scheduler, draining pending operations.
167			void shutdown() override;
168
169			/// Apply runtime configuration, resizing the event buffer.
170			void configure_reactor(
171			unsigned max_events,
172			unsigned budget_init,
173			unsigned budget_max,
174			unsigned unassisted) override;
175
176			/** Return the kqueue file descriptor.
177
178			Used by socket services to register file descriptors
179			for I/O event notification.
180
181			@return The kqueue file descriptor.
182			*/
183			int kq_fd() const noexcept
184			{
185			return kq_fd_;
186			}
187
188			/** Register a descriptor for persistent monitoring.
189
190			Adds EVFILT_READ (EV_CLEAR) for @a fd and stores @a desc in the
191			kevent udata field so that the reactor can dispatch events to
192			the correct reactor_descriptor_state. EVFILT_WRITE is added later
193			by ensure_write_registered.
194
195			@param fd The file descriptor to register.
196			@param desc Pointer to the caller-owned reactor_descriptor_state.
197
198			@return The error if kevent(EV_ADD) fails, otherwise a default
199			constructed error code.
200			*/
201			std::error_code
202			register_descriptor(int fd, reactor_descriptor_state* desc) const;
203
204			/** Add EVFILT_WRITE for @a fd if it is not registered yet.
205
206			Called with `desc->mutex` held, when a write-direction
207			operation is about to park.
208
209			@param fd The registered file descriptor.
210			@param desc Its reactor_descriptor_state.
211
212			@return The kernel's refusal of the filter, otherwise a default
213			constructed error code.
214			*/
215			std::error_code ensure_write_registered(
216			int fd, reactor_descriptor_state* desc) const noexcept;
217
218			/** Deregister a persistently registered descriptor.
219
220			Issues kevent(EV_DELETE) for both EVFILT_READ and EVFILT_WRITE.
221			Errors are silently ignored because the fd may already be
222			closed and kqueue automatically removes closed descriptors.
223
224			@param fd The file descriptor to deregister.
225			*/
226			void deregister_descriptor(int fd) const;
227
228			/// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
229		289x	[[nodiscard]] std::error_code register_signal_reader(int read_fd) override
230			{
231		289x	return register_descriptor(read_fd, signal_pipe_reader_.arm());
232			}
233
234			private:
235			void run_task(lock_type& lock, context_type& ctx, long timeout_us) override;
236			void interrupt_reactor() const override;
237			long calculate_timeout(long requested_timeout_us) const;
238
239			int kq_fd_;
240
241			// Watches the global signal self-pipe's read end (armed lazily by
242			// register_signal_reader on the first signal registration).
243			reactor_signal_pipe_reader signal_pipe_reader_;
244
245			// EVFILT_USER idempotency
246		2714x	mutable std::atomic<bool> user_event_armed_{false};
247
248			// Event buffer sized from max_events_per_poll_.
249			std::vector<kqueue_event> event_buffer_;
250			};
251
252		8142x	inline kqueue_scheduler::kqueue_scheduler(capy::execution_context&, int)
253		2714x	: kq_fd_(-1)
254	1/2 ✓ Branch 0 taken 2714 times. ✗ Branch 1 not taken.	2714x	, event_buffer_(max_events_per_poll_)
255		5428x	{
256	1/2 ✓ Branch 0 taken 2714 times. ✗ Branch 1 not taken.	2714x	kq_fd_ = ::kqueue();
257	2/2 ✓ Branch 0 taken 1 time. ✓ Branch 1 taken 2713 times.	2714x	if (kq_fd_ < 0)
258	2/4 ✓ Branch 0 taken 1 time. ✗ Branch 1 not taken. ✗ Branch 2 not taken. ✓ Branch 3 taken 1 time.	1x	detail::throw_system_error(make_err(errno), "kqueue");
259
260	3/4 ✓ Branch 0 taken 2713 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 1 time. ✓ Branch 3 taken 2712 times.	2713x	if (::fcntl(kq_fd_, F_SETFD, FD_CLOEXEC) == -1)
261			{
262	1/2 ✓ Branch 0 taken 1 time. ✗ Branch 1 not taken.	1x	int errn = errno;
263	1/2 ✓ Branch 0 taken 1 time. ✗ Branch 1 not taken.	1x	::close(kq_fd_);
264	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 1 time.	1x	detail::throw_system_error(make_err(errn), "fcntl (kqueue FD_CLOEXEC)");
265			}
266
267			kqueue_event ev;
268		2712x	kqueue_set(ev, 0, EVFILT_USER, EV_ADD \| EV_CLEAR, 0, nullptr);
269	2/2 ✓ Branch 0 taken 1 time. ✓ Branch 1 taken 2711 times.	2712x	if (kqueue_call(kq_fd_, &ev, 1, nullptr, 0, nullptr) < 0)
270			{
271	1/2 ✓ Branch 0 taken 1 time. ✗ Branch 1 not taken.	1x	int errn = errno;
272	1/2 ✓ Branch 0 taken 1 time. ✗ Branch 1 not taken.	1x	::close(kq_fd_);
273	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 1 time.	1x	detail::throw_system_error(make_err(errn), "kevent (EVFILT_USER)");
274			}
275
276		2711x	completed_ops_.push(&task_op_);
277		5428x	}
278
279		8132x	inline kqueue_scheduler::~kqueue_scheduler()
280		5422x	{
281	2/2 ✓ Branch 0 taken 1 time. ✓ Branch 1 taken 2711 times.	2710x	if (kq_fd_ >= 0)
282	1/2 ✓ Branch 0 taken 2711 times. ✗ Branch 1 not taken.	2711x	::close(kq_fd_);
283		8134x	}
284
285			inline void
286		2711x	kqueue_scheduler::shutdown()
287			{
288		2711x	shutdown_drain();
289
290	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 2709 times.	2711x	if (kq_fd_ >= 0)
291		2709x	interrupt_reactor();
292		2711x	}
293
294			inline void
295		29x	kqueue_scheduler::configure_reactor(
296			unsigned max_events,
297			unsigned budget_init,
298			unsigned budget_max,
299			unsigned unassisted)
300			{
301		29x	reactor_scheduler::configure_reactor(
302		29x	max_events, budget_init, budget_max, unassisted);
303		29x	event_buffer_.resize(max_events_per_poll_);
304		29x	}
305
306			/// Add one edge-triggered filter for @p fd; returns 0 or the errno.
307			inline int
308		37896x	kqueue_add_filter(int kq, int fd, short filter, void* udata) noexcept
309			{
310			kqueue_event ch;
311		37896x	kqueue_set(ch, fd, filter, EV_ADD \| EV_CLEAR \| EV_RECEIPT, 0, udata);
312			kqueue_event receipt;
313	2/2 ✓ Branch 0 taken 37887 times. ✓ Branch 1 taken 9 times.	37896x	if (kqueue_call(kq, &ch, 1, &receipt, 1, nullptr) < 0)
314	1/2 ✓ Branch 0 taken 9 times. ✗ Branch 1 not taken.	9x	return errno;
315	1/2 ✓ Branch 0 taken 37887 times. ✗ Branch 1 not taken.	37887x	return (receipt.flags & EV_ERROR) ? static_cast<int>(receipt.data) : 0;
316		37896x	}
317
318			inline std::error_code
319		25501x	kqueue_scheduler::register_descriptor(
320			int fd, reactor_descriptor_state* desc) const
321			{
322			// Initialized before the add: another run thread's reactor can
323			// dispatch an event for the descriptor as soon as it is added.
324			// A completion queued by an earlier registration of the same state
325			// may still be running, so the fields it reads change under its
326			// mutex, and the mutex mode is only written when it differs.
327	2/2 ✓ Branch 0 taken 25498 times. ✓ Branch 1 taken 3 times.	25501x	if (desc->mutex.enabled() != reactor_io_locking_)
328		3x	desc->mutex.set_enabled(reactor_io_locking_);
329			{
330		25501x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
331		25501x	desc->registered_events = reactor_event_read;
332		25501x	desc->unpollable = false;
333		25501x	desc->fd = fd;
334		25501x	desc->scheduler_ = this;
335		25501x	desc->ready_events_.store(0, std::memory_order_release);
336			// object_ref_ is not touched: a queued invocation from an
337			// earlier registration still owns the reference in it.
338		25501x	desc->read_ready = false;
339		25501x	desc->write_ready = false;
340		25501x	}
341
342	2/2 ✓ Branch 0 taken 25489 times. ✓ Branch 1 taken 12 times.	25501x	if (int const err = kqueue_add_filter(kq_fd_, fd, EVFILT_READ, desc))
343			{
344		12x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
345		12x	desc->registered_events = 0;
346			// EINVAL/ENODEV: a device with no kqfilter. As on epoll's
347			// EPERM, adopt it unwatched.
348	3/4 ✓ Branch 0 taken 8 times. ✓ Branch 1 taken 4 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 8 times.	12x	if (err != EINVAL && err != ENODEV)
349		8x	return make_err(err);
350		4x	desc->unpollable = true;
351		12x	}
352		25493x	return {};
353		25501x	}
354
355			inline std::error_code
356		12900x	kqueue_scheduler::ensure_write_registered(
357			int fd, reactor_descriptor_state* desc) const noexcept
358			{
359			// Added on first need, as asio does: adoption never asks for a
360			// write filter a read-only device would refuse. Adding a filter
361			// reports the current state, so an fd already writable fires at
362			// once and no edge is lost.
363	2/2 ✓ Branch 0 taken 506 times. ✓ Branch 1 taken 12394 times.	12900x	if (desc->registered_events & reactor_event_write)
364		506x	return {};
365	2/2 ✓ Branch 0 taken 1 time. ✓ Branch 1 taken 12393 times.	12394x	if (int err = kqueue_add_filter(kq_fd_, fd, EVFILT_WRITE, desc))
366		1x	return make_err(err);
367		12393x	desc->registered_events \|= reactor_event_write;
368		12393x	return {};
369		12900x	}
370
371			inline void
372		25197x	kqueue_scheduler::deregister_descriptor(int fd) const
373			{
374			// EV_RECEIPT reports each change separately, so a never-added
375			// EVFILT_WRITE (ENOENT) does not stop the EVFILT_READ delete.
376			kqueue_event changes[2];
377		25197x	kqueue_set(changes[0], fd, EVFILT_READ, EV_DELETE \| EV_RECEIPT, 0, nullptr);
378		25197x	kqueue_set(
379		25197x	changes[1], fd, EVFILT_WRITE, EV_DELETE \| EV_RECEIPT, 0, nullptr);
380			kqueue_event receipts[2];
381		25197x	kqueue_call(kq_fd_, changes, 2, receipts, 2, nullptr);
382		25197x	}
383
384			inline void
385		22115x	kqueue_scheduler::interrupt_reactor() const
386			{
387		22115x	bool expected = false;
388	2/2 ✓ Branch 0 taken 2592 times. ✓ Branch 1 taken 19523 times.	22115x	if (user_event_armed_.compare_exchange_strong(
389			expected, true, std::memory_order_acq_rel,
390			std::memory_order_acquire))
391			{
392			kqueue_event ev;
393		19523x	kqueue_set(ev, 0, EVFILT_USER, 0, NOTE_TRIGGER, nullptr);
394	2/2 ✓ Branch 0 taken 19521 times. ✓ Branch 1 taken 2 times.	19523x	if (kqueue_call(kq_fd_, &ev, 1, nullptr, 0, nullptr) < 0)
395			{
396			// The flag is what coalesces later interrupts into a
397			// trigger already queued on the kqueue; a kevent that
398			// failed queued nothing, so leaving it armed would swallow
399			// every interrupt that follows. Disarming keeps the cost to
400			// the interrupts already in flight -- the next one arms and
401			// triggers again, instead of every one after this
402			// coalescing into a trigger that does not exist.
403		2x	user_event_armed_.store(false, std::memory_order_release);
404		2x	}
405		19523x	}
406		22115x	}
407
408			inline long
409		66673x	kqueue_scheduler::calculate_timeout(long requested_timeout_us) const
410			{
411	1/2 ✓ Branch 0 taken 66673 times. ✗ Branch 1 not taken.	66673x	if (requested_timeout_us == 0)
412		✗	return 0;
413
414		66673x	auto* tsvc = timers();
415	2/2 ✓ Branch 0 taken 53877 times. ✓ Branch 1 taken 12796 times.	66673x	if (!tsvc)
416		12796x	return requested_timeout_us;
417		53877x	auto nearest = tsvc->nearest_expiry();
418	2/2 ✓ Branch 0 taken 976 times. ✓ Branch 1 taken 52901 times.	53877x	if (nearest == timer_service::time_point::max())
419		976x	return requested_timeout_us;
420
421		52901x	auto now = std::chrono::steady_clock::now();
422	2/2 ✓ Branch 0 taken 138 times. ✓ Branch 1 taken 52763 times.	52901x	if (nearest <= now)
423		138x	return 0;
424
425		52763x	auto timer_timeout_us =
426		52763x	std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
427		52763x	.count();
428
429		52763x	constexpr auto long_max =
430			static_cast<long long>((std::numeric_limits<long>::max)());
431		52763x	auto capped_timer_us = std::min(
432		52763x	std::max(timer_timeout_us, static_cast<long long>(0)), long_max);
433
434	2/2 ✓ Branch 0 taken 6 times. ✓ Branch 1 taken 52757 times.	52763x	if (requested_timeout_us < 0)
435		52757x	return static_cast<long>(capped_timer_us);
436
437		6x	return static_cast<long>(std::min(
438		6x	static_cast<long long>(requested_timeout_us), capped_timer_us));
439		66673x	}
440
441			inline void
442		214512x	kqueue_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
443			{
444		214512x	long effective_timeout_us =
445	2/2 ✓ Branch 0 taken 147842 times. ✓ Branch 1 taken 66670 times.	214512x	task_interrupted_ ? 0 : calculate_timeout(timeout_us);
446
447	2/2 ✓ Branch 0 taken 147841 times. ✓ Branch 1 taken 66671 times.	214512x	if (lock.owns_lock())
448		66671x	lock.unlock();
449
450		214512x	task_cleanup on_exit{this, &lock, ctx};
451
452			struct timespec ts;
453		214512x	struct timespec* ts_ptr = nullptr;
454	2/2 ✓ Branch 0 taken 13284 times. ✓ Branch 1 taken 201228 times.	214512x	if (effective_timeout_us >= 0)
455			{
456		201228x	ts.tv_sec = effective_timeout_us / 1000000;
457		201228x	ts.tv_nsec = (effective_timeout_us % 1000000) * 1000;
458		201228x	ts_ptr = &ts;
459		201228x	}
460
461		214512x	int nev = kqueue_call(
462		214512x	kq_fd_, nullptr, 0, event_buffer_.data(),
463		214512x	static_cast<int>(event_buffer_.size()), ts_ptr);
464	2/2 ✓ Branch 0 taken 214522 times. ✓ Branch 1 taken 10 times.	214512x	int saved_errno = errno;
465
466	4/4 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 214520 times. ✓ Branch 2 taken 1 time. ✓ Branch 3 taken 1 time.	214522x	if (nev < 0 && saved_errno != EINTR)
467	1/2 ✓ Branch 0 taken 1 time. ✗ Branch 1 not taken.	1x	detail::throw_system_error(make_err(saved_errno), "kevent");
468
469		214521x	ready_queue local_ops;
470
471	2/2 ✓ Branch 0 taken 214521 times. ✓ Branch 1 taken 98797 times.	313318x	for (int i = 0; i < nev; ++i)
472			{
473	2/2 ✓ Branch 0 taken 16811 times. ✓ Branch 1 taken 81986 times.	98797x	if (event_buffer_[i].filter == EVFILT_USER)
474			{
475		16811x	user_event_armed_.store(false, std::memory_order_release);
476		16811x	continue;
477			}
478
479		81986x	auto* desc =
480			static_cast<reactor_descriptor_state*>(
481		81986x	kqueue_udata(event_buffer_[i]));
482	1/2 ✓ Branch 0 taken 81986 times. ✗ Branch 1 not taken.	81986x	if (!desc)
483		✗	continue;
484
485		81986x	std::uint32_t ready = 0;
486
487	2/2 ✓ Branch 0 taken 14085 times. ✓ Branch 1 taken 67901 times.	81986x	if (event_buffer_[i].filter == EVFILT_READ)
488		67901x	ready \|= reactor_event_read;
489	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 14085 times.	14085x	else if (event_buffer_[i].filter == EVFILT_WRITE)
490		14085x	ready \|= reactor_event_write;
491
492	1/2 ✓ Branch 0 taken 81986 times. ✗ Branch 1 not taken.	81986x	if (event_buffer_[i].flags & EV_ERROR)
493		✗	ready \|= reactor_event_error;
494
495	2/2 ✓ Branch 0 taken 80441 times. ✓ Branch 1 taken 1545 times.	81986x	if (event_buffer_[i].flags & EV_EOF)
496			{
497	2/2 ✓ Branch 0 taken 88 times. ✓ Branch 1 taken 1457 times.	1545x	if (event_buffer_[i].filter == EVFILT_READ)
498		1457x	ready \|= reactor_event_read;
499	2/2 ✓ Branch 0 taken 373 times. ✓ Branch 1 taken 1172 times.	1545x	if (event_buffer_[i].fflags != 0)
500		1172x	ready \|= reactor_event_error;
501		1545x	}
502
503			#if BOOST_COROSIO_TSAN
504			// kevent hands the descriptor over in the kernel, where the
505			// sanitizer sees no ordering, and it checks a read-modify-write
506			// before applying its acquire. This load gives it the acquire
507			// that pairs with register_descriptor's release store first.
508			std::ignore = desc->ready_events_.load(std::memory_order_acquire);
509			#endif
510		81986x	desc->add_ready_events(ready);
511
512		81986x	bool expected = false;
513	2/2 ✓ Branch 0 taken 243 times. ✓ Branch 1 taken 81743 times.	81986x	if (desc->is_enqueued_.compare_exchange_strong(
514			expected, true, std::memory_order_acq_rel,
515			std::memory_order_acquire))
516			{
517		81743x	local_ops.push(desc);
518		81743x	}
519		81986x	}
520
521	2/2 ✓ Branch 0 taken 86217 times. ✓ Branch 1 taken 128304 times.	214521x	if (auto* tsvc = timers())
522	1/2 ✓ Branch 0 taken 86217 times. ✗ Branch 1 not taken.	86217x	tsvc->process_expired();
523
524	2/2 ✓ Branch 0 taken 214511 times. ✓ Branch 1 taken 10 times.	214521x	lock.lock();
525
526		214511x	completed_ops_.splice(local_ops);
527		214532x	}
528
529			} // namespace boost::corosio::detail
530
531			#endif // BOOST_COROSIO_HAS_KQUEUE
532
533			#endif // BOOST_COROSIO_NATIVE_DETAIL_KQUEUE_KQUEUE_SCHEDULER_HPP
534