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