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

96.0% Lines (316/2/331) 100.0% List of functions (42/2/44)

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

reactor_scheduler.hpp

f(x) Functions (44)

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/capy/ex/execution_context.hpp>
15
16		#include <boost/corosio/detail/ready_queue.hpp>
17		#include <boost/corosio/detail/scheduler.hpp>
18		#include <boost/corosio/detail/scheduler_op.hpp>
19		#include <boost/corosio/detail/thread_local_ptr.hpp>
20
21		#include <atomic>
22		#include <chrono>
23		#include <coroutine>
24		#include <cstddef>
25		#include <cstdint>
26		#include <limits>
27		#include <memory>
28		#include <stdexcept>
29
30		#include <boost/corosio/detail/conditionally_enabled_mutex.hpp>
31		#include <boost/corosio/detail/conditionally_enabled_event.hpp>
32
33		namespace boost::corosio::detail {
34
35		// Forward declarations
36		class reactor_scheduler;
37		class timer_service;
38
39		/** Per-thread state for a reactor scheduler.
40
41		Each thread running a scheduler's event loop has one of these
42		on a thread-local stack. It holds a private work queue and
43		inline completion budget for speculative I/O fast paths.
44		*/
45		struct BOOST_COROSIO_SYMBOL_VISIBLE reactor_scheduler_context
46		{
47		/// Scheduler this context belongs to.
48		reactor_scheduler const* key;
49
50		/// Next context frame on this thread's stack.
51		reactor_scheduler_context* next;
52
53		/// Private work queue for reduced contention.
54		ready_queue private_queue;
55
56		/// Unflushed work count for the private queue.
57		std::int64_t private_outstanding_work;
58
59		/// Remaining inline completions allowed this cycle.
60		int inline_budget;
61
62		/// Maximum inline budget (adaptive, 2-16).
63		int inline_budget_max;
64
65		/// True if no other thread absorbed queued work last cycle.
66		bool unassisted;
67
68		/// Construct a context frame linked to @a n.
69		reactor_scheduler_context(
70		reactor_scheduler const* k, reactor_scheduler_context* n);
71		};
72
73		/// Thread-local context stack for reactor schedulers.
74		inline thread_local_ptr<reactor_scheduler_context> reactor_context_stack;
75
76		/// Find the context frame for a scheduler on this thread.
77		inline reactor_scheduler_context*
78	960856x	reactor_find_context(reactor_scheduler const* self) noexcept
79		{
80	960856x	for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
81		{
82	939422x	if (c->key == self)
83	939422x	return c;
84		}
85	21434x	return nullptr;
86		}
87
88		/** Non-template base for reactor-backed scheduler implementations.
89
90		Provides the complete threading model shared by epoll, kqueue,
91		and select schedulers: signal state machine, inline completion
92		budget, work counting, run/poll methods, and the do_one event
93		loop.
94
95		Derived classes provide platform-specific hooks by overriding:
96		- `run_task(lock, ctx)` to run the reactor poll
97		- `interrupt_reactor()` to wake a blocked reactor
98
99		De-templated from the original CRTP design to eliminate
100		duplicate instantiations when multiple backends are compiled
101		into the same binary. Virtual dispatch for run_task (called
102		once per reactor cycle, before a blocking syscall) has
103		negligible overhead.
104
105		@par Thread Safety
106		All public member functions are thread-safe.
107		*/
108		class reactor_scheduler
109		: public scheduler
110		, public capy::execution_context::service
111		{
112		public:
113		using key_type = scheduler;
114		using context_type = reactor_scheduler_context;
115		using mutex_type = conditionally_enabled_mutex;
116		using lock_type = mutex_type::scoped_lock;
117		using event_type = conditionally_enabled_event;
118
119		/// Post a coroutine for deferred execution.
120		void post(std::coroutine_handle<> h) const override;
121
122		/// Post a scheduler operation for deferred execution.
123		void post(scheduler_op* h) const override;
124
125		/// Post a continuation for deferred execution.
126		void post(capy::continuation&) const override;
127
128		/// Return true if called from a thread running this scheduler.
129		bool running_in_this_thread() const noexcept override;
130
131		/// Request the scheduler to stop dispatching handlers.
132		void stop() override;
133
134		/// Return true if the scheduler has been stopped.
135		bool stopped() const noexcept override;
136
137		/// Reset the stopped state so `run()` can resume.
138		void restart() override;
139
140		/// Run the event loop until no work remains.
141		std::size_t run() override;
142
143		/// Run until one handler completes or no work remains.
144		std::size_t run_one() override;
145
146		/// Run until one handler completes or @a usec elapses.
147		std::size_t wait_one(long usec) override;
148
149		/// Run ready handlers without blocking.
150		std::size_t poll() override;
151
152		/// Run at most one ready handler without blocking.
153		std::size_t poll_one() override;
154
155		/// Increment the outstanding work count.
156		void work_started() noexcept override;
157
158		/// Decrement the outstanding work count, stopping on zero.
159		void work_finished() noexcept override;
160
161		/** Reset the thread's inline completion budget.
162
163		Called at the start of each posted completion handler to
164		grant a fresh budget for speculative inline completions.
165		*/
166		void reset_inline_budget() const noexcept;
167
168		/** Consume one unit of inline budget if available.
169
170		@return True if budget was available and consumed.
171		*/
172		bool try_consume_inline_budget() const noexcept;
173
174		/** Offset a forthcoming work_finished from work_cleanup.
175
176		Called by descriptor_state when all I/O returned EAGAIN and
177		no handler will be executed. Must be called from a scheduler
178		thread.
179		*/
180		void compensating_work_started() const noexcept;
181
182		/** Post completed operations for deferred invocation.
183
184		If called from a thread running this scheduler, operations
185		go to the thread's private queue (fast path). Otherwise,
186		operations are added to the global queue under mutex and a
187		waiter is signaled.
188
189		@par Preconditions
190		work_started() must have been called for each operation.
191
192		@param ops Queue of operations to post.
193		*/
194		void post_deferred_completions(ready_queue& ops) const;
195
196		/** Apply runtime configuration to the scheduler.
197
198		Called by `io_context` after construction. Values that do
199		not apply to this backend are silently ignored.
200
201		@param max_events Event buffer size for epoll/kqueue.
202		@param budget_init Starting inline completion budget.
203		@param budget_max Hard ceiling on adaptive budget ramp-up.
204		@param unassisted Budget when single-threaded.
205		*/
206		virtual void configure_reactor(
207		unsigned max_events,
208		unsigned budget_init,
209		unsigned budget_max,
210		unsigned unassisted);
211
212		/// Return the configured initial inline budget.
213	1982x	unsigned inline_budget_initial() const noexcept
214		{
215	1982x	return inline_budget_initial_;
216		}
217
218		/// Return true when scheduler locking is disabled (fully-lockless tier).
219	302x	bool scheduler_locking_disabled() const noexcept override
220		{
221	302x	return scheduler_locking_disabled_;
222		}
223
224	2106x	void configure_threading(threading_config cfg) noexcept override
225		{
226	2106x	scheduler_locking_disabled_ = !cfg.scheduler_locking;
227		// reactor_io_locking takes effect at descriptor registration (see the
228		// register_descriptor overrides), not here.
229	2106x	reactor_io_locking_ = cfg.reactor_io_locking;
230	2106x	one_thread_ = cfg.one_thread;
231	2106x	mutex_.set_enabled(cfg.scheduler_locking);
232	2106x	cond_.set_enabled(cfg.scheduler_locking);
233	2106x	}
234
235		protected:
236		timer_service* timer_svc_ = nullptr;
237		bool scheduler_locking_disabled_ = false;
238		bool reactor_io_locking_ = true;
239		bool one_thread_ = false;
240
241	2118x	reactor_scheduler() = default;
242
243		/** Drain completed_ops during shutdown.
244
245		Pops all operations from the global queue and destroys them,
246		skipping the task sentinel. Signals all waiting threads.
247		Derived classes call this from their shutdown() override
248		before performing platform-specific cleanup.
249		*/
250		void shutdown_drain();
251
252		/// RAII guard that re-inserts the task sentinel after `run_task`.
253		struct task_cleanup
254		{
255		reactor_scheduler const* sched;
256		lock_type* lock;
257		context_type& ctx;
258		~task_cleanup();
259		};
260
261		mutable mutex_type mutex_{true};
262		mutable event_type cond_{true};
263		mutable ready_queue completed_ops_;
264		mutable std::atomic<std::int64_t> outstanding_work_{0};
265		std::atomic<bool> stopped_{false};
266		mutable std::atomic<bool> task_running_{false};
267		mutable bool task_interrupted_ = false;
268
269		// Runtime-configurable reactor tuning parameters.
270		// Defaults match the library's built-in values.
271		unsigned max_events_per_poll_ = 128;
272		unsigned inline_budget_initial_ = 2;
273		unsigned inline_budget_max_ = 16;
274		unsigned unassisted_budget_ = 4;
275
276		/// Bit 0 of `state_`: set when the condvar should be signaled.
277		static constexpr std::size_t signaled_bit = 1;
278
279		/// Increment per waiting thread in `state_`.
280		static constexpr std::size_t waiter_increment = 2;
281		mutable std::size_t state_ = 0;
282
283		/// Sentinel op that triggers a reactor poll when dequeued.
284		struct task_op final : scheduler_op
285		{
286		// LCOV_EXCL_START: the sentinel is intercepted by pointer
287		// identity; its virtuals exist for vtable completeness.
288	−	void operator()() override {}
289	−	void destroy() override {}
290		// LCOV_EXCL_STOP
291		};
292		task_op task_op_;
293
294		/** Run the platform-specific reactor poll.
295
296		@par Postconditions
297		`lock` is owned on return, however the poll ended. An
298		implementation that unlocks around the blocking call owes the
299		caller a matching re-acquire on every path out, including the
300		errors it retries rather than reports.
301		*/
302		virtual void
303		run_task(lock_type& lock, context_type& ctx, long timeout_us) = 0;
304
305		/// Wake a blocked reactor (e.g. write to eventfd or pipe).
306		virtual void interrupt_reactor() const = 0;
307
308		private:
309		struct work_cleanup
310		{
311		reactor_scheduler* sched;
312		lock_type* lock;
313		context_type& ctx;
314		~work_cleanup();
315		};
316
317		std::size_t do_one(lock_type& lock, long timeout_us, context_type& ctx);
318
319		void signal_all(lock_type& lock) const;
320		bool maybe_unlock_and_signal_one(lock_type& lock) const;
321		bool unlock_and_signal_one(lock_type& lock) const;
322		void clear_signal() const;
323		void wait_for_signal(lock_type& lock) const;
324		void wait_for_signal_for(lock_type& lock, long timeout_us) const;
325		void wake_one_thread_and_unlock(lock_type& lock) const;
326		};
327
328		/** RAII guard that pushes/pops a scheduler context frame.
329
330		On construction, pushes a new context frame onto the
331		thread-local stack. On destruction, drains any remaining
332		private queue items to the global queue and pops the frame.
333		*/
334		struct reactor_thread_context_guard
335		{
336		/// The context frame managed by this guard.
337		reactor_scheduler_context frame_;
338
339		/// Construct the guard, pushing a frame for @a sched.
340	1982x	explicit reactor_thread_context_guard(
341		reactor_scheduler const* sched) noexcept
342	1982x	: frame_(sched, reactor_context_stack.get())
343		{
344	1982x	reactor_context_stack.set(&frame_);
345	1982x	}
346
347		/** Destroy the guard, popping the frame.
348
349		The private queue is empty here by invariant: work_cleanup and
350		task_cleanup splice it to the global queue after every handler
351		and every reactor pass.
352		*/
353	1982x	~reactor_thread_context_guard() noexcept
354		{
355	1982x	reactor_context_stack.set(frame_.next);
356	1982x	}
357		};
358
359		// ---- Inline implementations ------------------------------------------------
360
361	1982x	inline reactor_scheduler_context::reactor_scheduler_context(
362	1982x	reactor_scheduler const* k, reactor_scheduler_context* n)
363	1982x	: key(k)
364	1982x	, next(n)
365	1982x	, private_outstanding_work(0)
366	1982x	, inline_budget(0)
367	1982x	, inline_budget_max(static_cast<int>(k->inline_budget_initial()))
368	1982x	, unassisted(false)
369		{
370	1982x	}
371
372		inline void
373	36x	reactor_scheduler::configure_reactor(
374		unsigned max_events,
375		unsigned budget_init,
376		unsigned budget_max,
377		unsigned unassisted)
378		{
379	70x	if (max_events < 1 \|\|
380	34x	max_events > static_cast<unsigned>(std::numeric_limits<int>::max()))
381	2x	throw std::out_of_range("max_events_per_poll must be in [1, INT_MAX]");
382	34x	if (budget_max > static_cast<unsigned>(std::numeric_limits<int>::max()))
383	2x	throw std::out_of_range("inline_budget_max must be in [0, INT_MAX]");
384
385		// Clamp initial and unassisted to budget_max.
386	32x	if (budget_init > budget_max)
387	8x	budget_init = budget_max;
388	32x	if (unassisted > budget_max)
389	8x	unassisted = budget_max;
390
391	32x	max_events_per_poll_ = max_events;
392	32x	inline_budget_initial_ = budget_init;
393	32x	inline_budget_max_ = budget_max;
394	32x	unassisted_budget_ = unassisted;
395	32x	}
396
397		inline void
398	91583x	reactor_scheduler::reset_inline_budget() const noexcept
399		{
400		// When budget is disabled (max==0), all paths below would no-op
401		// (inline_budget stays 0). Skip the TLS lookup entirely.
402	91583x	if (inline_budget_max_ == 0)
403	30x	return;
404	91553x	if (auto* ctx = reactor_find_context(this))
405		{
406		// Cap when no other thread absorbed queued work
407	91553x	if (ctx->unassisted)
408		{
409	91553x	ctx->inline_budget_max = static_cast<int>(unassisted_budget_);
410	91553x	ctx->inline_budget = static_cast<int>(unassisted_budget_);
411	91553x	return;
412		}
413		// Ramp up when previous cycle fully consumed budget.
414		// max(1, ...) ensures the doubling escapes zero.
415	✗	if (ctx->inline_budget == 0)
416	✗	ctx->inline_budget_max =
417	✗	(std::min)((std::max)(1, ctx->inline_budget_max) * 2,
418	✗	static_cast<int>(inline_budget_max_));
419	✗	else if (ctx->inline_budget < ctx->inline_budget_max)
420	✗	ctx->inline_budget_max = static_cast<int>(inline_budget_initial_);
421	✗	ctx->inline_budget = ctx->inline_budget_max;
422		}
423		}
424
425		inline bool
426	405078x	reactor_scheduler::try_consume_inline_budget() const noexcept
427		{
428	405078x	if (inline_budget_max_ == 0)
429	26x	return false;
430	405052x	if (auto* ctx = reactor_find_context(this))
431		{
432	405052x	if (ctx->inline_budget > 0)
433		{
434	323865x	--ctx->inline_budget;
435	323865x	return true;
436		}
437		}
438	81187x	return false;
439		}
440
441		inline void
442	3756x	reactor_scheduler::post(std::coroutine_handle<> h) const
443		{
444		struct post_handler final : scheduler_op
445		{
446		std::coroutine_handle<> h_;
447
448	3756x	explicit post_handler(std::coroutine_handle<> h) : h_(h) {}
449	7512x	~post_handler() override = default;
450
451	3744x	void operator()() override
452		{
453	3744x	auto saved = h_;
454	3744x	delete this;
455	3744x	saved.resume();
456	3744x	}
457
458	12x	void destroy() override
459		{
460	12x	auto saved = h_;
461	12x	delete this;
462	12x	saved.destroy();
463	12x	}
464		};
465
466	3756x	auto ph = std::make_unique<post_handler>(h);
467
468	3756x	if (auto* ctx = reactor_find_context(this))
469		{
470	96x	++ctx->private_outstanding_work;
471	96x	ctx->private_queue.push(ph.release());
472	96x	return;
473		}
474
475	3660x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
476
477	3660x	lock_type lock(mutex_);
478	3660x	completed_ops_.push(ph.release());
479	3660x	wake_one_thread_and_unlock(lock);
480	3756x	}
481
482		inline void
483	101153x	reactor_scheduler::post(scheduler_op* h) const
484		{
485	101153x	if (auto* ctx = reactor_find_context(this))
486		{
487	100390x	++ctx->private_outstanding_work;
488	100390x	ctx->private_queue.push(h);
489	100390x	return;
490		}
491
492	763x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
493
494	763x	lock_type lock(mutex_);
495	763x	completed_ops_.push(h);
496	763x	wake_one_thread_and_unlock(lock);
497	763x	}
498
499		inline void
500	23939x	reactor_scheduler::post(capy::continuation& c) const
501		{
502	23939x	if (auto* ctx = reactor_find_context(this))
503		{
504	15432x	++ctx->private_outstanding_work;
505	15432x	ctx->private_queue.push(c);
506	15432x	return;
507		}
508
509	8507x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
510
511	8507x	lock_type lock(mutex_);
512	8507x	completed_ops_.push(c);
513	8507x	wake_one_thread_and_unlock(lock);
514	8507x	}
515
516		inline bool
517	9448x	reactor_scheduler::running_in_this_thread() const noexcept
518		{
519	9448x	return reactor_find_context(this) != nullptr;
520		}
521
522		inline void
523	1850x	reactor_scheduler::stop()
524		{
525	1850x	lock_type lock(mutex_);
526	1850x	if (!stopped_.load(std::memory_order_acquire))
527		{
528	1753x	stopped_.store(true, std::memory_order_release);
529	1753x	signal_all(lock);
530	1753x	interrupt_reactor();
531		}
532	1850x	}
533
534		inline bool
535	139x	reactor_scheduler::stopped() const noexcept
536		{
537	139x	return stopped_.load(std::memory_order_acquire);
538		}
539
540		inline void
541	457x	reactor_scheduler::restart()
542		{
543	457x	stopped_.store(false, std::memory_order_release);
544	457x	}
545
546		inline std::size_t
547	1790x	reactor_scheduler::run()
548		{
549	3580x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
550		{
551	97x	stop();
552	97x	return 0;
553		}
554
555	1693x	reactor_thread_context_guard ctx(this);
556	1693x	lock_type lock(mutex_);
557
558	1693x	std::size_t n = 0;
559		for (;;)
560		{
561	465912x	if (!do_one(lock, -1, ctx.frame_))
562	1690x	break;
563	464219x	if (n != (std::numeric_limits<std::size_t>::max)())
564	464219x	++n;
565	464219x	if (!lock.owns_lock())
566	365232x	lock.lock();
567		}
568	1690x	return n;
569	1696x	}
570
571		inline std::size_t
572	112x	reactor_scheduler::run_one()
573		{
574	224x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
575		{
576	3x	stop();
577	3x	return 0;
578		}
579
580	109x	reactor_thread_context_guard ctx(this);
581	109x	lock_type lock(mutex_);
582	109x	return do_one(lock, -1, ctx.frame_);
583	109x	}
584
585		inline std::size_t
586	165x	reactor_scheduler::wait_one(long usec)
587		{
588	330x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
589		{
590	25x	stop();
591	25x	return 0;
592		}
593
594	140x	reactor_thread_context_guard ctx(this);
595	140x	lock_type lock(mutex_);
596	140x	return do_one(lock, usec, ctx.frame_);
597	140x	}
598
599		inline std::size_t
600	49x	reactor_scheduler::poll()
601		{
602	98x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
603		{
604	15x	stop();
605	15x	return 0;
606		}
607
608	34x	reactor_thread_context_guard ctx(this);
609	34x	lock_type lock(mutex_);
610
611	34x	std::size_t n = 0;
612		for (;;)
613		{
614	75x	if (!do_one(lock, 0, ctx.frame_))
615	34x	break;
616	41x	if (n != (std::numeric_limits<std::size_t>::max)())
617	41x	++n;
618	41x	if (!lock.owns_lock())
619	41x	lock.lock();
620		}
621	34x	return n;
622	34x	}
623
624		inline std::size_t
625	11x	reactor_scheduler::poll_one()
626		{
627	22x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
628		{
629	5x	stop();
630	5x	return 0;
631		}
632
633	6x	reactor_thread_context_guard ctx(this);
634	6x	lock_type lock(mutex_);
635	6x	return do_one(lock, 0, ctx.frame_);
636	6x	}
637
638		inline void
639	34582x	reactor_scheduler::work_started() noexcept
640		{
641	34582x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
642	34582x	}
643
644		inline void
645	64577x	reactor_scheduler::work_finished() noexcept
646		{
647	129154x	if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
648	1690x	stop();
649	64577x	}
650
651		inline void
652	325953x	reactor_scheduler::compensating_work_started() const noexcept
653		{
654	325953x	auto* ctx = reactor_find_context(this);
655	325953x	if (ctx)
656	325953x	++ctx->private_outstanding_work;
657	325953x	}
658
659		inline void
660	9736x	reactor_scheduler::post_deferred_completions(ready_queue& ops) const
661		{
662	9736x	if (ops.empty())
663	9736x	return;
664
665	2x	if (auto* ctx = reactor_find_context(this))
666		{
667	2x	ctx->private_queue.splice(ops);
668	2x	return;
669		}
670
671	✗	lock_type lock(mutex_);
672	✗	completed_ops_.splice(ops);
673	✗	wake_one_thread_and_unlock(lock);
674	✗	}
675
676		inline void
677	2106x	reactor_scheduler::shutdown_drain()
678		{
679	2106x	lock_type lock(mutex_);
680
681	4588x	while (auto e = completed_ops_.pop())
682		{
683	2482x	if (ready_is_continuation(e))
684		{
685	8x	lock.unlock();
686	8x	if (auto h = ready_as_cont(e)->h)
687	8x	h.destroy();
688	8x	lock.lock();
689		}
690		else
691		{
692	2474x	auto* op = ready_as_op(e);
693	2474x	if (op == &task_op_)
694	2103x	continue;
695	371x	lock.unlock();
696	371x	op->destroy();
697	371x	lock.lock();
698		}
699	2482x	}
700
701	2106x	signal_all(lock);
702	2106x	}
703
704		inline void
705	3859x	reactor_scheduler::signal_all(lock_type&) const
706		{
707	3859x	state_ \|= signaled_bit;
708	3859x	cond_.notify_all();
709	3859x	}
710
711		inline bool
712	12930x	reactor_scheduler::maybe_unlock_and_signal_one(lock_type& lock) const
713		{
714	12930x	state_ \|= signaled_bit;
715	12930x	if (state_ > signaled_bit)
716		{
717	38x	lock.unlock();
718	38x	cond_.notify_one();
719	38x	return true;
720		}
721	12892x	return false;
722		}
723
724		inline bool
725	516662x	reactor_scheduler::unlock_and_signal_one(lock_type& lock) const
726		{
727	516662x	state_ \|= signaled_bit;
728	516662x	bool have_waiters = state_ > signaled_bit;
729	516662x	lock.unlock();
730	516662x	if (have_waiters)
731	6x	cond_.notify_one();
732	516662x	return have_waiters;
733		}
734
735		inline void
736	38x	reactor_scheduler::clear_signal() const
737		{
738	38x	state_ &= ~signaled_bit;
739	38x	}
740
741		inline void
742	7x	reactor_scheduler::wait_for_signal(lock_type& lock) const
743		{
744	16x	while ((state_ & signaled_bit) == 0)
745		{
746	9x	state_ += waiter_increment;
747	9x	cond_.wait(lock);
748	9x	state_ -= waiter_increment;
749		}
750	7x	}
751
752		inline void
753	31x	reactor_scheduler::wait_for_signal_for(lock_type& lock, long timeout_us) const
754		{
755	31x	if ((state_ & signaled_bit) == 0)
756		{
757	31x	state_ += waiter_increment;
758	31x	cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
759	31x	state_ -= waiter_increment;
760		}
761	31x	}
762
763		inline void
764	12930x	reactor_scheduler::wake_one_thread_and_unlock(lock_type& lock) const
765		{
766	12930x	if (maybe_unlock_and_signal_one(lock))
767	38x	return;
768
769	12892x	if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
770		{
771	222x	task_interrupted_ = true;
772	222x	lock.unlock();
773	222x	interrupt_reactor();
774		}
775		else
776		{
777	12670x	lock.unlock();
778		}
779		}
780
781	464455x	inline reactor_scheduler::work_cleanup::~work_cleanup()
782		{
783	464455x	std::int64_t produced = ctx.private_outstanding_work;
784	464455x	if (produced > 1)
785	339x	sched->outstanding_work_.fetch_add(
786		produced - 1, std::memory_order_relaxed);
787	464116x	else if (produced < 1)
788	39801x	sched->work_finished();
789	464455x	ctx.private_outstanding_work = 0;
790
791	464455x	if (!ctx.private_queue.empty())
792		{
793	99011x	lock->lock();
794	99011x	sched->completed_ops_.splice(ctx.private_queue);
795		}
796	464455x	}
797
798	355146x	inline reactor_scheduler::task_cleanup::~task_cleanup()
799		{
800	355146x	if (ctx.private_outstanding_work > 0)
801		{
802	9730x	sched->outstanding_work_.fetch_add(
803	9730x	ctx.private_outstanding_work, std::memory_order_relaxed);
804	9730x	ctx.private_outstanding_work = 0;
805		}
806
807	355146x	if (!ctx.private_queue.empty())
808		{
809	9730x	if (!lock->owns_lock())
810	✗	lock->lock();
811	9730x	sched->completed_ops_.splice(ctx.private_queue);
812		}
813	355146x	}
814
815		inline std::size_t
816	466242x	reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx)
817		{
818		for (;;)
819		{
820	821366x	if (stopped_.load(std::memory_order_acquire))
821	1693x	return 0;
822
823	819673x	std::uintptr_t e = completed_ops_.pop();
824	819673x	scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
825
826		// Handle reactor sentinel — time to poll for I/O
827	819673x	if (op == &task_op_)
828		{
829	355180x	bool more_handlers = !completed_ops_.empty();
830
831	658102x	if (!more_handlers &&
832	605844x	(outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
833		timeout_us == 0))
834		{
835	34x	completed_ops_.push(&task_op_);
836	34x	return 0;
837		}
838
839	355146x	long task_timeout_us = more_handlers ? 0 : timeout_us;
840	355146x	task_interrupted_ = task_timeout_us == 0;
841	355146x	task_running_.store(true, std::memory_order_release);
842
843		// Wake a peer to take the pending handlers while this thread
844		// polls the reactor; skipped when one_thread_ (no peer exists).
845	355146x	if (more_handlers && !one_thread_)
846	52251x	unlock_and_signal_one(lock);
847
848		try
849		{
850	355146x	run_task(lock, ctx, task_timeout_us);
851		}
852	3x	catch (...)
853		{
854	3x	task_running_.store(false, std::memory_order_relaxed);
855	3x	throw;
856	3x	}
857
858	355143x	task_running_.store(false, std::memory_order_relaxed);
859	355143x	completed_ops_.push(&task_op_);
860	355143x	if (timeout_us > 0)
861	57x	return 0;
862	355086x	continue;
863	355086x	}
864
865		// Handle ready entry (op or continuation)
866	464493x	if (e != 0)
867		{
868	464455x	bool more = !completed_ops_.empty();
869
870	464455x	if (more && !one_thread_)
871		{
872		// Wake a peer for the remaining work; unassisted if none
873		// was parked to take it.
874	464411x	ctx.unassisted = !unlock_and_signal_one(lock);
875		}
876		else
877		{
878		// No peer to wake (one_thread_, or nothing more queued).
879	44x	ctx.unassisted = more;
880	44x	lock.unlock();
881		}
882
883	464455x	[[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
884
885	464455x	if (ready_is_continuation(e))
886	23931x	ready_as_cont(e)->h.resume();
887		else
888	440524x	(*op)();
889	464455x	return 1;
890	464455x	}
891
892	76x	if (outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
893		timeout_us == 0)
894	✗	return 0;
895
896	38x	clear_signal();
897	38x	if (timeout_us < 0)
898	7x	wait_for_signal(lock);
899		else
900	31x	wait_for_signal_for(lock, timeout_us);
901	355124x	}
902		}
903
904		} // namespace boost::corosio::detail
905
906		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
907