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 881228 : reactor_find_context(reactor_scheduler const* self) noexcept
80 : {
81 881228 : for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
82 : {
83 866266 : if (c->key == self)
84 866266 : return c;
85 : }
86 14962 : 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 917 : unsigned inline_budget_initial() const noexcept
257 : {
258 917 : return inline_budget_initial_;
259 : }
260 :
261 : /// Return true when scheduler locking is disabled (fully-lockless tier).
262 162 : bool scheduler_locking_disabled() const noexcept override
263 : {
264 162 : return scheduler_locking_disabled_;
265 : }
266 :
267 1225 : void configure_threading(threading_config cfg) noexcept override
268 : {
269 1225 : scheduler_locking_disabled_ = !cfg.scheduler_locking;
270 : // reactor_io_locking takes effect at descriptor registration (see the
271 : // register_descriptor overrides), not here.
272 1225 : reactor_io_locking_ = cfg.reactor_io_locking;
273 1225 : one_thread_ = cfg.one_thread;
274 1225 : mutex_.set_enabled(cfg.scheduler_locking);
275 1225 : cond_.set_enabled(cfg.scheduler_locking);
276 1225 : }
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 1225 : 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 : virtual void
336 : run_task(lock_type& lock, context_type* ctx,
337 : long timeout_us) = 0;
338 :
339 : /// Wake a blocked reactor (e.g. write to eventfd or pipe).
340 : virtual void interrupt_reactor() const = 0;
341 :
342 : private:
343 : struct work_cleanup
344 : {
345 : reactor_scheduler* sched;
346 : lock_type* lock;
347 : context_type* ctx;
348 : ~work_cleanup();
349 : };
350 :
351 : std::size_t do_one(
352 : lock_type& lock, long timeout_us, context_type* ctx);
353 :
354 : void signal_all(lock_type& lock) const;
355 : bool maybe_unlock_and_signal_one(lock_type& lock) const;
356 : bool unlock_and_signal_one(lock_type& lock) const;
357 : void clear_signal() const;
358 : void wait_for_signal(lock_type& lock) const;
359 : void wait_for_signal_for(
360 : lock_type& lock, long timeout_us) const;
361 : void wake_one_thread_and_unlock(lock_type& lock) const;
362 : };
363 :
364 : /** RAII guard that pushes/pops a scheduler context frame.
365 :
366 : On construction, pushes a new context frame onto the
367 : thread-local stack. On destruction, drains any remaining
368 : private queue items to the global queue and pops the frame.
369 : */
370 : struct reactor_thread_context_guard
371 : {
372 : /// The context frame managed by this guard.
373 : reactor_scheduler_context frame_;
374 :
375 : /// Construct the guard, pushing a frame for @a sched.
376 HIT 917 : explicit reactor_thread_context_guard(
377 : reactor_scheduler const* sched) noexcept
378 917 : : frame_(sched, reactor_context_stack.get())
379 : {
380 917 : reactor_context_stack.set(&frame_);
381 917 : }
382 :
383 : /// Destroy the guard, draining private work and popping the frame.
384 917 : ~reactor_thread_context_guard() noexcept
385 : {
386 917 : if (!frame_.private_queue.empty())
387 MIS 0 : frame_.key->drain_thread_queue(
388 0 : frame_.private_queue, frame_.private_outstanding_work);
389 HIT 917 : reactor_context_stack.set(frame_.next);
390 917 : }
391 : };
392 :
393 : // ---- Inline implementations ------------------------------------------------
394 :
395 : inline
396 917 : reactor_scheduler_context::reactor_scheduler_context(
397 : reactor_scheduler const* k,
398 917 : reactor_scheduler_context* n)
399 917 : : key(k)
400 917 : , next(n)
401 917 : , private_outstanding_work(0)
402 917 : , inline_budget(0)
403 917 : , inline_budget_max(
404 917 : static_cast<int>(k->inline_budget_initial()))
405 917 : , unassisted(false)
406 : {
407 917 : }
408 :
409 : inline void
410 25 : reactor_scheduler::configure_reactor(
411 : unsigned max_events,
412 : unsigned budget_init,
413 : unsigned budget_max,
414 : unsigned unassisted)
415 : {
416 48 : if (max_events < 1 ||
417 23 : max_events > static_cast<unsigned>(std::numeric_limits<int>::max()))
418 : throw std::out_of_range(
419 2 : "max_events_per_poll must be in [1, INT_MAX]");
420 23 : if (budget_max > static_cast<unsigned>(std::numeric_limits<int>::max()))
421 : throw std::out_of_range(
422 MIS 0 : "inline_budget_max must be in [0, INT_MAX]");
423 :
424 : // Clamp initial and unassisted to budget_max.
425 HIT 23 : if (budget_init > budget_max)
426 2 : budget_init = budget_max;
427 23 : if (unassisted > budget_max)
428 2 : unassisted = budget_max;
429 :
430 23 : max_events_per_poll_ = max_events;
431 23 : inline_budget_initial_ = budget_init;
432 23 : inline_budget_max_ = budget_max;
433 23 : unassisted_budget_ = unassisted;
434 23 : }
435 :
436 : inline void
437 96437 : reactor_scheduler::reset_inline_budget() const noexcept
438 : {
439 : // When budget is disabled (max==0), all paths below would no-op
440 : // (inline_budget stays 0). Skip the TLS lookup entirely.
441 96437 : if (inline_budget_max_ == 0)
442 MIS 0 : return;
443 HIT 96437 : if (auto* ctx = reactor_find_context(this))
444 : {
445 : // Cap when no other thread absorbed queued work
446 96437 : if (ctx->unassisted)
447 : {
448 96437 : ctx->inline_budget_max =
449 96437 : static_cast<int>(unassisted_budget_);
450 96437 : ctx->inline_budget =
451 96437 : static_cast<int>(unassisted_budget_);
452 96437 : return;
453 : }
454 : // Ramp up when previous cycle fully consumed budget.
455 : // max(1, ...) ensures the doubling escapes zero.
456 MIS 0 : if (ctx->inline_budget == 0)
457 0 : ctx->inline_budget_max = (std::min)(
458 0 : (std::max)(1, ctx->inline_budget_max) * 2,
459 0 : static_cast<int>(inline_budget_max_));
460 0 : else if (ctx->inline_budget < ctx->inline_budget_max)
461 0 : ctx->inline_budget_max =
462 0 : static_cast<int>(inline_budget_initial_);
463 0 : ctx->inline_budget = ctx->inline_budget_max;
464 : }
465 : }
466 :
467 : inline bool
468 HIT 418869 : reactor_scheduler::try_consume_inline_budget() const noexcept
469 : {
470 418869 : if (inline_budget_max_ == 0)
471 MIS 0 : return false;
472 HIT 418869 : if (auto* ctx = reactor_find_context(this))
473 : {
474 418869 : if (ctx->inline_budget > 0)
475 : {
476 335053 : --ctx->inline_budget;
477 335053 : return true;
478 : }
479 : }
480 83816 : return false;
481 : }
482 :
483 : inline void
484 3686 : reactor_scheduler::post(std::coroutine_handle<> h) const
485 : {
486 : struct post_handler final : scheduler_op
487 : {
488 : std::coroutine_handle<> h_;
489 :
490 3686 : explicit post_handler(std::coroutine_handle<> h) : h_(h) {}
491 7372 : ~post_handler() override = default;
492 :
493 3674 : void operator()() override
494 : {
495 3674 : auto saved = h_;
496 3674 : delete this;
497 3674 : saved.resume();
498 3674 : }
499 :
500 12 : void destroy() override
501 : {
502 12 : auto saved = h_;
503 12 : delete this;
504 12 : saved.destroy();
505 12 : }
506 : };
507 :
508 3686 : auto ph = std::make_unique<post_handler>(h);
509 :
510 3686 : if (auto* ctx = reactor_find_context(this))
511 : {
512 26 : ++ctx->private_outstanding_work;
513 26 : ctx->private_queue.push(ph.release());
514 26 : return;
515 : }
516 :
517 3660 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
518 :
519 3660 : lock_type lock(mutex_);
520 3660 : completed_ops_.push(ph.release());
521 3660 : wake_one_thread_and_unlock(lock);
522 3686 : }
523 :
524 : inline void
525 92494 : reactor_scheduler::post(scheduler_op* h) const
526 : {
527 92494 : if (auto* ctx = reactor_find_context(this))
528 : {
529 92108 : ++ctx->private_outstanding_work;
530 92108 : ctx->private_queue.push(h);
531 92108 : return;
532 : }
533 :
534 386 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
535 :
536 386 : lock_type lock(mutex_);
537 386 : completed_ops_.push(h);
538 386 : wake_one_thread_and_unlock(lock);
539 386 : }
540 :
541 : inline void
542 14215 : reactor_scheduler::post(capy::continuation& c) const
543 : {
544 14215 : if (auto* ctx = reactor_find_context(this))
545 : {
546 8755 : ++ctx->private_outstanding_work;
547 8755 : ctx->private_queue.push(c);
548 8755 : return;
549 : }
550 :
551 5460 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
552 :
553 5460 : lock_type lock(mutex_);
554 5460 : completed_ops_.push(c);
555 5460 : wake_one_thread_and_unlock(lock);
556 5460 : }
557 :
558 : inline bool
559 6131 : reactor_scheduler::running_in_this_thread() const noexcept
560 : {
561 6131 : return reactor_find_context(this) != nullptr;
562 : }
563 :
564 : inline void
565 937 : reactor_scheduler::stop()
566 : {
567 937 : lock_type lock(mutex_);
568 937 : if (!stopped_.load(std::memory_order_acquire))
569 : {
570 860 : stopped_.store(true, std::memory_order_release);
571 860 : signal_all(lock);
572 860 : interrupt_reactor();
573 : }
574 937 : }
575 :
576 : inline bool
577 90 : reactor_scheduler::stopped() const noexcept
578 : {
579 90 : return stopped_.load(std::memory_order_acquire);
580 : }
581 :
582 : inline void
583 183 : reactor_scheduler::restart()
584 : {
585 183 : stopped_.store(false, std::memory_order_release);
586 183 : }
587 :
588 : inline std::size_t
589 904 : reactor_scheduler::run()
590 : {
591 1808 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
592 : {
593 69 : stop();
594 69 : return 0;
595 : }
596 :
597 835 : reactor_thread_context_guard ctx(this);
598 835 : lock_type lock(mutex_);
599 :
600 835 : std::size_t n = 0;
601 : for (;;)
602 : {
603 372712 : if (!do_one(lock, -1, &ctx.frame_))
604 835 : break;
605 371877 : if (n != (std::numeric_limits<std::size_t>::max)())
606 371877 : ++n;
607 371877 : if (!lock.owns_lock())
608 277188 : lock.lock();
609 : }
610 835 : return n;
611 835 : }
612 :
613 : inline std::size_t
614 22 : reactor_scheduler::run_one()
615 : {
616 44 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
617 : {
618 MIS 0 : stop();
619 0 : return 0;
620 : }
621 :
622 HIT 22 : reactor_thread_context_guard ctx(this);
623 22 : lock_type lock(mutex_);
624 22 : return do_one(lock, -1, &ctx.frame_);
625 22 : }
626 :
627 : inline std::size_t
628 62 : reactor_scheduler::wait_one(long usec)
629 : {
630 124 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
631 : {
632 20 : stop();
633 20 : return 0;
634 : }
635 :
636 42 : reactor_thread_context_guard ctx(this);
637 42 : lock_type lock(mutex_);
638 42 : return do_one(lock, usec, &ctx.frame_);
639 42 : }
640 :
641 : inline std::size_t
642 16 : reactor_scheduler::poll()
643 : {
644 32 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
645 : {
646 2 : stop();
647 2 : return 0;
648 : }
649 :
650 14 : reactor_thread_context_guard ctx(this);
651 14 : lock_type lock(mutex_);
652 :
653 14 : std::size_t n = 0;
654 : for (;;)
655 : {
656 46 : if (!do_one(lock, 0, &ctx.frame_))
657 14 : break;
658 32 : if (n != (std::numeric_limits<std::size_t>::max)())
659 32 : ++n;
660 32 : if (!lock.owns_lock())
661 32 : lock.lock();
662 : }
663 14 : return n;
664 14 : }
665 :
666 : inline std::size_t
667 8 : reactor_scheduler::poll_one()
668 : {
669 16 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
670 : {
671 4 : stop();
672 4 : return 0;
673 : }
674 :
675 4 : reactor_thread_context_guard ctx(this);
676 4 : lock_type lock(mutex_);
677 4 : return do_one(lock, 0, &ctx.frame_);
678 4 : }
679 :
680 : inline void
681 26939 : reactor_scheduler::work_started() noexcept
682 : {
683 26939 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
684 26939 : }
685 :
686 : inline void
687 42905 : reactor_scheduler::work_finished() noexcept
688 : {
689 85810 : if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
690 836 : stop();
691 42905 : }
692 :
693 : inline void
694 249396 : reactor_scheduler::compensating_work_started() const noexcept
695 : {
696 249396 : auto* ctx = reactor_find_context(this);
697 249396 : if (ctx)
698 249396 : ++ctx->private_outstanding_work;
699 249396 : }
700 :
701 : inline void
702 MIS 0 : reactor_scheduler::drain_thread_queue(
703 : ready_queue& queue, std::int64_t count) const
704 : {
705 0 : if (count > 0)
706 0 : outstanding_work_.fetch_add(count, std::memory_order_relaxed);
707 :
708 0 : lock_type lock(mutex_);
709 0 : completed_ops_.splice(queue);
710 0 : if (count > 0)
711 0 : maybe_unlock_and_signal_one(lock);
712 0 : }
713 :
714 : inline void
715 HIT 12196 : reactor_scheduler::post_deferred_completions(ready_queue& ops) const
716 : {
717 12196 : if (ops.empty())
718 12196 : return;
719 :
720 MIS 0 : if (auto* ctx = reactor_find_context(this))
721 : {
722 0 : ctx->private_queue.splice(ops);
723 0 : return;
724 : }
725 :
726 0 : lock_type lock(mutex_);
727 0 : completed_ops_.splice(ops);
728 0 : wake_one_thread_and_unlock(lock);
729 0 : }
730 :
731 : inline void
732 HIT 1225 : reactor_scheduler::shutdown_drain()
733 : {
734 1225 : lock_type lock(mutex_);
735 :
736 2631 : while (auto e = completed_ops_.pop())
737 : {
738 1406 : if (ready_is_continuation(e))
739 : {
740 4 : lock.unlock();
741 4 : if (auto h = ready_as_cont(e)->h)
742 4 : h.destroy();
743 4 : lock.lock();
744 : }
745 : else
746 : {
747 1402 : auto* op = ready_as_op(e);
748 1402 : if (op == &task_op_)
749 1225 : continue;
750 177 : lock.unlock();
751 177 : op->destroy();
752 177 : lock.lock();
753 : }
754 1406 : }
755 :
756 1225 : signal_all(lock);
757 1225 : }
758 :
759 : inline void
760 2085 : reactor_scheduler::signal_all(lock_type&) const
761 : {
762 2085 : state_ |= signaled_bit;
763 2085 : cond_.notify_all();
764 2085 : }
765 :
766 : inline bool
767 9506 : reactor_scheduler::maybe_unlock_and_signal_one(
768 : lock_type& lock) const
769 : {
770 9506 : state_ |= signaled_bit;
771 9506 : if (state_ > signaled_bit)
772 : {
773 MIS 0 : lock.unlock();
774 0 : cond_.notify_one();
775 0 : return true;
776 : }
777 HIT 9506 : return false;
778 : }
779 :
780 : inline bool
781 417026 : reactor_scheduler::unlock_and_signal_one(
782 : lock_type& lock) const
783 : {
784 417026 : state_ |= signaled_bit;
785 417026 : bool have_waiters = state_ > signaled_bit;
786 417026 : lock.unlock();
787 417026 : if (have_waiters)
788 6 : cond_.notify_one();
789 417026 : return have_waiters;
790 : }
791 :
792 : inline void
793 6 : reactor_scheduler::clear_signal() const
794 : {
795 6 : state_ &= ~signaled_bit;
796 6 : }
797 :
798 : inline void
799 6 : reactor_scheduler::wait_for_signal(
800 : lock_type& lock) const
801 : {
802 14 : while ((state_ & signaled_bit) == 0)
803 : {
804 8 : state_ += waiter_increment;
805 8 : cond_.wait(lock);
806 8 : state_ -= waiter_increment;
807 : }
808 6 : }
809 :
810 : inline void
811 MIS 0 : reactor_scheduler::wait_for_signal_for(
812 : lock_type& lock, long timeout_us) const
813 : {
814 0 : if ((state_ & signaled_bit) == 0)
815 : {
816 0 : state_ += waiter_increment;
817 0 : cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
818 0 : state_ -= waiter_increment;
819 : }
820 0 : }
821 :
822 : inline void
823 HIT 9506 : reactor_scheduler::wake_one_thread_and_unlock(
824 : lock_type& lock) const
825 : {
826 9506 : if (maybe_unlock_and_signal_one(lock))
827 MIS 0 : return;
828 :
829 HIT 9506 : if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
830 : {
831 130 : task_interrupted_ = true;
832 130 : lock.unlock();
833 130 : interrupt_reactor();
834 : }
835 : else
836 : {
837 9376 : lock.unlock();
838 : }
839 : }
840 :
841 371959 : inline reactor_scheduler::work_cleanup::~work_cleanup()
842 : {
843 371959 : if (ctx)
844 : {
845 371959 : std::int64_t produced = ctx->private_outstanding_work;
846 371959 : if (produced > 1)
847 306 : sched->outstanding_work_.fetch_add(
848 : produced - 1, std::memory_order_relaxed);
849 371653 : else if (produced < 1)
850 28174 : sched->work_finished();
851 371959 : ctx->private_outstanding_work = 0;
852 :
853 371959 : if (!ctx->private_queue.empty())
854 : {
855 94689 : lock->lock();
856 94689 : sched->completed_ops_.splice(ctx->private_queue);
857 : }
858 : }
859 : else
860 : {
861 MIS 0 : sched->work_finished();
862 : }
863 HIT 371959 : }
864 :
865 546794 : inline reactor_scheduler::task_cleanup::~task_cleanup()
866 : {
867 273397 : if (!ctx)
868 MIS 0 : return;
869 :
870 HIT 273397 : if (ctx->private_outstanding_work > 0)
871 : {
872 6185 : sched->outstanding_work_.fetch_add(
873 6185 : ctx->private_outstanding_work, std::memory_order_relaxed);
874 6185 : ctx->private_outstanding_work = 0;
875 : }
876 :
877 273397 : if (!ctx->private_queue.empty())
878 : {
879 6185 : if (!lock->owns_lock())
880 MIS 0 : lock->lock();
881 HIT 6185 : sched->completed_ops_.splice(ctx->private_queue);
882 : }
883 273397 : }
884 :
885 : inline std::size_t
886 372826 : reactor_scheduler::do_one(
887 : lock_type& lock, long timeout_us, context_type* ctx)
888 : {
889 : for (;;)
890 : {
891 646211 : if (stopped_.load(std::memory_order_acquire))
892 839 : return 0;
893 :
894 645372 : std::uintptr_t e = completed_ops_.pop();
895 645372 : scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
896 :
897 : // Handle reactor sentinel — time to poll for I/O
898 645372 : if (op == &task_op_)
899 : {
900 : bool more_handlers =
901 273407 : !completed_ops_.empty() || (ctx && !ctx->private_queue.empty());
902 :
903 501723 : if (!more_handlers &&
904 456632 : (outstanding_work_.load(std::memory_order_acquire) == 0 ||
905 : timeout_us == 0))
906 : {
907 10 : completed_ops_.push(&task_op_);
908 10 : return 0;
909 : }
910 :
911 273397 : long task_timeout_us = more_handlers ? 0 : timeout_us;
912 273397 : task_interrupted_ = task_timeout_us == 0;
913 273397 : task_running_.store(true, std::memory_order_release);
914 :
915 : // Wake a peer to take the pending handlers while this thread
916 : // polls the reactor; skipped when one_thread_ (no peer exists).
917 273397 : if (more_handlers && !one_thread_)
918 45084 : unlock_and_signal_one(lock);
919 :
920 : try
921 : {
922 273397 : run_task(lock, ctx, task_timeout_us);
923 : }
924 MIS 0 : catch (...)
925 : {
926 0 : task_running_.store(false, std::memory_order_relaxed);
927 0 : throw;
928 0 : }
929 :
930 HIT 273397 : task_running_.store(false, std::memory_order_relaxed);
931 273397 : completed_ops_.push(&task_op_);
932 273397 : if (timeout_us > 0)
933 18 : return 0;
934 273379 : continue;
935 273379 : }
936 :
937 : // Handle ready entry (op or continuation)
938 371965 : if (e != 0)
939 : {
940 371959 : bool more = !completed_ops_.empty();
941 :
942 371959 : if (more && !one_thread_)
943 : {
944 : // Wake a peer for the remaining work; unassisted if none
945 : // was parked to take it.
946 371942 : ctx->unassisted = !unlock_and_signal_one(lock);
947 : }
948 : else
949 : {
950 : // No peer to wake (one_thread_, or nothing more queued).
951 17 : ctx->unassisted = more;
952 17 : lock.unlock();
953 : }
954 :
955 371959 : work_cleanup on_exit{this, &lock, ctx};
956 : (void)on_exit;
957 :
958 371959 : if (ready_is_continuation(e))
959 14211 : ready_as_cont(e)->h.resume();
960 : else
961 357748 : (*op)();
962 371959 : return 1;
963 371959 : }
964 :
965 : // Try private queue before blocking
966 6 : if (reactor_drain_private_queue(ctx, outstanding_work_, completed_ops_))
967 MIS 0 : continue;
968 :
969 HIT 12 : if (outstanding_work_.load(std::memory_order_acquire) == 0 ||
970 : timeout_us == 0)
971 MIS 0 : return 0;
972 :
973 HIT 6 : clear_signal();
974 6 : if (timeout_us < 0)
975 6 : wait_for_signal(lock);
976 : else
977 MIS 0 : wait_for_signal_for(lock, timeout_us);
978 HIT 273385 : }
979 : }
980 :
981 : } // namespace boost::corosio::detail
982 :
983 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
|
