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