TLA Line data 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_REACTOR_REACTOR_SCHEDULER_HPP
11 : #define BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
12 :
13 : #include <boost/corosio/detail/config.hpp>
14 : #include <boost/capy/ex/execution_context.hpp>
15 :
16 : #include <boost/corosio/detail/ready_queue.hpp>
17 : #include <boost/corosio/detail/scheduler.hpp>
18 : #include <boost/corosio/detail/scheduler_op.hpp>
19 : #include <boost/corosio/detail/thread_local_ptr.hpp>
20 :
21 : #include <atomic>
22 : #include <chrono>
23 : #include <coroutine>
24 : #include <cstddef>
25 : #include <cstdint>
26 : #include <limits>
27 : #include <memory>
28 : #include <stdexcept>
29 :
30 : #include <boost/corosio/detail/conditionally_enabled_mutex.hpp>
31 : #include <boost/corosio/detail/conditionally_enabled_event.hpp>
32 :
33 : namespace boost::corosio::detail {
34 :
35 : // Forward declarations
36 : class reactor_scheduler;
37 : class timer_service;
38 :
39 : /** Per-thread state for a reactor scheduler.
40 :
41 : Each thread running a scheduler's event loop has one of these
42 : on a thread-local stack. It holds a private work queue and
43 : inline completion budget for speculative I/O fast paths.
44 : */
45 : struct BOOST_COROSIO_SYMBOL_VISIBLE reactor_scheduler_context
46 : {
47 : /// Scheduler this context belongs to.
48 : reactor_scheduler const* key;
49 :
50 : /// Next context frame on this thread's stack.
51 : reactor_scheduler_context* next;
52 :
53 : /// Private work queue for reduced contention.
54 : ready_queue private_queue;
55 :
56 : /// Unflushed work count for the private queue.
57 : std::int64_t private_outstanding_work;
58 :
59 : /// Remaining inline completions allowed this cycle.
60 : int inline_budget;
61 :
62 : /// Maximum inline budget (adaptive, 2-16).
63 : int inline_budget_max;
64 :
65 : /// True if no other thread absorbed queued work last cycle.
66 : bool unassisted;
67 :
68 : /// Construct a context frame linked to @a n.
69 : reactor_scheduler_context(
70 : reactor_scheduler const* k, reactor_scheduler_context* n);
71 : };
72 :
73 : /// Thread-local context stack for reactor schedulers.
74 : inline thread_local_ptr<reactor_scheduler_context> reactor_context_stack;
75 :
76 : /// Find the context frame for a scheduler on this thread.
77 : inline reactor_scheduler_context*
78 HIT 993472 : reactor_find_context(reactor_scheduler const* self) noexcept
79 : {
80 993472 : for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
81 : {
82 968566 : if (c->key == self)
83 968566 : return c;
84 : }
85 24906 : return nullptr;
86 : }
87 :
88 : /** Non-template base for reactor-backed scheduler implementations.
89 :
90 : Provides the complete threading model shared by epoll, kqueue,
91 : and select schedulers: signal state machine, inline completion
92 : budget, work counting, run/poll methods, and the do_one event
93 : loop.
94 :
95 : Derived classes provide platform-specific hooks by overriding:
96 : - `run_task(lock, ctx)` to run the reactor poll
97 : - `interrupt_reactor()` to wake a blocked reactor
98 :
99 : De-templated from the original CRTP design to eliminate
100 : duplicate instantiations when multiple backends are compiled
101 : into the same binary. Virtual dispatch for run_task (called
102 : once per reactor cycle, before a blocking syscall) has
103 : negligible overhead.
104 :
105 : @par Thread Safety
106 : All public member functions are thread-safe.
107 : */
108 : class reactor_scheduler
109 : : public scheduler
110 : , public capy::execution_context::service
111 : {
112 : public:
113 : using key_type = scheduler;
114 : using context_type = reactor_scheduler_context;
115 : using mutex_type = conditionally_enabled_mutex;
116 : using lock_type = mutex_type::scoped_lock;
117 : using event_type = conditionally_enabled_event;
118 :
119 : /// Post a coroutine for deferred execution.
120 : void post(std::coroutine_handle<> h) const override;
121 :
122 : /// Post a scheduler operation for deferred execution.
123 : void post(scheduler_op* h) const override;
124 :
125 : /// Post a continuation for deferred execution.
126 : void post(capy::continuation&) const override;
127 :
128 : /// Return true if called from a thread running this scheduler.
129 : bool running_in_this_thread() const noexcept override;
130 :
131 : /// Request the scheduler to stop dispatching handlers.
132 : void stop() override;
133 :
134 : /// Return true if the scheduler has been stopped.
135 : bool stopped() const noexcept override;
136 :
137 : /// Reset the stopped state so `run()` can resume.
138 : void restart() override;
139 :
140 : /// Run the event loop until no work remains.
141 : std::size_t run() override;
142 :
143 : /// Run until one handler completes or no work remains.
144 : std::size_t run_one() override;
145 :
146 : /// Run until one handler completes or @a usec elapses.
147 : std::size_t wait_one(long usec) override;
148 :
149 : /// Run ready handlers without blocking.
150 : std::size_t poll() override;
151 :
152 : /// Run at most one ready handler without blocking.
153 : std::size_t poll_one() override;
154 :
155 : /// Increment the outstanding work count.
156 : void work_started() noexcept override;
157 :
158 : /// Decrement the outstanding work count, stopping on zero.
159 : void work_finished() noexcept override;
160 :
161 : /** Reset the thread's inline completion budget.
162 :
163 : Called at the start of each posted completion handler to
164 : grant a fresh budget for speculative inline completions.
165 : */
166 : void reset_inline_budget() const noexcept;
167 :
168 : /** Consume one unit of inline budget if available.
169 :
170 : @return True if budget was available and consumed.
171 : */
172 : bool try_consume_inline_budget() const noexcept;
173 :
174 : /** Offset a forthcoming work_finished from work_cleanup.
175 :
176 : Called by descriptor_state when all I/O returned EAGAIN and
177 : no handler will be executed. Must be called from a scheduler
178 : thread.
179 : */
180 : void compensating_work_started() const noexcept;
181 :
182 : /** Post completed operations for deferred invocation.
183 :
184 : If called from a thread running this scheduler, operations
185 : go to the thread's private queue (fast path). Otherwise,
186 : operations are added to the global queue under mutex and a
187 : waiter is signaled.
188 :
189 : @par Preconditions
190 : work_started() must have been called for each operation.
191 :
192 : @param ops Queue of operations to post.
193 : */
194 : void post_deferred_completions(ready_queue& ops) const;
195 :
196 : /** Apply runtime configuration to the scheduler.
197 :
198 : Called by `io_context` after construction. Values that do
199 : not apply to this backend are silently ignored.
200 :
201 : @param max_events Event buffer size for epoll/kqueue.
202 : @param budget_init Starting inline completion budget.
203 : @param budget_max Hard ceiling on adaptive budget ramp-up.
204 : @param unassisted Budget when single-threaded.
205 : */
206 : virtual void configure_reactor(
207 : unsigned max_events,
208 : unsigned budget_init,
209 : unsigned budget_max,
210 : unsigned unassisted);
211 :
212 : /// Return the configured initial inline budget.
213 5451 : unsigned inline_budget_initial() const noexcept
214 : {
215 5451 : return inline_budget_initial_;
216 : }
217 :
218 : /// Return true when scheduler locking is disabled (fully-lockless tier).
219 502 : bool scheduler_locking_disabled() const noexcept override
220 : {
221 502 : return scheduler_locking_disabled_;
222 : }
223 :
224 2241 : void configure_threading(threading_config cfg) noexcept override
225 : {
226 2241 : scheduler_locking_disabled_ = !cfg.scheduler_locking;
227 : // reactor_io_locking takes effect at descriptor registration (see the
228 : // register_descriptor overrides), not here.
229 2241 : reactor_io_locking_ = cfg.reactor_io_locking;
230 2241 : one_thread_ = cfg.one_thread;
231 2241 : mutex_.set_enabled(cfg.scheduler_locking);
232 2241 : cond_.set_enabled(cfg.scheduler_locking);
233 2241 : }
234 :
235 : protected:
236 : timer_service* timer_svc_ = nullptr;
237 : bool scheduler_locking_disabled_ = false;
238 : bool reactor_io_locking_ = true;
239 : bool one_thread_ = false;
240 :
241 2253 : reactor_scheduler() = default;
242 :
243 : /** Drain completed_ops during shutdown.
244 :
245 : Pops all operations from the global queue and destroys them,
246 : skipping the task sentinel. Signals all waiting threads.
247 : Derived classes call this from their shutdown() override
248 : before performing platform-specific cleanup.
249 : */
250 : void shutdown_drain();
251 :
252 : /// RAII guard that re-inserts the task sentinel after `run_task`.
253 : struct task_cleanup
254 : {
255 : reactor_scheduler const* sched;
256 : lock_type* lock;
257 : context_type& ctx;
258 : ~task_cleanup();
259 : };
260 :
261 : mutable mutex_type mutex_{true};
262 : mutable event_type cond_{true};
263 : mutable ready_queue completed_ops_;
264 : mutable std::atomic<std::int64_t> outstanding_work_{0};
265 : std::atomic<bool> stopped_{false};
266 : mutable std::atomic<bool> task_running_{false};
267 : mutable bool task_interrupted_ = false;
268 :
269 : // Runtime-configurable reactor tuning parameters.
270 : // Defaults match the library's built-in values.
271 : unsigned max_events_per_poll_ = 128;
272 : unsigned inline_budget_initial_ = 2;
273 : unsigned inline_budget_max_ = 16;
274 : unsigned unassisted_budget_ = 4;
275 :
276 : /// Bit 0 of `state_`: set when the condvar should be signaled.
277 : static constexpr std::size_t signaled_bit = 1;
278 :
279 : /// Increment per waiting thread in `state_`.
280 : static constexpr std::size_t waiter_increment = 2;
281 : mutable std::size_t state_ = 0;
282 :
283 : /// Sentinel op that triggers a reactor poll when dequeued.
284 : struct task_op final : scheduler_op
285 : {
286 : // LCOV_EXCL_START: the sentinel is intercepted by pointer
287 : // identity; its virtuals exist for vtable completeness.
288 : void operator()() override {}
289 : void destroy() override {}
290 : // LCOV_EXCL_STOP
291 : };
292 : task_op task_op_;
293 :
294 : /** Run the platform-specific reactor poll.
295 :
296 : @par Postconditions
297 : `lock` is owned on return, however the poll ended. An
298 : implementation that unlocks around the blocking call owes the
299 : caller a matching re-acquire on every path out, including the
300 : errors it retries rather than reports.
301 : */
302 : virtual void
303 : run_task(lock_type& lock, context_type& ctx, long timeout_us) = 0;
304 :
305 : /// Wake a blocked reactor (e.g. write to eventfd or pipe).
306 : virtual void interrupt_reactor() const = 0;
307 :
308 : private:
309 : struct work_cleanup
310 : {
311 : reactor_scheduler* sched;
312 : lock_type* lock;
313 : context_type& ctx;
314 : ~work_cleanup();
315 : };
316 :
317 : std::size_t do_one(lock_type& lock, long timeout_us, context_type& ctx);
318 :
319 : void signal_all(lock_type& lock) const;
320 : bool maybe_unlock_and_signal_one(lock_type& lock) const;
321 : bool unlock_and_signal_one(lock_type& lock) const;
322 : void clear_signal() const;
323 : void wait_for_signal(lock_type& lock) const;
324 : void wait_for_signal_for(lock_type& lock, long timeout_us) const;
325 : void wake_one_thread_and_unlock(lock_type& lock) const;
326 : };
327 :
328 : /** RAII guard that pushes/pops a scheduler context frame.
329 :
330 : On construction, pushes a new context frame onto the
331 : thread-local stack. On destruction, drains any remaining
332 : private queue items to the global queue and pops the frame.
333 : */
334 : struct reactor_thread_context_guard
335 : {
336 : /// The context frame managed by this guard.
337 : reactor_scheduler_context frame_;
338 :
339 : /// Construct the guard, pushing a frame for @a sched.
340 5451 : explicit reactor_thread_context_guard(
341 : reactor_scheduler const* sched) noexcept
342 5451 : : frame_(sched, reactor_context_stack.get())
343 : {
344 5451 : reactor_context_stack.set(&frame_);
345 5451 : }
346 :
347 : /** Destroy the guard, popping the frame.
348 :
349 : The private queue is empty here by invariant: work_cleanup and
350 : task_cleanup splice it to the global queue after every handler
351 : and every reactor pass.
352 : */
353 5451 : ~reactor_thread_context_guard() noexcept
354 : {
355 5451 : reactor_context_stack.set(frame_.next);
356 5451 : }
357 : };
358 :
359 : // ---- Inline implementations ------------------------------------------------
360 :
361 5451 : inline reactor_scheduler_context::reactor_scheduler_context(
362 5451 : reactor_scheduler const* k, reactor_scheduler_context* n)
363 5451 : : key(k)
364 5451 : , next(n)
365 5451 : , private_outstanding_work(0)
366 5451 : , inline_budget(0)
367 5451 : , inline_budget_max(static_cast<int>(k->inline_budget_initial()))
368 5451 : , unassisted(false)
369 : {
370 5451 : }
371 :
372 : inline void
373 44 : reactor_scheduler::configure_reactor(
374 : unsigned max_events,
375 : unsigned budget_init,
376 : unsigned budget_max,
377 : unsigned unassisted)
378 : {
379 86 : if (max_events < 1 ||
380 42 : max_events > static_cast<unsigned>(std::numeric_limits<int>::max()))
381 2 : throw std::out_of_range("max_events_per_poll must be in [1, INT_MAX]");
382 42 : if (budget_max > static_cast<unsigned>(std::numeric_limits<int>::max()))
383 2 : throw std::out_of_range("inline_budget_max must be in [0, INT_MAX]");
384 :
385 : // Clamp initial and unassisted to budget_max.
386 40 : if (budget_init > budget_max)
387 16 : budget_init = budget_max;
388 40 : if (unassisted > budget_max)
389 16 : unassisted = budget_max;
390 :
391 40 : max_events_per_poll_ = max_events;
392 40 : inline_budget_initial_ = budget_init;
393 40 : inline_budget_max_ = budget_max;
394 40 : unassisted_budget_ = unassisted;
395 40 : }
396 :
397 : inline void
398 98175 : reactor_scheduler::reset_inline_budget() const noexcept
399 : {
400 : // When budget is disabled (max==0), all paths below would no-op
401 : // (inline_budget stays 0). Skip the TLS lookup entirely.
402 98175 : if (inline_budget_max_ == 0)
403 56 : return;
404 98119 : if (auto* ctx = reactor_find_context(this))
405 : {
406 : // Cap when no other thread absorbed queued work
407 98119 : if (ctx->unassisted)
408 : {
409 98119 : ctx->inline_budget_max = static_cast<int>(unassisted_budget_);
410 98119 : ctx->inline_budget = static_cast<int>(unassisted_budget_);
411 98119 : return;
412 : }
413 : // Ramp up when previous cycle fully consumed budget.
414 : // max(1, ...) ensures the doubling escapes zero.
415 MIS 0 : if (ctx->inline_budget == 0)
416 0 : ctx->inline_budget_max =
417 0 : (std::min)((std::max)(1, ctx->inline_budget_max) * 2,
418 0 : static_cast<int>(inline_budget_max_));
419 0 : else if (ctx->inline_budget < ctx->inline_budget_max)
420 0 : ctx->inline_budget_max = static_cast<int>(inline_budget_initial_);
421 0 : ctx->inline_budget = ctx->inline_budget_max;
422 : }
423 : }
424 :
425 : inline bool
426 HIT 440182 : reactor_scheduler::try_consume_inline_budget() const noexcept
427 : {
428 440182 : if (inline_budget_max_ == 0)
429 40 : return false;
430 440142 : if (auto* ctx = reactor_find_context(this))
431 : {
432 440142 : if (ctx->inline_budget > 0)
433 : {
434 351948 : --ctx->inline_budget;
435 351948 : return true;
436 : }
437 : }
438 88194 : return false;
439 : }
440 :
441 : inline void
442 3756 : reactor_scheduler::post(std::coroutine_handle<> h) const
443 : {
444 : struct post_handler final : scheduler_op
445 : {
446 : std::coroutine_handle<> h_;
447 :
448 3756 : explicit post_handler(std::coroutine_handle<> h) : h_(h) {}
449 7512 : ~post_handler() override = default;
450 :
451 3744 : void operator()() override
452 : {
453 3744 : auto saved = h_;
454 3744 : delete this;
455 3744 : saved.resume();
456 3744 : }
457 :
458 12 : void destroy() override
459 : {
460 12 : auto saved = h_;
461 12 : delete this;
462 12 : saved.destroy();
463 12 : }
464 : };
465 :
466 3756 : auto ph = std::make_unique<post_handler>(h);
467 :
468 3756 : if (auto* ctx = reactor_find_context(this))
469 : {
470 96 : ++ctx->private_outstanding_work;
471 96 : ctx->private_queue.push(ph.release());
472 96 : return;
473 : }
474 :
475 3660 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
476 :
477 3660 : lock_type lock(mutex_);
478 3660 : completed_ops_.push(ph.release());
479 3660 : wake_one_thread_and_unlock(lock);
480 3756 : }
481 :
482 : inline void
483 107973 : reactor_scheduler::post(scheduler_op* h) const
484 : {
485 107973 : if (auto* ctx = reactor_find_context(this))
486 : {
487 106935 : ++ctx->private_outstanding_work;
488 106935 : ctx->private_queue.push(h);
489 106935 : return;
490 : }
491 :
492 1038 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
493 :
494 1038 : lock_type lock(mutex_);
495 1038 : completed_ops_.push(h);
496 1038 : wake_one_thread_and_unlock(lock);
497 1038 : }
498 :
499 : inline void
500 25908 : reactor_scheduler::post(capy::continuation& c) const
501 : {
502 25908 : if (auto* ctx = reactor_find_context(this))
503 : {
504 15548 : ++ctx->private_outstanding_work;
505 15548 : ctx->private_queue.push(c);
506 15548 : return;
507 : }
508 :
509 10360 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
510 :
511 10360 : lock_type lock(mutex_);
512 10360 : completed_ops_.push(c);
513 10360 : wake_one_thread_and_unlock(lock);
514 10360 : }
515 :
516 : inline bool
517 10789 : reactor_scheduler::running_in_this_thread() const noexcept
518 : {
519 10789 : return reactor_find_context(this) != nullptr;
520 : }
521 :
522 : inline void
523 3722 : reactor_scheduler::stop()
524 : {
525 3722 : lock_type lock(mutex_);
526 3722 : if (!stopped_.load(std::memory_order_acquire))
527 : {
528 2859 : stopped_.store(true, std::memory_order_release);
529 2859 : signal_all(lock);
530 2859 : interrupt_reactor();
531 : }
532 3722 : }
533 :
534 : inline bool
535 2503 : reactor_scheduler::stopped() const noexcept
536 : {
537 2503 : return stopped_.load(std::memory_order_acquire);
538 : }
539 :
540 : inline void
541 1439 : reactor_scheduler::restart()
542 : {
543 1439 : stopped_.store(false, std::memory_order_release);
544 1439 : }
545 :
546 : inline std::size_t
547 2096 : reactor_scheduler::run()
548 : {
549 4192 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
550 : {
551 103 : stop();
552 103 : return 0;
553 : }
554 :
555 1993 : reactor_thread_context_guard ctx(this);
556 1993 : lock_type lock(mutex_);
557 :
558 1993 : std::size_t n = 0;
559 : for (;;)
560 : {
561 453901 : if (!do_one(lock, -1, ctx.frame_))
562 1990 : break;
563 451908 : if (n != (std::numeric_limits<std::size_t>::max)())
564 451908 : ++n;
565 451908 : if (!lock.owns_lock())
566 346110 : lock.lock();
567 : }
568 1990 : return n;
569 1996 : }
570 :
571 : inline std::size_t
572 112 : reactor_scheduler::run_one()
573 : {
574 224 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
575 : {
576 3 : stop();
577 3 : return 0;
578 : }
579 :
580 109 : reactor_thread_context_guard ctx(this);
581 109 : lock_type lock(mutex_);
582 109 : return do_one(lock, -1, ctx.frame_);
583 109 : }
584 :
585 : inline std::size_t
586 4094 : reactor_scheduler::wait_one(long usec)
587 : {
588 8188 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
589 : {
590 785 : stop();
591 785 : return 0;
592 : }
593 :
594 3309 : reactor_thread_context_guard ctx(this);
595 3309 : lock_type lock(mutex_);
596 3309 : return do_one(lock, usec, ctx.frame_);
597 3309 : }
598 :
599 : inline std::size_t
600 49 : reactor_scheduler::poll()
601 : {
602 98 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
603 : {
604 15 : stop();
605 15 : return 0;
606 : }
607 :
608 34 : reactor_thread_context_guard ctx(this);
609 34 : lock_type lock(mutex_);
610 :
611 34 : std::size_t n = 0;
612 : for (;;)
613 : {
614 75 : if (!do_one(lock, 0, ctx.frame_))
615 34 : break;
616 41 : if (n != (std::numeric_limits<std::size_t>::max)())
617 41 : ++n;
618 41 : if (!lock.owns_lock())
619 41 : lock.lock();
620 : }
621 34 : return n;
622 34 : }
623 :
624 : inline std::size_t
625 11 : reactor_scheduler::poll_one()
626 : {
627 22 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
628 : {
629 5 : stop();
630 5 : return 0;
631 : }
632 :
633 6 : reactor_thread_context_guard ctx(this);
634 6 : lock_type lock(mutex_);
635 6 : return do_one(lock, 0, ctx.frame_);
636 6 : }
637 :
638 : inline void
639 36090 : reactor_scheduler::work_started() noexcept
640 : {
641 36090 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
642 36090 : }
643 :
644 : inline void
645 67816 : reactor_scheduler::work_finished() noexcept
646 : {
647 135632 : if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
648 2796 : stop();
649 67816 : }
650 :
651 : inline void
652 306783 : reactor_scheduler::compensating_work_started() const noexcept
653 : {
654 306783 : auto* ctx = reactor_find_context(this);
655 306783 : if (ctx)
656 306783 : ++ctx->private_outstanding_work;
657 306783 : }
658 :
659 : inline void
660 9371 : reactor_scheduler::post_deferred_completions(ready_queue& ops) const
661 : {
662 9371 : if (ops.empty())
663 9371 : return;
664 :
665 2 : if (auto* ctx = reactor_find_context(this))
666 : {
667 2 : ctx->private_queue.splice(ops);
668 2 : return;
669 : }
670 :
671 MIS 0 : lock_type lock(mutex_);
672 0 : completed_ops_.splice(ops);
673 0 : wake_one_thread_and_unlock(lock);
674 0 : }
675 :
676 : inline void
677 HIT 2241 : reactor_scheduler::shutdown_drain()
678 : {
679 2241 : lock_type lock(mutex_);
680 :
681 4862 : while (auto e = completed_ops_.pop())
682 : {
683 2621 : if (ready_is_continuation(e))
684 : {
685 8 : lock.unlock();
686 8 : if (auto h = ready_as_cont(e)->h)
687 8 : h.destroy();
688 8 : lock.lock();
689 : }
690 : else
691 : {
692 2613 : auto* op = ready_as_op(e);
693 2613 : if (op == &task_op_)
694 2238 : continue;
695 375 : lock.unlock();
696 375 : op->destroy();
697 375 : lock.lock();
698 : }
699 2621 : }
700 :
701 2241 : signal_all(lock);
702 2241 : }
703 :
704 : inline void
705 5100 : reactor_scheduler::signal_all(lock_type&) const
706 : {
707 5100 : state_ |= signaled_bit;
708 5100 : cond_.notify_all();
709 5100 : }
710 :
711 : inline bool
712 15058 : reactor_scheduler::maybe_unlock_and_signal_one(lock_type& lock) const
713 : {
714 15058 : state_ |= signaled_bit;
715 15058 : if (state_ > signaled_bit)
716 : {
717 33 : lock.unlock();
718 33 : cond_.notify_one();
719 33 : return true;
720 : }
721 15025 : return false;
722 : }
723 :
724 : inline bool
725 507698 : reactor_scheduler::unlock_and_signal_one(lock_type& lock) const
726 : {
727 507698 : state_ |= signaled_bit;
728 507698 : bool have_waiters = state_ > signaled_bit;
729 507698 : lock.unlock();
730 507698 : if (have_waiters)
731 6 : cond_.notify_one();
732 507698 : return have_waiters;
733 : }
734 :
735 : inline void
736 19 : reactor_scheduler::clear_signal() const
737 : {
738 19 : state_ &= ~signaled_bit;
739 19 : }
740 :
741 : inline void
742 6 : reactor_scheduler::wait_for_signal(lock_type& lock) const
743 : {
744 14 : while ((state_ & signaled_bit) == 0)
745 : {
746 8 : state_ += waiter_increment;
747 8 : cond_.wait(lock);
748 8 : state_ -= waiter_increment;
749 : }
750 6 : }
751 :
752 : inline void
753 13 : reactor_scheduler::wait_for_signal_for(lock_type& lock, long timeout_us) const
754 : {
755 13 : if ((state_ & signaled_bit) == 0)
756 : {
757 13 : state_ += waiter_increment;
758 13 : cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
759 13 : state_ -= waiter_increment;
760 : }
761 13 : }
762 :
763 : inline void
764 15058 : reactor_scheduler::wake_one_thread_and_unlock(lock_type& lock) const
765 : {
766 15058 : if (maybe_unlock_and_signal_one(lock))
767 33 : return;
768 :
769 15025 : if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
770 : {
771 841 : task_interrupted_ = true;
772 841 : lock.unlock();
773 841 : interrupt_reactor();
774 : }
775 : else
776 : {
777 14184 : lock.unlock();
778 : }
779 : }
780 :
781 453709 : inline reactor_scheduler::work_cleanup::~work_cleanup()
782 : {
783 453709 : std::int64_t produced = ctx.private_outstanding_work;
784 453709 : if (produced > 1)
785 345 : sched->outstanding_work_.fetch_add(
786 : produced - 1, std::memory_order_relaxed);
787 453364 : else if (produced < 1)
788 41167 : sched->work_finished();
789 453709 : ctx.private_outstanding_work = 0;
790 :
791 453709 : if (!ctx.private_queue.empty())
792 : {
793 106078 : lock->lock();
794 106078 : sched->completed_ops_.splice(ctx.private_queue);
795 : }
796 453709 : }
797 :
798 334383 : inline reactor_scheduler::task_cleanup::~task_cleanup()
799 : {
800 334383 : if (ctx.private_outstanding_work > 0)
801 : {
802 9194 : sched->outstanding_work_.fetch_add(
803 9194 : ctx.private_outstanding_work, std::memory_order_relaxed);
804 9194 : ctx.private_outstanding_work = 0;
805 : }
806 :
807 334383 : if (!ctx.private_queue.empty())
808 : {
809 9194 : if (!lock->owns_lock())
810 MIS 0 : lock->lock();
811 HIT 9194 : sched->completed_ops_.splice(ctx.private_queue);
812 : }
813 334383 : }
814 :
815 : inline std::size_t
816 457400 : reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx)
817 : {
818 : for (;;)
819 : {
820 790137 : if (stopped_.load(std::memory_order_acquire))
821 1994 : return 0;
822 :
823 788143 : std::uintptr_t e = completed_ops_.pop();
824 788143 : scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
825 :
826 : // Handle reactor sentinel — time to poll for I/O
827 788143 : if (op == &task_op_)
828 : {
829 334415 : bool more_handlers = !completed_ops_.empty();
830 :
831 614809 : if (!more_handlers &&
832 560788 : (outstanding_work_.load(std::memory_order_acquire) == 0 ||
833 : timeout_us == 0))
834 : {
835 32 : completed_ops_.push(&task_op_);
836 32 : return 0;
837 : }
838 :
839 334383 : long task_timeout_us = more_handlers ? 0 : timeout_us;
840 334383 : task_interrupted_ = task_timeout_us == 0;
841 334383 : task_running_.store(true, std::memory_order_release);
842 :
843 : // Wake a peer to take the pending handlers while this thread
844 : // polls the reactor; skipped when one_thread_ (no peer exists).
845 334383 : if (more_handlers && !one_thread_)
846 54014 : unlock_and_signal_one(lock);
847 :
848 : try
849 : {
850 334383 : run_task(lock, ctx, task_timeout_us);
851 : }
852 3 : catch (...)
853 : {
854 3 : task_running_.store(false, std::memory_order_relaxed);
855 3 : throw;
856 3 : }
857 :
858 334380 : task_running_.store(false, std::memory_order_relaxed);
859 334380 : completed_ops_.push(&task_op_);
860 334380 : if (timeout_us > 0)
861 1662 : return 0;
862 332718 : continue;
863 332718 : }
864 :
865 : // Handle ready entry (op or continuation)
866 453728 : if (e != 0)
867 : {
868 453709 : bool more = !completed_ops_.empty();
869 :
870 453709 : if (more && !one_thread_)
871 : {
872 : // Wake a peer for the remaining work; unassisted if none
873 : // was parked to take it.
874 453684 : ctx.unassisted = !unlock_and_signal_one(lock);
875 : }
876 : else
877 : {
878 : // No peer to wake (one_thread_, or nothing more queued).
879 25 : ctx.unassisted = more;
880 25 : lock.unlock();
881 : }
882 :
883 453709 : [[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
884 :
885 453709 : if (ready_is_continuation(e))
886 25900 : ready_as_cont(e)->h.resume();
887 : else
888 427809 : (*op)();
889 453709 : return 1;
890 453709 : }
891 :
892 38 : if (outstanding_work_.load(std::memory_order_acquire) == 0 ||
893 : timeout_us == 0)
894 MIS 0 : return 0;
895 :
896 HIT 19 : clear_signal();
897 19 : if (timeout_us < 0)
898 6 : wait_for_signal(lock);
899 : else
900 13 : wait_for_signal_for(lock, timeout_us);
901 332737 : }
902 : }
903 :
904 : } // namespace boost::corosio::detail
905 :
906 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
|
