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

95.8% Lines (319/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	1041465x	reactor_find_context(reactor_scheduler const* self) noexcept
80		{
81	1041465x	for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
82		{
83	1019880x	if (c->key == self)
84	1019880x	return c;
85		}
86	21585x	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	1964x	unsigned inline_budget_initial() const noexcept
217		{
218	1964x	return inline_budget_initial_;
219		}
220
221		/// Return true when scheduler locking is disabled (fully-lockless tier).
222	298x	bool scheduler_locking_disabled() const noexcept override
223		{
224	298x	return scheduler_locking_disabled_;
225		}
226
227	2095x	void configure_threading(threading_config cfg) noexcept override
228		{
229	2095x	scheduler_locking_disabled_ = !cfg.scheduler_locking;
230		// reactor_io_locking takes effect at descriptor registration (see the
231		// register_descriptor overrides), not here.
232	2095x	reactor_io_locking_ = cfg.reactor_io_locking;
233	2095x	one_thread_ = cfg.one_thread;
234	2095x	mutex_.set_enabled(cfg.scheduler_locking);
235	2095x	cond_.set_enabled(cfg.scheduler_locking);
236	2095x	}
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	2107x	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	1964x	explicit reactor_thread_context_guard(
347		reactor_scheduler const* sched) noexcept
348	1964x	: frame_(sched, reactor_context_stack.get())
349		{
350	1964x	reactor_context_stack.set(&frame_);
351	1964x	}
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	1964x	~reactor_thread_context_guard() noexcept
360		{
361	1964x	reactor_context_stack.set(frame_.next);
362	1964x	}
363		};
364
365		// ---- Inline implementations ------------------------------------------------
366
367		inline
368	1964x	reactor_scheduler_context::reactor_scheduler_context(
369		reactor_scheduler const* k,
370	1964x	reactor_scheduler_context* n)
371	1964x	: key(k)
372	1964x	, next(n)
373	1964x	, private_outstanding_work(0)
374	1964x	, inline_budget(0)
375	1964x	, inline_budget_max(
376	1964x	static_cast<int>(k->inline_budget_initial()))
377	1964x	, unassisted(false)
378		{
379	1964x	}
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	100371x	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	100371x	if (inline_budget_max_ == 0)
414	30x	return;
415	100341x	if (auto* ctx = reactor_find_context(this))
416		{
417		// Cap when no other thread absorbed queued work
418	100341x	if (ctx->unassisted)
419		{
420	100341x	ctx->inline_budget_max =
421	100341x	static_cast<int>(unassisted_budget_);
422	100341x	ctx->inline_budget =
423	100341x	static_cast<int>(unassisted_budget_);
424	100341x	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	428621x	reactor_scheduler::try_consume_inline_budget() const noexcept
441		{
442	428621x	if (inline_budget_max_ == 0)
443	26x	return false;
444	428595x	if (auto* ctx = reactor_find_context(this))
445		{
446	428595x	if (ctx->inline_budget > 0)
447		{
448	342700x	--ctx->inline_budget;
449	342700x	return true;
450		}
451		}
452	85895x	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	108008x	reactor_scheduler::post(scheduler_op* h) const
498		{
499	108008x	if (auto* ctx = reactor_find_context(this))
500		{
501	107082x	++ctx->private_outstanding_work;
502	107082x	ctx->private_queue.push(h);
503	107082x	return;
504		}
505
506	926x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
507
508	926x	lock_type lock(mutex_);
509	926x	completed_ops_.push(h);
510	926x	wake_one_thread_and_unlock(lock);
511	926x	}
512
513		inline void
514	26090x	reactor_scheduler::post(capy::continuation& c) const
515		{
516	26090x	if (auto* ctx = reactor_find_context(this))
517		{
518	17589x	++ctx->private_outstanding_work;
519	17589x	ctx->private_queue.push(c);
520	17589x	return;
521		}
522
523	8501x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
524
525	8501x	lock_type lock(mutex_);
526	8501x	completed_ops_.push(c);
527	8501x	wake_one_thread_and_unlock(lock);
528	8501x	}
529
530		inline bool
531	9438x	reactor_scheduler::running_in_this_thread() const noexcept
532		{
533	9438x	return reactor_find_context(this) != nullptr;
534		}
535
536		inline void
537	1846x	reactor_scheduler::stop()
538		{
539	1846x	lock_type lock(mutex_);
540	1846x	if (!stopped_.load(std::memory_order_acquire))
541		{
542	1743x	stopped_.store(true, std::memory_order_release);
543	1743x	signal_all(lock);
544	1743x	interrupt_reactor();
545		}
546	1846x	}
547
548		inline bool
549	137x	reactor_scheduler::stopped() const noexcept
550		{
551	137x	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	1780x	reactor_scheduler::run()
562		{
563	3560x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
564		{
565	103x	stop();
566	103x	return 0;
567		}
568
569	1677x	reactor_thread_context_guard ctx(this);
570	1677x	lock_type lock(mutex_);
571
572	1677x	std::size_t n = 0;
573		for (;;)
574		{
575	518271x	if (!do_one(lock, -1, ctx.frame_))
576	1674x	break;
577	516594x	if (n != (std::numeric_limits<std::size_t>::max)())
578	516594x	++n;
579	516594x	if (!lock.owns_lock())
580	410809x	lock.lock();
581		}
582	1674x	return n;
583	1680x	}
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	163x	reactor_scheduler::wait_one(long usec)
601		{
602	326x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
603		{
604	25x	stop();
605	25x	return 0;
606		}
607
608	138x	reactor_thread_context_guard ctx(this);
609	138x	lock_type lock(mutex_);
610	138x	return do_one(lock, usec, ctx.frame_);
611	138x	}
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	40802x	reactor_scheduler::work_started() noexcept
654		{
655	40802x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
656	40802x	}
657
658		inline void
659	73005x	reactor_scheduler::work_finished() noexcept
660		{
661	146010x	if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
662	1680x	stop();
663	73005x	}
664
665		inline void
666	365235x	reactor_scheduler::compensating_work_started() const noexcept
667		{
668	365235x	auto* ctx = reactor_find_context(this);
669	365235x	if (ctx)
670	365235x	++ctx->private_outstanding_work;
671	365235x	}
672
673
674		inline void
675	13823x	reactor_scheduler::post_deferred_completions(ready_queue& ops) const
676		{
677	13823x	if (ops.empty())
678	13823x	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	2095x	reactor_scheduler::shutdown_drain()
693		{
694	2095x	lock_type lock(mutex_);
695
696	4561x	while (auto e = completed_ops_.pop())
697		{
698	2466x	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	2458x	auto* op = ready_as_op(e);
708	2458x	if (op == &task_op_)
709	2092x	continue;
710	366x	lock.unlock();
711	366x	op->destroy();
712	366x	lock.lock();
713		}
714	2466x	}
715
716	2095x	signal_all(lock);
717	2095x	}
718
719		inline void
720	3838x	reactor_scheduler::signal_all(lock_type&) const
721		{
722	3838x	state_ \|= signaled_bit;
723	3838x	cond_.notify_all();
724	3838x	}
725
726		inline bool
727	13087x	reactor_scheduler::maybe_unlock_and_signal_one(
728		lock_type& lock) const
729		{
730	13087x	state_ \|= signaled_bit;
731	13087x	if (state_ > signaled_bit)
732		{
733	40x	lock.unlock();
734	40x	cond_.notify_one();
735	40x	return true;
736		}
737	13047x	return false;
738		}
739
740		inline bool
741	571824x	reactor_scheduler::unlock_and_signal_one(
742		lock_type& lock) const
743		{
744	571824x	state_ \|= signaled_bit;
745	571824x	bool have_waiters = state_ > signaled_bit;
746	571824x	lock.unlock();
747	571824x	if (have_waiters)
748	5x	cond_.notify_one();
749	571824x	return have_waiters;
750		}
751
752		inline void
753	55x	reactor_scheduler::clear_signal() const
754		{
755	55x	state_ &= ~signaled_bit;
756	55x	}
757
758		inline void
759	7x	reactor_scheduler::wait_for_signal(
760		lock_type& lock) const
761		{
762	15x	while ((state_ & signaled_bit) == 0)
763		{
764	8x	state_ += waiter_increment;
765	8x	cond_.wait(lock);
766	8x	state_ -= waiter_increment;
767		}
768	7x	}
769
770		inline void
771	48x	reactor_scheduler::wait_for_signal_for(
772		lock_type& lock, long timeout_us) const
773		{
774	48x	if ((state_ & signaled_bit) == 0)
775		{
776	48x	state_ += waiter_increment;
777	48x	cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
778	48x	state_ -= waiter_increment;
779		}
780	48x	}
781
782		inline void
783	13087x	reactor_scheduler::wake_one_thread_and_unlock(
784		lock_type& lock) const
785		{
786	13087x	if (maybe_unlock_and_signal_one(lock))
787	40x	return;
788
789	13047x	if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
790		{
791	221x	task_interrupted_ = true;
792	221x	lock.unlock();
793	221x	interrupt_reactor();
794		}
795		else
796		{
797	12826x	lock.unlock();
798		}
799		}
800
801	516830x	inline reactor_scheduler::work_cleanup::~work_cleanup()
802		{
803	516830x	std::int64_t produced = ctx.private_outstanding_work;
804	516830x	if (produced > 1)
805	340x	sched->outstanding_work_.fetch_add(
806		produced - 1, std::memory_order_relaxed);
807	516490x	else if (produced < 1)
808	46096x	sched->work_finished();
809	516830x	ctx.private_outstanding_work = 0;
810
811	516830x	if (!ctx.private_queue.empty())
812		{
813	105809x	lock->lock();
814	105809x	sched->completed_ops_.splice(ctx.private_queue);
815		}
816	516830x	}
817
818	403673x	inline reactor_scheduler::task_cleanup::~task_cleanup()
819		{
820	403673x	if (ctx.private_outstanding_work > 0)
821		{
822	11902x	sched->outstanding_work_.fetch_add(
823	11902x	ctx.private_outstanding_work, std::memory_order_relaxed);
824	11902x	ctx.private_outstanding_work = 0;
825		}
826
827	403673x	if (!ctx.private_queue.empty())
828		{
829	11902x	if (!lock->owns_lock())
830	✗	lock->lock();
831	11902x	sched->completed_ops_.splice(ctx.private_queue);
832		}
833	403673x	}
834
835		inline std::size_t
836	518599x	reactor_scheduler::do_one(
837		lock_type& lock, long timeout_us, context_type& ctx)
838		{
839		for (;;)
840		{
841	922269x	if (stopped_.load(std::memory_order_acquire))
842	1675x	return 0;
843
844	920594x	std::uintptr_t e = completed_ops_.pop();
845	920594x	scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
846
847		// Handle reactor sentinel — time to poll for I/O
848	920594x	if (op == &task_op_)
849		{
850	403709x	bool more_handlers = !completed_ops_.empty();
851
852	752359x	if (!more_handlers &&
853	697300x	(outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
854		timeout_us == 0))
855		{
856	36x	completed_ops_.push(&task_op_);
857	36x	return 0;
858		}
859
860	403673x	long task_timeout_us = more_handlers ? 0 : timeout_us;
861	403673x	task_interrupted_ = task_timeout_us == 0;
862	403673x	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	403673x	if (more_handlers && !one_thread_)
867	55052x	unlock_and_signal_one(lock);
868
869		try
870		{
871	403673x	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	403670x	task_running_.store(false, std::memory_order_relaxed);
880	403670x	completed_ops_.push(&task_op_);
881	403670x	if (timeout_us > 0)
882	55x	return 0;
883	403615x	continue;
884	403615x	}
885
886		// Handle ready entry (op or continuation)
887	516885x	if (e != 0)
888		{
889	516830x	bool more = !completed_ops_.empty();
890
891	516830x	if (more && !one_thread_)
892		{
893		// Wake a peer for the remaining work; unassisted if none
894		// was parked to take it.
895	516772x	ctx.unassisted = !unlock_and_signal_one(lock);
896		}
897		else
898		{
899		// No peer to wake (one_thread_, or nothing more queued).
900	58x	ctx.unassisted = more;
901	58x	lock.unlock();
902		}
903
904	516830x	[[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
905
906	516830x	if (ready_is_continuation(e))
907	26082x	ready_as_cont(e)->h.resume();
908		else
909	490748x	(*op)();
910	516830x	return 1;
911	516830x	}
912
913	110x	if (outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
914		timeout_us == 0)
915	✗	return 0;
916
917	55x	clear_signal();
918	55x	if (timeout_us < 0)
919	7x	wait_for_signal(lock);
920		else
921	48x	wait_for_signal_for(lock, timeout_us);
922	403670x	}
923		}
924
925		} // namespace boost::corosio::detail
926
927		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
928