include/boost/corosio/native/detail/reactor/reactor_scheduler.hpp

98.5% Lines (382 / 388, 2 excl) 100.0% Functions (51 / 51, 2 excl)

[ ] prev/next file n p prev/next uncovered line

reactor_scheduler.hpp

f(x) Functions (53)

Function Calls Lines Blocks

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