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

96.0% Lines (316 / 329, 2 excl) 100.0% Functions (42 / 42, 2 excl)

[ ] 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	993472x	reactor_find_context(reactor_scheduler const* self) noexcept
79		{
80	993472x	for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
81		{
82	968566x	if (c->key == self)
83	968566x	return c;
84		}
85	24906x	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	5451x	unsigned inline_budget_initial() const noexcept
214		{
215	5451x	return inline_budget_initial_;
216		}
217
218		/// Return true when scheduler locking is disabled (fully-lockless tier).
219	502x	bool scheduler_locking_disabled() const noexcept override
220		{
221	502x	return scheduler_locking_disabled_;
222		}
223
224	2241x	void configure_threading(threading_config cfg) noexcept override
225		{
226	2241x	scheduler_locking_disabled_ = !cfg.scheduler_locking;
227		// reactor_io_locking takes effect at descriptor registration (see the
228		// register_descriptor overrides), not here.
229	2241x	reactor_io_locking_ = cfg.reactor_io_locking;
230	2241x	one_thread_ = cfg.one_thread;
231	2241x	mutex_.set_enabled(cfg.scheduler_locking);
232	2241x	cond_.set_enabled(cfg.scheduler_locking);
233	2241x	}
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	2253x	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	5451x	explicit reactor_thread_context_guard(
341		reactor_scheduler const* sched) noexcept
342	5451x	: frame_(sched, reactor_context_stack.get())
343		{
344	5451x	reactor_context_stack.set(&frame_);
345	5451x	}
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	5451x	~reactor_thread_context_guard() noexcept
354		{
355	5451x	reactor_context_stack.set(frame_.next);
356	5451x	}
357		};
358
359		// ---- Inline implementations ------------------------------------------------
360
361	5451x	inline reactor_scheduler_context::reactor_scheduler_context(
362	5451x	reactor_scheduler const* k, reactor_scheduler_context* n)
363	5451x	: key(k)
364	5451x	, next(n)
365	5451x	, private_outstanding_work(0)
366	5451x	, inline_budget(0)
367	5451x	, inline_budget_max(static_cast<int>(k->inline_budget_initial()))
368	5451x	, unassisted(false)
369		{
370	5451x	}
371
372		inline void
373	44x	reactor_scheduler::configure_reactor(
374		unsigned max_events,
375		unsigned budget_init,
376		unsigned budget_max,
377		unsigned unassisted)
378		{
379	86x	if (max_events < 1 \|\|
380	42x	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	42x	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	40x	if (budget_init > budget_max)
387	16x	budget_init = budget_max;
388	40x	if (unassisted > budget_max)
389	16x	unassisted = budget_max;
390
391	40x	max_events_per_poll_ = max_events;
392	40x	inline_budget_initial_ = budget_init;
393	40x	inline_budget_max_ = budget_max;
394	40x	unassisted_budget_ = unassisted;
395	40x	}
396
397		inline void
398	98175x	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	98175x	if (inline_budget_max_ == 0)
403	56x	return;
404	98119x	if (auto* ctx = reactor_find_context(this))
405		{
406		// Cap when no other thread absorbed queued work
407	98119x	if (ctx->unassisted)
408		{
409	98119x	ctx->inline_budget_max = static_cast<int>(unassisted_budget_);
410	98119x	ctx->inline_budget = static_cast<int>(unassisted_budget_);
411	98119x	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	440182x	reactor_scheduler::try_consume_inline_budget() const noexcept
427		{
428	440182x	if (inline_budget_max_ == 0)
429	40x	return false;
430	440142x	if (auto* ctx = reactor_find_context(this))
431		{
432	440142x	if (ctx->inline_budget > 0)
433		{
434	351948x	--ctx->inline_budget;
435	351948x	return true;
436		}
437		}
438	88194x	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	107973x	reactor_scheduler::post(scheduler_op* h) const
484		{
485	107973x	if (auto* ctx = reactor_find_context(this))
486		{
487	106935x	++ctx->private_outstanding_work;
488	106935x	ctx->private_queue.push(h);
489	106935x	return;
490		}
491
492	1038x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
493
494	1038x	lock_type lock(mutex_);
495	1038x	completed_ops_.push(h);
496	1038x	wake_one_thread_and_unlock(lock);
497	1038x	}
498
499		inline void
500	25908x	reactor_scheduler::post(capy::continuation& c) const
501		{
502	25908x	if (auto* ctx = reactor_find_context(this))
503		{
504	15548x	++ctx->private_outstanding_work;
505	15548x	ctx->private_queue.push(c);
506	15548x	return;
507		}
508
509	10360x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
510
511	10360x	lock_type lock(mutex_);
512	10360x	completed_ops_.push(c);
513	10360x	wake_one_thread_and_unlock(lock);
514	10360x	}
515
516		inline bool
517	10789x	reactor_scheduler::running_in_this_thread() const noexcept
518		{
519	10789x	return reactor_find_context(this) != nullptr;
520		}
521
522		inline void
523	3722x	reactor_scheduler::stop()
524		{
525	3722x	lock_type lock(mutex_);
526	3722x	if (!stopped_.load(std::memory_order_acquire))
527		{
528	2859x	stopped_.store(true, std::memory_order_release);
529	2859x	signal_all(lock);
530	2859x	interrupt_reactor();
531		}
532	3722x	}
533
534		inline bool
535	2503x	reactor_scheduler::stopped() const noexcept
536		{
537	2503x	return stopped_.load(std::memory_order_acquire);
538		}
539
540		inline void
541	1439x	reactor_scheduler::restart()
542		{
543	1439x	stopped_.store(false, std::memory_order_release);
544	1439x	}
545
546		inline std::size_t
547	2096x	reactor_scheduler::run()
548		{
549	4192x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
550		{
551	103x	stop();
552	103x	return 0;
553		}
554
555	1993x	reactor_thread_context_guard ctx(this);
556	1993x	lock_type lock(mutex_);
557
558	1993x	std::size_t n = 0;
559		for (;;)
560		{
561	453901x	if (!do_one(lock, -1, ctx.frame_))
562	1990x	break;
563	451908x	if (n != (std::numeric_limits<std::size_t>::max)())
564	451908x	++n;
565	451908x	if (!lock.owns_lock())
566	346110x	lock.lock();
567		}
568	1990x	return n;
569	1996x	}
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	4094x	reactor_scheduler::wait_one(long usec)
587		{
588	8188x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
589		{
590	785x	stop();
591	785x	return 0;
592		}
593
594	3309x	reactor_thread_context_guard ctx(this);
595	3309x	lock_type lock(mutex_);
596	3309x	return do_one(lock, usec, ctx.frame_);
597	3309x	}
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	36090x	reactor_scheduler::work_started() noexcept
640		{
641	36090x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
642	36090x	}
643
644		inline void
645	67816x	reactor_scheduler::work_finished() noexcept
646		{
647	135632x	if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
648	2796x	stop();
649	67816x	}
650
651		inline void
652	306783x	reactor_scheduler::compensating_work_started() const noexcept
653		{
654	306783x	auto* ctx = reactor_find_context(this);
655	306783x	if (ctx)
656	306783x	++ctx->private_outstanding_work;
657	306783x	}
658
659		inline void
660	9371x	reactor_scheduler::post_deferred_completions(ready_queue& ops) const
661		{
662	9371x	if (ops.empty())
663	9371x	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	2241x	reactor_scheduler::shutdown_drain()
678		{
679	2241x	lock_type lock(mutex_);
680
681	4862x	while (auto e = completed_ops_.pop())
682		{
683	2621x	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	2613x	auto* op = ready_as_op(e);
693	2613x	if (op == &task_op_)
694	2238x	continue;
695	375x	lock.unlock();
696	375x	op->destroy();
697	375x	lock.lock();
698		}
699	2621x	}
700
701	2241x	signal_all(lock);
702	2241x	}
703
704		inline void
705	5100x	reactor_scheduler::signal_all(lock_type&) const
706		{
707	5100x	state_ \|= signaled_bit;
708	5100x	cond_.notify_all();
709	5100x	}
710
711		inline bool
712	15058x	reactor_scheduler::maybe_unlock_and_signal_one(lock_type& lock) const
713		{
714	15058x	state_ \|= signaled_bit;
715	15058x	if (state_ > signaled_bit)
716		{
717	33x	lock.unlock();
718	33x	cond_.notify_one();
719	33x	return true;
720		}
721	15025x	return false;
722		}
723
724		inline bool
725	507698x	reactor_scheduler::unlock_and_signal_one(lock_type& lock) const
726		{
727	507698x	state_ \|= signaled_bit;
728	507698x	bool have_waiters = state_ > signaled_bit;
729	507698x	lock.unlock();
730	507698x	if (have_waiters)
731	6x	cond_.notify_one();
732	507698x	return have_waiters;
733		}
734
735		inline void
736	19x	reactor_scheduler::clear_signal() const
737		{
738	19x	state_ &= ~signaled_bit;
739	19x	}
740
741		inline void
742	6x	reactor_scheduler::wait_for_signal(lock_type& lock) const
743		{
744	14x	while ((state_ & signaled_bit) == 0)
745		{
746	8x	state_ += waiter_increment;
747	8x	cond_.wait(lock);
748	8x	state_ -= waiter_increment;
749		}
750	6x	}
751
752		inline void
753	13x	reactor_scheduler::wait_for_signal_for(lock_type& lock, long timeout_us) const
754		{
755	13x	if ((state_ & signaled_bit) == 0)
756		{
757	13x	state_ += waiter_increment;
758	13x	cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
759	13x	state_ -= waiter_increment;
760		}
761	13x	}
762
763		inline void
764	15058x	reactor_scheduler::wake_one_thread_and_unlock(lock_type& lock) const
765		{
766	15058x	if (maybe_unlock_and_signal_one(lock))
767	33x	return;
768
769	15025x	if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
770		{
771	841x	task_interrupted_ = true;
772	841x	lock.unlock();
773	841x	interrupt_reactor();
774		}
775		else
776		{
777	14184x	lock.unlock();
778		}
779		}
780
781	453709x	inline reactor_scheduler::work_cleanup::~work_cleanup()
782		{
783	453709x	std::int64_t produced = ctx.private_outstanding_work;
784	453709x	if (produced > 1)
785	345x	sched->outstanding_work_.fetch_add(
786		produced - 1, std::memory_order_relaxed);
787	453364x	else if (produced < 1)
788	41167x	sched->work_finished();
789	453709x	ctx.private_outstanding_work = 0;
790
791	453709x	if (!ctx.private_queue.empty())
792		{
793	106078x	lock->lock();
794	106078x	sched->completed_ops_.splice(ctx.private_queue);
795		}
796	453709x	}
797
798	334383x	inline reactor_scheduler::task_cleanup::~task_cleanup()
799		{
800	334383x	if (ctx.private_outstanding_work > 0)
801		{
802	9194x	sched->outstanding_work_.fetch_add(
803	9194x	ctx.private_outstanding_work, std::memory_order_relaxed);
804	9194x	ctx.private_outstanding_work = 0;
805		}
806
807	334383x	if (!ctx.private_queue.empty())
808		{
809	9194x	if (!lock->owns_lock())
810	✗	lock->lock();
811	9194x	sched->completed_ops_.splice(ctx.private_queue);
812		}
813	334383x	}
814
815		inline std::size_t
816	457400x	reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx)
817		{
818		for (;;)
819		{
820	790137x	if (stopped_.load(std::memory_order_acquire))
821	1994x	return 0;
822
823	788143x	std::uintptr_t e = completed_ops_.pop();
824	788143x	scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
825
826		// Handle reactor sentinel — time to poll for I/O
827	788143x	if (op == &task_op_)
828		{
829	334415x	bool more_handlers = !completed_ops_.empty();
830
831	614809x	if (!more_handlers &&
832	560788x	(outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
833		timeout_us == 0))
834		{
835	32x	completed_ops_.push(&task_op_);
836	32x	return 0;
837		}
838
839	334383x	long task_timeout_us = more_handlers ? 0 : timeout_us;
840	334383x	task_interrupted_ = task_timeout_us == 0;
841	334383x	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	334383x	if (more_handlers && !one_thread_)
846	54014x	unlock_and_signal_one(lock);
847
848		try
849		{
850	334383x	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	334380x	task_running_.store(false, std::memory_order_relaxed);
859	334380x	completed_ops_.push(&task_op_);
860	334380x	if (timeout_us > 0)
861	1662x	return 0;
862	332718x	continue;
863	332718x	}
864
865		// Handle ready entry (op or continuation)
866	453728x	if (e != 0)
867		{
868	453709x	bool more = !completed_ops_.empty();
869
870	453709x	if (more && !one_thread_)
871		{
872		// Wake a peer for the remaining work; unassisted if none
873		// was parked to take it.
874	453684x	ctx.unassisted = !unlock_and_signal_one(lock);
875		}
876		else
877		{
878		// No peer to wake (one_thread_, or nothing more queued).
879	25x	ctx.unassisted = more;
880	25x	lock.unlock();
881		}
882
883	453709x	[[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
884
885	453709x	if (ready_is_continuation(e))
886	25900x	ready_as_cont(e)->h.resume();
887		else
888	427809x	(*op)();
889	453709x	return 1;
890	453709x	}
891
892	38x	if (outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
893		timeout_us == 0)
894	✗	return 0;
895
896	19x	clear_signal();
897	19x	if (timeout_us < 0)
898	6x	wait_for_signal(lock);
899		else
900	13x	wait_for_signal_for(lock, timeout_us);
901	332737x	}
902		}
903
904		} // namespace boost::corosio::detail
905
906		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
907