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	1001470x	reactor_find_context(reactor_scheduler const* self) noexcept
80		{
81	1001470x	for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
82		{
83	980125x	if (c->key == self)
84	980125x	return c;
85		}
86	21345x	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	1980x	unsigned inline_budget_initial() const noexcept
217		{
218	1980x	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	1980x	explicit reactor_thread_context_guard(
347		reactor_scheduler const* sched) noexcept
348	1980x	: frame_(sched, reactor_context_stack.get())
349		{
350	1980x	reactor_context_stack.set(&frame_);
351	1980x	}
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	1980x	~reactor_thread_context_guard() noexcept
360		{
361	1980x	reactor_context_stack.set(frame_.next);
362	1980x	}
363		};
364
365		// ---- Inline implementations ------------------------------------------------
366
367		inline
368	1980x	reactor_scheduler_context::reactor_scheduler_context(
369		reactor_scheduler const* k,
370	1980x	reactor_scheduler_context* n)
371	1980x	: key(k)
372	1980x	, next(n)
373	1980x	, private_outstanding_work(0)
374	1980x	, inline_budget(0)
375	1980x	, inline_budget_max(
376	1980x	static_cast<int>(k->inline_budget_initial()))
377	1980x	, unassisted(false)
378		{
379	1980x	}
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	92865x	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	92865x	if (inline_budget_max_ == 0)
414	30x	return;
415	92835x	if (auto* ctx = reactor_find_context(this))
416		{
417		// Cap when no other thread absorbed queued work
418	92835x	if (ctx->unassisted)
419		{
420	92835x	ctx->inline_budget_max =
421	92835x	static_cast<int>(unassisted_budget_);
422	92835x	ctx->inline_budget =
423	92835x	static_cast<int>(unassisted_budget_);
424	92835x	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	393158x	reactor_scheduler::try_consume_inline_budget() const noexcept
441		{
442	393158x	if (inline_budget_max_ == 0)
443	26x	return false;
444	393132x	if (auto* ctx = reactor_find_context(this))
445		{
446	393132x	if (ctx->inline_budget > 0)
447		{
448	314330x	--ctx->inline_budget;
449	314330x	return true;
450		}
451		}
452	78802x	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	100537x	reactor_scheduler::post(scheduler_op* h) const
498		{
499	100537x	if (auto* ctx = reactor_find_context(this))
500		{
501	99851x	++ctx->private_outstanding_work;
502	99851x	ctx->private_queue.push(h);
503	99851x	return;
504		}
505
506	686x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
507
508	686x	lock_type lock(mutex_);
509	686x	completed_ops_.push(h);
510	686x	wake_one_thread_and_unlock(lock);
511	686x	}
512
513		inline void
514	25723x	reactor_scheduler::post(capy::continuation& c) const
515		{
516	25723x	if (auto* ctx = reactor_find_context(this))
517		{
518	17222x	++ctx->private_outstanding_work;
519	17222x	ctx->private_queue.push(c);
520	17222x	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	9439x	reactor_scheduler::running_in_this_thread() const noexcept
532		{
533	9439x	return reactor_find_context(this) != nullptr;
534		}
535
536		inline void
537	1832x	reactor_scheduler::stop()
538		{
539	1832x	lock_type lock(mutex_);
540	1832x	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	1832x	}
547
548		inline bool
549	139x	reactor_scheduler::stopped() const noexcept
550		{
551	139x	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	89x	stop();
566	89x	return 0;
567		}
568
569	1691x	reactor_thread_context_guard ctx(this);
570	1691x	lock_type lock(mutex_);
571
572	1691x	std::size_t n = 0;
573		for (;;)
574		{
575	520844x	if (!do_one(lock, -1, ctx.frame_))
576	1688x	break;
577	519153x	if (n != (std::numeric_limits<std::size_t>::max)())
578	519153x	++n;
579	519153x	if (!lock.owns_lock())
580	420745x	lock.lock();
581		}
582	1688x	return n;
583	1694x	}
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	165x	reactor_scheduler::wait_one(long usec)
601		{
602	330x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
603		{
604	25x	stop();
605	25x	return 0;
606		}
607
608	140x	reactor_thread_context_guard ctx(this);
609	140x	lock_type lock(mutex_);
610	140x	return do_one(lock, usec, ctx.frame_);
611	140x	}
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	40023x	reactor_scheduler::work_started() noexcept
654		{
655	40023x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
656	40023x	}
657
658		inline void
659	71765x	reactor_scheduler::work_finished() noexcept
660		{
661	143530x	if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
662	1680x	stop();
663	71765x	}
664
665		inline void
666	376046x	reactor_scheduler::compensating_work_started() const noexcept
667		{
668	376046x	auto* ctx = reactor_find_context(this);
669	376046x	if (ctx)
670	376046x	++ctx->private_outstanding_work;
671	376046x	}
672
673
674		inline void
675	13409x	reactor_scheduler::post_deferred_completions(ready_queue& ops) const
676		{
677	13409x	if (ops.empty())
678	13409x	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	4563x	while (auto e = completed_ops_.pop())
697		{
698	2468x	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	2460x	auto* op = ready_as_op(e);
708	2460x	if (op == &task_op_)
709	2092x	continue;
710	368x	lock.unlock();
711	368x	op->destroy();
712	368x	lock.lock();
713		}
714	2468x	}
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	12847x	reactor_scheduler::maybe_unlock_and_signal_one(
728		lock_type& lock) const
729		{
730	12847x	state_ \|= signaled_bit;
731	12847x	if (state_ > signaled_bit)
732		{
733	37x	lock.unlock();
734	37x	cond_.notify_one();
735	37x	return true;
736		}
737	12810x	return false;
738		}
739
740		inline bool
741	569087x	reactor_scheduler::unlock_and_signal_one(
742		lock_type& lock) const
743		{
744	569087x	state_ \|= signaled_bit;
745	569087x	bool have_waiters = state_ > signaled_bit;
746	569087x	lock.unlock();
747	569087x	if (have_waiters)
748	6x	cond_.notify_one();
749	569087x	return have_waiters;
750		}
751
752		inline void
753	40x	reactor_scheduler::clear_signal() const
754		{
755	40x	state_ &= ~signaled_bit;
756	40x	}
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	34x	reactor_scheduler::wait_for_signal_for(
772		lock_type& lock, long timeout_us) const
773		{
774	34x	if ((state_ & signaled_bit) == 0)
775		{
776	34x	state_ += waiter_increment;
777	34x	cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
778	34x	state_ -= waiter_increment;
779		}
780	34x	}
781
782		inline void
783	12847x	reactor_scheduler::wake_one_thread_and_unlock(
784		lock_type& lock) const
785		{
786	12847x	if (maybe_unlock_and_signal_one(lock))
787	37x	return;
788
789	12810x	if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
790		{
791	222x	task_interrupted_ = true;
792	222x	lock.unlock();
793	222x	interrupt_reactor();
794		}
795		else
796		{
797	12588x	lock.unlock();
798		}
799		}
800
801	519389x	inline reactor_scheduler::work_cleanup::~work_cleanup()
802		{
803	519389x	std::int64_t produced = ctx.private_outstanding_work;
804	519389x	if (produced > 1)
805	337x	sched->outstanding_work_.fetch_add(
806		produced - 1, std::memory_order_relaxed);
807	519052x	else if (produced < 1)
808	45221x	sched->work_finished();
809	519389x	ctx.private_outstanding_work = 0;
810
811	519389x	if (!ctx.private_queue.empty())
812		{
813	98432x	lock->lock();
814	98432x	sched->completed_ops_.splice(ctx.private_queue);
815		}
816	519389x	}
817
818	408857x	inline reactor_scheduler::task_cleanup::~task_cleanup()
819		{
820	408857x	if (ctx.private_outstanding_work > 0)
821		{
822	11363x	sched->outstanding_work_.fetch_add(
823	11363x	ctx.private_outstanding_work, std::memory_order_relaxed);
824	11363x	ctx.private_outstanding_work = 0;
825		}
826
827	408857x	if (!ctx.private_queue.empty())
828		{
829	11363x	if (!lock->owns_lock())
830	✗	lock->lock();
831	11363x	sched->completed_ops_.splice(ctx.private_queue);
832		}
833	408857x	}
834
835		inline std::size_t
836	521174x	reactor_scheduler::do_one(
837		lock_type& lock, long timeout_us, context_type& ctx)
838		{
839		for (;;)
840		{
841	930010x	if (stopped_.load(std::memory_order_acquire))
842	1688x	return 0;
843
844	928322x	std::uintptr_t e = completed_ops_.pop();
845	928322x	scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
846
847		// Handle reactor sentinel — time to poll for I/O
848	928322x	if (op == &task_op_)
849		{
850	408890x	bool more_handlers = !completed_ops_.empty();
851
852	768029x	if (!more_handlers &&
853	718278x	(outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
854		timeout_us == 0))
855		{
856	33x	completed_ops_.push(&task_op_);
857	33x	return 0;
858		}
859
860	408857x	long task_timeout_us = more_handlers ? 0 : timeout_us;
861	408857x	task_interrupted_ = task_timeout_us == 0;
862	408857x	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	408857x	if (more_handlers && !one_thread_)
867	49744x	unlock_and_signal_one(lock);
868
869		try
870		{
871	408857x	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	408854x	task_running_.store(false, std::memory_order_relaxed);
880	408854x	completed_ops_.push(&task_op_);
881	408854x	if (timeout_us > 0)
882	58x	return 0;
883	408796x	continue;
884	408796x	}
885
886		// Handle ready entry (op or continuation)
887	519432x	if (e != 0)
888		{
889	519389x	bool more = !completed_ops_.empty();
890
891	519389x	if (more && !one_thread_)
892		{
893		// Wake a peer for the remaining work; unassisted if none
894		// was parked to take it.
895	519343x	ctx.unassisted = !unlock_and_signal_one(lock);
896		}
897		else
898		{
899		// No peer to wake (one_thread_, or nothing more queued).
900	46x	ctx.unassisted = more;
901	46x	lock.unlock();
902		}
903
904	519389x	[[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
905
906	519389x	if (ready_is_continuation(e))
907	25715x	ready_as_cont(e)->h.resume();
908		else
909	493674x	(*op)();
910	519389x	return 1;
911	519389x	}
912
913	86x	if (outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
914		timeout_us == 0)
915	3x	return 0;
916
917	40x	clear_signal();
918	40x	if (timeout_us < 0)
919	6x	wait_for_signal(lock);
920		else
921	34x	wait_for_signal_for(lock, timeout_us);
922	408836x	}
923		}
924
925		} // namespace boost::corosio::detail
926
927		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
928