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

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

90.2% Lines (129/143) 100.0% List of functions (13/13) 53.1% Branches (52/98)
[ ] 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) :152 51x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::kqueue_scheduler(boost::capy::execution_context&, int) :177 2484x 61.1% 25.0% 100.0% boost::corosio::detail::kqueue_scheduler::kqueue_scheduler(boost::capy::execution_context&, int)::'lambda'(void*)::__invoke(void*) :203 6105x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::kqueue_scheduler(boost::capy::execution_context&, int)::'lambda'(void*)::operator void (*)(void*)() const :203 1242x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::kqueue_scheduler(boost::capy::execution_context&, int)::'lambda'(void*)::operator()(void*) const :203 6105x 100.0% 50.0% 100.0% boost::corosio::detail::kqueue_scheduler::~kqueue_scheduler() :215 3726x 100.0% 50.0% 100.0% boost::corosio::detail::kqueue_scheduler::shutdown() :222 1242x 100.0% 50.0% 100.0% boost::corosio::detail::kqueue_scheduler::configure_reactor(unsigned int, unsigned int, unsigned int, unsigned int) :231 18x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::register_descriptor(int, boost::corosio::detail::reactor_descriptor_state*) const :243 13875x 93.8% 50.0% 87.0% boost::corosio::detail::kqueue_scheduler::deregister_descriptor(int) const :270 13824x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::interrupt_reactor() const :283 11758x 100.0% 100.0% 100.0% boost::corosio::detail::kqueue_scheduler::calculate_timeout(long) const :297 53143x 85.0% 75.0% 80.0% boost::corosio::detail::kqueue_scheduler::run_task(boost::corosio::detail::conditionally_enabled_mutex::scoped_lock&, boost::corosio::detail::reactor_scheduler_context*, long) :327 101064x 94.0% 70.0% 82.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 [[nodiscard]] std::error_code
152 51x register_signal_reader(int read_fd) override
153 {
154 51x return register_descriptor(read_fd, signal_pipe_reader_.arm());
155 }
156
157 private:
158 void
159 run_task(lock_type& lock, context_type* ctx,
160 long timeout_us) override;
161 void interrupt_reactor() const override;
162 long calculate_timeout(long requested_timeout_us) const;
163
164 int kq_fd_;
165
166 // Watches the global signal self-pipe's read end (armed lazily by
167 // register_signal_reader on the first signal registration).
168 reactor_signal_pipe_reader signal_pipe_reader_;
169
170 // EVFILT_USER idempotency
171 1242x mutable std::atomic<bool> user_event_armed_{false};
172
173 // Event buffer sized from max_events_per_poll_.
174 std::vector<struct kevent> event_buffer_;
175 };
176
177 3726x inline kqueue_scheduler::kqueue_scheduler(capy::execution_context& ctx, int)
178 1242x : kq_fd_(-1)
179
1/2
✓ Branch 0 taken 1242 times.
✗ Branch 1 not taken.
 1242x , event_buffer_(max_events_per_poll_)
180 2484x {
181
1/2
✓ Branch 0 taken 1242 times.
✗ Branch 1 not taken.
 1242x kq_fd_ = ::kqueue();
182
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1242 times.
 1242x if (kq_fd_ < 0)
183 ✗ detail::throw_system_error(make_err(errno), "kqueue");
184
185
2/4
✓ Branch 0 taken 1242 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 1242 times.
 1242x if (::fcntl(kq_fd_, F_SETFD, FD_CLOEXEC) == -1)
186 {
187 ✗ int errn = errno;
188 ✗ ::close(kq_fd_);
189 ✗ detail::throw_system_error(make_err(errn), "fcntl (kqueue FD_CLOEXEC)");
190 }
191
192 struct kevent ev;
193 1242x EV_SET(&ev, 0, EVFILT_USER, EV_ADD | EV_CLEAR, 0, 0, nullptr);
194
2/4
✓ Branch 0 taken 1242 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 1242 times.
 1242x if (::kevent(kq_fd_, &ev, 1, nullptr, 0, nullptr) < 0)
195 {
196 ✗ int errn = errno;
197 ✗ ::close(kq_fd_);
198 ✗ detail::throw_system_error(make_err(errn), "kevent (EVFILT_USER)");
199 }
200
201
1/2
✓ Branch 0 taken 1242 times.
✗ Branch 1 not taken.
 1242x timer_svc_ = &get_timer_service(ctx, *this);
202 2484x timer_svc_->set_on_earliest_changed(
203 7347x timer_service::callback(this, [](void* p) {
204 6105x static_cast<kqueue_scheduler*>(p)->interrupt_reactor();
205 6105x }));
206
207
1/2
✓ Branch 0 taken 1242 times.
✗ Branch 1 not taken.
 1242x get_resolver_service(ctx, *this);
208
1/2
✓ Branch 0 taken 1242 times.
✗ Branch 1 not taken.
 1242x get_signal_service(ctx, *this);
209
1/2
✓ Branch 0 taken 1242 times.
✗ Branch 1 not taken.
 1242x get_stream_file_service(ctx, *this);
210
1/2
✓ Branch 0 taken 1242 times.
✗ Branch 1 not taken.
 1242x get_random_access_file_service(ctx, *this);
211
212 1242x completed_ops_.push(&task_op_);
213 2484x }
214
215 3726x inline kqueue_scheduler::~kqueue_scheduler()
216 2484x {
217
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1242 times.
 1242x if (kq_fd_ >= 0)
218
1/2
✓ Branch 0 taken 1242 times.
✗ Branch 1 not taken.
 1242x ::close(kq_fd_);
219 3726x }
220
221 inline void
222 1242x kqueue_scheduler::shutdown()
223 {
224 1242x shutdown_drain();
225
226
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1242 times.
 1242x if (kq_fd_ >= 0)
227 1242x interrupt_reactor();
228 1242x }
229
230 inline void
231 18x kqueue_scheduler::configure_reactor(
232 unsigned max_events,
233 unsigned budget_init,
234 unsigned budget_max,
235 unsigned unassisted)
236 {
237 18x reactor_scheduler::configure_reactor(
238 18x max_events, budget_init, budget_max, unassisted);
239 18x event_buffer_.resize(max_events_per_poll_);
240 18x }
241
242 inline std::error_code
243 13875x kqueue_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) const
244 {
245 struct kevent changes[2];
246 13875x EV_SET(
247 &changes[0], static_cast<uintptr_t>(fd), EVFILT_READ, EV_ADD | EV_CLEAR,
248 0, 0, desc);
249 13875x EV_SET(
250 &changes[1], static_cast<uintptr_t>(fd), EVFILT_WRITE,
251 EV_ADD | EV_CLEAR, 0, 0, desc);
252
253
1/2
✓ Branch 0 taken 13875 times.
✗ Branch 1 not taken.
 13875x if (::kevent(kq_fd_, changes, 2, nullptr, 0, nullptr) < 0)
254 ✗ return make_err(errno);
255
256 13875x desc->registered_events = reactor_event_read | reactor_event_write;
257 13875x desc->fd = fd;
258 13875x desc->scheduler_ = this;
259 13875x desc->mutex.set_enabled(reactor_io_locking_);
260 13875x desc->ready_events_.store(0, std::memory_order_relaxed);
261
262 13875x conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
263 13875x desc->impl_ref_.reset();
264 13875x desc->read_ready = false;
265 13875x desc->write_ready = false;
266 13875x return {};
267 13875x }
268
269 inline void
270 13824x kqueue_scheduler::deregister_descriptor(int fd) const
271 {
272 struct kevent changes[2];
273 13824x EV_SET(
274 &changes[0], static_cast<uintptr_t>(fd), EVFILT_READ, EV_DELETE, 0, 0,
275 nullptr);
276 13824x EV_SET(
277 &changes[1], static_cast<uintptr_t>(fd), EVFILT_WRITE, EV_DELETE, 0, 0,
278 nullptr);
279 13824x ::kevent(kq_fd_, changes, 2, nullptr, 0, nullptr);
280 13824x }
281
282 inline void
283 11758x kqueue_scheduler::interrupt_reactor() const
284 {
285 11758x bool expected = false;
286
2/2
✓ Branch 0 taken 1577 times.
✓ Branch 1 taken 10181 times.
 11758x if (user_event_armed_.compare_exchange_strong(
287 expected, true, std::memory_order_acq_rel,
288 std::memory_order_acquire))
289 {
290 struct kevent ev;
291 10181x EV_SET(&ev, 0, EVFILT_USER, 0, NOTE_TRIGGER, 0, nullptr);
292 10181x ::kevent(kq_fd_, &ev, 1, nullptr, 0, nullptr);
293 10181x }
294 11758x }
295
296 inline long
297 53143x kqueue_scheduler::calculate_timeout(long requested_timeout_us) const
298 {
299
1/2
✓ Branch 0 taken 53143 times.
✗ Branch 1 not taken.
 53143x if (requested_timeout_us == 0)
300 ✗ return 0;
301
302 53143x auto nearest = timer_svc_->nearest_expiry();
303
2/2
✓ Branch 0 taken 4374 times.
✓ Branch 1 taken 48769 times.
 53143x if (nearest == timer_service::time_point::max())
304 4374x return requested_timeout_us;
305
306 48769x auto now = std::chrono::steady_clock::now();
307
2/2
✓ Branch 0 taken 100 times.
✓ Branch 1 taken 48669 times.
 48769x if (nearest <= now)
308 100x return 0;
309
310 48669x auto timer_timeout_us =
311 48669x std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
312 48669x .count();
313
314 48669x constexpr auto long_max =
315 static_cast<long long>((std::numeric_limits<long>::max)());
316 48669x auto capped_timer_us = std::min(
317 48669x std::max(timer_timeout_us, static_cast<long long>(0)), long_max);
318
319
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 48669 times.
 48669x if (requested_timeout_us < 0)
320 48669x return static_cast<long>(capped_timer_us);
321
322 ✗ return static_cast<long>(std::min(
323 ✗ static_cast<long long>(requested_timeout_us), capped_timer_us));
324 53143x }
325
326 inline void
327 101064x kqueue_scheduler::run_task(
328 lock_type& lock, context_type* ctx, long timeout_us)
329 {
330 101064x long effective_timeout_us =
331
2/2
✓ Branch 0 taken 47921 times.
✓ Branch 1 taken 53143 times.
 101064x task_interrupted_ ? 0 : calculate_timeout(timeout_us);
332
333
2/2
✓ Branch 0 taken 47919 times.
✓ Branch 1 taken 53145 times.
 101064x if (lock.owns_lock())
334 53145x lock.unlock();
335
336 101064x task_cleanup on_exit{this, &lock, ctx};
337
338 struct timespec ts;
339 101064x struct timespec* ts_ptr = nullptr;
340
2/2
✓ Branch 0 taken 4370 times.
✓ Branch 1 taken 96694 times.
 101064x if (effective_timeout_us >= 0)
341 {
342 96694x ts.tv_sec = effective_timeout_us / 1000000;
343 96694x ts.tv_nsec = (effective_timeout_us % 1000000) * 1000;
344 96694x ts_ptr = &ts;
345 96694x }
346
347
1/2
✓ Branch 0 taken 101064 times.
✗ Branch 1 not taken.
 101064x int nev = ::kevent(
348 101064x kq_fd_, nullptr, 0, event_buffer_.data(),
349 101064x static_cast<int>(event_buffer_.size()), ts_ptr);
350
1/2
✓ Branch 0 taken 101064 times.
✗ Branch 1 not taken.
 101064x int saved_errno = errno;
351
352
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 101064 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 101064x if (nev < 0 && saved_errno != EINTR)
353 ✗ detail::throw_system_error(make_err(saved_errno), "kevent");
354
355 101064x ready_queue local_ops;
356
357
2/2
✓ Branch 0 taken 101064 times.
✓ Branch 1 taken 137681 times.
 238745x for (int i = 0; i < nev; ++i)
358 {
359
2/2
✓ Branch 0 taken 8939 times.
✓ Branch 1 taken 128742 times.
 137681x if (event_buffer_[i].filter == EVFILT_USER)
360 {
361 8939x user_event_armed_.store(false, std::memory_order_release);
362 8939x continue;
363 }
364
365 128742x auto* desc =
366 128742x static_cast<reactor_descriptor_state*>(event_buffer_[i].udata);
367
1/2
✓ Branch 0 taken 128742 times.
✗ Branch 1 not taken.
 128742x if (!desc)
368 ✗ continue;
369
370 128742x std::uint32_t ready = 0;
371
372
2/2
✓ Branch 0 taken 66222 times.
✓ Branch 1 taken 62520 times.
 128742x if (event_buffer_[i].filter == EVFILT_READ)
373 62520x ready |= reactor_event_read;
374
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 66222 times.
 66222x else if (event_buffer_[i].filter == EVFILT_WRITE)
375 66222x ready |= reactor_event_write;
376
377
1/2
✓ Branch 0 taken 128742 times.
✗ Branch 1 not taken.
 128742x if (event_buffer_[i].flags & EV_ERROR)
378 ✗ ready |= reactor_event_error;
379
380
2/2
✓ Branch 0 taken 128628 times.
✓ Branch 1 taken 114 times.
 128742x if (event_buffer_[i].flags & EV_EOF)
381 {
382
2/2
✓ Branch 0 taken 49 times.
✓ Branch 1 taken 65 times.
 114x if (event_buffer_[i].filter == EVFILT_READ)
383 65x ready |= reactor_event_read;
384
2/2
✓ Branch 0 taken 52 times.
✓ Branch 1 taken 62 times.
 114x if (event_buffer_[i].fflags != 0)
385 62x ready |= reactor_event_error;
386 114x }
387
388 128742x desc->add_ready_events(ready);
389
390 128742x bool expected = false;
391
2/2
✓ Branch 0 taken 219 times.
✓ Branch 1 taken 128523 times.
 128742x if (desc->is_enqueued_.compare_exchange_strong(
392 expected, true, std::memory_order_acq_rel,
393 std::memory_order_acquire))
394 {
395 128523x local_ops.push(desc);
396 128523x }
397 128742x }
398
399
1/2
✓ Branch 0 taken 101064 times.
✗ Branch 1 not taken.
 101064x timer_svc_->process_expired();
400
401
1/2
✓ Branch 0 taken 101064 times.
✗ Branch 1 not taken.
 101064x lock.lock();
402
403 101064x completed_ops_.splice(local_ops);
404 101064x }
405
406 } // namespace boost::corosio::detail
407
408 #endif // BOOST_COROSIO_HAS_KQUEUE
409
410 #endif // BOOST_COROSIO_NATIVE_DETAIL_KQUEUE_KQUEUE_SCHEDULER_HPP
411