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

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