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) :149 41x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::kqueue_scheduler(boost::capy::execution_context&, int) :174 1490x 61.1% 25.0% 100.0% boost::corosio::detail::kqueue_scheduler::kqueue_scheduler(boost::capy::execution_context&, int)::'lambda'(void*)::__invoke(void*) :200 5433x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::kqueue_scheduler(boost::capy::execution_context&, int)::'lambda'(void*)::operator void (*)(void*)() const :200 745x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::kqueue_scheduler(boost::capy::execution_context&, int)::'lambda'(void*)::operator()(void*) const :200 5433x 100.0% 50.0% 100.0% boost::corosio::detail::kqueue_scheduler::~kqueue_scheduler() :212 2235x 100.0% 50.0% 100.0% boost::corosio::detail::kqueue_scheduler::shutdown() :219 745x 100.0% 50.0% 100.0% boost::corosio::detail::kqueue_scheduler::configure_reactor(unsigned int, unsigned int, unsigned int, unsigned int) :228 16x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::register_descriptor(int, boost::corosio::detail::reactor_descriptor_state*) const :240 11129x 93.3% 50.0% 85.0% boost::corosio::detail::kqueue_scheduler::deregister_descriptor(int) const :266 11088x 100.0% – 100.0% boost::corosio::detail::kqueue_scheduler::interrupt_reactor() const :279 8957x 100.0% 100.0% 100.0% boost::corosio::detail::kqueue_scheduler::calculate_timeout(long) const :293 50712x 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) :323 87043x 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 @throws std::system_error if kevent(EV_ADD) fails.
135 */
136 void register_descriptor(int fd, reactor_descriptor_state* desc) const;
137
138 /** Deregister a persistently registered descriptor.
139
140 Issues kevent(EV_DELETE) for both EVFILT_READ and EVFILT_WRITE.
141 Errors are silently ignored because the fd may already be
142 closed and kqueue automatically removes closed descriptors.
143
144 @param fd The file descriptor to deregister.
145 */
146 void deregister_descriptor(int fd) const;
147
148 /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
149 41x void register_signal_reader(int read_fd) override
150 {
151 41x register_descriptor(read_fd, signal_pipe_reader_.arm());
152 41x }
153
154 private:
155 void
156 run_task(lock_type& lock, context_type* ctx,
157 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 745x 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 2235x inline kqueue_scheduler::kqueue_scheduler(capy::execution_context& ctx, int)
175 745x : kq_fd_(-1)
176
1/2
✓ Branch 0 taken 745 times.
✗ Branch 1 not taken.
 745x , event_buffer_(max_events_per_poll_)
177 1490x {
178
1/2
✓ Branch 0 taken 745 times.
✗ Branch 1 not taken.
 745x kq_fd_ = ::kqueue();
179
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 745 times.
 745x if (kq_fd_ < 0)
180 ✗ detail::throw_system_error(make_err(errno), "kqueue");
181
182
2/4
✓ Branch 0 taken 745 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 745 times.
 745x if (::fcntl(kq_fd_, F_SETFD, FD_CLOEXEC) == -1)
183 {
184 ✗ int errn = errno;
185 ✗ ::close(kq_fd_);
186 ✗ detail::throw_system_error(make_err(errn), "fcntl (kqueue FD_CLOEXEC)");
187 }
188
189 struct kevent ev;
190 745x EV_SET(&ev, 0, EVFILT_USER, EV_ADD | EV_CLEAR, 0, 0, nullptr);
191
2/4
✓ Branch 0 taken 745 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 745 times.
 745x if (::kevent(kq_fd_, &ev, 1, nullptr, 0, nullptr) < 0)
192 {
193 ✗ int errn = errno;
194 ✗ ::close(kq_fd_);
195 ✗ detail::throw_system_error(make_err(errn), "kevent (EVFILT_USER)");
196 }
197
198
1/2
✓ Branch 0 taken 745 times.
✗ Branch 1 not taken.
 745x timer_svc_ = &get_timer_service(ctx, *this);
199 1490x timer_svc_->set_on_earliest_changed(
200 6178x timer_service::callback(this, [](void* p) {
201 5433x static_cast<kqueue_scheduler*>(p)->interrupt_reactor();
202 5433x }));
203
204
1/2
✓ Branch 0 taken 745 times.
✗ Branch 1 not taken.
 745x get_resolver_service(ctx, *this);
205
1/2
✓ Branch 0 taken 745 times.
✗ Branch 1 not taken.
 745x get_signal_service(ctx, *this);
206
1/2
✓ Branch 0 taken 745 times.
✗ Branch 1 not taken.
 745x get_stream_file_service(ctx, *this);
207
1/2
✓ Branch 0 taken 745 times.
✗ Branch 1 not taken.
 745x get_random_access_file_service(ctx, *this);
208
209 745x completed_ops_.push(&task_op_);
210 1490x }
211
212 2235x inline kqueue_scheduler::~kqueue_scheduler()
213 1490x {
214
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 745 times.
 745x if (kq_fd_ >= 0)
215
1/2
✓ Branch 0 taken 745 times.
✗ Branch 1 not taken.
 745x ::close(kq_fd_);
216 2235x }
217
218 inline void
219 745x kqueue_scheduler::shutdown()
220 {
221 745x shutdown_drain();
222
223
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 745 times.
 745x if (kq_fd_ >= 0)
224 745x interrupt_reactor();
225 745x }
226
227 inline void
228 16x kqueue_scheduler::configure_reactor(
229 unsigned max_events,
230 unsigned budget_init,
231 unsigned budget_max,
232 unsigned unassisted)
233 {
234 16x reactor_scheduler::configure_reactor(
235 16x max_events, budget_init, budget_max, unassisted);
236 16x event_buffer_.resize(max_events_per_poll_);
237 16x }
238
239 inline void
240 11129x kqueue_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) const
241 {
242 struct kevent changes[2];
243 11129x EV_SET(
244 &changes[0], static_cast<uintptr_t>(fd), EVFILT_READ, EV_ADD | EV_CLEAR,
245 0, 0, desc);
246 11129x EV_SET(
247 &changes[1], static_cast<uintptr_t>(fd), EVFILT_WRITE,
248 EV_ADD | EV_CLEAR, 0, 0, desc);
249
250
1/2
✓ Branch 0 taken 11129 times.
✗ Branch 1 not taken.
 11129x if (::kevent(kq_fd_, changes, 2, nullptr, 0, nullptr) < 0)
251 ✗ detail::throw_system_error(make_err(errno), "kevent (register)");
252
253 11129x desc->registered_events = reactor_event_read | reactor_event_write;
254 11129x desc->fd = fd;
255 11129x desc->scheduler_ = this;
256 11129x desc->mutex.set_enabled(reactor_io_locking_);
257 11129x desc->ready_events_.store(0, std::memory_order_relaxed);
258
259 11129x conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
260 11129x desc->impl_ref_.reset();
261 11129x desc->read_ready = false;
262 11129x desc->write_ready = false;
263 11129x }
264
265 inline void
266 11088x kqueue_scheduler::deregister_descriptor(int fd) const
267 {
268 struct kevent changes[2];
269 11088x EV_SET(
270 &changes[0], static_cast<uintptr_t>(fd), EVFILT_READ, EV_DELETE, 0, 0,
271 nullptr);
272 11088x EV_SET(
273 &changes[1], static_cast<uintptr_t>(fd), EVFILT_WRITE, EV_DELETE, 0, 0,
274 nullptr);
275 11088x ::kevent(kq_fd_, changes, 2, nullptr, 0, nullptr);
276 11088x }
277
278 inline void
279 8957x kqueue_scheduler::interrupt_reactor() const
280 {
281 8957x bool expected = false;
282
2/2
✓ Branch 0 taken 1128 times.
✓ Branch 1 taken 7829 times.
 8957x if (user_event_armed_.compare_exchange_strong(
283 expected, true, std::memory_order_acq_rel,
284 std::memory_order_acquire))
285 {
286 struct kevent ev;
287 7829x EV_SET(&ev, 0, EVFILT_USER, 0, NOTE_TRIGGER, 0, nullptr);
288 7829x ::kevent(kq_fd_, &ev, 1, nullptr, 0, nullptr);
289 7829x }
290 8957x }
291
292 inline long
293 50712x kqueue_scheduler::calculate_timeout(long requested_timeout_us) const
294 {
295
1/2
✓ Branch 0 taken 50712 times.
✗ Branch 1 not taken.
 50712x if (requested_timeout_us == 0)
296 ✗ return 0;
297
298 50712x auto nearest = timer_svc_->nearest_expiry();
299
2/2
✓ Branch 0 taken 3215 times.
✓ Branch 1 taken 47497 times.
 50712x if (nearest == timer_service::time_point::max())
300 3215x return requested_timeout_us;
301
302 47497x auto now = std::chrono::steady_clock::now();
303
2/2
✓ Branch 0 taken 102 times.
✓ Branch 1 taken 47395 times.
 47497x if (nearest <= now)
304 102x return 0;
305
306 47395x auto timer_timeout_us =
307 47395x std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
308 47395x .count();
309
310 47395x constexpr auto long_max =
311 static_cast<long long>((std::numeric_limits<long>::max)());
312 47395x auto capped_timer_us = std::min(
313 47395x std::max(timer_timeout_us, static_cast<long long>(0)), long_max);
314
315
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 47395 times.
 47395x if (requested_timeout_us < 0)
316 47395x return static_cast<long>(capped_timer_us);
317
318 ✗ return static_cast<long>(std::min(
319 ✗ static_cast<long long>(requested_timeout_us), capped_timer_us));
320 50712x }
321
322 inline void
323 87043x kqueue_scheduler::run_task(
324 lock_type& lock, context_type* ctx, long timeout_us)
325 {
326 87043x long effective_timeout_us =
327
2/2
✓ Branch 0 taken 36331 times.
✓ Branch 1 taken 50712 times.
 87043x task_interrupted_ ? 0 : calculate_timeout(timeout_us);
328
329
2/2
✓ Branch 0 taken 36329 times.
✓ Branch 1 taken 50714 times.
 87043x if (lock.owns_lock())
330 50714x lock.unlock();
331
332 87043x task_cleanup on_exit{this, &lock, ctx};
333
334 struct timespec ts;
335 87043x struct timespec* ts_ptr = nullptr;
336
2/2
✓ Branch 0 taken 3211 times.
✓ Branch 1 taken 83832 times.
 87043x if (effective_timeout_us >= 0)
337 {
338 83832x ts.tv_sec = effective_timeout_us / 1000000;
339 83832x ts.tv_nsec = (effective_timeout_us % 1000000) * 1000;
340 83832x ts_ptr = &ts;
341 83832x }
342
343
1/2
✓ Branch 0 taken 87043 times.
✗ Branch 1 not taken.
 87043x int nev = ::kevent(
344 87043x kq_fd_, nullptr, 0, event_buffer_.data(),
345 87043x static_cast<int>(event_buffer_.size()), ts_ptr);
346
1/2
✓ Branch 0 taken 87043 times.
✗ Branch 1 not taken.
 87043x int saved_errno = errno;
347
348
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 87043 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 87043x if (nev < 0 && saved_errno != EINTR)
349 ✗ detail::throw_system_error(make_err(saved_errno), "kevent");
350
351 87043x ready_queue local_ops;
352
353
2/2
✓ Branch 0 taken 87043 times.
✓ Branch 1 taken 123590 times.
 210633x for (int i = 0; i < nev; ++i)
354 {
355
2/2
✓ Branch 0 taken 7084 times.
✓ Branch 1 taken 116506 times.
 123590x if (event_buffer_[i].filter == EVFILT_USER)
356 {
357 7084x user_event_armed_.store(false, std::memory_order_release);
358 7084x continue;
359 }
360
361 116506x auto* desc =
362 116506x static_cast<reactor_descriptor_state*>(event_buffer_[i].udata);
363
1/2
✓ Branch 0 taken 116506 times.
✗ Branch 1 not taken.
 116506x if (!desc)
364 ✗ continue;
365
366 116506x std::uint32_t ready = 0;
367
368
2/2
✓ Branch 0 taken 59467 times.
✓ Branch 1 taken 57039 times.
 116506x if (event_buffer_[i].filter == EVFILT_READ)
369 57039x ready |= reactor_event_read;
370
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 59467 times.
 59467x else if (event_buffer_[i].filter == EVFILT_WRITE)
371 59467x ready |= reactor_event_write;
372
373
1/2
✓ Branch 0 taken 116506 times.
✗ Branch 1 not taken.
 116506x if (event_buffer_[i].flags & EV_ERROR)
374 ✗ ready |= reactor_event_error;
375
376
2/2
✓ Branch 0 taken 116440 times.
✓ Branch 1 taken 66 times.
 116506x if (event_buffer_[i].flags & EV_EOF)
377 {
378
2/2
✓ Branch 0 taken 23 times.
✓ Branch 1 taken 43 times.
 66x if (event_buffer_[i].filter == EVFILT_READ)
379 43x ready |= reactor_event_read;
380
2/2
✓ Branch 0 taken 32 times.
✓ Branch 1 taken 34 times.
 66x if (event_buffer_[i].fflags != 0)
381 34x ready |= reactor_event_error;
382 66x }
383
384 116506x desc->add_ready_events(ready);
385
386 116506x bool expected = false;
387
2/2
✓ Branch 0 taken 129 times.
✓ Branch 1 taken 116377 times.
 116506x if (desc->is_enqueued_.compare_exchange_strong(
388 expected, true, std::memory_order_acq_rel,
389 std::memory_order_acquire))
390 {
391 116377x local_ops.push(desc);
392 116377x }
393 116506x }
394
395
1/2
✓ Branch 0 taken 87043 times.
✗ Branch 1 not taken.
 87043x timer_svc_->process_expired();
396
397
1/2
✓ Branch 0 taken 87043 times.
✗ Branch 1 not taken.
 87043x lock.lock();
398
399 87043x completed_ops_.splice(local_ops);
400 87043x }
401
402 } // namespace boost::corosio::detail
403
404 #endif // BOOST_COROSIO_HAS_KQUEUE
405
406 #endif // BOOST_COROSIO_NATIVE_DETAIL_KQUEUE_KQUEUE_SCHEDULER_HPP
407