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