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 11915 : ~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 1002195 : reactor_find_context(reactor_scheduler const* self) noexcept
112 : {
113 1002718 : for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
114 : {
115 940389 : if (c->key == self)
116 939866 : return c;
117 : }
118 62329 : 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 : public scheduler
142 : {
143 : public:
144 : using context_type = reactor_scheduler_context;
145 : using mutex_type = conditionally_enabled_mutex;
146 : using lock_type = mutex_type::scoped_lock;
147 : using event_type = conditionally_enabled_event;
148 :
149 : /// Post a coroutine for deferred execution.
150 : void post(std::coroutine_handle<> h) const override;
151 :
152 : /// Post a scheduler operation for deferred execution.
153 : void post(scheduler_op* h) const override;
154 :
155 : /** Release an owner reference once no reactor pass can hold `d`.
156 :
157 : Called after a descriptor is deregistered. If a reactor pass is
158 : running, `keep` is parked on `d` until that pass ends. Then, or
159 : at once if no pass is running, it moves to `d`'s queued
160 : invocation if `d` is queued, and is dropped otherwise.
161 :
162 : @param d The deregistered descriptor.
163 : @param keep A reference to the descriptor's owner.
164 : */
165 : void retire_descriptor(reactor_retirable& d, object_ref keep) const;
166 :
167 : /// Post a continuation for deferred execution.
168 : void post(capy::continuation&) const override;
169 :
170 : /// Return true if called from a thread running this scheduler.
171 : bool running_in_this_thread() const noexcept override;
172 :
173 : /// Request the scheduler to stop dispatching handlers.
174 : void stop() override;
175 :
176 : /// Return true if the scheduler has been stopped.
177 : bool stopped() const noexcept override;
178 :
179 : /// Reset the stopped state so `run()` can resume.
180 : void restart() override;
181 :
182 : /// Run the event loop until no work remains.
183 : std::size_t run() override;
184 :
185 : /// Run until one handler completes or no work remains.
186 : std::size_t run_one() override;
187 :
188 : /// Run until one handler completes or @a usec elapses.
189 : std::size_t wait_one(long usec) override;
190 :
191 : /// Run ready handlers without blocking.
192 : std::size_t poll() override;
193 :
194 : /// Run at most one ready handler without blocking.
195 : std::size_t poll_one() override;
196 :
197 : /// Increment the outstanding work count.
198 : void work_started() noexcept override;
199 :
200 : /// Decrement the outstanding work count, stopping on zero.
201 : void work_finished() noexcept override;
202 :
203 : /** Reset the thread's inline completion budget.
204 :
205 : Called at the start of each posted completion handler to
206 : grant a fresh budget for speculative inline completions.
207 : */
208 : void reset_inline_budget() const noexcept;
209 :
210 : /** Consume one unit of inline budget if available.
211 :
212 : @return True if budget was available and consumed.
213 : */
214 : bool try_consume_inline_budget() const noexcept;
215 :
216 : /** Offset a forthcoming work_finished from work_cleanup.
217 :
218 : Called by descriptor_state when all I/O returned EAGAIN and
219 : no handler will be executed. Must be called from a scheduler
220 : thread.
221 : */
222 : /** Note that an op was parked outside the usual registration path.
223 :
224 : A reactor that builds its poll set from parked ops must rebuild
225 : it; one the kernel tracks needs nothing.
226 : */
227 455 : virtual void notify_reparked() const {}
228 :
229 : void compensating_work_started() const noexcept;
230 :
231 : /** Post an operation whose work is already counted.
232 :
233 : The posting thread touches nothing of the scheduler after this
234 : returns, unlike a @ref post followed by @ref work_finished: once
235 : posted, the operation can complete and let its context be
236 : destroyed.
237 :
238 : @pre work_started() was called for @p op.
239 :
240 : @param op The operation to post.
241 : */
242 : void post_counted(scheduler_op* op) const;
243 :
244 : /** Post completed operations for deferred invocation.
245 :
246 : If called from a thread running this scheduler, operations
247 : go to the thread's private queue (fast path). Otherwise,
248 : operations are added to the global queue under mutex and a
249 : waiter is signaled.
250 :
251 : @pre work_started() must have been called for each operation.
252 :
253 : @param ops Queue of operations to post.
254 : */
255 : void post_deferred_completions(ready_queue& ops) const;
256 :
257 : /** Apply runtime configuration to the scheduler.
258 :
259 : Called by `io_context` after construction. Values that do
260 : not apply to this backend are silently ignored.
261 :
262 : @param max_events Event buffer size for epoll/kqueue.
263 : @param budget_init Starting inline completion budget.
264 : @param budget_max Hard ceiling on adaptive budget ramp-up.
265 : @param unassisted Budget when single-threaded.
266 : */
267 : virtual void configure_reactor(
268 : unsigned max_events,
269 : unsigned budget_init,
270 : unsigned budget_max,
271 : unsigned unassisted);
272 :
273 : /// Return the configured initial inline budget.
274 12733 : unsigned inline_budget_initial() const noexcept
275 : {
276 12733 : return inline_budget_initial_;
277 : }
278 :
279 : /// Return true when scheduler locking is disabled (fully-lockless tier).
280 1448 : bool scheduler_locking_disabled() const noexcept override
281 : {
282 1448 : return scheduler_locking_disabled_;
283 : }
284 :
285 4415 : void configure_threading(threading_config cfg) noexcept override
286 : {
287 4415 : scheduler_locking_disabled_ = !cfg.scheduler_locking;
288 : // reactor_io_locking takes effect at descriptor registration (see the
289 : // register_descriptor overrides), not here.
290 4415 : reactor_io_locking_ = cfg.reactor_io_locking;
291 4415 : one_thread_ = cfg.one_thread;
292 4415 : mutex_.set_enabled(cfg.scheduler_locking);
293 4415 : cond_.set_enabled(cfg.scheduler_locking);
294 4415 : }
295 :
296 : protected:
297 : timer_service* timer_svc_ = nullptr;
298 : bool scheduler_locking_disabled_ = false;
299 : bool reactor_io_locking_ = true;
300 : bool one_thread_ = false;
301 :
302 4427 : 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 : private:
384 : struct work_cleanup
385 : {
386 : reactor_scheduler* sched;
387 : lock_type* lock;
388 : context_type& ctx;
389 : ~work_cleanup();
390 : };
391 :
392 : std::size_t do_one(lock_type& lock, long timeout_us, context_type& ctx);
393 :
394 : void signal_all(lock_type& lock) const;
395 : bool maybe_unlock_and_signal_one(lock_type& lock) const;
396 : bool unlock_and_signal_one(lock_type& lock) const;
397 : void clear_signal() const;
398 : void wait_for_signal(lock_type& lock) const;
399 : void wait_for_signal_for(lock_type& lock, long timeout_us) const;
400 : void wake_one_thread_and_unlock(lock_type& lock) const;
401 : };
402 :
403 : /** RAII guard that pushes/pops a scheduler context frame.
404 :
405 : On construction, pushes a new context frame onto the
406 : thread-local stack. On destruction, drains any remaining
407 : private queue items to the global queue and pops the frame.
408 : */
409 : struct reactor_thread_context_guard
410 : {
411 : /// The context frame managed by this guard.
412 : reactor_scheduler_context frame_;
413 :
414 : /// Marks the scheduler innermost for the executor's fast check.
415 : running_scheduler_guard running_;
416 :
417 : /// Construct the guard, pushing a frame for @a sched.
418 12733 : explicit reactor_thread_context_guard(
419 : reactor_scheduler const* sched) noexcept
420 12733 : : frame_(sched, reactor_context_stack.get())
421 12733 : , running_(sched)
422 : {
423 12733 : reactor_context_stack.set(&frame_);
424 12733 : }
425 :
426 : /** Destroy the guard, popping the frame.
427 :
428 : The private queue is empty here by invariant: work_cleanup and
429 : task_cleanup splice it to the global queue after every handler
430 : and every reactor pass.
431 : */
432 12733 : ~reactor_thread_context_guard() noexcept
433 : {
434 12733 : reactor_context_stack.set(frame_.next);
435 12733 : }
436 : };
437 :
438 : // ---- Inline implementations ------------------------------------------------
439 :
440 12733 : inline reactor_scheduler_context::reactor_scheduler_context(
441 12733 : reactor_scheduler const* k, reactor_scheduler_context* n)
442 12733 : : key(k)
443 12733 : , next(n)
444 12733 : , private_outstanding_work(0)
445 12733 : , inline_budget(0)
446 12733 : , inline_budget_max(static_cast<int>(k->inline_budget_initial()))
447 12733 : , unassisted(false)
448 : {
449 12733 : }
450 :
451 : inline void
452 54 : reactor_scheduler::configure_reactor(
453 : unsigned max_events,
454 : unsigned budget_init,
455 : unsigned budget_max,
456 : unsigned unassisted)
457 : {
458 106 : if (max_events < 1 ||
459 52 : max_events > static_cast<unsigned>(std::numeric_limits<int>::max()))
460 2 : throw std::out_of_range("max_events_per_poll must be in [1, INT_MAX]");
461 52 : if (budget_max > static_cast<unsigned>(std::numeric_limits<int>::max()))
462 2 : throw std::out_of_range("inline_budget_max must be in [0, INT_MAX]");
463 :
464 : // Clamp initial and unassisted to budget_max.
465 50 : if (budget_init > budget_max)
466 20 : budget_init = budget_max;
467 50 : if (unassisted > budget_max)
468 20 : unassisted = budget_max;
469 :
470 50 : max_events_per_poll_ = max_events;
471 50 : inline_budget_initial_ = budget_init;
472 50 : inline_budget_max_ = budget_max;
473 50 : unassisted_budget_ = unassisted;
474 50 : }
475 :
476 : inline void
477 107544 : reactor_scheduler::reset_inline_budget() const noexcept
478 : {
479 : // When budget is disabled (max==0), all paths below would no-op
480 : // (inline_budget stays 0). Skip the TLS lookup entirely.
481 107544 : if (inline_budget_max_ == 0)
482 58 : return;
483 107486 : if (auto* ctx = reactor_find_context(this))
484 : {
485 : // Cap when no other thread absorbed queued work
486 107486 : if (ctx->unassisted)
487 : {
488 106897 : ctx->inline_budget_max = static_cast<int>(unassisted_budget_);
489 106897 : ctx->inline_budget = static_cast<int>(unassisted_budget_);
490 106897 : return;
491 : }
492 : // Ramp up when previous cycle fully consumed budget.
493 : // max(1, ...) ensures the doubling escapes zero.
494 589 : if (ctx->inline_budget == 0)
495 231 : ctx->inline_budget_max =
496 231 : (std::min)((std::max)(1, ctx->inline_budget_max) * 2,
497 462 : static_cast<int>(inline_budget_max_));
498 358 : else if (ctx->inline_budget < ctx->inline_budget_max)
499 144 : ctx->inline_budget_max = static_cast<int>(inline_budget_initial_);
500 589 : ctx->inline_budget = ctx->inline_budget_max;
501 : }
502 : }
503 :
504 : inline bool
505 458616 : reactor_scheduler::try_consume_inline_budget() const noexcept
506 : {
507 458616 : if (inline_budget_max_ == 0)
508 44 : return false;
509 458572 : if (auto* ctx = reactor_find_context(this))
510 : {
511 458572 : if (ctx->inline_budget > 0)
512 : {
513 366623 : --ctx->inline_budget;
514 366623 : return true;
515 : }
516 : }
517 91949 : return false;
518 : }
519 :
520 : inline void
521 3672 : reactor_scheduler::post(std::coroutine_handle<> h) const
522 : {
523 : struct post_handler final : scheduler_op
524 : {
525 : std::coroutine_handle<> h_;
526 :
527 3672 : explicit post_handler(std::coroutine_handle<> h) : h_(h) {}
528 7344 : ~post_handler() override = default;
529 :
530 3656 : void operator()() override
531 : {
532 3656 : auto saved = h_;
533 3656 : delete this;
534 3656 : saved.resume();
535 3656 : }
536 :
537 16 : void destroy() override
538 : {
539 : // A bare handle carries no owner to unwind from: it is
540 : // destroyed directly.
541 16 : auto saved = h_;
542 16 : delete this;
543 16 : saved.destroy();
544 16 : }
545 : };
546 :
547 3672 : auto ph = std::make_unique<post_handler>(h);
548 :
549 3672 : if (auto* ctx = reactor_find_context(this))
550 : {
551 8 : ++ctx->private_outstanding_work;
552 8 : ctx->private_queue.push(ph.release());
553 8 : return;
554 : }
555 :
556 3664 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
557 :
558 3664 : lock_type lock(mutex_);
559 3664 : completed_ops_.push(ph.release());
560 3664 : wake_one_thread_and_unlock(lock);
561 3672 : }
562 :
563 : inline void
564 116342 : reactor_scheduler::post(scheduler_op* h) const
565 : {
566 116342 : if (auto* ctx = reactor_find_context(this))
567 : {
568 114223 : ++ctx->private_outstanding_work;
569 114223 : ctx->private_queue.push(h);
570 114223 : return;
571 : }
572 :
573 2119 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
574 :
575 2119 : lock_type lock(mutex_);
576 2119 : completed_ops_.push(h);
577 2119 : wake_one_thread_and_unlock(lock);
578 2119 : }
579 :
580 : inline void
581 251906 : reactor_scheduler::post(capy::continuation& c) const
582 : {
583 251906 : if (auto* ctx = reactor_find_context(this))
584 : {
585 220128 : ++ctx->private_outstanding_work;
586 220128 : ctx->private_queue.push(c);
587 220128 : return;
588 : }
589 :
590 31778 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
591 :
592 31778 : lock_type lock(mutex_);
593 31778 : completed_ops_.push(c);
594 31778 : wake_one_thread_and_unlock(lock);
595 31778 : }
596 :
597 : inline bool
598 23388 : reactor_scheduler::running_in_this_thread() const noexcept
599 : {
600 23388 : return reactor_find_context(this) != nullptr;
601 : }
602 :
603 : inline void
604 8035 : reactor_scheduler::stop()
605 : {
606 8035 : lock_type lock(mutex_);
607 8035 : if (!stopped_.load(std::memory_order_acquire))
608 : {
609 6999 : stopped_.store(true, std::memory_order_release);
610 6999 : signal_all(lock);
611 6999 : interrupt_reactor();
612 : }
613 8035 : }
614 :
615 : inline bool
616 3548 : reactor_scheduler::stopped() const noexcept
617 : {
618 3548 : return stopped_.load(std::memory_order_acquire);
619 : }
620 :
621 : inline void
622 5406 : reactor_scheduler::restart()
623 : {
624 5406 : stopped_.store(false, std::memory_order_release);
625 5406 : }
626 :
627 : inline std::size_t
628 6171 : reactor_scheduler::run()
629 : {
630 12342 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
631 : {
632 242 : stop();
633 242 : return 0;
634 : }
635 :
636 5929 : reactor_thread_context_guard ctx(this);
637 5929 : lock_type lock(mutex_);
638 :
639 5929 : std::size_t n = 0;
640 : for (;;)
641 : {
642 413301 : if (!do_one(lock, -1, ctx.frame_))
643 5926 : break;
644 407372 : if (n != (std::numeric_limits<std::size_t>::max)())
645 407372 : ++n;
646 407372 : if (!lock.owns_lock())
647 91667 : lock.lock();
648 : }
649 5926 : return n;
650 5932 : }
651 :
652 : inline std::size_t
653 130 : reactor_scheduler::run_one()
654 : {
655 260 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
656 : {
657 3 : stop();
658 3 : return 0;
659 : }
660 :
661 127 : reactor_thread_context_guard ctx(this);
662 127 : lock_type lock(mutex_);
663 127 : return do_one(lock, -1, ctx.frame_);
664 127 : }
665 :
666 : inline std::size_t
667 6146 : reactor_scheduler::wait_one(long usec)
668 : {
669 12292 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
670 : {
671 820 : stop();
672 820 : return 0;
673 : }
674 :
675 5326 : reactor_thread_context_guard ctx(this);
676 5326 : lock_type lock(mutex_);
677 5326 : return do_one(lock, usec, ctx.frame_);
678 5326 : }
679 :
680 : inline std::size_t
681 1327 : reactor_scheduler::poll()
682 : {
683 2654 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
684 : {
685 29 : stop();
686 29 : return 0;
687 : }
688 :
689 1298 : reactor_thread_context_guard ctx(this);
690 1298 : lock_type lock(mutex_);
691 :
692 1298 : std::size_t n = 0;
693 : for (;;)
694 : {
695 9275 : if (!do_one(lock, 0, ctx.frame_))
696 1298 : break;
697 7977 : if (n != (std::numeric_limits<std::size_t>::max)())
698 7977 : ++n;
699 7977 : if (!lock.owns_lock())
700 7975 : lock.lock();
701 : }
702 1298 : return n;
703 1298 : }
704 :
705 : inline std::size_t
706 58 : reactor_scheduler::poll_one()
707 : {
708 116 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
709 : {
710 5 : stop();
711 5 : return 0;
712 : }
713 :
714 53 : reactor_thread_context_guard ctx(this);
715 53 : lock_type lock(mutex_);
716 53 : return do_one(lock, 0, ctx.frame_);
717 53 : }
718 :
719 : inline void
720 66220 : reactor_scheduler::work_started() noexcept
721 : {
722 66220 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
723 66220 : }
724 :
725 : inline void
726 111054 : reactor_scheduler::work_finished() noexcept
727 : {
728 222108 : if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
729 6871 : stop();
730 111054 : }
731 :
732 : inline void
733 38848 : reactor_scheduler::compensating_work_started() const noexcept
734 : {
735 38848 : auto* ctx = reactor_find_context(this);
736 38848 : if (ctx)
737 38846 : ++ctx->private_outstanding_work;
738 38848 : }
739 :
740 : inline void
741 1979 : reactor_scheduler::post_counted(scheduler_op* op) const
742 : {
743 1979 : if (auto* ctx = reactor_find_context(this))
744 : {
745 601 : ctx->private_queue.push(op);
746 601 : return;
747 : }
748 :
749 1378 : lock_type lock(mutex_);
750 1378 : completed_ops_.push(op);
751 1378 : wake_one_thread_and_unlock(lock);
752 1378 : }
753 :
754 : inline void
755 15438 : reactor_scheduler::post_deferred_completions(ready_queue& ops) const
756 : {
757 15438 : if (ops.empty())
758 15438 : return;
759 :
760 2 : if (auto* ctx = reactor_find_context(this))
761 : {
762 2 : ctx->private_queue.splice(ops);
763 2 : return;
764 : }
765 :
766 MIS 0 : lock_type lock(mutex_);
767 0 : completed_ops_.splice(ops);
768 0 : wake_one_thread_and_unlock(lock);
769 0 : }
770 :
771 : inline void
772 HIT 22130 : reactor_scheduler::retire_descriptor(
773 : reactor_retirable& d, object_ref keep) const
774 : {
775 22130 : lock_type lock(mutex_);
776 22130 : if (!task_running_.load(std::memory_order_relaxed))
777 : {
778 : // No pass can hold an event, but an earlier one may have
779 : // queued d.
780 16841 : d.hand_off(keep);
781 16841 : return;
782 : }
783 : // Already parked for this pass: the held reference suffices.
784 5289 : if (d.retired_ref_)
785 5 : return;
786 5284 : d.retired_ref_ = std::move(keep);
787 5284 : d.retired_next_ = retired_;
788 5284 : retired_ = &d;
789 22130 : }
790 :
791 : inline void
792 304304 : reactor_scheduler::settle_retired(lock_type& lock) const
793 : {
794 : // Handed off under the lock, since a later pass may retire the same
795 : // descriptor again as soon as it is released. A reference that is
796 : // not handed off may be the owner's last, and its release recycles
797 : // through the pool mutex, so it is dropped with the lock released.
798 304304 : reactor_retirable* dropped = nullptr;
799 309588 : while (auto* d = retired_)
800 : {
801 5284 : retired_ = d->retired_next_;
802 5284 : d->retired_next_ = nullptr;
803 5284 : auto keep = std::move(d->retired_ref_);
804 5284 : d->hand_off(keep);
805 : // A drop already pending keeps the owner alive, so this one
806 : // cannot be the last and is released here.
807 5284 : if (!keep || d->dropped_ref_)
808 508 : continue;
809 4776 : d->dropped_ref_ = std::move(keep);
810 4776 : d->dropped_next_ = dropped;
811 4776 : dropped = d;
812 10568 : }
813 309080 : while (auto* d = dropped)
814 : {
815 4776 : dropped = d->dropped_next_;
816 4776 : d->dropped_next_ = nullptr;
817 4776 : auto keep = std::move(d->dropped_ref_);
818 4776 : lock.unlock();
819 4776 : keep.reset();
820 4776 : lock.lock();
821 4776 : }
822 304304 : }
823 :
824 : inline void
825 4415 : reactor_scheduler::shutdown_drain()
826 : {
827 : // This scheduler's constructor registered the timer service, so
828 : // the context shuts it down later; its waiters' frames must be
829 : // destroyed before the drain, which then destroys whatever their
830 : // destructors post. The context's own call later finds no waiter.
831 4415 : if (timer_svc_)
832 4415 : timer_svc_->shutdown();
833 :
834 4415 : wait_cancels_posting();
835 :
836 4415 : lock_type lock(mutex_);
837 4415 : settle_retired(lock);
838 4415 : destroy_queued(lock);
839 4415 : signal_all(lock);
840 4415 : }
841 :
842 4427 : inline reactor_scheduler::~reactor_scheduler()
843 : {
844 : // Service destructors run after the drain and may still post.
845 4427 : lock_type lock(mutex_);
846 4427 : destroy_queued(lock);
847 4427 : }
848 :
849 : inline void
850 8842 : reactor_scheduler::destroy_queued(lock_type& lock) noexcept
851 : {
852 22604 : while (auto e = completed_ops_.pop())
853 : {
854 13762 : if (ready_is_continuation(e))
855 : {
856 7808 : lock.unlock();
857 7808 : if (auto* c = ready_as_cont(e); c->h)
858 7808 : capy::abandon(*c);
859 7808 : lock.lock();
860 : }
861 : else
862 : {
863 5954 : auto* op = ready_as_op(e);
864 5954 : if (op == &task_op_)
865 4412 : continue;
866 1542 : lock.unlock();
867 1542 : op->destroy();
868 1542 : lock.lock();
869 : }
870 13762 : }
871 8842 : }
872 :
873 : inline void
874 11414 : reactor_scheduler::signal_all(lock_type&) const
875 : {
876 11414 : state_ |= signaled_bit;
877 11414 : cond_.notify_all();
878 11414 : }
879 :
880 : inline bool
881 38939 : reactor_scheduler::maybe_unlock_and_signal_one(lock_type& lock) const
882 : {
883 38939 : state_ |= signaled_bit;
884 38939 : if (state_ > signaled_bit)
885 : {
886 : // Signalled before unlocking: a post from another context's
887 : // thread may be the last touch of this scheduler before the
888 : // posted work runs, finishes and lets it be destroyed.
889 196 : cond_.notify_one();
890 196 : lock.unlock();
891 196 : return true;
892 : }
893 38743 : return false;
894 : }
895 :
896 : inline bool
897 679604 : reactor_scheduler::unlock_and_signal_one(lock_type& lock) const
898 : {
899 679604 : state_ |= signaled_bit;
900 679604 : bool have_waiters = state_ > signaled_bit;
901 679604 : lock.unlock();
902 679604 : if (have_waiters)
903 2631 : cond_.notify_one();
904 679604 : return have_waiters;
905 : }
906 :
907 : inline void
908 886 : reactor_scheduler::clear_signal() const
909 : {
910 886 : state_ &= ~signaled_bit;
911 886 : }
912 :
913 : inline void
914 873 : reactor_scheduler::wait_for_signal(lock_type& lock) const
915 : {
916 1825 : while ((state_ & signaled_bit) == 0)
917 : {
918 952 : state_ += waiter_increment;
919 952 : cond_.wait(lock);
920 952 : state_ -= waiter_increment;
921 : }
922 873 : }
923 :
924 : inline void
925 13 : reactor_scheduler::wait_for_signal_for(lock_type& lock, long timeout_us) const
926 : {
927 13 : if ((state_ & signaled_bit) == 0)
928 : {
929 13 : state_ += waiter_increment;
930 13 : cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
931 13 : state_ -= waiter_increment;
932 : }
933 13 : }
934 :
935 : inline void
936 38939 : reactor_scheduler::wake_one_thread_and_unlock(lock_type& lock) const
937 : {
938 38939 : if (maybe_unlock_and_signal_one(lock))
939 196 : return;
940 :
941 38743 : if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
942 : {
943 1647 : task_interrupted_ = true;
944 : // Under the lock, for the same reason as the signal above.
945 1647 : interrupt_reactor();
946 1647 : lock.unlock();
947 : }
948 : else
949 : {
950 37096 : lock.unlock();
951 : }
952 : }
953 :
954 418199 : inline reactor_scheduler::work_cleanup::~work_cleanup()
955 : {
956 418199 : std::int64_t produced = ctx.private_outstanding_work;
957 418199 : if (produced > 1)
958 892 : sched->outstanding_work_.fetch_add(
959 : produced - 1, std::memory_order_relaxed);
960 417307 : else if (produced < 1)
961 64597 : sched->work_finished();
962 418199 : ctx.private_outstanding_work = 0;
963 :
964 418199 : if (!ctx.private_queue.empty())
965 : {
966 316073 : lock->lock();
967 316073 : sched->completed_ops_.splice(ctx.private_queue);
968 : }
969 418199 : }
970 :
971 299892 : inline reactor_scheduler::task_cleanup::~task_cleanup()
972 : {
973 299892 : if (ctx.private_outstanding_work > 0)
974 : {
975 10531 : sched->outstanding_work_.fetch_add(
976 10531 : ctx.private_outstanding_work, std::memory_order_relaxed);
977 10531 : ctx.private_outstanding_work = 0;
978 : }
979 :
980 299892 : if (!ctx.private_queue.empty())
981 : {
982 10531 : if (!lock->owns_lock())
983 MIS 0 : lock->lock();
984 HIT 10531 : sched->completed_ops_.splice(ctx.private_queue);
985 : }
986 299892 : }
987 :
988 : inline std::size_t
989 428082 : reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx)
990 : {
991 : for (;;)
992 : {
993 726205 : if (stopped_.load(std::memory_order_acquire))
994 6101 : return 0;
995 :
996 720104 : auto const e = completed_ops_.pop();
997 720104 : scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
998 :
999 : // Handle reactor sentinel — time to poll for I/O
1000 720104 : if (op == &task_op_)
1001 : {
1002 301019 : bool more_handlers = !completed_ops_.empty();
1003 :
1004 338822 : if (!more_handlers &&
1005 75606 : (outstanding_work_.load(std::memory_order_acquire) == 0 ||
1006 : timeout_us == 0))
1007 : {
1008 1127 : completed_ops_.push(&task_op_);
1009 1127 : return 0;
1010 : }
1011 :
1012 299892 : long task_timeout_us = more_handlers ? 0 : timeout_us;
1013 299892 : task_interrupted_ = task_timeout_us == 0;
1014 299892 : task_running_.store(true, std::memory_order_release);
1015 :
1016 : // Wake a peer to take the pending handlers while this thread
1017 : // polls the reactor; skipped when one_thread_ (no peer exists).
1018 299892 : if (more_handlers && !one_thread_)
1019 263205 : unlock_and_signal_one(lock);
1020 :
1021 : try
1022 : {
1023 299892 : run_task(lock, ctx, task_timeout_us);
1024 : }
1025 3 : catch (...)
1026 : {
1027 3 : task_running_.store(false, std::memory_order_relaxed);
1028 3 : throw;
1029 3 : }
1030 :
1031 299889 : task_running_.store(false, std::memory_order_relaxed);
1032 299889 : completed_ops_.push(&task_op_);
1033 299889 : settle_retired(lock);
1034 299889 : if (timeout_us > 0)
1035 2652 : return 0;
1036 297237 : continue;
1037 297237 : }
1038 :
1039 : // Handle ready entry (op or continuation)
1040 419085 : if (e != nullptr)
1041 : {
1042 418199 : bool more = !completed_ops_.empty();
1043 :
1044 418199 : if (more && !one_thread_)
1045 : {
1046 : // Wake a peer for the remaining work; unassisted if none
1047 : // was parked to take it.
1048 416399 : ctx.unassisted = !unlock_and_signal_one(lock);
1049 : }
1050 : else
1051 : {
1052 : // No peer to wake (one_thread_, or nothing more queued).
1053 1800 : ctx.unassisted = more;
1054 1800 : lock.unlock();
1055 : }
1056 :
1057 418199 : [[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
1058 :
1059 418199 : if (ready_is_continuation(e))
1060 244098 : ready_as_cont(e)->h.resume();
1061 : else
1062 174101 : (*op)();
1063 418199 : return 1;
1064 418199 : }
1065 :
1066 1772 : if (outstanding_work_.load(std::memory_order_acquire) == 0 ||
1067 : timeout_us == 0)
1068 MIS 0 : return 0;
1069 :
1070 HIT 886 : clear_signal();
1071 886 : if (timeout_us < 0)
1072 873 : wait_for_signal(lock);
1073 : else
1074 13 : wait_for_signal_for(lock, timeout_us);
1075 298123 : }
1076 : }
1077 :
1078 : } // namespace boost::corosio::detail
1079 :
1080 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
|
