Index/include / boost / corosio / native / detail / select / select_scheduler.hpp

include/boost/corosio/native/detail/select/select_scheduler.hpp

77.7% Lines (278/358) 87.9% Functions (29/33)
include/boost/corosio/native/detail/select/select_scheduler.hpp
Prev Index Next
Line TLA Hits Source Code
1 //
2 // Copyright (c) 2026 Steve Gerbino
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_SELECT_SELECT_SCHEDULER_HPP
11 #define BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP
12
13 #include <boost/corosio/detail/platform.hpp>
14
15 #if BOOST_COROSIO_HAS_SELECT
16
17 #include <boost/corosio/detail/config.hpp>
18 #include <boost/capy/ex/execution_context.hpp>
19
20 #include <boost/corosio/native/native_scheduler.hpp>
21 #include <boost/corosio/detail/scheduler_op.hpp>
22
23 #include <boost/corosio/native/detail/select/select_op.hpp>
24 #include <boost/corosio/detail/timer_service.hpp>
25 #include <boost/corosio/native/detail/make_err.hpp>
26 #include <boost/corosio/native/detail/posix/posix_resolver_service.hpp>
27 #include <boost/corosio/native/detail/posix/posix_signal_service.hpp>
28
29 #include <boost/corosio/detail/except.hpp>
30 #include <boost/corosio/detail/thread_local_ptr.hpp>
31
32 #include <sys/select.h>
33 #include <sys/socket.h>
34 #include <unistd.h>
35 #include <errno.h>
36 #include <fcntl.h>
37
38 #include <algorithm>
39 #include <atomic>
40 #include <chrono>
41 #include <condition_variable>
42 #include <cstddef>
43 #include <limits>
44 #include <mutex>
45 #include <unordered_map>
46
47 namespace boost::corosio::detail {
48
49 struct select_op;
50
51 /** POSIX scheduler using select() for I/O multiplexing.
52
53 This scheduler implements the scheduler interface using the POSIX select()
54 call for I/O event notification. It uses a single reactor model
55 where one thread runs select() while other threads wait on a condition
56 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 - Portability: Works on all POSIX systems
61
62 The design mirrors epoll_scheduler for behavioral consistency:
63 - Same single-reactor thread coordination model
64 - Same work counting semantics
65 - Same timer integration pattern
66
67 Known Limitations:
68 - FD_SETSIZE (~1024) limits maximum concurrent connections
69 - O(n) scanning: rebuilds fd_sets each iteration
70 - Level-triggered only (no edge-triggered mode)
71
72 @par Thread Safety
73 All public member functions are thread-safe.
74 */
75 class BOOST_COROSIO_DECL select_scheduler final
76 : public native_scheduler
77 , public capy::execution_context::service
78 {
79 public:
80 using key_type = scheduler;
81
82 /** Construct the scheduler.
83
84 Creates a self-pipe for reactor interruption.
85
86 @param ctx Reference to the owning execution_context.
87 @param concurrency_hint Hint for expected thread count (unused).
88 */
89 select_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
90
91 ~select_scheduler() override;
92
93 select_scheduler(select_scheduler const&) = delete;
94 select_scheduler& operator=(select_scheduler const&) = delete;
95
96 void shutdown() override;
97 void post(std::coroutine_handle<> h) const override;
98 void post(scheduler_op* h) const override;
99 bool running_in_this_thread() const noexcept override;
100 void stop() override;
101 bool stopped() const noexcept override;
102 void restart() override;
103 std::size_t run() override;
104 std::size_t run_one() override;
105 std::size_t wait_one(long usec) override;
106 std::size_t poll() override;
107 std::size_t poll_one() override;
108
109 /** Return the maximum file descriptor value supported.
110
111 Returns FD_SETSIZE - 1, the maximum fd value that can be
112 monitored by select(). Operations with fd >= FD_SETSIZE
113 will fail with EINVAL.
114
115 @return The maximum supported file descriptor value.
116 */
117 static constexpr int max_fd() noexcept
118 {
119 return FD_SETSIZE - 1;
120 }
121
122 /** Register a file descriptor for monitoring.
123
124 @param fd The file descriptor to register.
125 @param op The operation associated with this fd.
126 @param events Event mask: 1 = read, 2 = write, 3 = both.
127 */
128 void register_fd(int fd, select_op* op, int events) const;
129
130 /** Unregister a file descriptor from monitoring.
131
132 @param fd The file descriptor to unregister.
133 @param events Event mask to remove: 1 = read, 2 = write, 3 = both.
134 */
135 void deregister_fd(int fd, int events) const;
136
137 void work_started() noexcept override;
138 void work_finished() noexcept override;
139
140 // Event flags for register_fd/deregister_fd
141 static constexpr int event_read = 1;
142 static constexpr int event_write = 2;
143
144 private:
145 std::size_t do_one(long timeout_us);
146 void run_reactor(std::unique_lock<std::mutex>& lock);
147 void wake_one_thread_and_unlock(std::unique_lock<std::mutex>& lock) const;
148 void interrupt_reactor() const;
149 long calculate_timeout(long requested_timeout_us) const;
150
151 // Self-pipe for interrupting select()
152 int pipe_fds_[2]; // [0]=read, [1]=write
153
154 mutable std::mutex mutex_;
155 mutable std::condition_variable wakeup_event_;
156 mutable op_queue completed_ops_;
157 mutable std::atomic<long> outstanding_work_;
158 std::atomic<bool> stopped_;
159
160 // Per-fd state for tracking registered operations
161 struct fd_state
162 {
163 select_op* read_op = nullptr;
164 select_op* write_op = nullptr;
165 };
166 mutable std::unordered_map<int, fd_state> registered_fds_;
167 mutable int max_fd_ = -1;
168
169 // Single reactor thread coordination
170 mutable bool reactor_running_ = false;
171 mutable bool reactor_interrupted_ = false;
172 mutable int idle_thread_count_ = 0;
173
174 // Sentinel operation for interleaving reactor runs with handler execution.
175 // Ensures the reactor runs periodically even when handlers are continuously
176 // posted, preventing timer starvation.
177 struct task_op final : scheduler_op
178 {
179 ✗ void operator()() override {}
180 ✗ void destroy() override {}
181 };
182 task_op task_op_;
183 };
184
185 /*
186 select Scheduler - Single Reactor Model
187 =======================================
188
189 This scheduler mirrors the epoll_scheduler design but uses select() instead
190 of epoll for I/O multiplexing. The thread coordination strategy is identical:
191 one thread becomes the "reactor" while others wait on a condition variable.
192
193 Thread Model
194 ------------
195 - ONE thread runs select() at a time (the reactor thread)
196 - OTHER threads wait on wakeup_event_ (condition variable) for handlers
197 - When work is posted, exactly one waiting thread wakes via notify_one()
198
199 Key Differences from epoll
200 --------------------------
201 - Uses self-pipe instead of eventfd for interruption (more portable)
202 - fd_set rebuilding each iteration (O(n) vs O(1) for epoll)
203 - FD_SETSIZE limit (~1024 fds on most systems)
204 - Level-triggered only (no edge-triggered mode)
205
206 Self-Pipe Pattern
207 -----------------
208 To interrupt a blocking select() call (e.g., when work is posted or a timer
209 expires), we write a byte to pipe_fds_[1]. The read end pipe_fds_[0] is
210 always in the read_fds set, so select() returns immediately. We drain the
211 pipe to clear the readable state.
212
213 fd-to-op Mapping
214 ----------------
215 We use an unordered_map<int, fd_state> to track which operations are
216 registered for each fd. This allows O(1) lookup when select() returns
217 ready fds. Each fd can have at most one read op and one write op registered.
218 */
219
220 namespace select {
221
222 struct BOOST_COROSIO_SYMBOL_VISIBLE scheduler_context
223 {
224 select_scheduler const* key;
225 scheduler_context* next;
226 };
227
228 inline thread_local_ptr<scheduler_context> context_stack;
229
230 struct thread_context_guard
231 {
232 scheduler_context frame_;
233
234 148 explicit thread_context_guard(select_scheduler const* ctx) noexcept
235 148 : frame_{ctx, context_stack.get()}
236 {
237 148 context_stack.set(&frame_);
238 148 }
239
240 148 ~thread_context_guard() noexcept
241 {
242 148 context_stack.set(frame_.next);
243 148 }
244 };
245
246 struct work_guard
247 {
248 select_scheduler* self;
249 170058 ~work_guard()
250 {
251 170058 self->work_finished();
252 170058 }
253 };
254
255 } // namespace select
256
257 168 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int)
258 168 : pipe_fds_{-1, -1}
259 168 , outstanding_work_(0)
260 168 , stopped_(false)
261 168 , max_fd_(-1)
262 168 , reactor_running_(false)
263 168 , reactor_interrupted_(false)
264 336 , idle_thread_count_(0)
265 {
266 // Create self-pipe for interrupting select()
267 168 if (::pipe(pipe_fds_) < 0)
268 ✗ detail::throw_system_error(make_err(errno), "pipe");
269
270 // Set both ends to non-blocking and close-on-exec
271 504 for (int i = 0; i < 2; ++i)
272 {
273 336 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0);
274 336 if (flags == -1)
275 {
276 ✗ int errn = errno;
277 ✗ ::close(pipe_fds_[0]);
278 ✗ ::close(pipe_fds_[1]);
279 ✗ detail::throw_system_error(make_err(errn), "fcntl F_GETFL");
280 }
281 336 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1)
282 {
283 ✗ int errn = errno;
284 ✗ ::close(pipe_fds_[0]);
285 ✗ ::close(pipe_fds_[1]);
286 ✗ detail::throw_system_error(make_err(errn), "fcntl F_SETFL");
287 }
288 336 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1)
289 {
290 ✗ int errn = errno;
291 ✗ ::close(pipe_fds_[0]);
292 ✗ ::close(pipe_fds_[1]);
293 ✗ detail::throw_system_error(make_err(errn), "fcntl F_SETFD");
294 }
295 }
296
297 168 timer_svc_ = &get_timer_service(ctx, *this);
298 168 timer_svc_->set_on_earliest_changed(
299 3888 timer_service::callback(this, [](void* p) {
300 3720 static_cast<select_scheduler*>(p)->interrupt_reactor();
301 3720 }));
302
303 // Initialize resolver service
304 168 get_resolver_service(ctx, *this);
305
306 // Initialize signal service
307 168 get_signal_service(ctx, *this);
308
309 // Push task sentinel to interleave reactor runs with handler execution
310 168 completed_ops_.push(&task_op_);
311 168 }
312
313 336 inline select_scheduler::~select_scheduler()
314 {
315 168 if (pipe_fds_[0] >= 0)
316 168 ::close(pipe_fds_[0]);
317 168 if (pipe_fds_[1] >= 0)
318 168 ::close(pipe_fds_[1]);
319 336 }
320
321 inline void
322 168 select_scheduler::shutdown()
323 {
324 {
325 168 std::unique_lock lock(mutex_);
326
327 343 while (auto* h = completed_ops_.pop())
328 {
329 175 if (h == &task_op_)
330 168 continue;
331 7 lock.unlock();
332 7 h->destroy();
333 7 lock.lock();
334 175 }
335 168 }
336
337 168 if (pipe_fds_[1] >= 0)
338 168 interrupt_reactor();
339
340 168 wakeup_event_.notify_all();
341 168 }
342
343 inline void
344 4109 select_scheduler::post(std::coroutine_handle<> h) const
345 {
346 struct post_handler final : scheduler_op
347 {
348 std::coroutine_handle<> h_;
349
350 4109 explicit post_handler(std::coroutine_handle<> h) : h_(h) {}
351
352 8218 ~post_handler() override = default;
353
354 4106 void operator()() override
355 {
356 4106 auto h = h_;
357 4106 delete this;
358 4106 h.resume();
359 4106 }
360
361 3 void destroy() override
362 {
363 3 auto h = h_;
364 3 delete this;
365 3 h.destroy();
366 3 }
367 };
368
369 4109 auto ph = std::make_unique<post_handler>(h);
370 4109 outstanding_work_.fetch_add(1, std::memory_order_relaxed);
371
372 4109 std::unique_lock lock(mutex_);
373 4109 completed_ops_.push(ph.release());
374 4109 wake_one_thread_and_unlock(lock);
375 4109 }
376
377 inline void
378 158845 select_scheduler::post(scheduler_op* h) const
379 {
380 158845 outstanding_work_.fetch_add(1, std::memory_order_relaxed);
381
382 158845 std::unique_lock lock(mutex_);
383 158845 completed_ops_.push(h);
384 158845 wake_one_thread_and_unlock(lock);
385 158845 }
386
387 inline bool
388 592 select_scheduler::running_in_this_thread() const noexcept
389 {
390 592 for (auto* c = select::context_stack.get(); c != nullptr; c = c->next)
391 362 if (c->key == this)
392 362 return true;
393 230 return false;
394 }
395
396 inline void
397 126 select_scheduler::stop()
398 {
399 126 bool expected = false;
400 126 if (stopped_.compare_exchange_strong(
401 expected, true, std::memory_order_release,
402 std::memory_order_relaxed))
403 {
404 // Wake all threads so they notice stopped_ and exit
405 {
406 126 std::lock_guard lock(mutex_);
407 126 wakeup_event_.notify_all();
408 126 }
409 126 interrupt_reactor();
410 }
411 126 }
412
413 inline bool
414 3 select_scheduler::stopped() const noexcept
415 {
416 3 return stopped_.load(std::memory_order_acquire);
417 }
418
419 inline void
420 38 select_scheduler::restart()
421 {
422 38 stopped_.store(false, std::memory_order_release);
423 38 }
424
425 inline std::size_t
426 122 select_scheduler::run()
427 {
428 122 if (stopped_.load(std::memory_order_acquire))
429 ✗ return 0;
430
431 244 if (outstanding_work_.load(std::memory_order_acquire) == 0)
432 {
433 ✗ stop();
434 ✗ return 0;
435 }
436
437 122 select::thread_context_guard ctx(this);
438
439 122 std::size_t n = 0;
440 170154 while (do_one(-1))
441 170032 if (n != (std::numeric_limits<std::size_t>::max)())
442 170032 ++n;
443 122 return n;
444 122 }
445
446 inline std::size_t
447 ✗ select_scheduler::run_one()
448 {
449 ✗ if (stopped_.load(std::memory_order_acquire))
450 ✗ return 0;
451
452 ✗ if (outstanding_work_.load(std::memory_order_acquire) == 0)
453 {
454 ✗ stop();
455 ✗ return 0;
456 }
457
458 ✗ select::thread_context_guard ctx(this);
459 ✗ return do_one(-1);
460 ✗ }
461
462 inline std::size_t
463 27 select_scheduler::wait_one(long usec)
464 {
465 27 if (stopped_.load(std::memory_order_acquire))
466 3 return 0;
467
468 48 if (outstanding_work_.load(std::memory_order_acquire) == 0)
469 {
470 ✗ stop();
471 ✗ return 0;
472 }
473
474 24 select::thread_context_guard ctx(this);
475 24 return do_one(usec);
476 24 }
477
478 inline std::size_t
479 2 select_scheduler::poll()
480 {
481 2 if (stopped_.load(std::memory_order_acquire))
482 ✗ return 0;
483
484 4 if (outstanding_work_.load(std::memory_order_acquire) == 0)
485 {
486 ✗ stop();
487 ✗ return 0;
488 }
489
490 2 select::thread_context_guard ctx(this);
491
492 2 std::size_t n = 0;
493 4 while (do_one(0))
494 2 if (n != (std::numeric_limits<std::size_t>::max)())
495 2 ++n;
496 2 return n;
497 2 }
498
499 inline std::size_t
500 ✗ select_scheduler::poll_one()
501 {
502 ✗ if (stopped_.load(std::memory_order_acquire))
503 ✗ return 0;
504
505 ✗ if (outstanding_work_.load(std::memory_order_acquire) == 0)
506 {
507 ✗ stop();
508 ✗ return 0;
509 }
510
511 ✗ select::thread_context_guard ctx(this);
512 ✗ return do_one(0);
513 ✗ }
514
515 inline void
516 7271 select_scheduler::register_fd(int fd, select_op* op, int events) const
517 {
518 // Validate fd is within select() limits
519 7271 if (fd < 0 || fd >= FD_SETSIZE)
520 ✗ detail::throw_system_error(make_err(EINVAL), "select: fd out of range");
521
522 {
523 7271 std::lock_guard lock(mutex_);
524
525 7271 auto& state = registered_fds_[fd];
526 7271 if (events & event_read)
527 3774 state.read_op = op;
528 7271 if (events & event_write)
529 3497 state.write_op = op;
530
531 7271 if (fd > max_fd_)
532 248 max_fd_ = fd;
533 7271 }
534
535 // Wake the reactor so a thread blocked in select() rebuilds its fd_sets
536 // with the newly registered fd.
537 7271 interrupt_reactor();
538 7271 }
539
540 inline void
541 7226 select_scheduler::deregister_fd(int fd, int events) const
542 {
543 7226 std::lock_guard lock(mutex_);
544
545 7226 auto it = registered_fds_.find(fd);
546 7226 if (it == registered_fds_.end())
547 7066 return;
548
549 160 if (events & event_read)
550 160 it->second.read_op = nullptr;
551 160 if (events & event_write)
552 ✗ it->second.write_op = nullptr;
553
554 // Remove entry if both are null
555 160 if (!it->second.read_op && !it->second.write_op)
556 {
557 160 registered_fds_.erase(it);
558
559 // Recalculate max_fd_ if needed
560 160 if (fd == max_fd_)
561 {
562 159 max_fd_ = pipe_fds_[0]; // At minimum, the pipe read end
563 159 for (auto& [registered_fd, state] : registered_fds_)
564 {
565 ✗ if (registered_fd > max_fd_)
566 ✗ max_fd_ = registered_fd;
567 }
568 }
569 }
570 7226 }
571
572 inline void
573 11613 select_scheduler::work_started() noexcept
574 {
575 11613 outstanding_work_.fetch_add(1, std::memory_order_relaxed);
576 11613 }
577
578 inline void
579 174560 select_scheduler::work_finished() noexcept
580 {
581 349120 if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
582 125 stop();
583 174560 }
584
585 inline void
586 14985 select_scheduler::interrupt_reactor() const
587 {
588 14985 char byte = 1;
589 14985 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1);
590 14985 }
591
592 inline void
593 162954 select_scheduler::wake_one_thread_and_unlock(
594 std::unique_lock<std::mutex>& lock) const
595 {
596 162954 if (idle_thread_count_ > 0)
597 {
598 // Idle worker exists - wake it via condvar
599 ✗ wakeup_event_.notify_one();
600 ✗ lock.unlock();
601 }
602 162954 else if (reactor_running_ && !reactor_interrupted_)
603 {
604 // No idle workers but reactor is running - interrupt it
605 3700 reactor_interrupted_ = true;
606 3700 lock.unlock();
607 3700 interrupt_reactor();
608 }
609 else
610 {
611 // No one to wake
612 159254 lock.unlock();
613 }
614 162954 }
615
616 inline long
617 10690 select_scheduler::calculate_timeout(long requested_timeout_us) const
618 {
619 10690 if (requested_timeout_us == 0)
620 ✗ return 0;
621
622 10690 auto nearest = timer_svc_->nearest_expiry();
623 10690 if (nearest == timer_service::time_point::max())
624 46 return requested_timeout_us;
625
626 10644 auto now = std::chrono::steady_clock::now();
627 10644 if (nearest <= now)
628 133 return 0;
629
630 auto timer_timeout_us =
631 10511 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
632 10511 .count();
633
634 // Clamp to [0, LONG_MAX] to prevent truncation on 32-bit long platforms
635 10511 constexpr auto long_max =
636 static_cast<long long>((std::numeric_limits<long>::max)());
637 auto capped_timer_us =
638 10511 (std::min)((std::max)(static_cast<long long>(timer_timeout_us),
639 10511 static_cast<long long>(0)),
640 10511 long_max);
641
642 10511 if (requested_timeout_us < 0)
643 10511 return static_cast<long>(capped_timer_us);
644
645 // requested_timeout_us is already long, so min() result fits in long
646 return static_cast<long>(
647 ✗ (std::min)(static_cast<long long>(requested_timeout_us),
648 ✗ capped_timer_us));
649 }
650
651 inline void
652 97640 select_scheduler::run_reactor(std::unique_lock<std::mutex>& lock)
653 {
654 // Calculate timeout considering timers, use 0 if interrupted
655 long effective_timeout_us =
656 97640 reactor_interrupted_ ? 0 : calculate_timeout(-1);
657
658 // Build fd_sets from registered_fds_
659 fd_set read_fds, write_fds, except_fds;
660 1659880 FD_ZERO(&read_fds);
661 1659880 FD_ZERO(&write_fds);
662 1659880 FD_ZERO(&except_fds);
663
664 // Always include the interrupt pipe
665 97640 FD_SET(pipe_fds_[0], &read_fds);
666 97640 int nfds = pipe_fds_[0];
667
668 // Add registered fds
669 115026 for (auto& [fd, state] : registered_fds_)
670 {
671 17386 if (state.read_op)
672 13889 FD_SET(fd, &read_fds);
673 17386 if (state.write_op)
674 {
675 3497 FD_SET(fd, &write_fds);
676 // Also monitor for errors on connect operations
677 3497 FD_SET(fd, &except_fds);
678 }
679 17386 if (fd > nfds)
680 13893 nfds = fd;
681 }
682
683 // Convert timeout to timeval
684 struct timeval tv;
685 97640 struct timeval* tv_ptr = nullptr;
686 97640 if (effective_timeout_us >= 0)
687 {
688 97594 tv.tv_sec = effective_timeout_us / 1000000;
689 97594 tv.tv_usec = effective_timeout_us % 1000000;
690 97594 tv_ptr = &tv;
691 }
692
693 97640 lock.unlock();
694
695 97640 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr);
696 97640 int saved_errno = errno;
697
698 // Process timers outside the lock
699 97640 timer_svc_->process_expired();
700
701 97640 if (ready < 0 && saved_errno != EINTR)
702 ✗ detail::throw_system_error(make_err(saved_errno), "select");
703
704 // Re-acquire lock before modifying completed_ops_
705 97640 lock.lock();
706
707 // Drain the interrupt pipe if readable
708 97640 if (ready > 0 && FD_ISSET(pipe_fds_[0], &read_fds))
709 {
710 char buf[256];
711 22102 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0)
712 {
713 }
714 }
715
716 // Process I/O completions
717 97640 int completions_queued = 0;
718 97640 if (ready > 0)
719 {
720 // Iterate over registered fds (copy keys to avoid iterator invalidation)
721 11051 std::vector<int> fds_to_check;
722 11051 fds_to_check.reserve(registered_fds_.size());
723 24985 for (auto& [fd, state] : registered_fds_)
724 13934 fds_to_check.push_back(fd);
725
726 24985 for (int fd : fds_to_check)
727 {
728 13934 auto it = registered_fds_.find(fd);
729 13934 if (it == registered_fds_.end())
730 ✗ continue;
731
732 13934 auto& state = it->second;
733
734 // Check for errors (especially for connect operations)
735 13934 bool has_error = FD_ISSET(fd, &except_fds);
736
737 // Process read readiness
738 13934 if (state.read_op && (FD_ISSET(fd, &read_fds) || has_error))
739 {
740 3614 auto* op = state.read_op;
741 // Claim the op by exchanging to unregistered. Both registering and
742 // registered states mean the op is ours to complete.
743 3614 auto prev = op->registered.exchange(
744 select_registration_state::unregistered,
745 std::memory_order_acq_rel);
746 3614 if (prev != select_registration_state::unregistered)
747 {
748 3614 state.read_op = nullptr;
749
750 3614 if (has_error)
751 {
752 ✗ int errn = 0;
753 ✗ socklen_t len = sizeof(errn);
754 ✗ if (::getsockopt(
755 ✗ fd, SOL_SOCKET, SO_ERROR, &errn, &len) < 0)
756 ✗ errn = errno;
757 ✗ if (errn == 0)
758 ✗ errn = EIO;
759 ✗ op->complete(errn, 0);
760 }
761 else
762 {
763 3614 op->perform_io();
764 }
765
766 3614 completed_ops_.push(op);
767 3614 ++completions_queued;
768 }
769 }
770
771 // Process write readiness
772 13934 if (state.write_op && (FD_ISSET(fd, &write_fds) || has_error))
773 {
774 3497 auto* op = state.write_op;
775 // Claim the op by exchanging to unregistered. Both registering and
776 // registered states mean the op is ours to complete.
777 3497 auto prev = op->registered.exchange(
778 select_registration_state::unregistered,
779 std::memory_order_acq_rel);
780 3497 if (prev != select_registration_state::unregistered)
781 {
782 3497 state.write_op = nullptr;
783
784 3497 if (has_error)
785 {
786 ✗ int errn = 0;
787 ✗ socklen_t len = sizeof(errn);
788 ✗ if (::getsockopt(
789 ✗ fd, SOL_SOCKET, SO_ERROR, &errn, &len) < 0)
790 ✗ errn = errno;
791 ✗ if (errn == 0)
792 ✗ errn = EIO;
793 ✗ op->complete(errn, 0);
794 }
795 else
796 {
797 3497 op->perform_io();
798 }
799
800 3497 completed_ops_.push(op);
801 3497 ++completions_queued;
802 }
803 }
804
805 // Clean up empty entries
806 13934 if (!state.read_op && !state.write_op)
807 7111 registered_fds_.erase(it);
808 }
809 11051 }
810
811 97640 if (completions_queued > 0)
812 {
813 3618 if (completions_queued == 1)
814 125 wakeup_event_.notify_one();
815 else
816 3493 wakeup_event_.notify_all();
817 }
818 97640 }
819
820 inline std::size_t
821 170182 select_scheduler::do_one(long timeout_us)
822 {
823 170182 std::unique_lock lock(mutex_);
824
825 for (;;)
826 {
827 267822 if (stopped_.load(std::memory_order_acquire))
828 122 return 0;
829
830 267700 scheduler_op* op = completed_ops_.pop();
831
832 267700 if (op == &task_op_)
833 {
834 97642 bool more_handlers = !completed_ops_.empty();
835
836 97642 if (!more_handlers)
837 {
838 21384 if (outstanding_work_.load(std::memory_order_acquire) == 0)
839 {
840 ✗ completed_ops_.push(&task_op_);
841 ✗ return 0;
842 }
843 10692 if (timeout_us == 0)
844 {
845 2 completed_ops_.push(&task_op_);
846 2 return 0;
847 }
848 }
849
850 97640 reactor_interrupted_ = more_handlers || timeout_us == 0;
851 97640 reactor_running_ = true;
852
853 97640 if (more_handlers && idle_thread_count_ > 0)
854 ✗ wakeup_event_.notify_one();
855
856 97640 run_reactor(lock);
857
858 97640 reactor_running_ = false;
859 97640 completed_ops_.push(&task_op_);
860 97640 continue;
861 97640 }
862
863 170058 if (op != nullptr)
864 {
865 170058 lock.unlock();
866 170058 select::work_guard g{this};
867 170058 (*op)();
868 170058 return 1;
869 170058 }
870
871 ✗ if (outstanding_work_.load(std::memory_order_acquire) == 0)
872 ✗ return 0;
873
874 ✗ if (timeout_us == 0)
875 ✗ return 0;
876
877 ✗ ++idle_thread_count_;
878 ✗ if (timeout_us < 0)
879 ✗ wakeup_event_.wait(lock);
880 else
881 ✗ wakeup_event_.wait_for(lock, std::chrono::microseconds(timeout_us));
882 ✗ --idle_thread_count_;
883 97640 }
884 170182 }
885
886 } // namespace boost::corosio::detail
887
888 #endif // BOOST_COROSIO_HAS_SELECT
889
890 #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP
891