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 11890 : ~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 1009565 : reactor_find_context(reactor_scheduler const* self) noexcept
112 : {
113 1010087 : for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
114 : {
115 947327 : if (c->key == self)
116 946805 : return c;
117 : }
118 62760 : 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 421 : 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 13056 : unsigned inline_budget_initial() const noexcept
276 : {
277 13056 : 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 4415 : void configure_threading(threading_config cfg) noexcept override
287 : {
288 4415 : scheduler_locking_disabled_ = !cfg.scheduler_locking;
289 : // reactor_io_locking takes effect at descriptor registration (see the
290 : // register_descriptor overrides), not here.
291 4415 : reactor_io_locking_ = cfg.reactor_io_locking;
292 4415 : one_thread_ = cfg.one_thread;
293 4415 : mutex_.set_enabled(cfg.scheduler_locking);
294 4415 : cond_.set_enabled(cfg.scheduler_locking);
295 4415 : }
296 :
297 : protected:
298 : timer_service* timer_svc_ = nullptr;
299 : bool scheduler_locking_disabled_ = false;
300 : bool reactor_io_locking_ = true;
301 : bool one_thread_ = false;
302 :
303 4427 : reactor_scheduler() = default;
304 :
305 : /// Destroy, destroying whatever was posted after the drain.
306 : ~reactor_scheduler() override;
307 :
308 : /** Drain completed_ops during shutdown.
309 :
310 : Pops all operations from the global queue and destroys them,
311 : skipping the task sentinel. Signals all waiting threads.
312 : Derived classes call this from their shutdown() override
313 : before performing platform-specific cleanup.
314 : */
315 : void shutdown_drain();
316 :
317 : /// Destroy every queued operation without running it.
318 : void destroy_queued(lock_type& lock) noexcept;
319 :
320 : /// RAII guard that re-inserts the task sentinel after `run_task`.
321 : struct task_cleanup
322 : {
323 : reactor_scheduler const* sched;
324 : lock_type* lock;
325 : context_type& ctx;
326 : ~task_cleanup();
327 : };
328 :
329 : mutable mutex_type mutex_{true};
330 : mutable event_type cond_{true};
331 : mutable ready_queue completed_ops_;
332 : mutable std::atomic<std::int64_t> outstanding_work_{0};
333 : std::atomic<bool> stopped_{false};
334 : mutable std::atomic<bool> task_running_{false};
335 : mutable bool task_interrupted_ = false;
336 :
337 : // Descriptors retired during the running reactor pass (guarded by
338 : // mutex_).
339 : mutable reactor_retirable* retired_ = nullptr;
340 :
341 : /// Settle every descriptor retired before the current pass ended.
342 : /// @pre `lock` holds `mutex_`; it is held again on return.
343 : void settle_retired(lock_type& lock) const;
344 :
345 : // Runtime-configurable reactor tuning parameters.
346 : // Defaults match the library's built-in values.
347 : unsigned max_events_per_poll_ = 128;
348 : unsigned inline_budget_initial_ = 2;
349 : unsigned inline_budget_max_ = 16;
350 : unsigned unassisted_budget_ = 4;
351 :
352 : /// Bit 0 of `state_`: set when the condvar should be signaled.
353 : static constexpr std::size_t signaled_bit = 1;
354 :
355 : /// Increment per waiting thread in `state_`.
356 : static constexpr std::size_t waiter_increment = 2;
357 : mutable std::size_t state_ = 0;
358 :
359 : /// Sentinel op that triggers a reactor poll when dequeued.
360 : struct task_op final : scheduler_op
361 : {
362 : // LCOV_EXCL_START: the sentinel is intercepted by pointer
363 : // identity; its virtuals exist for vtable completeness.
364 : void operator()() override {}
365 : void destroy() override {}
366 : // LCOV_EXCL_STOP
367 : };
368 : task_op task_op_;
369 :
370 : /** Run the platform-specific reactor poll.
371 :
372 : @par Postconditions
373 : `lock` is owned on return, however the poll ended. An
374 : implementation that unlocks around the blocking call owes the
375 : caller a matching re-acquire on every path out, including the
376 : errors it retries rather than reports.
377 : */
378 : virtual void
379 : run_task(lock_type& lock, context_type& ctx, long timeout_us) = 0;
380 :
381 : /// Wake a blocked reactor (e.g. write to eventfd or pipe).
382 : virtual void interrupt_reactor() const = 0;
383 :
384 : private:
385 : struct work_cleanup
386 : {
387 : reactor_scheduler* sched;
388 : lock_type* lock;
389 : context_type& ctx;
390 : ~work_cleanup();
391 : };
392 :
393 : std::size_t do_one(lock_type& lock, long timeout_us, context_type& ctx);
394 :
395 : void signal_all(lock_type& lock) const;
396 : bool maybe_unlock_and_signal_one(lock_type& lock) const;
397 : bool unlock_and_signal_one(lock_type& lock) const;
398 : void clear_signal() const;
399 : void wait_for_signal(lock_type& lock) const;
400 : void wait_for_signal_for(lock_type& lock, long timeout_us) const;
401 : void wake_one_thread_and_unlock(lock_type& lock) const;
402 : };
403 :
404 : /** RAII guard that pushes/pops a scheduler context frame.
405 :
406 : On construction, pushes a new context frame onto the
407 : thread-local stack. On destruction, drains any remaining
408 : private queue items to the global queue and pops the frame.
409 : */
410 : struct reactor_thread_context_guard
411 : {
412 : /// The context frame managed by this guard.
413 : reactor_scheduler_context frame_;
414 :
415 : /// Marks the scheduler innermost for the executor's fast check.
416 : running_scheduler_guard running_;
417 :
418 : /// Construct the guard, pushing a frame for @a sched.
419 13056 : explicit reactor_thread_context_guard(
420 : reactor_scheduler const* sched) noexcept
421 13056 : : frame_(sched, reactor_context_stack.get())
422 13056 : , running_(sched)
423 : {
424 13056 : reactor_context_stack.set(&frame_);
425 13056 : }
426 :
427 : /** Destroy the guard, popping the frame.
428 :
429 : The private queue is empty here by invariant: work_cleanup and
430 : task_cleanup splice it to the global queue after every handler
431 : and every reactor pass.
432 : */
433 13056 : ~reactor_thread_context_guard() noexcept
434 : {
435 13056 : reactor_context_stack.set(frame_.next);
436 13056 : }
437 : };
438 :
439 : // ---- Inline implementations ------------------------------------------------
440 :
441 13056 : inline reactor_scheduler_context::reactor_scheduler_context(
442 13056 : reactor_scheduler const* k, reactor_scheduler_context* n)
443 13056 : : key(k)
444 13056 : , next(n)
445 13056 : , private_outstanding_work(0)
446 13056 : , inline_budget(0)
447 13056 : , inline_budget_max(static_cast<int>(k->inline_budget_initial()))
448 13056 : , unassisted(false)
449 : {
450 13056 : }
451 :
452 : inline void
453 54 : reactor_scheduler::configure_reactor(
454 : unsigned max_events,
455 : unsigned budget_init,
456 : unsigned budget_max,
457 : unsigned unassisted)
458 : {
459 106 : if (max_events < 1 ||
460 52 : max_events > static_cast<unsigned>(std::numeric_limits<int>::max()))
461 2 : throw std::out_of_range("max_events_per_poll must be in [1, INT_MAX]");
462 52 : if (budget_max > static_cast<unsigned>(std::numeric_limits<int>::max()))
463 2 : throw std::out_of_range("inline_budget_max must be in [0, INT_MAX]");
464 :
465 : // Clamp initial and unassisted to budget_max.
466 50 : if (budget_init > budget_max)
467 20 : budget_init = budget_max;
468 50 : if (unassisted > budget_max)
469 20 : unassisted = budget_max;
470 :
471 50 : max_events_per_poll_ = max_events;
472 50 : inline_budget_initial_ = budget_init;
473 50 : inline_budget_max_ = budget_max;
474 50 : unassisted_budget_ = unassisted;
475 50 : }
476 :
477 : inline void
478 109020 : reactor_scheduler::reset_inline_budget() const noexcept
479 : {
480 : // When budget is disabled (max==0), all paths below would no-op
481 : // (inline_budget stays 0). Skip the TLS lookup entirely.
482 109020 : if (inline_budget_max_ == 0)
483 58 : return;
484 108962 : if (auto* ctx = reactor_find_context(this))
485 : {
486 : // Cap when no other thread absorbed queued work
487 108962 : if (ctx->unassisted)
488 : {
489 108090 : ctx->inline_budget_max = static_cast<int>(unassisted_budget_);
490 108090 : ctx->inline_budget = static_cast<int>(unassisted_budget_);
491 108090 : return;
492 : }
493 : // Ramp up when previous cycle fully consumed budget.
494 : // max(1, ...) ensures the doubling escapes zero.
495 872 : if (ctx->inline_budget == 0)
496 272 : ctx->inline_budget_max =
497 272 : (std::min)((std::max)(1, ctx->inline_budget_max) * 2,
498 544 : static_cast<int>(inline_budget_max_));
499 600 : else if (ctx->inline_budget < ctx->inline_budget_max)
500 205 : ctx->inline_budget_max = static_cast<int>(inline_budget_initial_);
501 872 : ctx->inline_budget = ctx->inline_budget_max;
502 : }
503 : }
504 :
505 : inline bool
506 463453 : reactor_scheduler::try_consume_inline_budget() const noexcept
507 : {
508 463453 : if (inline_budget_max_ == 0)
509 44 : return false;
510 463409 : if (auto* ctx = reactor_find_context(this))
511 : {
512 463403 : if (ctx->inline_budget > 0)
513 : {
514 370452 : --ctx->inline_budget;
515 370452 : return true;
516 : }
517 : }
518 92957 : return false;
519 : }
520 :
521 : inline void
522 3672 : reactor_scheduler::post(std::coroutine_handle<> h) const
523 : {
524 : struct post_handler final : scheduler_op
525 : {
526 : std::coroutine_handle<> h_;
527 :
528 3672 : explicit post_handler(std::coroutine_handle<> h) : h_(h) {}
529 7344 : ~post_handler() override = default;
530 :
531 3656 : void operator()() override
532 : {
533 3656 : auto saved = h_;
534 3656 : delete this;
535 3656 : saved.resume();
536 3656 : }
537 :
538 16 : void destroy() override
539 : {
540 : // A bare handle carries no owner to unwind from: it is
541 : // destroyed directly.
542 16 : auto saved = h_;
543 16 : delete this;
544 16 : saved.destroy();
545 16 : }
546 : };
547 :
548 3672 : auto ph = std::make_unique<post_handler>(h);
549 :
550 3672 : if (auto* ctx = reactor_find_context(this))
551 : {
552 8 : ++ctx->private_outstanding_work;
553 8 : ctx->private_queue.push(ph.release());
554 8 : return;
555 : }
556 :
557 3664 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
558 :
559 3664 : lock_type lock(mutex_);
560 3664 : completed_ops_.push(ph.release());
561 3664 : wake_one_thread_and_unlock(lock);
562 3672 : }
563 :
564 : inline void
565 117352 : reactor_scheduler::post(scheduler_op* h) const
566 : {
567 117352 : if (auto* ctx = reactor_find_context(this))
568 : {
569 115468 : ++ctx->private_outstanding_work;
570 115468 : ctx->private_queue.push(h);
571 115468 : return;
572 : }
573 :
574 1884 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
575 :
576 1884 : lock_type lock(mutex_);
577 1884 : completed_ops_.push(h);
578 1884 : wake_one_thread_and_unlock(lock);
579 1884 : }
580 :
581 : inline void
582 252232 : reactor_scheduler::post(capy::continuation& c) const
583 : {
584 252232 : if (auto* ctx = reactor_find_context(this))
585 : {
586 220061 : ++ctx->private_outstanding_work;
587 220061 : ctx->private_queue.push(c);
588 220061 : return;
589 : }
590 :
591 32171 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
592 :
593 32171 : lock_type lock(mutex_);
594 32171 : completed_ops_.push(c);
595 32171 : wake_one_thread_and_unlock(lock);
596 32171 : }
597 :
598 : inline bool
599 23656 : reactor_scheduler::running_in_this_thread() const noexcept
600 : {
601 23656 : return reactor_find_context(this) != nullptr;
602 : }
603 :
604 : inline void
605 8228 : reactor_scheduler::stop()
606 : {
607 8228 : lock_type lock(mutex_);
608 8228 : if (!stopped_.load(std::memory_order_acquire))
609 : {
610 7248 : stopped_.store(true, std::memory_order_release);
611 7248 : signal_all(lock);
612 7248 : interrupt_reactor();
613 : }
614 8228 : }
615 :
616 : inline bool
617 3553 : reactor_scheduler::stopped() const noexcept
618 : {
619 3553 : return stopped_.load(std::memory_order_acquire);
620 : }
621 :
622 : inline void
623 5632 : reactor_scheduler::restart()
624 : {
625 5632 : stopped_.store(false, std::memory_order_release);
626 5632 : }
627 :
628 : inline std::size_t
629 6421 : reactor_scheduler::run()
630 : {
631 12842 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
632 : {
633 227 : stop();
634 227 : return 0;
635 : }
636 :
637 6194 : reactor_thread_context_guard ctx(this);
638 6194 : lock_type lock(mutex_);
639 :
640 6194 : std::size_t n = 0;
641 : for (;;)
642 : {
643 414605 : if (!do_one(lock, -1, ctx.frame_))
644 6191 : break;
645 408411 : if (n != (std::numeric_limits<std::size_t>::max)())
646 408411 : ++n;
647 408411 : if (!lock.owns_lock())
648 91536 : lock.lock();
649 : }
650 6191 : return n;
651 6197 : }
652 :
653 : inline std::size_t
654 130 : reactor_scheduler::run_one()
655 : {
656 260 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
657 : {
658 3 : stop();
659 3 : return 0;
660 : }
661 :
662 127 : reactor_thread_context_guard ctx(this);
663 127 : lock_type lock(mutex_);
664 127 : return do_one(lock, -1, ctx.frame_);
665 127 : }
666 :
667 : inline std::size_t
668 6161 : reactor_scheduler::wait_one(long usec)
669 : {
670 12322 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
671 : {
672 778 : stop();
673 778 : return 0;
674 : }
675 :
676 5383 : reactor_thread_context_guard ctx(this);
677 5383 : lock_type lock(mutex_);
678 5383 : return do_one(lock, usec, ctx.frame_);
679 5383 : }
680 :
681 : inline std::size_t
682 1329 : reactor_scheduler::poll()
683 : {
684 2658 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
685 : {
686 30 : stop();
687 30 : return 0;
688 : }
689 :
690 1299 : reactor_thread_context_guard ctx(this);
691 1299 : lock_type lock(mutex_);
692 :
693 1299 : std::size_t n = 0;
694 : for (;;)
695 : {
696 9295 : if (!do_one(lock, 0, ctx.frame_))
697 1299 : break;
698 7996 : if (n != (std::numeric_limits<std::size_t>::max)())
699 7996 : ++n;
700 7996 : if (!lock.owns_lock())
701 7994 : lock.lock();
702 : }
703 1299 : return n;
704 1299 : }
705 :
706 : inline std::size_t
707 58 : reactor_scheduler::poll_one()
708 : {
709 116 : if (outstanding_work_.load(std::memory_order_acquire) == 0)
710 : {
711 5 : stop();
712 5 : return 0;
713 : }
714 :
715 53 : reactor_thread_context_guard ctx(this);
716 53 : lock_type lock(mutex_);
717 53 : return do_one(lock, 0, ctx.frame_);
718 53 : }
719 :
720 : inline void
721 66767 : reactor_scheduler::work_started() noexcept
722 : {
723 66767 : outstanding_work_.fetch_add(1, std::memory_order_relaxed);
724 66767 : }
725 :
726 : inline void
727 111817 : reactor_scheduler::work_finished() noexcept
728 : {
729 223634 : if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
730 7120 : stop();
731 111817 : }
732 :
733 : inline void
734 38300 : reactor_scheduler::compensating_work_started() const noexcept
735 : {
736 38300 : auto* ctx = reactor_find_context(this);
737 38300 : if (ctx)
738 38298 : ++ctx->private_outstanding_work;
739 38300 : }
740 :
741 : inline void
742 1980 : reactor_scheduler::post_counted(scheduler_op* op) const
743 : {
744 1980 : if (auto* ctx = reactor_find_context(this))
745 : {
746 603 : ctx->private_queue.push(op);
747 603 : return;
748 : }
749 :
750 1377 : lock_type lock(mutex_);
751 1377 : completed_ops_.push(op);
752 1377 : wake_one_thread_and_unlock(lock);
753 1377 : }
754 :
755 : inline void
756 15719 : reactor_scheduler::post_deferred_completions(ready_queue& ops) const
757 : {
758 15719 : if (ops.empty())
759 15719 : return;
760 :
761 2 : if (auto* ctx = reactor_find_context(this))
762 : {
763 2 : ctx->private_queue.splice(ops);
764 2 : return;
765 : }
766 :
767 MIS 0 : lock_type lock(mutex_);
768 0 : completed_ops_.splice(ops);
769 0 : wake_one_thread_and_unlock(lock);
770 0 : }
771 :
772 : inline void
773 HIT 22598 : reactor_scheduler::retire_descriptor(
774 : reactor_retirable& d, object_ref keep) const
775 : {
776 22598 : lock_type lock(mutex_);
777 22598 : if (!task_running_.load(std::memory_order_relaxed))
778 : {
779 : // No pass can hold an event, but an earlier one may have
780 : // queued d.
781 17420 : d.hand_off(keep);
782 17420 : return;
783 : }
784 : // Already parked for this pass: the held reference suffices.
785 5178 : if (d.retired_ref_)
786 126 : return;
787 5052 : d.retired_ref_ = std::move(keep);
788 5052 : d.retired_next_ = retired_;
789 5052 : retired_ = &d;
790 22598 : }
791 :
792 : inline void
793 305468 : reactor_scheduler::settle_retired(lock_type& lock) const
794 : {
795 : // Handed off under the lock, since a later pass may retire the same
796 : // descriptor again as soon as it is released. A reference that is
797 : // not handed off may be the owner's last, and its release recycles
798 : // through the pool mutex, so it is dropped with the lock released.
799 305468 : reactor_retirable* dropped = nullptr;
800 310520 : while (auto* d = retired_)
801 : {
802 5052 : retired_ = d->retired_next_;
803 5052 : d->retired_next_ = nullptr;
804 5052 : auto keep = std::move(d->retired_ref_);
805 5052 : d->hand_off(keep);
806 : // A drop already pending keeps the owner alive, so this one
807 : // cannot be the last and is released here.
808 5052 : if (!keep || d->dropped_ref_)
809 328 : continue;
810 4724 : d->dropped_ref_ = std::move(keep);
811 4724 : d->dropped_next_ = dropped;
812 4724 : dropped = d;
813 10104 : }
814 310192 : while (auto* d = dropped)
815 : {
816 4724 : dropped = d->dropped_next_;
817 4724 : d->dropped_next_ = nullptr;
818 4724 : auto keep = std::move(d->dropped_ref_);
819 4724 : lock.unlock();
820 4724 : keep.reset();
821 4724 : lock.lock();
822 4724 : }
823 305468 : }
824 :
825 : inline void
826 4415 : reactor_scheduler::shutdown_drain()
827 : {
828 : // This scheduler's constructor registered the timer service, so
829 : // the context shuts it down later; its waiters' frames must be
830 : // destroyed before the drain, which then destroys whatever their
831 : // destructors post. The context's own call later finds no waiter.
832 4415 : if (timer_svc_)
833 4415 : timer_svc_->shutdown();
834 :
835 4415 : wait_cancels_posting();
836 :
837 4415 : lock_type lock(mutex_);
838 4415 : settle_retired(lock);
839 4415 : destroy_queued(lock);
840 4415 : signal_all(lock);
841 4415 : }
842 :
843 4427 : inline reactor_scheduler::~reactor_scheduler()
844 : {
845 : // Services give up their work in shutdown(), but anything posted
846 : // after the drain, such as from a destructor, is still destroyed.
847 4427 : lock_type lock(mutex_);
848 4427 : destroy_queued(lock);
849 4427 : }
850 :
851 : inline void
852 8842 : reactor_scheduler::destroy_queued(lock_type& lock) noexcept
853 : {
854 22628 : while (auto e = completed_ops_.pop())
855 : {
856 13786 : if (ready_is_continuation(e))
857 : {
858 7826 : lock.unlock();
859 7826 : if (auto* c = ready_as_cont(e); c->h)
860 7826 : capy::abandon(*c);
861 7826 : lock.lock();
862 : }
863 : else
864 : {
865 5960 : auto* op = ready_as_op(e);
866 5960 : if (op == &task_op_)
867 4412 : continue;
868 1548 : lock.unlock();
869 1548 : op->destroy();
870 1548 : lock.lock();
871 : }
872 13786 : }
873 8842 : }
874 :
875 : inline void
876 11663 : reactor_scheduler::signal_all(lock_type&) const
877 : {
878 11663 : state_ |= signaled_bit;
879 11663 : cond_.notify_all();
880 11663 : }
881 :
882 : inline bool
883 39096 : reactor_scheduler::maybe_unlock_and_signal_one(lock_type& lock) const
884 : {
885 39096 : state_ |= signaled_bit;
886 39096 : if (state_ > signaled_bit)
887 : {
888 : // Signalled before unlocking: a post from another context's
889 : // thread may be the last touch of this scheduler before the
890 : // posted work runs, finishes and lets it be destroyed.
891 192 : cond_.notify_one();
892 192 : lock.unlock();
893 192 : return true;
894 : }
895 38904 : return false;
896 : }
897 :
898 : inline bool
899 685409 : reactor_scheduler::unlock_and_signal_one(lock_type& lock) const
900 : {
901 685409 : state_ |= signaled_bit;
902 685409 : bool have_waiters = state_ > signaled_bit;
903 685409 : lock.unlock();
904 685409 : if (have_waiters)
905 3164 : cond_.notify_one();
906 685409 : return have_waiters;
907 : }
908 :
909 : inline void
910 812 : reactor_scheduler::clear_signal() const
911 : {
912 812 : state_ &= ~signaled_bit;
913 812 : }
914 :
915 : inline void
916 798 : reactor_scheduler::wait_for_signal(lock_type& lock) const
917 : {
918 1669 : while ((state_ & signaled_bit) == 0)
919 : {
920 871 : state_ += waiter_increment;
921 871 : cond_.wait(lock);
922 871 : state_ -= waiter_increment;
923 : }
924 798 : }
925 :
926 : inline void
927 14 : reactor_scheduler::wait_for_signal_for(lock_type& lock, long timeout_us) const
928 : {
929 14 : if ((state_ & signaled_bit) == 0)
930 : {
931 14 : state_ += waiter_increment;
932 14 : cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
933 14 : state_ -= waiter_increment;
934 : }
935 14 : }
936 :
937 : inline void
938 39096 : reactor_scheduler::wake_one_thread_and_unlock(lock_type& lock) const
939 : {
940 39096 : if (maybe_unlock_and_signal_one(lock))
941 192 : return;
942 :
943 38904 : if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
944 : {
945 1520 : task_interrupted_ = true;
946 : // Under the lock, for the same reason as the signal above.
947 1520 : interrupt_reactor();
948 1520 : lock.unlock();
949 : }
950 : else
951 : {
952 37384 : lock.unlock();
953 : }
954 : }
955 :
956 419267 : inline reactor_scheduler::work_cleanup::~work_cleanup()
957 : {
958 419267 : std::int64_t produced = ctx.private_outstanding_work;
959 419267 : if (produced > 1)
960 941 : sched->outstanding_work_.fetch_add(
961 : produced - 1, std::memory_order_relaxed);
962 418326 : else if (produced < 1)
963 65126 : sched->work_finished();
964 419267 : ctx.private_outstanding_work = 0;
965 :
966 419267 : if (!ctx.private_queue.empty())
967 : {
968 317155 : lock->lock();
969 317155 : sched->completed_ops_.splice(ctx.private_queue);
970 : }
971 419267 : }
972 :
973 301056 : inline reactor_scheduler::task_cleanup::~task_cleanup()
974 : {
975 301056 : if (ctx.private_outstanding_work > 0)
976 : {
977 10574 : sched->outstanding_work_.fetch_add(
978 10574 : ctx.private_outstanding_work, std::memory_order_relaxed);
979 10574 : ctx.private_outstanding_work = 0;
980 : }
981 :
982 301056 : if (!ctx.private_queue.empty())
983 : {
984 10574 : if (!lock->owns_lock())
985 MIS 0 : lock->lock();
986 HIT 10574 : sched->completed_ops_.splice(ctx.private_queue);
987 : }
988 301056 : }
989 :
990 : inline std::size_t
991 429463 : reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx)
992 : {
993 : for (;;)
994 : {
995 728627 : if (stopped_.load(std::memory_order_acquire))
996 6360 : return 0;
997 :
998 722267 : std::uintptr_t e = completed_ops_.pop();
999 722267 : scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
1000 :
1001 : // Handle reactor sentinel — time to poll for I/O
1002 722267 : if (op == &task_op_)
1003 : {
1004 302188 : bool more_handlers = !completed_ops_.empty();
1005 :
1006 336487 : if (!more_handlers &&
1007 68598 : (outstanding_work_.load(std::memory_order_acquire) == 0 ||
1008 : timeout_us == 0))
1009 : {
1010 1132 : completed_ops_.push(&task_op_);
1011 1132 : return 0;
1012 : }
1013 :
1014 301056 : long task_timeout_us = more_handlers ? 0 : timeout_us;
1015 301056 : task_interrupted_ = task_timeout_us == 0;
1016 301056 : task_running_.store(true, std::memory_order_release);
1017 :
1018 : // Wake a peer to take the pending handlers while this thread
1019 : // polls the reactor; skipped when one_thread_ (no peer exists).
1020 301056 : if (more_handlers && !one_thread_)
1021 267878 : unlock_and_signal_one(lock);
1022 :
1023 : try
1024 : {
1025 301056 : run_task(lock, ctx, task_timeout_us);
1026 : }
1027 3 : catch (...)
1028 : {
1029 3 : task_running_.store(false, std::memory_order_relaxed);
1030 3 : throw;
1031 3 : }
1032 :
1033 301053 : task_running_.store(false, std::memory_order_relaxed);
1034 301053 : completed_ops_.push(&task_op_);
1035 301053 : settle_retired(lock);
1036 301053 : if (timeout_us > 0)
1037 2701 : return 0;
1038 298352 : continue;
1039 298352 : }
1040 :
1041 : // Handle ready entry (op or continuation)
1042 420079 : if (e != 0)
1043 : {
1044 419267 : bool more = !completed_ops_.empty();
1045 :
1046 419267 : if (more && !one_thread_)
1047 : {
1048 : // Wake a peer for the remaining work; unassisted if none
1049 : // was parked to take it.
1050 417531 : ctx.unassisted = !unlock_and_signal_one(lock);
1051 : }
1052 : else
1053 : {
1054 : // No peer to wake (one_thread_, or nothing more queued).
1055 1736 : ctx.unassisted = more;
1056 1736 : lock.unlock();
1057 : }
1058 :
1059 419267 : [[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
1060 :
1061 419267 : if (ready_is_continuation(e))
1062 244406 : ready_as_cont(e)->h.resume();
1063 : else
1064 174861 : (*op)();
1065 419267 : return 1;
1066 419267 : }
1067 :
1068 1624 : if (outstanding_work_.load(std::memory_order_acquire) == 0 ||
1069 : timeout_us == 0)
1070 MIS 0 : return 0;
1071 :
1072 HIT 812 : clear_signal();
1073 812 : if (timeout_us < 0)
1074 798 : wait_for_signal(lock);
1075 : else
1076 14 : wait_for_signal_for(lock, timeout_us);
1077 299164 : }
1078 : }
1079 :
1080 : } // namespace boost::corosio::detail
1081 :
1082 : #endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
|
