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

98.4% Lines (379 / 385, 2 excl) 100.0% Functions (49 / 49, 2 excl)

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

reactor_scheduler.hpp

f(x) Functions (51)

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	11890x	~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	1009565x	reactor_find_context(reactor_scheduler const* self) noexcept
112		{
113	1010087x	for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
114		{
115	947327x	if (c->key == self)
116	946805x	return c;
117		}
118	62760x	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	421x	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	13056x	unsigned inline_budget_initial() const noexcept
276		{
277	13056x	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	4415x	void configure_threading(threading_config cfg) noexcept override
287		{
288	4415x	scheduler_locking_disabled_ = !cfg.scheduler_locking;
289		// reactor_io_locking takes effect at descriptor registration (see the
290		// register_descriptor overrides), not here.
291	4415x	reactor_io_locking_ = cfg.reactor_io_locking;
292	4415x	one_thread_ = cfg.one_thread;
293	4415x	mutex_.set_enabled(cfg.scheduler_locking);
294	4415x	cond_.set_enabled(cfg.scheduler_locking);
295	4415x	}
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	4427x	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	13056x	explicit reactor_thread_context_guard(
420		reactor_scheduler const* sched) noexcept
421	13056x	: frame_(sched, reactor_context_stack.get())
422	13056x	, running_(sched)
423		{
424	13056x	reactor_context_stack.set(&frame_);
425	13056x	}
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	13056x	~reactor_thread_context_guard() noexcept
434		{
435	13056x	reactor_context_stack.set(frame_.next);
436	13056x	}
437		};
438
439		// ---- Inline implementations ------------------------------------------------
440
441	13056x	inline reactor_scheduler_context::reactor_scheduler_context(
442	13056x	reactor_scheduler const* k, reactor_scheduler_context* n)
443	13056x	: key(k)
444	13056x	, next(n)
445	13056x	, private_outstanding_work(0)
446	13056x	, inline_budget(0)
447	13056x	, inline_budget_max(static_cast<int>(k->inline_budget_initial()))
448	13056x	, unassisted(false)
449		{
450	13056x	}
451
452		inline void
453	54x	reactor_scheduler::configure_reactor(
454		unsigned max_events,
455		unsigned budget_init,
456		unsigned budget_max,
457		unsigned unassisted)
458		{
459	106x	if (max_events < 1 \|\|
460	52x	max_events > static_cast<unsigned>(std::numeric_limits<int>::max()))
461	2x	throw std::out_of_range("max_events_per_poll must be in [1, INT_MAX]");
462	52x	if (budget_max > static_cast<unsigned>(std::numeric_limits<int>::max()))
463	2x	throw std::out_of_range("inline_budget_max must be in [0, INT_MAX]");
464
465		// Clamp initial and unassisted to budget_max.
466	50x	if (budget_init > budget_max)
467	20x	budget_init = budget_max;
468	50x	if (unassisted > budget_max)
469	20x	unassisted = budget_max;
470
471	50x	max_events_per_poll_ = max_events;
472	50x	inline_budget_initial_ = budget_init;
473	50x	inline_budget_max_ = budget_max;
474	50x	unassisted_budget_ = unassisted;
475	50x	}
476
477		inline void
478	109020x	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	109020x	if (inline_budget_max_ == 0)
483	58x	return;
484	108962x	if (auto* ctx = reactor_find_context(this))
485		{
486		// Cap when no other thread absorbed queued work
487	108962x	if (ctx->unassisted)
488		{
489	108090x	ctx->inline_budget_max = static_cast<int>(unassisted_budget_);
490	108090x	ctx->inline_budget = static_cast<int>(unassisted_budget_);
491	108090x	return;
492		}
493		// Ramp up when previous cycle fully consumed budget.
494		// max(1, ...) ensures the doubling escapes zero.
495	872x	if (ctx->inline_budget == 0)
496	272x	ctx->inline_budget_max =
497	272x	(std::min)((std::max)(1, ctx->inline_budget_max) * 2,
498	544x	static_cast<int>(inline_budget_max_));
499	600x	else if (ctx->inline_budget < ctx->inline_budget_max)
500	205x	ctx->inline_budget_max = static_cast<int>(inline_budget_initial_);
501	872x	ctx->inline_budget = ctx->inline_budget_max;
502		}
503		}
504
505		inline bool
506	463453x	reactor_scheduler::try_consume_inline_budget() const noexcept
507		{
508	463453x	if (inline_budget_max_ == 0)
509	44x	return false;
510	463409x	if (auto* ctx = reactor_find_context(this))
511		{
512	463403x	if (ctx->inline_budget > 0)
513		{
514	370452x	--ctx->inline_budget;
515	370452x	return true;
516		}
517		}
518	92957x	return false;
519		}
520
521		inline void
522	3672x	reactor_scheduler::post(std::coroutine_handle<> h) const
523		{
524		struct post_handler final : scheduler_op
525		{
526		std::coroutine_handle<> h_;
527
528	3672x	explicit post_handler(std::coroutine_handle<> h) : h_(h) {}
529	7344x	~post_handler() override = default;
530
531	3656x	void operator()() override
532		{
533	3656x	auto saved = h_;
534	3656x	delete this;
535	3656x	saved.resume();
536	3656x	}
537
538	16x	void destroy() override
539		{
540		// A bare handle carries no owner to unwind from: it is
541		// destroyed directly.
542	16x	auto saved = h_;
543	16x	delete this;
544	16x	saved.destroy();
545	16x	}
546		};
547
548	3672x	auto ph = std::make_unique<post_handler>(h);
549
550	3672x	if (auto* ctx = reactor_find_context(this))
551		{
552	8x	++ctx->private_outstanding_work;
553	8x	ctx->private_queue.push(ph.release());
554	8x	return;
555		}
556
557	3664x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
558
559	3664x	lock_type lock(mutex_);
560	3664x	completed_ops_.push(ph.release());
561	3664x	wake_one_thread_and_unlock(lock);
562	3672x	}
563
564		inline void
565	117352x	reactor_scheduler::post(scheduler_op* h) const
566		{
567	117352x	if (auto* ctx = reactor_find_context(this))
568		{
569	115468x	++ctx->private_outstanding_work;
570	115468x	ctx->private_queue.push(h);
571	115468x	return;
572		}
573
574	1884x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
575
576	1884x	lock_type lock(mutex_);
577	1884x	completed_ops_.push(h);
578	1884x	wake_one_thread_and_unlock(lock);
579	1884x	}
580
581		inline void
582	252232x	reactor_scheduler::post(capy::continuation& c) const
583		{
584	252232x	if (auto* ctx = reactor_find_context(this))
585		{
586	220061x	++ctx->private_outstanding_work;
587	220061x	ctx->private_queue.push(c);
588	220061x	return;
589		}
590
591	32171x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
592
593	32171x	lock_type lock(mutex_);
594	32171x	completed_ops_.push(c);
595	32171x	wake_one_thread_and_unlock(lock);
596	32171x	}
597
598		inline bool
599	23656x	reactor_scheduler::running_in_this_thread() const noexcept
600		{
601	23656x	return reactor_find_context(this) != nullptr;
602		}
603
604		inline void
605	8228x	reactor_scheduler::stop()
606		{
607	8228x	lock_type lock(mutex_);
608	8228x	if (!stopped_.load(std::memory_order_acquire))
609		{
610	7248x	stopped_.store(true, std::memory_order_release);
611	7248x	signal_all(lock);
612	7248x	interrupt_reactor();
613		}
614	8228x	}
615
616		inline bool
617	3553x	reactor_scheduler::stopped() const noexcept
618		{
619	3553x	return stopped_.load(std::memory_order_acquire);
620		}
621
622		inline void
623	5632x	reactor_scheduler::restart()
624		{
625	5632x	stopped_.store(false, std::memory_order_release);
626	5632x	}
627
628		inline std::size_t
629	6421x	reactor_scheduler::run()
630		{
631	12842x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
632		{
633	227x	stop();
634	227x	return 0;
635		}
636
637	6194x	reactor_thread_context_guard ctx(this);
638	6194x	lock_type lock(mutex_);
639
640	6194x	std::size_t n = 0;
641		for (;;)
642		{
643	414605x	if (!do_one(lock, -1, ctx.frame_))
644	6191x	break;
645	408411x	if (n != (std::numeric_limits<std::size_t>::max)())
646	408411x	++n;
647	408411x	if (!lock.owns_lock())
648	91536x	lock.lock();
649		}
650	6191x	return n;
651	6197x	}
652
653		inline std::size_t
654	130x	reactor_scheduler::run_one()
655		{
656	260x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
657		{
658	3x	stop();
659	3x	return 0;
660		}
661
662	127x	reactor_thread_context_guard ctx(this);
663	127x	lock_type lock(mutex_);
664	127x	return do_one(lock, -1, ctx.frame_);
665	127x	}
666
667		inline std::size_t
668	6161x	reactor_scheduler::wait_one(long usec)
669		{
670	12322x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
671		{
672	778x	stop();
673	778x	return 0;
674		}
675
676	5383x	reactor_thread_context_guard ctx(this);
677	5383x	lock_type lock(mutex_);
678	5383x	return do_one(lock, usec, ctx.frame_);
679	5383x	}
680
681		inline std::size_t
682	1329x	reactor_scheduler::poll()
683		{
684	2658x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
685		{
686	30x	stop();
687	30x	return 0;
688		}
689
690	1299x	reactor_thread_context_guard ctx(this);
691	1299x	lock_type lock(mutex_);
692
693	1299x	std::size_t n = 0;
694		for (;;)
695		{
696	9295x	if (!do_one(lock, 0, ctx.frame_))
697	1299x	break;
698	7996x	if (n != (std::numeric_limits<std::size_t>::max)())
699	7996x	++n;
700	7996x	if (!lock.owns_lock())
701	7994x	lock.lock();
702		}
703	1299x	return n;
704	1299x	}
705
706		inline std::size_t
707	58x	reactor_scheduler::poll_one()
708		{
709	116x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
710		{
711	5x	stop();
712	5x	return 0;
713		}
714
715	53x	reactor_thread_context_guard ctx(this);
716	53x	lock_type lock(mutex_);
717	53x	return do_one(lock, 0, ctx.frame_);
718	53x	}
719
720		inline void
721	66767x	reactor_scheduler::work_started() noexcept
722		{
723	66767x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
724	66767x	}
725
726		inline void
727	111817x	reactor_scheduler::work_finished() noexcept
728		{
729	223634x	if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
730	7120x	stop();
731	111817x	}
732
733		inline void
734	38300x	reactor_scheduler::compensating_work_started() const noexcept
735		{
736	38300x	auto* ctx = reactor_find_context(this);
737	38300x	if (ctx)
738	38298x	++ctx->private_outstanding_work;
739	38300x	}
740
741		inline void
742	1980x	reactor_scheduler::post_counted(scheduler_op* op) const
743		{
744	1980x	if (auto* ctx = reactor_find_context(this))
745		{
746	603x	ctx->private_queue.push(op);
747	603x	return;
748		}
749
750	1377x	lock_type lock(mutex_);
751	1377x	completed_ops_.push(op);
752	1377x	wake_one_thread_and_unlock(lock);
753	1377x	}
754
755		inline void
756	15719x	reactor_scheduler::post_deferred_completions(ready_queue& ops) const
757		{
758	15719x	if (ops.empty())
759	15719x	return;
760
761	2x	if (auto* ctx = reactor_find_context(this))
762		{
763	2x	ctx->private_queue.splice(ops);
764	2x	return;
765		}
766
767	✗	lock_type lock(mutex_);
768	✗	completed_ops_.splice(ops);
769	✗	wake_one_thread_and_unlock(lock);
770	✗	}
771
772		inline void
773	22598x	reactor_scheduler::retire_descriptor(
774		reactor_retirable& d, object_ref keep) const
775		{
776	22598x	lock_type lock(mutex_);
777	22598x	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	17420x	d.hand_off(keep);
782	17420x	return;
783		}
784		// Already parked for this pass: the held reference suffices.
785	5178x	if (d.retired_ref_)
786	126x	return;
787	5052x	d.retired_ref_ = std::move(keep);
788	5052x	d.retired_next_ = retired_;
789	5052x	retired_ = &d;
790	22598x	}
791
792		inline void
793	305468x	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	305468x	reactor_retirable* dropped = nullptr;
800	310520x	while (auto* d = retired_)
801		{
802	5052x	retired_ = d->retired_next_;
803	5052x	d->retired_next_ = nullptr;
804	5052x	auto keep = std::move(d->retired_ref_);
805	5052x	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	5052x	if (!keep \|\| d->dropped_ref_)
809	328x	continue;
810	4724x	d->dropped_ref_ = std::move(keep);
811	4724x	d->dropped_next_ = dropped;
812	4724x	dropped = d;
813	10104x	}
814	310192x	while (auto* d = dropped)
815		{
816	4724x	dropped = d->dropped_next_;
817	4724x	d->dropped_next_ = nullptr;
818	4724x	auto keep = std::move(d->dropped_ref_);
819	4724x	lock.unlock();
820	4724x	keep.reset();
821	4724x	lock.lock();
822	4724x	}
823	305468x	}
824
825		inline void
826	4415x	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	4415x	if (timer_svc_)
833	4415x	timer_svc_->shutdown();
834
835	4415x	wait_cancels_posting();
836
837	4415x	lock_type lock(mutex_);
838	4415x	settle_retired(lock);
839	4415x	destroy_queued(lock);
840	4415x	signal_all(lock);
841	4415x	}
842
843	4427x	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	4427x	lock_type lock(mutex_);
848	4427x	destroy_queued(lock);
849	4427x	}
850
851		inline void
852	8842x	reactor_scheduler::destroy_queued(lock_type& lock) noexcept
853		{
854	22628x	while (auto e = completed_ops_.pop())
855		{
856	13786x	if (ready_is_continuation(e))
857		{
858	7826x	lock.unlock();
859	7826x	if (auto* c = ready_as_cont(e); c->h)
860	7826x	capy::abandon(*c);
861	7826x	lock.lock();
862		}
863		else
864		{
865	5960x	auto* op = ready_as_op(e);
866	5960x	if (op == &task_op_)
867	4412x	continue;
868	1548x	lock.unlock();
869	1548x	op->destroy();
870	1548x	lock.lock();
871		}
872	13786x	}
873	8842x	}
874
875		inline void
876	11663x	reactor_scheduler::signal_all(lock_type&) const
877		{
878	11663x	state_ \|= signaled_bit;
879	11663x	cond_.notify_all();
880	11663x	}
881
882		inline bool
883	39096x	reactor_scheduler::maybe_unlock_and_signal_one(lock_type& lock) const
884		{
885	39096x	state_ \|= signaled_bit;
886	39096x	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	192x	cond_.notify_one();
892	192x	lock.unlock();
893	192x	return true;
894		}
895	38904x	return false;
896		}
897
898		inline bool
899	685409x	reactor_scheduler::unlock_and_signal_one(lock_type& lock) const
900		{
901	685409x	state_ \|= signaled_bit;
902	685409x	bool have_waiters = state_ > signaled_bit;
903	685409x	lock.unlock();
904	685409x	if (have_waiters)
905	3164x	cond_.notify_one();
906	685409x	return have_waiters;
907		}
908
909		inline void
910	812x	reactor_scheduler::clear_signal() const
911		{
912	812x	state_ &= ~signaled_bit;
913	812x	}
914
915		inline void
916	798x	reactor_scheduler::wait_for_signal(lock_type& lock) const
917		{
918	1669x	while ((state_ & signaled_bit) == 0)
919		{
920	871x	state_ += waiter_increment;
921	871x	cond_.wait(lock);
922	871x	state_ -= waiter_increment;
923		}
924	798x	}
925
926		inline void
927	14x	reactor_scheduler::wait_for_signal_for(lock_type& lock, long timeout_us) const
928		{
929	14x	if ((state_ & signaled_bit) == 0)
930		{
931	14x	state_ += waiter_increment;
932	14x	cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
933	14x	state_ -= waiter_increment;
934		}
935	14x	}
936
937		inline void
938	39096x	reactor_scheduler::wake_one_thread_and_unlock(lock_type& lock) const
939		{
940	39096x	if (maybe_unlock_and_signal_one(lock))
941	192x	return;
942
943	38904x	if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
944		{
945	1520x	task_interrupted_ = true;
946		// Under the lock, for the same reason as the signal above.
947	1520x	interrupt_reactor();
948	1520x	lock.unlock();
949		}
950		else
951		{
952	37384x	lock.unlock();
953		}
954		}
955
956	419267x	inline reactor_scheduler::work_cleanup::~work_cleanup()
957		{
958	419267x	std::int64_t produced = ctx.private_outstanding_work;
959	419267x	if (produced > 1)
960	941x	sched->outstanding_work_.fetch_add(
961		produced - 1, std::memory_order_relaxed);
962	418326x	else if (produced < 1)
963	65126x	sched->work_finished();
964	419267x	ctx.private_outstanding_work = 0;
965
966	419267x	if (!ctx.private_queue.empty())
967		{
968	317155x	lock->lock();
969	317155x	sched->completed_ops_.splice(ctx.private_queue);
970		}
971	419267x	}
972
973	301056x	inline reactor_scheduler::task_cleanup::~task_cleanup()
974		{
975	301056x	if (ctx.private_outstanding_work > 0)
976		{
977	10574x	sched->outstanding_work_.fetch_add(
978	10574x	ctx.private_outstanding_work, std::memory_order_relaxed);
979	10574x	ctx.private_outstanding_work = 0;
980		}
981
982	301056x	if (!ctx.private_queue.empty())
983		{
984	10574x	if (!lock->owns_lock())
985	✗	lock->lock();
986	10574x	sched->completed_ops_.splice(ctx.private_queue);
987		}
988	301056x	}
989
990		inline std::size_t
991	429463x	reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx)
992		{
993		for (;;)
994		{
995	728627x	if (stopped_.load(std::memory_order_acquire))
996	6360x	return 0;
997
998	722267x	std::uintptr_t e = completed_ops_.pop();
999	722267x	scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
1000
1001		// Handle reactor sentinel — time to poll for I/O
1002	722267x	if (op == &task_op_)
1003		{
1004	302188x	bool more_handlers = !completed_ops_.empty();
1005
1006	336487x	if (!more_handlers &&
1007	68598x	(outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
1008		timeout_us == 0))
1009		{
1010	1132x	completed_ops_.push(&task_op_);
1011	1132x	return 0;
1012		}
1013
1014	301056x	long task_timeout_us = more_handlers ? 0 : timeout_us;
1015	301056x	task_interrupted_ = task_timeout_us == 0;
1016	301056x	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	301056x	if (more_handlers && !one_thread_)
1021	267878x	unlock_and_signal_one(lock);
1022
1023		try
1024		{
1025	301056x	run_task(lock, ctx, task_timeout_us);
1026		}
1027	3x	catch (...)
1028		{
1029	3x	task_running_.store(false, std::memory_order_relaxed);
1030	3x	throw;
1031	3x	}
1032
1033	301053x	task_running_.store(false, std::memory_order_relaxed);
1034	301053x	completed_ops_.push(&task_op_);
1035	301053x	settle_retired(lock);
1036	301053x	if (timeout_us > 0)
1037	2701x	return 0;
1038	298352x	continue;
1039	298352x	}
1040
1041		// Handle ready entry (op or continuation)
1042	420079x	if (e != 0)
1043		{
1044	419267x	bool more = !completed_ops_.empty();
1045
1046	419267x	if (more && !one_thread_)
1047		{
1048		// Wake a peer for the remaining work; unassisted if none
1049		// was parked to take it.
1050	417531x	ctx.unassisted = !unlock_and_signal_one(lock);
1051		}
1052		else
1053		{
1054		// No peer to wake (one_thread_, or nothing more queued).
1055	1736x	ctx.unassisted = more;
1056	1736x	lock.unlock();
1057		}
1058
1059	419267x	[[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
1060
1061	419267x	if (ready_is_continuation(e))
1062	244406x	ready_as_cont(e)->h.resume();
1063		else
1064	174861x	(*op)();
1065	419267x	return 1;
1066	419267x	}
1067
1068	1624x	if (outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
1069		timeout_us == 0)
1070	✗	return 0;
1071
1072	812x	clear_signal();
1073	812x	if (timeout_us < 0)
1074	798x	wait_for_signal(lock);
1075		else
1076	14x	wait_for_signal_for(lock, timeout_us);
1077	299164x	}
1078		}
1079
1080		} // namespace boost::corosio::detail
1081
1082		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
1083