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

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

97.9% Lines (142/145) 100.0% List of functions (13/13) 71.0% Branches (71/100)
[ ] prev/next file n p prev/next uncovered line
kqueue_scheduler.hpp
f(x) Functions (13)
Function Calls Lines Branches Blocks
boost::corosio::detail::kqueue_scheduler::register_signal_reader(int) :151 71x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::kqueue_scheduler(boost::capy::execution_context&, int) :174 2930x 100.0% 59.4% 100.0% boost::corosio::detail::kqueue_scheduler::kqueue_scheduler(boost::capy::execution_context&, int)::'lambda'(void*)::__invoke(void*) :200 5443x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::kqueue_scheduler(boost::capy::execution_context&, int)::'lambda'(void*)::operator void (*)(void*)() const :200 1462x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::kqueue_scheduler(boost::capy::execution_context&, int)::'lambda'(void*)::operator()(void*) const :200 5443x 100.0% 50.0% 100.0% boost::corosio::detail::kqueue_scheduler::~kqueue_scheduler() :212 4386x 100.0% 50.0% 100.0% boost::corosio::detail::kqueue_scheduler::shutdown() :219 1462x 100.0% 50.0% 100.0% boost::corosio::detail::kqueue_scheduler::configure_reactor(unsigned int, unsigned int, unsigned int, unsigned int) :228 22x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::register_descriptor(int, boost::corosio::detail::reactor_descriptor_state*) const :240 11121x 100.0% 100.0% 100.0% boost::corosio::detail::kqueue_scheduler::deregister_descriptor(int) const :268 11044x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::interrupt_reactor() const :281 10321x 100.0% 100.0% 100.0% boost::corosio::detail::kqueue_scheduler::calculate_timeout(long) const :305 45927x 95.0% 87.5% 90.0% boost::corosio::detail::kqueue_scheduler::run_task(boost::corosio::detail::conditionally_enabled_mutex::scoped_lock&, boost::corosio::detail::reactor_scheduler_context&, long) :335 73824x 96.0% 80.0% 92.0%
Line Branch TLA 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 #include <boost/corosio/native/detail/posix/posix_resolver_service.hpp>
28 #include <boost/corosio/native/detail/posix/posix_signal_service.hpp>
29 #include <boost/corosio/native/detail/posix/posix_stream_file_service.hpp>
30 #include <boost/corosio/native/detail/posix/posix_random_access_file_service.hpp>
31
32 #include <boost/corosio/detail/except.hpp>
33
34 #include <atomic>
35 #include <chrono>
36 #include <cstdint>
37 #include <limits>
38 #include <mutex>
39 #include <vector>
40
41 #include <errno.h>
42 #include <fcntl.h>
43 #include <sys/event.h>
44 #include <sys/time.h>
45 #include <unistd.h>
46
47 namespace boost::corosio::detail {
48
49 struct kqueue_op;
50
51 /** macOS/BSD scheduler using kqueue for I/O multiplexing.
52
53 This scheduler implements the scheduler interface using the BSD kqueue
54 API for efficient I/O event notification. It uses a single reactor model
55 where one thread runs kevent() while other threads
56 wait on a condition variable for handler work. This design provides:
57
58 - Handler parallelism: N posted handlers can execute on N threads
59 - No thundering herd: condition_variable wakes exactly one thread
60 - IOCP parity: Behavior matches Windows I/O completion port semantics
61
62 When threads call run(), they first try to execute queued handlers.
63 If the queue is empty and no reactor is running, one thread becomes
64 the reactor and runs kevent(). Other threads wait on a condition
65 variable until handlers are available.
66
67 kqueue uses EV_CLEAR for edge-triggered semantics (equivalent to
68 epoll's EPOLLET). File descriptors are registered once with both
69 EVFILT_READ and EVFILT_WRITE and stay registered until closed.
70
71 @par Thread Safety
72 All public member functions are thread-safe.
73 */
74 class BOOST_COROSIO_DECL kqueue_scheduler final : public reactor_scheduler
75 {
76 public:
77 /** Construct the scheduler.
78
79 Creates a kqueue file descriptor via kqueue(), sets
80 close-on-exec, and registers EVFILT_USER for reactor
81 interruption. On failure the kqueue fd is closed before
82 throwing.
83
84 @param ctx Reference to the owning execution_context.
85 @param concurrency_hint Hint for expected thread count (unused).
86
87 @throws std::system_error if kqueue() fails, if setting
88 FD_CLOEXEC on the kqueue fd fails, or if registering
89 the EVFILT_USER event fails. The error code contains
90 the errno from the failed syscall.
91 */
92 kqueue_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
93
94 /** Destructor.
95
96 Closes the kqueue file descriptor if valid. Does not throw.
97 */
98 ~kqueue_scheduler();
99
100 kqueue_scheduler(kqueue_scheduler const&) = delete;
101 kqueue_scheduler& operator=(kqueue_scheduler const&) = delete;
102
103 /// Shut down the scheduler, draining pending operations.
104 void shutdown() override;
105
106 /// Apply runtime configuration, resizing the event buffer.
107 void configure_reactor(
108 unsigned max_events,
109 unsigned budget_init,
110 unsigned budget_max,
111 unsigned unassisted) override;
112
113 /** Return the kqueue file descriptor.
114
115 Used by socket services to register file descriptors
116 for I/O event notification.
117
118 @return The kqueue file descriptor.
119 */
120 int kq_fd() const noexcept
121 {
122 return kq_fd_;
123 }
124
125 /** Register a descriptor for persistent monitoring.
126
127 Adds EVFILT_READ and EVFILT_WRITE (both EV_CLEAR) for @a fd
128 and stores @a desc in the kevent udata field so that the
129 reactor can dispatch events to the correct reactor_descriptor_state.
130
131 @param fd The file descriptor to register.
132 @param desc Pointer to the caller-owned reactor_descriptor_state.
133
134 @return The error if kevent(EV_ADD) fails, otherwise a default
135 constructed error code.
136 */
137 std::error_code
138 register_descriptor(int fd, reactor_descriptor_state* desc) const;
139
140 /** Deregister a persistently registered descriptor.
141
142 Issues kevent(EV_DELETE) for both EVFILT_READ and EVFILT_WRITE.
143 Errors are silently ignored because the fd may already be
144 closed and kqueue automatically removes closed descriptors.
145
146 @param fd The file descriptor to deregister.
147 */
148 void deregister_descriptor(int fd) const;
149
150 /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
151 71x [[nodiscard]] std::error_code register_signal_reader(int read_fd) override
152 {
153 71x return register_descriptor(read_fd, signal_pipe_reader_.arm());
154 }
155
156 private:
157 void run_task(lock_type& lock, context_type& ctx, long timeout_us) override;
158 void interrupt_reactor() const override;
159 long calculate_timeout(long requested_timeout_us) const;
160
161 int kq_fd_;
162
163 // Watches the global signal self-pipe's read end (armed lazily by
164 // register_signal_reader on the first signal registration).
165 reactor_signal_pipe_reader signal_pipe_reader_;
166
167 // EVFILT_USER idempotency
168 1465x mutable std::atomic<bool> user_event_armed_{false};
169
170 // Event buffer sized from max_events_per_poll_.
171 std::vector<struct kevent> event_buffer_;
172 };
173
174 4395x inline kqueue_scheduler::kqueue_scheduler(capy::execution_context& ctx, int)
175 1465x : kq_fd_(-1)
176
1/2
✓ Branch 0 taken 1465 times.
✗ Branch 1 not taken.
 1465x , event_buffer_(max_events_per_poll_)
177 2930x {
178
1/2
✓ Branch 0 taken 1465 times.
✗ Branch 1 not taken.
 1465x kq_fd_ = ::kqueue();
179
2/2
✓ Branch 0 taken 1 time.
✓ Branch 1 taken 1464 times.
 1465x if (kq_fd_ < 0)
180
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");
181
182
3/4
✓ Branch 0 taken 1464 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 time.
✓ Branch 3 taken 1463 times.
 1464x if (::fcntl(kq_fd_, F_SETFD, FD_CLOEXEC) == -1)
183 {
184
1/2
✓ Branch 0 taken 1 time.
✗ Branch 1 not taken.
 1x int errn = errno;
185
1/2
✓ Branch 0 taken 1 time.
✗ Branch 1 not taken.
 1x ::close(kq_fd_);
186
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 time.
 1x detail::throw_system_error(make_err(errn), "fcntl (kqueue FD_CLOEXEC)");
187 }
188
189 struct kevent ev;
190 1463x EV_SET(&ev, 0, EVFILT_USER, EV_ADD | EV_CLEAR, 0, 0, nullptr);
191
3/4
✓ Branch 0 taken 1463 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 1 time.
✓ Branch 3 taken 1462 times.
 1463x if (::kevent(kq_fd_, &ev, 1, nullptr, 0, nullptr) < 0)
192 {
193
1/2
✓ Branch 0 taken 1 time.
✗ Branch 1 not taken.
 1x int errn = errno;
194
1/2
✓ Branch 0 taken 1 time.
✗ Branch 1 not taken.
 1x ::close(kq_fd_);
195
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1 time.
 1x detail::throw_system_error(make_err(errn), "kevent (EVFILT_USER)");
196 }
197
198
1/2
✓ Branch 0 taken 1462 times.
✗ Branch 1 not taken.
 1462x timer_svc_ = &get_timer_service(ctx, *this);
199 2924x timer_svc_->set_on_earliest_changed(
200 6905x timer_service::callback(this, [](void* p) {
201 5443x static_cast<kqueue_scheduler*>(p)->interrupt_reactor();
202 5443x }));
203
204
1/2
✓ Branch 0 taken 1462 times.
✗ Branch 1 not taken.
 1462x get_resolver_service(ctx, *this);
205
1/2
✓ Branch 0 taken 1462 times.
✗ Branch 1 not taken.
 1462x get_signal_service(ctx, *this);
206
1/2
✓ Branch 0 taken 1462 times.
✗ Branch 1 not taken.
 1462x get_stream_file_service(ctx, *this);
207
1/2
✓ Branch 0 taken 1462 times.
✗ Branch 1 not taken.
 1462x get_random_access_file_service(ctx, *this);
208
209 1462x completed_ops_.push(&task_op_);
210 2930x }
211
212 4386x inline kqueue_scheduler::~kqueue_scheduler()
213 2924x {
214
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1462 times.
 1462x if (kq_fd_ >= 0)
215
1/2
✓ Branch 0 taken 1462 times.
✗ Branch 1 not taken.
 1462x ::close(kq_fd_);
216 4386x }
217
218 inline void
219 1462x kqueue_scheduler::shutdown()
220 {
221 1462x shutdown_drain();
222
223
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1462 times.
 1462x if (kq_fd_ >= 0)
224 1462x interrupt_reactor();
225 1462x }
226
227 inline void
228 22x kqueue_scheduler::configure_reactor(
229 unsigned max_events,
230 unsigned budget_init,
231 unsigned budget_max,
232 unsigned unassisted)
233 {
234 22x reactor_scheduler::configure_reactor(
235 22x max_events, budget_init, budget_max, unassisted);
236 22x event_buffer_.resize(max_events_per_poll_);
237 22x }
238
239 inline std::error_code
240 11121x kqueue_scheduler::register_descriptor(
241 int fd, reactor_descriptor_state* desc) const
242 {
243 struct kevent changes[2];
244 11121x EV_SET(
245 &changes[0], static_cast<uintptr_t>(fd), EVFILT_READ, EV_ADD | EV_CLEAR,
246 0, 0, desc);
247 11121x EV_SET(
248 &changes[1], static_cast<uintptr_t>(fd), EVFILT_WRITE,
249 EV_ADD | EV_CLEAR, 0, 0, desc);
250
251
2/2
✓ Branch 0 taken 11113 times.
✓ Branch 1 taken 8 times.
 11121x if (::kevent(kq_fd_, changes, 2, nullptr, 0, nullptr) < 0)
252 8x return make_err(errno);
253
254 11113x desc->registered_events = reactor_event_read | reactor_event_write;
255 11113x desc->fd = fd;
256 11113x desc->scheduler_ = this;
257 11113x desc->mutex.set_enabled(reactor_io_locking_);
258 11113x desc->ready_events_.store(0, std::memory_order_relaxed);
259
260 11113x conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
261 11113x desc->impl_ref_.reset();
262 11113x desc->read_ready = false;
263 11113x desc->write_ready = false;
264 11113x return {};
265 11121x }
266
267 inline void
268 11044x kqueue_scheduler::deregister_descriptor(int fd) const
269 {
270 struct kevent changes[2];
271 11044x EV_SET(
272 &changes[0], static_cast<uintptr_t>(fd), EVFILT_READ, EV_DELETE, 0, 0,
273 nullptr);
274 11044x EV_SET(
275 &changes[1], static_cast<uintptr_t>(fd), EVFILT_WRITE, EV_DELETE, 0, 0,
276 nullptr);
277 11044x ::kevent(kq_fd_, changes, 2, nullptr, 0, nullptr);
278 11044x }
279
280 inline void
281 10321x kqueue_scheduler::interrupt_reactor() const
282 {
283 10321x bool expected = false;
284
2/2
✓ Branch 0 taken 1768 times.
✓ Branch 1 taken 8553 times.
 10321x if (user_event_armed_.compare_exchange_strong(
285 expected, true, std::memory_order_acq_rel,
286 std::memory_order_acquire))
287 {
288 struct kevent ev;
289 8553x EV_SET(&ev, 0, EVFILT_USER, 0, NOTE_TRIGGER, 0, nullptr);
290
2/2
✓ Branch 0 taken 8551 times.
✓ Branch 1 taken 2 times.
 8553x if (::kevent(kq_fd_, &ev, 1, nullptr, 0, nullptr) < 0)
291 {
292 // The flag is what coalesces later interrupts into a
293 // trigger already queued on the kqueue; a kevent that
294 // failed queued nothing, so leaving it armed would swallow
295 // every interrupt that follows. Disarming keeps the cost to
296 // the interrupts already in flight -- the next one arms and
297 // triggers again, instead of every one after this
298 // coalescing into a trigger that does not exist.
299 2x user_event_armed_.store(false, std::memory_order_release);
300 2x }
301 8553x }
302 10321x }
303
304 inline long
305 45927x kqueue_scheduler::calculate_timeout(long requested_timeout_us) const
306 {
307
1/2
✓ Branch 0 taken 45927 times.
✗ Branch 1 not taken.
 45927x if (requested_timeout_us == 0)
308 ✗ return 0;
309
310 45927x auto nearest = timer_svc_->nearest_expiry();
311
2/2
✓ Branch 0 taken 3554 times.
✓ Branch 1 taken 42373 times.
 45927x if (nearest == timer_service::time_point::max())
312 3554x return requested_timeout_us;
313
314 42373x auto now = std::chrono::steady_clock::now();
315
2/2
✓ Branch 0 taken 147 times.
✓ Branch 1 taken 42226 times.
 42373x if (nearest <= now)
316 147x return 0;
317
318 42226x auto timer_timeout_us =
319 42226x std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
320 42226x .count();
321
322 42226x constexpr auto long_max =
323 static_cast<long long>((std::numeric_limits<long>::max)());
324 42226x auto capped_timer_us = std::min(
325 42226x std::max(timer_timeout_us, static_cast<long long>(0)), long_max);
326
327
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 42224 times.
 42226x if (requested_timeout_us < 0)
328 42224x return static_cast<long>(capped_timer_us);
329
330 2x return static_cast<long>(std::min(
331 2x static_cast<long long>(requested_timeout_us), capped_timer_us));
332 45927x }
333
334 inline void
335 73824x kqueue_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
336 {
337 73824x long effective_timeout_us =
338
2/2
✓ Branch 0 taken 27897 times.
✓ Branch 1 taken 45927 times.
 73824x task_interrupted_ ? 0 : calculate_timeout(timeout_us);
339
340
2/2
✓ Branch 0 taken 27895 times.
✓ Branch 1 taken 45929 times.
 73824x if (lock.owns_lock())
341 45929x lock.unlock();
342
343 73824x task_cleanup on_exit{this, &lock, ctx};
344
345 struct timespec ts;
346 73824x struct timespec* ts_ptr = nullptr;
347
2/2
✓ Branch 0 taken 3539 times.
✓ Branch 1 taken 70285 times.
 73824x if (effective_timeout_us >= 0)
348 {
349 70285x ts.tv_sec = effective_timeout_us / 1000000;
350 70285x ts.tv_nsec = (effective_timeout_us % 1000000) * 1000;
351 70285x ts_ptr = &ts;
352 70285x }
353
354
1/2
✓ Branch 0 taken 73824 times.
✗ Branch 1 not taken.
 73824x int nev = ::kevent(
355 73824x kq_fd_, nullptr, 0, event_buffer_.data(),
356 73824x static_cast<int>(event_buffer_.size()), ts_ptr);
357
1/2
✓ Branch 0 taken 73824 times.
✗ Branch 1 not taken.
 73824x int saved_errno = errno;
358
359
4/4
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 73822 times.
✓ Branch 2 taken 1 time.
✓ Branch 3 taken 1 time.
 73824x if (nev < 0 && saved_errno != EINTR)
360
1/2
✓ Branch 0 taken 1 time.
✗ Branch 1 not taken.
 1x detail::throw_system_error(make_err(saved_errno), "kevent");
361
362 73823x ready_queue local_ops;
363
364
2/2
✓ Branch 0 taken 73823 times.
✓ Branch 1 taken 91622 times.
 165445x for (int i = 0; i < nev; ++i)
365 {
366
2/2
✓ Branch 0 taken 7089 times.
✓ Branch 1 taken 84533 times.
 91622x if (event_buffer_[i].filter == EVFILT_USER)
367 {
368 7089x user_event_armed_.store(false, std::memory_order_release);
369 7089x continue;
370 }
371
372 84533x auto* desc =
373 84533x static_cast<reactor_descriptor_state*>(event_buffer_[i].udata);
374
1/2
✓ Branch 0 taken 84533 times.
✗ Branch 1 not taken.
 84533x if (!desc)
375 ✗ continue;
376
377 84533x std::uint32_t ready = 0;
378
379
2/2
✓ Branch 0 taken 43499 times.
✓ Branch 1 taken 41034 times.
 84533x if (event_buffer_[i].filter == EVFILT_READ)
380 41034x ready |= reactor_event_read;
381
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 43499 times.
 43499x else if (event_buffer_[i].filter == EVFILT_WRITE)
382 43499x ready |= reactor_event_write;
383
384
1/2
✓ Branch 0 taken 84533 times.
✗ Branch 1 not taken.
 84533x if (event_buffer_[i].flags & EV_ERROR)
385 ✗ ready |= reactor_event_error;
386
387
2/2
✓ Branch 0 taken 84363 times.
✓ Branch 1 taken 170 times.
 84533x if (event_buffer_[i].flags & EV_EOF)
388 {
389
2/2
✓ Branch 0 taken 54 times.
✓ Branch 1 taken 116 times.
 170x if (event_buffer_[i].filter == EVFILT_READ)
390 116x ready |= reactor_event_read;
391
2/2
✓ Branch 0 taken 84 times.
✓ Branch 1 taken 86 times.
 170x if (event_buffer_[i].fflags != 0)
392 86x ready |= reactor_event_error;
393 170x }
394
395 84533x desc->add_ready_events(ready);
396
397 84533x bool expected = false;
398
2/2
✓ Branch 0 taken 175 times.
✓ Branch 1 taken 84358 times.
 84533x if (desc->is_enqueued_.compare_exchange_strong(
399 expected, true, std::memory_order_acq_rel,
400 std::memory_order_acquire))
401 {
402 84358x local_ops.push(desc);
403 84358x }
404 84533x }
405
406
1/2
✓ Branch 0 taken 73823 times.
✗ Branch 1 not taken.
 73823x timer_svc_->process_expired();
407
408
1/2
✓ Branch 0 taken 73823 times.
✗ Branch 1 not taken.
 73823x lock.lock();
409
410 73823x completed_ops_.splice(local_ops);
411 73824x }
412
413 } // namespace boost::corosio::detail
414
415 #endif // BOOST_COROSIO_HAS_KQUEUE
416
417 #endif // BOOST_COROSIO_NATIVE_DETAIL_KQUEUE_KQUEUE_SCHEDULER_HPP
418