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/corosio/detail/object_ref.hpp>
15 : #include <boost/capy/ex/execution_context.hpp>
16 : #include <boost/capy/ex/abandon.hpp>
17 :
18 : #include <boost/corosio/detail/ready_queue.hpp>
19 : #include <boost/corosio/detail/scheduler.hpp>
20 : #include <boost/corosio/detail/scheduler_op.hpp>
21 : #include <boost/corosio/detail/thread_local_ptr.hpp>
22 : #include <boost/corosio/detail/timer_service.hpp>
23 :
24 : #include <atomic>
25 : #include <chrono>
26 : #include <coroutine>
27 : #include <cstddef>
28 : #include <cstdint>
29 : #include <limits>
30 : #include <memory>
31 : #include <stdexcept>
32 :
33 : #include <boost/corosio/detail/conditionally_enabled_mutex.hpp>
34 : #include <boost/corosio/detail/conditionally_enabled_event.hpp>
35 :
36 : namespace boost::corosio::detail {
37 :
38 : // Forward declarations
39 : class reactor_scheduler;
40 :
41 : /** A deregistered descriptor whose owner must outlive the reactor pass
42 : that may still hold it.
43 :
44 : An event the reactor has already read from the kernel can name a
45 : descriptor that another thread closes before the reactor queues it.
46 : The closing thread hands over a reference instead of dropping it, so
47 : the owner is neither freed nor recycled until that pass ends.
48 : */
49 : struct reactor_retirable
50 : {
51 : // All four fields are guarded by the scheduler's mutex.
52 :
53 : /// Next in the scheduler's retired list.
54 : reactor_retirable* retired_next_ = nullptr;
55 :
56 : /// Owner reference held until the pass ends.
57 : object_ref retired_ref_;
58 :
59 : /// Next in a settling thread's list of references to drop.
60 : reactor_retirable* dropped_next_ = nullptr;
61 :
62 : /// Owner reference waiting to be dropped outside the mutex.
63 : object_ref dropped_ref_;
64 :
65 : /// Move @p keep into a queued completion, if one is pending.
66 : virtual void hand_off(object_ref& keep) noexcept = 0;
67 :
68 : protected:
69 HIT 11998 : ~reactor_retirable() = default;
70 : };
71 :
72 : /** Per-thread state for a reactor scheduler.
73 :
74 : Each thread running a scheduler's event loop has one of these
75 : on a thread-local stack. It holds a private work queue and
76 : inline completion budget for speculative I/O fast paths.
77 : */
78 : struct BOOST_COROSIO_SYMBOL_VISIBLE reactor_scheduler_context
79 : {
80 : /// Scheduler this context belongs to.
81 : reactor_scheduler const* key;
82 :
83 : /// Next context frame on this thread's stack.
84 : reactor_scheduler_context* next;
85 :
86 : /// Private work queue for reduced contention.
87 : ready_queue private_queue;
88 :
89 : /// Unflushed work count for the private queue.
90 : std::int64_t private_outstanding_work;
91 :
92 : /// Remaining inline completions allowed this cycle.
93 : int inline_budget;
94 :
95 : /// Maximum inline budget (adaptive, 2-16).
96 : int inline_budget_max;
97 :
98 : /// True if no other thread absorbed queued work last cycle.
99 : bool unassisted;
100 :
101 : /// Construct a context frame linked to @a n.
102 : reactor_scheduler_context(
103 : reactor_scheduler const* k, reactor_scheduler_context* n);
104 : };
105 :
106 : /// Thread-local context stack for reactor schedulers.
107 : inline thread_local_ptr<reactor_scheduler_context> reactor_context_stack;
108 :
109 : /// Find the context frame for a scheduler on this thread.
110 : inline reactor_scheduler_context*
111 962057 : reactor_find_context(reactor_scheduler const* self) noexcept
112 : {
113 962650 : for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
114 : {
115 899775 : if (c->key == self)
116 899182 : return c;
117 : }
118 62875 : return nullptr;
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 : {
144 : public:
145 : using context_type = reactor_scheduler_context;
146 : using mutex_type = conditionally_enabled_mutex;
147 : using lock_type = mutex_type::scoped_lock;
148 : using event_type = conditionally_enabled_event;
149 :
150 : /// Post a coroutine for deferred execution.
151 : void post(std::coroutine_handle<> h) const override;
152 :
153 : /// Post a scheduler operation for deferred execution.
154 : void post(scheduler_op* h) const override;
155 :
156 : /** Release an owner reference once no reactor pass can hold `d`.
157 :
158 : Called after a descriptor is deregistered. If a reactor pass is
159 : running, `keep` is parked on `d` until that pass ends. Then, or
160 : at once if no pass is running, it moves to `d`'s queued
161 : invocation if `d` is queued, and is dropped otherwise.
162 :
163 : @param d The deregistered descriptor.
164 : @param keep A reference to the descriptor's owner.
165 : */
166 : void retire_descriptor(reactor_retirable& d, object_ref keep) const;
167 :
168 : /// Post a continuation for deferred execution.
169 : void post(capy::continuation&) const override;
170 :
171 : /// Return true if called from a thread running this scheduler.
172 : bool running_in_this_thread() const noexcept override;
173 :
174 : /// Request the scheduler to stop dispatching handlers.
175 : void stop() override;
176 :
177 : /// Return true if the scheduler has been stopped.
178 : bool stopped() const noexcept override;
179 :
180 : /// Reset the stopped state so `run()` can resume.
181 : void restart() override;
182 :
183 : /// Run the event loop until no work remains.
184 : std::size_t run() override;
185 :
186 : /// Run until one handler completes or no work remains.
187 : std::size_t run_one() override;
188 :
189 : /// Run until one handler completes or @a usec elapses.
190 : std::size_t wait_one(long usec) override;
191 :
192 : /// Run ready handlers without blocking.
193 : std::size_t poll() override;
194 :
195 : /// Run at most one ready handler without blocking.
196 : std::size_t poll_one() override;
197 :
198 : /// Increment the outstanding work count.
199 : void work_started() noexcept override;
200 :
201 : /// Decrement the outstanding work count, stopping on zero.
202 : void work_finished() noexcept override;
203 :
204 : /** Reset the thread's inline completion budget.
205 :
206 : Called at the start of each posted completion handler to
207 : grant a fresh budget for speculative inline completions.
208 : */
209 : void reset_inline_budget() const noexcept;
210 :
211 : /** Consume one unit of inline budget if available.
212 :
213 : @return True if budget was available and consumed.
214 : */
215 : bool try_consume_inline_budget() const noexcept;
216 :
217 : /** Offset a forthcoming work_finished from work_cleanup.
218 :
219 : Called by descriptor_state when all I/O returned EAGAIN and
220 : no handler will be executed. Must be called from a scheduler
221 : thread.
222 : */
223 : /** Note that an op was parked outside the usual registration path.
224 :
225 : A reactor that builds its poll set from parked ops must rebuild
226 : it; one the kernel tracks needs nothing.
227 : */
228 366 : virtual void notify_reparked() const {}
229 :
230 : void compensating_work_started() const noexcept;
231 :
232 : /** Post an operation whose work is already counted.
233 :
234 : The posting thread touches nothing of the scheduler after this
235 : returns, unlike a @ref post followed by @ref work_finished: once
236 : posted, the operation can complete and let its context be
237 : destroyed.
238 :
239 : @pre work_started() was called for @p op.
240 :
241 : @param op The operation to post.
242 : */
243 : void post_counted(scheduler_op* op) const;
244 :
245 : /** Post completed operations for deferred invocation.
246 :
247 : If called from a thread running this scheduler, operations
248 : go to the thread's private queue (fast path). Otherwise,
249 : operations are added to the global queue under mutex and a
250 : waiter is signaled.
251 :
252 : @pre work_started() must have been called for each operation.
253 :
254 : @param ops Queue of operations to post.
255 : */
256 : void post_deferred_completions(ready_queue& ops) const;
257 :
258 : /** Apply runtime configuration to the scheduler.
259 :
260 : Called by `io_context` after construction. Values that do
261 : not apply to this backend are silently ignored.
262 :
263 : @param max_events Event buffer size for epoll/kqueue.
264 : @param budget_init Starting inline completion budget.
265 : @param budget_max Hard ceiling on adaptive budget ramp-up.
266 : @param unassisted Budget when single-threaded.
267 : */
268 : virtual void configure_reactor(
269 : unsigned max_events,
270 : unsigned budget_init,
271 : unsigned budget_max,
272 : unsigned unassisted);
273 :
274 : /// Return the configured initial inline budget.
275 13011 : unsigned inline_budget_initial() const noexcept
276 : {
277 13011 : return inline_budget_initial_;
278 : }
279 :
280 : /// Return true when scheduler locking is disabled (fully-lockless tier).
281 1450 : bool scheduler_locking_disabled() const noexcept override
282 : {
283 1450 : return scheduler_locking_disabled_;
284 : }
285 :
286 4426 : void configure_threading(threading_config cfg) noexcept override
287 : {
288 4426 : scheduler_locking_disabled_ = !cfg.scheduler_locking;
289 : // reactor_io_locking takes effect at descriptor registration (see the
290 : // register_descriptor overrides), not here.
291 4426 : reactor_io_locking_ = cfg.reactor_io_locking;
292 4426 : one_thread_ = cfg.one_thread;
293 4426 : mutex_.set_enabled(cfg.scheduler_locking);
294 4426 : cond_.set_enabled(cfg.scheduler_locking);
295 4426 : }
296 :
297 : protected:
298 : bool scheduler_locking_disabled_ = false;
299 : bool reactor_io_locking_ = true;
300 : bool one_thread_ = false;
301 :
302 4442 : reactor_scheduler() = default;
303 :
304 : /// Destroy, destroying whatever was posted after the drain.
305 : ~reactor_scheduler() override;
306 :
307 : /** Drain completed_ops during shutdown.
308 :
309 : Pops all operations from the global queue and destroys them,
310 : skipping the task sentinel. Signals all waiting threads.
311 : Derived classes call this from their shutdown() override
312 : before performing platform-specific cleanup.
313 : */
314 : void shutdown_drain();
315 :
316 : /// Destroy every queued operation without running it.
317 : void destroy_queued(lock_type& lock) noexcept;
318 :
319 : /// RAII guard that re-inserts the task sentinel after `run_task`.
320 : struct task_cleanup
321 : {
322 : reactor_scheduler const* sched;
323 : lock_type* lock;
324 : context_type& ctx;
325 : ~task_cleanup();
326 : };
327 :
328 : mutable mutex_type mutex_{true};
329 : mutable event_type cond_{true};
330 : mutable ready_queue completed_ops_;
331 : mutable std::atomic<std::int64_t> outstanding_work_{0};
332 : std::atomic<bool> stopped_{false};
333 : mutable std::atomic<bool> task_running_{false};
334 : mutable bool task_interrupted_ = false;
335 :
336 : // Descriptors retired during the running reactor pass (guarded by
337 : // mutex_).
338 : mutable reactor_retirable* retired_ = nullptr;
339 :
340 : /// Settle every descriptor retired before the current pass ended.
341 : /// @pre `lock` holds `mutex_`; it is held again on return.
342 : void settle_retired(lock_type& lock) const;
343 :
344 : // Runtime-configurable reactor tuning parameters.
345 : // Defaults match the library's built-in values.
346 : unsigned max_events_per_poll_ = 128;
347 : unsigned inline_budget_initial_ = 2;
348 : unsigned inline_budget_max_ = 16;
349 : unsigned unassisted_budget_ = 4;
350 :
351 : /// Bit 0 of `state_`: set when the condvar should be signaled.
352 : static constexpr std::size_t signaled_bit = 1;
353 :
354 : /// Increment per waiting thread in `state_`.
355 : static constexpr std::size_t waiter_increment = 2;
356 : mutable std::size_t state_ = 0;
357 :
358 : /// Sentinel op that triggers a reactor poll when dequeued.
359 : struct task_op final : scheduler_op
360 : {
361 : // LCOV_EXCL_START: the sentinel is intercepted by pointer
362 : // identity; its virtuals exist for vtable completeness.
363 : void operator()() override {}
364 : void destroy() override {}
365 : // LCOV_EXCL_STOP
366 : };
367 : task_op task_op_;
368 :
369 : /** Run the platform-specific reactor poll.
370 :
371 : @par Postconditions
372 : `lock` is owned on return, however the poll ended. An
373 : implementation that unlocks around the blocking call owes the
374 : caller a matching re-acquire on every path out, including the
375 : errors it retries rather than reports.
376 : */
377 : virtual void
378 : run_task(lock_type& lock, context_type& ctx, long timeout_us) = 0;
379 :
380 : /// Wake a blocked reactor (e.g. write to eventfd or pipe).
381 : virtual void interrupt_reactor() const = 0;
382 :
383 : /// Wake the reactor whenever the earliest timer changes.
384 : void on_timer_service_attached(timer_service& svc) override;
385 :
386 : private:
387 : struct work_cleanup
388 : {
389 : reactor_scheduler* sched;
390 : lock_type* lock;
391 : context_type& ctx;
392 : ~work_cleanup();
393 : };
394 :
395 : std::size_t do_one(lock_type& lock, long timeout_us, context_type& ctx);
396 :
397 : void signal_all(lock_type& lock) const;
398 : bool maybe_unlock_and_signal_one(lock_type& lock) const;
399 : bool unlock_and_signal_one(lock_type& lock) const;
400 : void clear_signal() const;
401 : void wait_for_signal(lock_type& lock) const;
402 : void wait_for_signal_for(lock_type& lock, long timeout_us) const;
403 : void wake_one_thread_and_unlock(lock_type& lock) const;
404 : };
405 :
406 : /** RAII guard that pushes/pops a scheduler context frame.
407 :
408 : On construction, pushes a new context frame onto the
409 : thread-local stack. On destruction, drains any remaining
410 : private queue items to the global queue and pops the frame.
411 : */
412 : struct reactor_thread_context_guard
413 : {
414 : /// The context frame managed by this guard.
415 : reactor_scheduler_context frame_;
416 :
417 : /// Marks the scheduler innermost for the executor's fast check.
418 : running_scheduler_guard running_;
419 :
420 : /// Construct the guard, pushing a frame for @a sched.
421 13011 : explicit reactor_thread_context_guard(
422 : reactor_scheduler const* sched) noexcept
423 13011 : : frame_(sched, reactor_context_stack.get())
424 13011 : , running_(sched)
425 : {
426 13011 : reactor_context_stack.set(&frame_);
427 13011 : }
428 :
429 : /** Destroy the guard, popping the frame.
430 :
431 : The private queue is empty here by invariant: work_cleanup and
432 : task_cleanup splice it to the global queue after every handler
433 : and every reactor pass.
434 : */
435 13011 : ~reactor_thread_context_guard() noexcept
436 : {
437 13011 : reactor_context_stack.set(frame_.next);
438 13011 : }
439 : };
440 :
441 : // ---- Inline implementations ------------------------------------------------
442 :
443 13011 : inline reactor_scheduler_context::reactor_scheduler_context(
444 13011 : reactor_scheduler const* k, reactor_scheduler_context* n)
445 13011 : : key(k)
446 13011 : , next(n)
447 13011 : , private_outstanding_work(0)
448 13011 : , inline_budget(0)
449 13011 : , inline_budget_max(static_cast<int>(k->inline_budget_initial()))
450 13011 : , unassisted(false)
451 : {
452 13011 : }
453 :
454 : inline void
455 54 : reactor_scheduler::configure_reactor(
456 : unsigned max_events,
457 : unsigned budget_init,
458 : unsigned budget_max,
459 : unsigned unassisted)
460 : {
461 106 : if (max_events < 1 ||
462 52 : max_events > static_cast<unsigned>(std::numeric_limits<int>::max()))
463 2 : throw std::out_of_range("max_events_per_poll must be in [1, INT_MAX]");
464 52 : if (budget_max > static_cast<unsigned>(std::numeric_limits<int>::max()))
465 2 : throw std::out_of_range("inline_budget_max must be in [0, INT_MAX]");
466 :
467 : // Clamp initial and unassisted to budget_max.
468 50 : if (budget_init > budget_max)
469 20 : budget_init = budget_max;
470 50 : if (unassisted > budget_max)
471 20 : unassisted = budget_max;
472 :
473 50 : max_events_per_poll_ = max_events;
474 50 : inline_budget_initial_ = budget_init;
475 50 : inline_budget_max_ = budget_max;
476 50 : unassisted_budget_ = unassisted;
477 50 : }
478 :
479 : inline void
480 104726 : reactor_scheduler::reset_inline_budget() const noexcept
481 : {
482 : // When budget is disabled (max==0), all paths below would no-op
483 : // (inline_budget stays 0). Skip the TLS lookup entirely.
484 104726 : if (inline_budget_max_ == 0)
485 58 : return;
486 104668 : if (auto* ctx = reactor_find_context(this))
487 : {
488 : // Cap when no other thread absorbed queued work
489 104668 : if (ctx->unassisted)
490 : {
491 103690 : ctx->inline_budget_max = static_cast<int>(unassisted_budget_);
492 103690 : ctx->inline_budget = static_cast<int>(unassisted_budget_);
493 103690 : return;
494 : }
495 : // Ramp up when previous cycle fully consumed budget.
496 : // max(1, ...) ensures the doubling escapes zero.
497 978 : if (ctx->inline_budget == 0)
498 294 : ctx->inline_budget_max =
499 294 : (std::min)((std::max)(1, ctx->inline_budget_max) * 2,
500 588 : static_cast<int>(inline_budget_max_));
501 684 : else if (ctx->inline_budget < ctx->inline_budget_max)
502 208 : ctx->inline_budget_max = static_cast<int>(inline_budget_initial_);
503 978 : ctx->inline_budget = ctx->inline_budget_max;
504 : }
505 : }
506 :
507 : inline bool
508 441619 : reactor_scheduler::try_consume_inline_budget() const noexcept
509 : {
510 441619 : if (inline_budget_max_ == 0)
511 44 : return false;
512 441575 : if (auto* ctx = reactor_find_context(this))
513 : {
514 441537 : if (ctx->inline_budget > 0)
515 : {
516 352948 : --ctx->inline_budget;
517 352948 : return true;
518 : }
519 : }
520 88627 : return false;
521 : }
522 :
523 : inline void
524 3672 : reactor_scheduler::post(std::coroutine_handle<> h) const
525 : {
526 : struct post_handler final : scheduler_op
527 : {
528 : std::coroutine_handle<> h_;
529 :
530 3672 : explicit post_handler(std::coroutine_handle<> h) : h_(h) {}
531 7344 : ~post_handler() override = default;
532 :
533 3656 : void operator()() override
534 : {
535 3656 : auto saved = h_;
536 3656 : delete this;
537 3656 : saved.resume();
538 3656 : }
539 :
540 16 : void destroy() override
541 : {
542 : // A bare handle carries no owner to unwind from: it is
543 : // destroyed directly.
544 16 : auto saved = h_;
545 16 : delete this;
546 16 : saved.destroy();
547 16 : }
548 : };
549 :
550 3672 : auto ph = std::make_unique<post_handler>(h);
551 :
552 3672 : if (auto* ctx = reactor_find_context(this))
553 : {
554 8 : ++ctx->private_outstanding_work;
555 8 : ctx->private_queue.push(ph.release());
556 8 : return;
557 : }
558 :
559 3664 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
560 :
561 3664 : lock_type lock(mutex_);
562 3664 : completed_ops_.push(ph.release());
563 3664 : wake_one_thread_and_unlock(lock);
564 3672 : }
565 :
566 : inline void
567 113149 : reactor_scheduler::post(scheduler_op* h) const
568 : {
569 113149 : if (auto* ctx = reactor_find_context(this))
570 : {
571 111214 : ++ctx->private_outstanding_work;
572 111214 : ctx->private_queue.push(h);
573 111214 : return;
574 : }
575 :
576 1935 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
577 :
578 1935 : lock_type lock(mutex_);
579 1935 : completed_ops_.push(h);
580 1935 : wake_one_thread_and_unlock(lock);
581 1935 : }
582 :
583 : inline void
584 236425 : reactor_scheduler::post(capy::continuation& c) const
585 : {
586 236425 : if (auto* ctx = reactor_find_context(this))
587 : {
588 204377 : ++ctx->private_outstanding_work;
589 204377 : ctx->private_queue.push(c);
590 204377 : return;
591 : }
592 :
593 32048 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
594 :
595 32048 : lock_type lock(mutex_);
596 32048 : completed_ops_.push(c);
597 32048 : wake_one_thread_and_unlock(lock);
598 32048 : }
599 :
600 : inline bool
601 23720 : reactor_scheduler::running_in_this_thread() const noexcept
602 : {
603 23720 : return reactor_find_context(this) != nullptr;
604 : }
605 :
606 : inline void
607 8374 : reactor_scheduler::stop()
608 : {
609 8374 : lock_type lock(mutex_);
610 8374 : if (!stopped_.load(std::memory_order_acquire))
611 : {
612 7313 : stopped_.store(true, std::memory_order_release);
613 7313 : signal_all(lock);
614 7313 : interrupt_reactor();
615 : }
616 8374 : }
617 :
618 : inline bool
619 3540 : reactor_scheduler::stopped() const noexcept
620 : {
621 3540 : return stopped_.load(std::memory_order_acquire);
622 : }
623 :
624 : inline void
625 5691 : reactor_scheduler::restart()
626 : {
627 5691 : stopped_.store(false, std::memory_order_release);
628 5691 : }
629 :
630 : inline std::size_t
631 6483 : reactor_scheduler::run()
632 : {
633 12966 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
634 : {
635 271 : stop();
636 271 : return 0;
637 : }
638 :
639 6212 : reactor_thread_context_guard ctx(this);
640 6212 : lock_type lock(mutex_);
641 :
642 6212 : std::size_t n = 0;
643 : for (;;)
644 : {
645 393013 : if (!do_one(lock, -1, ctx.frame_))
646 6209 : break;
647 386801 : if (n != (std::numeric_limits<std::size_t>::max)())
648 386801 : ++n;
649 386801 : if (!lock.owns_lock())
650 90101 : lock.lock();
651 : }
652 6209 : return n;
653 6215 : }
654 :
655 : inline std::size_t
656 128 : reactor_scheduler::run_one()
657 : {
658 256 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
659 : {
660 3 : stop();
661 3 : return 0;
662 : }
663 :
664 125 : reactor_thread_context_guard ctx(this);
665 125 : lock_type lock(mutex_);
666 125 : return do_one(lock, -1, ctx.frame_);
667 125 : }
668 :
669 : inline std::size_t
670 6132 : reactor_scheduler::wait_one(long usec)
671 : {
672 12264 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
673 : {
674 815 : stop();
675 815 : return 0;
676 : }
677 :
678 5317 : reactor_thread_context_guard ctx(this);
679 5317 : lock_type lock(mutex_);
680 5317 : return do_one(lock, usec, ctx.frame_);
681 5317 : }
682 :
683 : inline std::size_t
684 1331 : reactor_scheduler::poll()
685 : {
686 2662 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
687 : {
688 29 : stop();
689 29 : return 0;
690 : }
691 :
692 1302 : reactor_thread_context_guard ctx(this);
693 1302 : lock_type lock(mutex_);
694 :
695 1302 : std::size_t n = 0;
696 : for (;;)
697 : {
698 9300 : if (!do_one(lock, 0, ctx.frame_))
699 1302 : break;
700 7998 : if (n != (std::numeric_limits<std::size_t>::max)())
701 7998 : ++n;
702 7998 : if (!lock.owns_lock())
703 7996 : lock.lock();
704 : }
705 1302 : return n;
706 1302 : }
707 :
708 : inline std::size_t
709 60 : reactor_scheduler::poll_one()
710 : {
711 120 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
712 : {
713 5 : stop();
714 5 : return 0;
715 : }
716 :
717 55 : reactor_thread_context_guard ctx(this);
718 55 : lock_type lock(mutex_);
719 55 : return do_one(lock, 0, ctx.frame_);
720 55 : }
721 :
722 : inline void
723 66684 : reactor_scheduler::work_started() noexcept
724 : {
725 66684 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
726 66684 : }
727 :
728 : inline void
729 111750 : reactor_scheduler::work_finished() noexcept
730 : {
731 223500 : if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
732 7186 : stop();
733 111750 : }
734 :
735 : inline void
736 36773 : reactor_scheduler::compensating_work_started() const noexcept
737 : {
738 36773 : auto* ctx = reactor_find_context(this);
739 36773 : if (ctx)
740 36771 : ++ctx->private_outstanding_work;
741 36773 : }
742 :
743 : inline void
744 2073 : reactor_scheduler::post_counted(scheduler_op* op) const
745 : {
746 2073 : if (auto* ctx = reactor_find_context(this))
747 : {
748 605 : ctx->private_queue.push(op);
749 605 : return;
750 : }
751 :
752 1468 : lock_type lock(mutex_);
753 1468 : completed_ops_.push(op);
754 1468 : wake_one_thread_and_unlock(lock);
755 1468 : }
756 :
757 : inline void
758 15527 : reactor_scheduler::post_deferred_completions(ready_queue& ops) const
759 : {
760 15527 : if (ops.empty())
761 15527 : return;
762 :
763 2 : if (auto* ctx = reactor_find_context(this))
764 : {
765 2 : ctx->private_queue.splice(ops);
766 2 : return;
767 : }
768 :
769 MIS 0 : lock_type lock(mutex_);
770 0 : completed_ops_.splice(ops);
771 0 : wake_one_thread_and_unlock(lock);
772 0 : }
773 :
774 : inline void
775 HIT 22564 : reactor_scheduler::retire_descriptor(
776 : reactor_retirable& d, object_ref keep) const
777 : {
778 22564 : lock_type lock(mutex_);
779 22564 : if (!task_running_.load(std::memory_order_relaxed))
780 : {
781 : // No pass can hold an event, but an earlier one may have
782 : // queued d.
783 17209 : d.hand_off(keep);
784 17209 : return;
785 : }
786 : // Already parked for this pass: the held reference suffices.
787 5355 : if (d.retired_ref_)
788 107 : return;
789 5248 : d.retired_ref_ = std::move(keep);
790 5248 : d.retired_next_ = retired_;
791 5248 : retired_ = &d;
792 22564 : }
793 :
794 : inline void
795 297203 : reactor_scheduler::settle_retired(lock_type& lock) const
796 : {
797 : // Handed off under the lock, since a later pass may retire the same
798 : // descriptor again as soon as it is released. A reference that is
799 : // not handed off may be the owner's last, and its release recycles
800 : // through the pool mutex, so it is dropped with the lock released.
801 297203 : reactor_retirable* dropped = nullptr;
802 302451 : while (auto* d = retired_)
803 : {
804 5248 : retired_ = d->retired_next_;
805 5248 : d->retired_next_ = nullptr;
806 5248 : auto keep = std::move(d->retired_ref_);
807 5248 : d->hand_off(keep);
808 : // A drop already pending keeps the owner alive, so this one
809 : // cannot be the last and is released here.
810 5248 : if (!keep || d->dropped_ref_)
811 392 : continue;
812 4856 : d->dropped_ref_ = std::move(keep);
813 4856 : d->dropped_next_ = dropped;
814 4856 : dropped = d;
815 10496 : }
816 302059 : while (auto* d = dropped)
817 : {
818 4856 : dropped = d->dropped_next_;
819 4856 : d->dropped_next_ = nullptr;
820 4856 : auto keep = std::move(d->dropped_ref_);
821 4856 : lock.unlock();
822 4856 : keep.reset();
823 4856 : lock.lock();
824 4856 : }
825 297203 : }
826 :
827 : inline void
828 393 : reactor_scheduler::on_timer_service_attached(timer_service& svc)
829 : {
830 393 : svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) {
831 3191 : static_cast<reactor_scheduler*>(p)->interrupt_reactor();
832 3191 : }));
833 393 : }
834 :
835 : inline void
836 4426 : reactor_scheduler::shutdown_drain()
837 : {
838 4426 : wait_cancels_posting();
839 :
840 4426 : lock_type lock(mutex_);
841 4426 : settle_retired(lock);
842 4426 : destroy_queued(lock);
843 4426 : signal_all(lock);
844 4426 : }
845 :
846 4442 : inline reactor_scheduler::~reactor_scheduler()
847 : {
848 : // Services give up their work in shutdown(), but anything posted
849 : // after the drain, such as from a destructor, is still destroyed.
850 4442 : lock_type lock(mutex_);
851 4442 : destroy_queued(lock);
852 4442 : }
853 :
854 : inline void
855 8868 : reactor_scheduler::destroy_queued(lock_type& lock) noexcept
856 : {
857 22660 : while (auto e = completed_ops_.pop())
858 : {
859 13792 : if (ready_is_continuation(e))
860 : {
861 7827 : lock.unlock();
862 7827 : if (auto* c = ready_as_cont(e); c->h)
863 7827 : capy::abandon(*c);
864 7827 : lock.lock();
865 : }
866 : else
867 : {
868 5965 : auto* op = ready_as_op(e);
869 5965 : if (op == &task_op_)
870 4423 : continue;
871 1542 : lock.unlock();
872 1542 : op->destroy();
873 1542 : lock.lock();
874 : }
875 13792 : }
876 8868 : }
877 :
878 : inline void
879 11739 : reactor_scheduler::signal_all(lock_type&) const
880 : {
881 11739 : state_ |= signaled_bit;
882 11739 : cond_.notify_all();
883 11739 : }
884 :
885 : inline bool
886 39115 : reactor_scheduler::maybe_unlock_and_signal_one(lock_type& lock) const
887 : {
888 39115 : state_ |= signaled_bit;
889 39115 : if (state_ > signaled_bit)
890 : {
891 : // Signalled before unlocking: a post from another context's
892 : // thread may be the last touch of this scheduler before the
893 : // posted work runs, finishes and lets it be destroyed.
894 136 : cond_.notify_one();
895 136 : lock.unlock();
896 136 : return true;
897 : }
898 38979 : return false;
899 : }
900 :
901 : inline bool
902 650828 : reactor_scheduler::unlock_and_signal_one(lock_type& lock) const
903 : {
904 650828 : state_ |= signaled_bit;
905 650828 : bool have_waiters = state_ > signaled_bit;
906 650828 : lock.unlock();
907 650828 : if (have_waiters)
908 3367 : cond_.notify_one();
909 650828 : return have_waiters;
910 : }
911 :
912 : inline void
913 750 : reactor_scheduler::clear_signal() const
914 : {
915 750 : state_ &= ~signaled_bit;
916 750 : }
917 :
918 : inline void
919 741 : reactor_scheduler::wait_for_signal(lock_type& lock) const
920 : {
921 1561 : while ((state_ & signaled_bit) == 0)
922 : {
923 820 : state_ += waiter_increment;
924 820 : cond_.wait(lock);
925 820 : state_ -= waiter_increment;
926 : }
927 741 : }
928 :
929 : inline void
930 9 : reactor_scheduler::wait_for_signal_for(lock_type& lock, long timeout_us) const
931 : {
932 9 : if ((state_ & signaled_bit) == 0)
933 : {
934 9 : state_ += waiter_increment;
935 9 : cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
936 9 : state_ -= waiter_increment;
937 : }
938 9 : }
939 :
940 : inline void
941 39115 : reactor_scheduler::wake_one_thread_and_unlock(lock_type& lock) const
942 : {
943 39115 : if (maybe_unlock_and_signal_one(lock))
944 136 : return;
945 :
946 38979 : if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
947 : {
948 1739 : task_interrupted_ = true;
949 : // Under the lock, for the same reason as the signal above.
950 1739 : interrupt_reactor();
951 1739 : lock.unlock();
952 : }
953 : else
954 : {
955 37240 : lock.unlock();
956 : }
957 : }
958 :
959 397643 : inline reactor_scheduler::work_cleanup::~work_cleanup()
960 : {
961 397643 : std::int64_t produced = ctx.private_outstanding_work;
962 397643 : if (produced > 1)
963 980 : sched->outstanding_work_.fetch_add(
964 : produced - 1, std::memory_order_relaxed);
965 396663 : else if (produced < 1)
966 65047 : sched->work_finished();
967 397643 : ctx.private_outstanding_work = 0;
968 :
969 397643 : if (!ctx.private_queue.empty())
970 : {
971 297145 : lock->lock();
972 297145 : sched->completed_ops_.splice(ctx.private_queue);
973 : }
974 397643 : }
975 :
976 292780 : inline reactor_scheduler::task_cleanup::~task_cleanup()
977 : {
978 292780 : if (ctx.private_outstanding_work > 0)
979 : {
980 10150 : sched->outstanding_work_.fetch_add(
981 10150 : ctx.private_outstanding_work, std::memory_order_relaxed);
982 10150 : ctx.private_outstanding_work = 0;
983 : }
984 :
985 292780 : if (!ctx.private_queue.empty())
986 : {
987 10150 : if (!lock->owns_lock())
988 MIS 0 : lock->lock();
989 HIT 10150 : sched->completed_ops_.splice(ctx.private_queue);
990 : }
991 292780 : }
992 :
993 : inline std::size_t
994 407810 : reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx)
995 : {
996 : for (;;)
997 : {
998 698687 : if (stopped_.load(std::memory_order_acquire))
999 6385 : return 0;
1000 :
1001 692302 : std::uintptr_t e = completed_ops_.pop();
1002 692302 : scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
1003 :
1004 : // Handle reactor sentinel — time to poll for I/O
1005 692302 : if (op == &task_op_)
1006 : {
1007 293909 : bool more_handlers = !completed_ops_.empty();
1008 :
1009 332874 : if (!more_handlers &&
1010 77930 : (outstanding_work_.load(std::memory_order_acquire) == 0 ||
1011 : timeout_us == 0))
1012 : {
1013 1129 : completed_ops_.push(&task_op_);
1014 1129 : return 0;
1015 : }
1016 :
1017 292780 : long task_timeout_us = more_handlers ? 0 : timeout_us;
1018 292780 : task_interrupted_ = task_timeout_us == 0;
1019 292780 : task_running_.store(true, std::memory_order_release);
1020 :
1021 : // Wake a peer to take the pending handlers while this thread
1022 : // polls the reactor; skipped when one_thread_ (no peer exists).
1023 292780 : if (more_handlers && !one_thread_)
1024 254933 : unlock_and_signal_one(lock);
1025 :
1026 : try
1027 : {
1028 292780 : run_task(lock, ctx, task_timeout_us);
1029 : }
1030 3 : catch (...)
1031 : {
1032 3 : task_running_.store(false, std::memory_order_relaxed);
1033 3 : throw;
1034 3 : }
1035 :
1036 292777 : task_running_.store(false, std::memory_order_relaxed);
1037 292777 : completed_ops_.push(&task_op_);
1038 292777 : settle_retired(lock);
1039 292777 : if (timeout_us > 0)
1040 2650 : return 0;
1041 290127 : continue;
1042 290127 : }
1043 :
1044 : // Handle ready entry (op or continuation)
1045 398393 : if (e != 0)
1046 : {
1047 397643 : bool more = !completed_ops_.empty();
1048 :
1049 397643 : if (more && !one_thread_)
1050 : {
1051 : // Wake a peer for the remaining work; unassisted if none
1052 : // was parked to take it.
1053 395895 : ctx.unassisted = !unlock_and_signal_one(lock);
1054 : }
1055 : else
1056 : {
1057 : // No peer to wake (one_thread_, or nothing more queued).
1058 1748 : ctx.unassisted = more;
1059 1748 : lock.unlock();
1060 : }
1061 :
1062 397643 : [[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
1063 :
1064 397643 : if (ready_is_continuation(e))
1065 228598 : ready_as_cont(e)->h.resume();
1066 : else
1067 169045 : (*op)();
1068 397643 : return 1;
1069 397643 : }
1070 :
1071 1500 : if (outstanding_work_.load(std::memory_order_acquire) == 0 ||
1072 : timeout_us == 0)
1073 MIS 0 : return 0;
1074 :
1075 HIT 750 : clear_signal();
1076 750 : if (timeout_us < 0)
1077 741 : wait_for_signal(lock);
1078 : else
1079 9 : wait_for_signal_for(lock, timeout_us);
1080 290877 : }
1081 : }
1082 :
1083 : } // namespace boost::corosio::detail
1084 :
1085 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
|
