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	1112121x	reactor_find_context(reactor_scheduler const* self) noexcept
80		{
81	1112121x	for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
82		{
83	1090719x	if (c->key == self)
84	1090719x	return c;
85		}
86	21402x	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	1974x	unsigned inline_budget_initial() const noexcept
217		{
218	1974x	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	1974x	explicit reactor_thread_context_guard(
347		reactor_scheduler const* sched) noexcept
348	1974x	: frame_(sched, reactor_context_stack.get())
349		{
350	1974x	reactor_context_stack.set(&frame_);
351	1974x	}
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	1974x	~reactor_thread_context_guard() noexcept
360		{
361	1974x	reactor_context_stack.set(frame_.next);
362	1974x	}
363		};
364
365		// ---- Inline implementations ------------------------------------------------
366
367		inline
368	1974x	reactor_scheduler_context::reactor_scheduler_context(
369		reactor_scheduler const* k,
370	1974x	reactor_scheduler_context* n)
371	1974x	: key(k)
372	1974x	, next(n)
373	1974x	, private_outstanding_work(0)
374	1974x	, inline_budget(0)
375	1974x	, inline_budget_max(
376	1974x	static_cast<int>(k->inline_budget_initial()))
377	1974x	, unassisted(false)
378		{
379	1974x	}
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	109141x	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	109141x	if (inline_budget_max_ == 0)
414	30x	return;
415	109111x	if (auto* ctx = reactor_find_context(this))
416		{
417		// Cap when no other thread absorbed queued work
418	109111x	if (ctx->unassisted)
419		{
420	109111x	ctx->inline_budget_max =
421	109111x	static_cast<int>(unassisted_budget_);
422	109111x	ctx->inline_budget =
423	109111x	static_cast<int>(unassisted_budget_);
424	109111x	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	473456x	reactor_scheduler::try_consume_inline_budget() const noexcept
441		{
442	473456x	if (inline_budget_max_ == 0)
443	26x	return false;
444	473430x	if (auto* ctx = reactor_find_context(this))
445		{
446	473430x	if (ctx->inline_budget > 0)
447		{
448	378567x	--ctx->inline_budget;
449	378567x	return true;
450		}
451		}
452	94863x	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	116559x	reactor_scheduler::post(scheduler_op* h) const
498		{
499	116559x	if (auto* ctx = reactor_find_context(this))
500		{
501	115848x	++ctx->private_outstanding_work;
502	115848x	ctx->private_queue.push(h);
503	115848x	return;
504		}
505
506	711x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
507
508	711x	lock_type lock(mutex_);
509	711x	completed_ops_.push(h);
510	711x	wake_one_thread_and_unlock(lock);
511	711x	}
512
513		inline void
514	25861x	reactor_scheduler::post(capy::continuation& c) const
515		{
516	25861x	if (auto* ctx = reactor_find_context(this))
517		{
518	17344x	++ctx->private_outstanding_work;
519	17344x	ctx->private_queue.push(c);
520	17344x	return;
521		}
522
523	8517x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
524
525	8517x	lock_type lock(mutex_);
526	8517x	completed_ops_.push(c);
527	8517x	wake_one_thread_and_unlock(lock);
528	8517x	}
529
530		inline bool
531	9455x	reactor_scheduler::running_in_this_thread() const noexcept
532		{
533	9455x	return reactor_find_context(this) != nullptr;
534		}
535
536		inline void
537	1855x	reactor_scheduler::stop()
538		{
539	1855x	lock_type lock(mutex_);
540	1855x	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	1855x	}
547
548		inline bool
549	136x	reactor_scheduler::stopped() const noexcept
550		{
551	136x	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	103x	stop();
566	103x	return 0;
567		}
568
569	1687x	reactor_thread_context_guard ctx(this);
570	1687x	lock_type lock(mutex_);
571
572	1687x	std::size_t n = 0;
573		for (;;)
574		{
575	535112x	if (!do_one(lock, -1, ctx.frame_))
576	1684x	break;
577	533425x	if (n != (std::numeric_limits<std::size_t>::max)())
578	533425x	++n;
579	533425x	if (!lock.owns_lock())
580	418834x	lock.lock();
581		}
582	1684x	return n;
583	1690x	}
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	162x	reactor_scheduler::wait_one(long usec)
601		{
602	324x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
603		{
604	24x	stop();
605	24x	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	40382x	reactor_scheduler::work_started() noexcept
654		{
655	40382x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
656	40382x	}
657
658		inline void
659	72101x	reactor_scheduler::work_finished() noexcept
660		{
661	144202x	if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
662	1690x	stop();
663	72101x	}
664
665		inline void
666	373947x	reactor_scheduler::compensating_work_started() const noexcept
667		{
668	373947x	auto* ctx = reactor_find_context(this);
669	373947x	if (ctx)
670	373947x	++ctx->private_outstanding_work;
671	373947x	}
672
673
674		inline void
675	13620x	reactor_scheduler::post_deferred_completions(ready_queue& ops) const
676		{
677	13620x	if (ops.empty())
678	13620x	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	4586x	while (auto e = completed_ops_.pop())
697		{
698	2480x	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	2472x	auto* op = ready_as_op(e);
708	2472x	if (op == &task_op_)
709	2103x	continue;
710	369x	lock.unlock();
711	369x	op->destroy();
712	369x	lock.lock();
713		}
714	2480x	}
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	12888x	reactor_scheduler::maybe_unlock_and_signal_one(
728		lock_type& lock) const
729		{
730	12888x	state_ \|= signaled_bit;
731	12888x	if (state_ > signaled_bit)
732		{
733	40x	lock.unlock();
734	40x	cond_.notify_one();
735	40x	return true;
736		}
737	12848x	return false;
738		}
739
740		inline bool
741	591293x	reactor_scheduler::unlock_and_signal_one(
742		lock_type& lock) const
743		{
744	591293x	state_ \|= signaled_bit;
745	591293x	bool have_waiters = state_ > signaled_bit;
746	591293x	lock.unlock();
747	591293x	if (have_waiters)
748	6x	cond_.notify_one();
749	591293x	return have_waiters;
750		}
751
752		inline void
753	54x	reactor_scheduler::clear_signal() const
754		{
755	54x	state_ &= ~signaled_bit;
756	54x	}
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	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	12888x	reactor_scheduler::wake_one_thread_and_unlock(
784		lock_type& lock) const
785		{
786	12888x	if (maybe_unlock_and_signal_one(lock))
787	40x	return;
788
789	12848x	if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
790		{
791	236x	task_interrupted_ = true;
792	236x	lock.unlock();
793	236x	interrupt_reactor();
794		}
795		else
796		{
797	12612x	lock.unlock();
798		}
799		}
800
801	533661x	inline reactor_scheduler::work_cleanup::~work_cleanup()
802		{
803	533661x	std::int64_t produced = ctx.private_outstanding_work;
804	533661x	if (produced > 1)
805	337x	sched->outstanding_work_.fetch_add(
806		produced - 1, std::memory_order_relaxed);
807	533324x	else if (produced < 1)
808	45409x	sched->work_finished();
809	533661x	ctx.private_outstanding_work = 0;
810
811	533661x	if (!ctx.private_queue.empty())
812		{
813	114615x	lock->lock();
814	114615x	sched->completed_ops_.splice(ctx.private_queue);
815		}
816	533661x	}
817
818	408469x	inline reactor_scheduler::task_cleanup::~task_cleanup()
819		{
820	408469x	if (ctx.private_outstanding_work > 0)
821		{
822	11270x	sched->outstanding_work_.fetch_add(
823	11270x	ctx.private_outstanding_work, std::memory_order_relaxed);
824	11270x	ctx.private_outstanding_work = 0;
825		}
826
827	408469x	if (!ctx.private_queue.empty())
828		{
829	11270x	if (!lock->owns_lock())
830	✗	lock->lock();
831	11270x	sched->completed_ops_.splice(ctx.private_queue);
832		}
833	408469x	}
834
835		inline std::size_t
836	535440x	reactor_scheduler::do_one(
837		lock_type& lock, long timeout_us, context_type& ctx)
838		{
839		for (;;)
840		{
841	943904x	if (stopped_.load(std::memory_order_acquire))
842	1685x	return 0;
843
844	942219x	std::uintptr_t e = completed_ops_.pop();
845	942219x	scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
846
847		// Handle reactor sentinel — time to poll for I/O
848	942219x	if (op == &task_op_)
849		{
850	408504x	bool more_handlers = !completed_ops_.empty();
851
852	759311x	if (!more_handlers &&
853	701614x	(outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
854		timeout_us == 0))
855		{
856	35x	completed_ops_.push(&task_op_);
857	35x	return 0;
858		}
859
860	408469x	long task_timeout_us = more_handlers ? 0 : timeout_us;
861	408469x	task_interrupted_ = task_timeout_us == 0;
862	408469x	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	408469x	if (more_handlers && !one_thread_)
867	57690x	unlock_and_signal_one(lock);
868
869		try
870		{
871	408469x	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	408466x	task_running_.store(false, std::memory_order_relaxed);
880	408466x	completed_ops_.push(&task_op_);
881	408466x	if (timeout_us > 0)
882	56x	return 0;
883	408410x	continue;
884	408410x	}
885
886		// Handle ready entry (op or continuation)
887	533715x	if (e != 0)
888		{
889	533661x	bool more = !completed_ops_.empty();
890
891	533661x	if (more && !one_thread_)
892		{
893		// Wake a peer for the remaining work; unassisted if none
894		// was parked to take it.
895	533603x	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	533661x	[[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
905
906	533661x	if (ready_is_continuation(e))
907	25853x	ready_as_cont(e)->h.resume();
908		else
909	507808x	(*op)();
910	533661x	return 1;
911	533661x	}
912
913	108x	if (outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
914		timeout_us == 0)
915	✗	return 0;
916
917	54x	clear_signal();
918	54x	if (timeout_us < 0)
919	6x	wait_for_signal(lock);
920		else
921	48x	wait_for_signal_for(lock, timeout_us);
922	408464x	}
923		}
924
925		} // namespace boost::corosio::detail
926
927		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
928