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