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

98.7% Lines (383 / 388, 2 excl) 100.0% Functions (51 / 51, 2 excl)

[ ] prev/next file n p prev/next uncovered line

reactor_scheduler.hpp

f(x) Functions (53)

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/corosio/detail/object_ref.hpp>
15		#include <boost/capy/ex/execution_context.hpp>
16		#include <boost/capy/ex/abandon.hpp>
17
18		#include <boost/corosio/detail/ready_queue.hpp>
19		#include <boost/corosio/detail/scheduler.hpp>
20		#include <boost/corosio/detail/scheduler_op.hpp>
21		#include <boost/corosio/detail/thread_local_ptr.hpp>
22		#include <boost/corosio/detail/timer_service.hpp>
23
24		#include <atomic>
25		#include <chrono>
26		#include <coroutine>
27		#include <cstddef>
28		#include <cstdint>
29		#include <limits>
30		#include <memory>
31
32		#include <boost/corosio/detail/conditionally_enabled_mutex.hpp>
33		#include <boost/corosio/detail/conditionally_enabled_event.hpp>
34
35		namespace boost::corosio::detail {
36
37		// Forward declarations
38		class reactor_scheduler;
39
40		/** A deregistered descriptor whose owner must outlive the reactor pass
41		that may still hold it.
42
43		An event the reactor has already read from the kernel can name a
44		descriptor that another thread closes before the reactor queues it.
45		The closing thread hands over a reference instead of dropping it, so
46		the owner is neither freed nor recycled until that pass ends.
47		*/
48		struct reactor_retirable
49		{
50		// All four fields are guarded by the scheduler's mutex.
51
52		/// Next in the scheduler's retired list.
53		reactor_retirable* retired_next_ = nullptr;
54
55		/// Owner reference held until the pass ends.
56		object_ref retired_ref_;
57
58		/// Next in a settling thread's list of references to drop.
59		reactor_retirable* dropped_next_ = nullptr;
60
61		/// Owner reference waiting to be dropped outside the mutex.
62		object_ref dropped_ref_;
63
64		/// Move @p keep into a queued completion, if one is pending.
65		virtual void hand_off(object_ref& keep) noexcept = 0;
66
67		protected:
68	12288x	~reactor_retirable() = default;
69		};
70
71		/** Per-thread state for a reactor scheduler.
72
73		Each thread running a scheduler's event loop has one of these
74		on a thread-local stack. It holds a private work queue and
75		inline completion budget for speculative I/O fast paths.
76		*/
77		struct BOOST_COROSIO_SYMBOL_VISIBLE reactor_scheduler_context
78		{
79		/// Scheduler this context belongs to.
80		reactor_scheduler const* key;
81
82		/// Next context frame on this thread's stack.
83		reactor_scheduler_context* next;
84
85		/// Private work queue for reduced contention.
86		ready_queue private_queue;
87
88		/// Unflushed work count for the private queue.
89		std::int64_t private_outstanding_work;
90
91		/// Remaining inline completions allowed this cycle.
92		int inline_budget;
93
94		/// Maximum inline budget (adaptive, 2-16).
95		int inline_budget_max;
96
97		/// True if no other thread absorbed queued work last cycle.
98		bool unassisted;
99
100		/// Construct a context frame linked to @a n.
101		reactor_scheduler_context(
102		reactor_scheduler const* k, reactor_scheduler_context* n);
103		};
104
105		/// Thread-local context stack for reactor schedulers.
106		inline thread_local_ptr<reactor_scheduler_context> reactor_context_stack;
107
108		/// Find the context frame for a scheduler on this thread.
109		inline reactor_scheduler_context*
110	831818x	reactor_find_context(reactor_scheduler const* self) noexcept
111		{
112	832406x	for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
113		{
114	769748x	if (c->key == self)
115	769160x	return c;
116		}
117	62658x	return nullptr;
118		}
119
120		/** Non-template base for reactor-backed scheduler implementations.
121
122		Provides the complete threading model shared by epoll, kqueue,
123		and select schedulers: signal state machine, inline completion
124		budget, work counting, run/poll methods, and the do_one event
125		loop.
126
127		Derived classes provide platform-specific hooks by overriding:
128		- `run_task(lock, ctx)` to run the reactor poll
129		- `interrupt_reactor()` to wake a blocked reactor
130
131		De-templated from the original CRTP design to eliminate
132		duplicate instantiations when multiple backends are compiled
133		into the same binary. Virtual dispatch for run_task (called
134		once per reactor cycle, before a blocking syscall) has
135		negligible overhead.
136
137		@par Thread Safety
138		All public member functions are thread-safe.
139		*/
140		class reactor_scheduler : public scheduler
141		{
142		public:
143		using context_type = reactor_scheduler_context;
144		using mutex_type = conditionally_enabled_mutex;
145		using lock_type = mutex_type::scoped_lock;
146		using event_type = conditionally_enabled_event;
147
148		/// Post a coroutine for deferred execution.
149		void post(std::coroutine_handle<> h) const override;
150
151		/// Post a scheduler operation for deferred execution.
152		void post(scheduler_op* h) const override;
153
154		/** Release an owner reference once no reactor pass can hold `d`.
155
156		Called after a descriptor is deregistered. If a reactor pass is
157		running, `keep` is parked on `d` until that pass ends. Then, or
158		at once if no pass is running, it moves to `d`'s queued
159		invocation if `d` is queued, and is dropped otherwise.
160
161		@param d The deregistered descriptor.
162		@param keep A reference to the descriptor's owner.
163		*/
164		void retire_descriptor(reactor_retirable& d, object_ref keep) const;
165
166		/// Post a continuation for deferred execution.
167		void post(capy::continuation&) const override;
168
169		/// Return true if called from a thread running this scheduler.
170		bool running_in_this_thread() const noexcept override;
171
172		/// Request the scheduler to stop dispatching handlers.
173		void stop() override;
174
175		/// Return true if the scheduler has been stopped.
176		bool stopped() const noexcept override;
177
178		/// Reset the stopped state so `run()` can resume.
179		void restart() override;
180
181		/// Run the event loop until no work remains.
182		std::size_t run() override;
183
184		/// Run until one handler completes or no work remains.
185		std::size_t run_one() override;
186
187		/// Run until one handler completes or @a usec elapses.
188		std::size_t wait_one(long usec) override;
189
190		/// Run ready handlers without blocking.
191		std::size_t poll() override;
192
193		/// Run at most one ready handler without blocking.
194		std::size_t poll_one() override;
195
196		/// Increment the outstanding work count.
197		void work_started() noexcept override;
198
199		/// Decrement the outstanding work count, stopping on zero.
200		void work_finished() noexcept override;
201
202		/** Reset the thread's inline completion budget.
203
204		Called at the start of each posted completion handler to
205		grant a fresh budget for speculative inline completions.
206		*/
207		void reset_inline_budget() const noexcept;
208
209		/** Consume one unit of inline budget if available.
210
211		@return True if budget was available and consumed.
212		*/
213		bool try_consume_inline_budget() const noexcept;
214
215		/** Offset a forthcoming work_finished from work_cleanup.
216
217		Called by descriptor_state when all I/O returned EAGAIN and
218		no handler will be executed. Must be called from a scheduler
219		thread.
220		*/
221		/** Note that an op was parked outside the usual registration path.
222
223		A reactor that builds its poll set from parked ops must rebuild
224		it; one the kernel tracks needs nothing.
225		*/
226	465x	virtual void notify_reparked() const {}
227
228		void compensating_work_started() const noexcept;
229
230		/** Post an operation whose work is already counted.
231
232		The posting thread touches nothing of the scheduler after this
233		returns, unlike a @ref post followed by @ref work_finished: once
234		posted, the operation can complete and let its context be
235		destroyed.
236
237		@pre work_started() was called for @p op.
238
239		@param op The operation to post.
240		*/
241		void post_counted(scheduler_op* op) const;
242
243		/** Post completed operations for deferred invocation.
244
245		If called from a thread running this scheduler, operations
246		go to the thread's private queue (fast path). Otherwise,
247		operations are added to the global queue under mutex and a
248		waiter is signaled.
249
250		@pre work_started() must have been called for each operation.
251
252		@param ops Queue of operations to post.
253		*/
254		void post_deferred_completions(ready_queue& ops) const;
255
256		/** Apply runtime configuration to the scheduler.
257
258		Called by `io_context` after construction. Values that do
259		not apply to this backend are silently ignored.
260
261		@pre `max_events` is in [ 1, `INT_MAX` ] and `budget_max` is at
262		most `INT_MAX`; `io_context` validates both.
263
264		@param max_events Event buffer size for epoll/kqueue.
265		@param budget_init Starting inline completion budget.
266		@param budget_max Hard ceiling on adaptive budget ramp-up.
267		@param unassisted Budget when single-threaded.
268		*/
269		virtual void configure_reactor(
270		unsigned max_events,
271		unsigned budget_init,
272		unsigned budget_max,
273		unsigned unassisted);
274
275		/// Return the configured initial inline budget.
276	13089x	unsigned inline_budget_initial() const noexcept
277		{
278	13089x	return inline_budget_initial_;
279		}
280
281		/// Return true when scheduler locking is disabled (fully-lockless tier).
282	1460x	bool scheduler_locking_disabled() const noexcept override
283		{
284	1460x	return scheduler_locking_disabled_;
285		}
286
287	4510x	void configure_threading(threading_config cfg) noexcept override
288		{
289	4510x	scheduler_locking_disabled_ = !cfg.scheduler_locking;
290		// reactor_io_locking takes effect at descriptor registration (see the
291		// register_descriptor overrides), not here.
292	4510x	reactor_io_locking_ = cfg.reactor_io_locking;
293	4510x	one_thread_ = cfg.one_thread;
294	4510x	mutex_.set_enabled(cfg.scheduler_locking);
295	4510x	cond_.set_enabled(cfg.scheduler_locking);
296	4510x	}
297
298		protected:
299		bool scheduler_locking_disabled_ = false;
300		bool reactor_io_locking_ = true;
301		bool one_thread_ = false;
302
303	4526x	reactor_scheduler() = default;
304
305		/// Destroy, destroying whatever was posted after the drain.
306		~reactor_scheduler() override;
307
308		/** Drain completed_ops during shutdown.
309
310		Pops all operations from the global queue and destroys them,
311		skipping the task sentinel. Signals all waiting threads.
312		Derived classes call this from their shutdown() override
313		before performing platform-specific cleanup.
314		*/
315		void shutdown_drain();
316
317		/// Destroy every queued operation without running it.
318		void destroy_queued(lock_type& lock) noexcept;
319
320		/// RAII guard that re-inserts the task sentinel after `run_task`.
321		struct task_cleanup
322		{
323		reactor_scheduler const* sched;
324		lock_type* lock;
325		context_type& ctx;
326		~task_cleanup();
327		};
328
329		mutable mutex_type mutex_{true};
330		mutable event_type cond_{true};
331		mutable ready_queue completed_ops_;
332		mutable std::atomic<std::int64_t> outstanding_work_{0};
333		std::atomic<bool> stopped_{false};
334		mutable std::atomic<bool> task_running_{false};
335		mutable bool task_interrupted_ = false;
336
337		// Descriptors retired during the running reactor pass (guarded by
338		// mutex_).
339		mutable reactor_retirable* retired_ = nullptr;
340
341		/// Settle every descriptor retired before the current pass ended.
342		/// @pre `lock` holds `mutex_`; it is held again on return.
343		void settle_retired(lock_type& lock) const;
344
345		// Runtime-configurable reactor tuning parameters.
346		// Defaults match the library's built-in values.
347		unsigned max_events_per_poll_ = 128;
348		unsigned inline_budget_initial_ = 2;
349		unsigned inline_budget_max_ = 16;
350		unsigned unassisted_budget_ = 4;
351
352		/// Bit 0 of `state_`: set when the condvar should be signaled.
353		static constexpr std::size_t signaled_bit = 1;
354
355		/// Increment per waiting thread in `state_`.
356		static constexpr std::size_t waiter_increment = 2;
357		mutable std::size_t state_ = 0;
358
359		/// Sentinel op that triggers a reactor poll when dequeued.
360		struct task_op final : scheduler_op
361		{
362		// LCOV_EXCL_START: the sentinel is intercepted by pointer
363		// identity; its virtuals exist for vtable completeness.
364	−	void operator()() override {}
365	−	void destroy() override {}
366		// LCOV_EXCL_STOP
367		};
368		task_op task_op_;
369
370		/** Run the platform-specific reactor poll.
371
372		@par Postconditions
373		`lock` is owned on return, however the poll ended. An
374		implementation that unlocks around the blocking call owes the
375		caller a matching re-acquire on every path out, including the
376		errors it retries rather than reports.
377		*/
378		virtual void
379		run_task(lock_type& lock, context_type& ctx, long timeout_us) = 0;
380
381		/// Wake a blocked reactor (e.g. write to eventfd or pipe).
382		virtual void interrupt_reactor() const = 0;
383
384		/// Wake the reactor whenever the earliest timer changes.
385		void on_timer_service_attached(timer_service& svc) override;
386
387		private:
388		struct work_cleanup
389		{
390		reactor_scheduler* sched;
391		lock_type* lock;
392		context_type& ctx;
393		~work_cleanup();
394		};
395
396		std::size_t do_one(lock_type& lock, long timeout_us, context_type& ctx);
397
398		void signal_all(lock_type& lock) const;
399		bool maybe_unlock_and_signal_one(lock_type& lock) const;
400		bool unlock_and_signal_one(lock_type& lock) const;
401		void clear_signal() const;
402		void wait_for_signal(lock_type& lock) const;
403		void wait_for_signal_for(lock_type& lock, long timeout_us) const;
404		void wake_one_thread_and_unlock(lock_type& lock) const;
405		};
406
407		/** RAII guard that pushes/pops a scheduler context frame.
408
409		On construction, pushes a new context frame onto the
410		thread-local stack. On destruction, drains any remaining
411		private queue items to the global queue and pops the frame.
412		*/
413		struct reactor_thread_context_guard
414		{
415		/// The context frame managed by this guard.
416		reactor_scheduler_context frame_;
417
418		/// Marks the scheduler innermost for the executor's fast check.
419		running_scheduler_guard running_;
420
421		/// Construct the guard, pushing a frame for @a sched.
422	13089x	explicit reactor_thread_context_guard(
423		reactor_scheduler const* sched) noexcept
424	13089x	: frame_(sched, reactor_context_stack.get())
425	13089x	, running_(sched)
426		{
427	13089x	reactor_context_stack.set(&frame_);
428	13089x	}
429
430		/** Destroy the guard, popping the frame.
431
432		The private queue is empty here by invariant: work_cleanup and
433		task_cleanup splice it to the global queue after every handler
434		and every reactor pass.
435		*/
436	13089x	~reactor_thread_context_guard() noexcept
437		{
438	13089x	reactor_context_stack.set(frame_.next);
439	13089x	}
440		};
441
442		// ---- Inline implementations ------------------------------------------------
443
444	13089x	inline reactor_scheduler_context::reactor_scheduler_context(
445	13089x	reactor_scheduler const* k, reactor_scheduler_context* n)
446	13089x	: key(k)
447	13089x	, next(n)
448	13089x	, private_outstanding_work(0)
449	13089x	, inline_budget(0)
450	13089x	, inline_budget_max(static_cast<int>(k->inline_budget_initial()))
451	13089x	, unassisted(false)
452		{
453	13089x	}
454
455		inline void
456	50x	reactor_scheduler::configure_reactor(
457		unsigned max_events,
458		unsigned budget_init,
459		unsigned budget_max,
460		unsigned unassisted)
461		{
462	50x	BOOST_COROSIO_ASSERT(
463		max_events >= 1 &&
464		max_events <= static_cast<unsigned>((std::numeric_limits<int>::max)()));
465	50x	BOOST_COROSIO_ASSERT(
466		budget_max <= static_cast<unsigned>((std::numeric_limits<int>::max)()));
467
468		// Clamp initial and unassisted to budget_max.
469	50x	if (budget_init > budget_max)
470	20x	budget_init = budget_max;
471	50x	if (unassisted > budget_max)
472	20x	unassisted = budget_max;
473
474	50x	max_events_per_poll_ = max_events;
475	50x	inline_budget_initial_ = budget_init;
476	50x	inline_budget_max_ = budget_max;
477	50x	unassisted_budget_ = unassisted;
478	50x	}
479
480		inline void
481	87185x	reactor_scheduler::reset_inline_budget() const noexcept
482		{
483		// When budget is disabled (max==0), all paths below would no-op
484		// (inline_budget stays 0). Skip the TLS lookup entirely.
485	87185x	if (inline_budget_max_ == 0)
486	58x	return;
487	87127x	if (auto* ctx = reactor_find_context(this))
488		{
489		// Cap when no other thread absorbed queued work
490	87127x	if (ctx->unassisted)
491		{
492	86088x	ctx->inline_budget_max = static_cast<int>(unassisted_budget_);
493	86088x	ctx->inline_budget = static_cast<int>(unassisted_budget_);
494	86088x	return;
495		}
496		// Ramp up when previous cycle fully consumed budget.
497		// max(1, ...) ensures the doubling escapes zero.
498		// Doubled in a wider type: the cap may be as large as INT_MAX.
499	1039x	if (ctx->inline_budget == 0)
500		{
501		long long const doubled =
502	339x	2LL * (std::max)(1, ctx->inline_budget_max);
503	339x	long long const cap = inline_budget_max_;
504	339x	ctx->inline_budget_max = static_cast<int>((std::min)(doubled, cap));
505		}
506	700x	else if (ctx->inline_budget < ctx->inline_budget_max)
507		{
508	234x	ctx->inline_budget_max = static_cast<int>(inline_budget_initial_);
509		}
510	1039x	ctx->inline_budget = ctx->inline_budget_max;
511		}
512		}
513
514		inline bool
515	352506x	reactor_scheduler::try_consume_inline_budget() const noexcept
516		{
517	352506x	if (inline_budget_max_ == 0)
518	44x	return false;
519	352462x	if (auto* ctx = reactor_find_context(this))
520		{
521	352445x	if (ctx->inline_budget > 0)
522		{
523	281693x	--ctx->inline_budget;
524	281693x	return true;
525		}
526		}
527	70769x	return false;
528		}
529
530		inline void
531	3678x	reactor_scheduler::post(std::coroutine_handle<> h) const
532		{
533		struct post_handler final : scheduler_op
534		{
535		std::coroutine_handle<> h_;
536
537	3678x	explicit post_handler(std::coroutine_handle<> h) : h_(h) {}
538	7356x	~post_handler() override = default;
539
540	3662x	void operator()() override
541		{
542	3662x	auto saved = h_;
543	3662x	delete this;
544	3662x	saved.resume();
545	3662x	}
546
547	16x	void destroy() override
548		{
549		// A bare handle carries no owner to unwind from: it is
550		// destroyed directly.
551	16x	auto saved = h_;
552	16x	delete this;
553	16x	saved.destroy();
554	16x	}
555		};
556
557	3678x	auto ph = std::make_unique<post_handler>(h);
558
559	3678x	if (auto* ctx = reactor_find_context(this))
560		{
561	12x	++ctx->private_outstanding_work;
562	12x	ctx->private_queue.push(ph.release());
563	12x	return;
564		}
565
566	3666x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
567
568	3666x	lock_type lock(mutex_);
569	3666x	completed_ops_.push(ph.release());
570	3666x	wake_one_thread_and_unlock(lock);
571	3678x	}
572
573		inline void
574	94782x	reactor_scheduler::post(scheduler_op* h) const
575		{
576	94782x	if (auto* ctx = reactor_find_context(this))
577		{
578	93270x	++ctx->private_outstanding_work;
579	93270x	ctx->private_queue.push(h);
580	93270x	return;
581		}
582
583	1512x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
584
585	1512x	lock_type lock(mutex_);
586	1512x	completed_ops_.push(h);
587	1512x	wake_one_thread_and_unlock(lock);
588	1512x	}
589
590		inline void
591	236794x	reactor_scheduler::post(capy::continuation& c) const
592		{
593	236794x	if (auto* ctx = reactor_find_context(this))
594		{
595	204508x	++ctx->private_outstanding_work;
596	204508x	ctx->private_queue.push(c);
597	204508x	return;
598		}
599
600	32286x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
601
602	32286x	lock_type lock(mutex_);
603	32286x	completed_ops_.push(c);
604	32286x	wake_one_thread_and_unlock(lock);
605	32286x	}
606
607		inline bool
608	23781x	reactor_scheduler::running_in_this_thread() const noexcept
609		{
610	23781x	return reactor_find_context(this) != nullptr;
611		}
612
613		inline void
614	8401x	reactor_scheduler::stop()
615		{
616	8401x	lock_type lock(mutex_);
617	8401x	if (!stopped_.load(std::memory_order_acquire))
618		{
619	7332x	stopped_.store(true, std::memory_order_release);
620	7332x	signal_all(lock);
621	7332x	interrupt_reactor();
622		}
623	8401x	}
624
625		inline bool
626	3569x	reactor_scheduler::stopped() const noexcept
627		{
628	3569x	return stopped_.load(std::memory_order_acquire);
629		}
630
631		inline void
632	5679x	reactor_scheduler::restart()
633		{
634	5679x	stopped_.store(false, std::memory_order_release);
635	5679x	}
636
637		inline std::size_t
638	6496x	reactor_scheduler::run()
639		{
640	12992x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
641		{
642	288x	stop();
643	288x	return 0;
644		}
645
646	6208x	reactor_thread_context_guard ctx(this);
647	6208x	lock_type lock(mutex_);
648
649	6208x	std::size_t n = 0;
650		for (;;)
651		{
652	369666x	if (!do_one(lock, -1, ctx.frame_))
653	6205x	break;
654	363458x	if (n != (std::numeric_limits<std::size_t>::max)())
655	363458x	++n;
656	363458x	if (!lock.owns_lock())
657	84207x	lock.lock();
658		}
659	6205x	return n;
660	6211x	}
661
662		inline std::size_t
663	128x	reactor_scheduler::run_one()
664		{
665	256x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
666		{
667	3x	stop();
668	3x	return 0;
669		}
670
671	125x	reactor_thread_context_guard ctx(this);
672	125x	lock_type lock(mutex_);
673	125x	return do_one(lock, -1, ctx.frame_);
674	125x	}
675
676		inline std::size_t
677	6187x	reactor_scheduler::wait_one(long usec)
678		{
679	12374x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
680		{
681	808x	stop();
682	808x	return 0;
683		}
684
685	5379x	reactor_thread_context_guard ctx(this);
686	5379x	lock_type lock(mutex_);
687	5379x	return do_one(lock, usec, ctx.frame_);
688	5379x	}
689
690		inline std::size_t
691	1349x	reactor_scheduler::poll()
692		{
693	2698x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
694		{
695	29x	stop();
696	29x	return 0;
697		}
698
699	1320x	reactor_thread_context_guard ctx(this);
700	1320x	lock_type lock(mutex_);
701
702	1320x	std::size_t n = 0;
703		for (;;)
704		{
705	9344x	if (!do_one(lock, 0, ctx.frame_))
706	1320x	break;
707	8024x	if (n != (std::numeric_limits<std::size_t>::max)())
708	8024x	++n;
709	8024x	if (!lock.owns_lock())
710	8022x	lock.lock();
711		}
712	1320x	return n;
713	1320x	}
714
715		inline std::size_t
716	62x	reactor_scheduler::poll_one()
717		{
718	124x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
719		{
720	5x	stop();
721	5x	return 0;
722		}
723
724	57x	reactor_thread_context_guard ctx(this);
725	57x	lock_type lock(mutex_);
726	57x	return do_one(lock, 0, ctx.frame_);
727	57x	}
728
729		inline void
730	66905x	reactor_scheduler::work_started() noexcept
731		{
732	66905x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
733	66905x	}
734
735		inline void
736	111566x	reactor_scheduler::work_finished() noexcept
737		{
738	223132x	if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
739	7203x	stop();
740	111566x	}
741
742		inline void
743	31188x	reactor_scheduler::compensating_work_started() const noexcept
744		{
745	31188x	auto* ctx = reactor_find_context(this);
746	31188x	if (ctx)
747	31186x	++ctx->private_outstanding_work;
748	31188x	}
749
750		inline void
751	2004x	reactor_scheduler::post_counted(scheduler_op* op) const
752		{
753	2004x	if (auto* ctx = reactor_find_context(this))
754		{
755	610x	ctx->private_queue.push(op);
756	610x	return;
757		}
758
759	1394x	lock_type lock(mutex_);
760	1394x	completed_ops_.push(op);
761	1394x	wake_one_thread_and_unlock(lock);
762	1394x	}
763
764		inline void
765	15934x	reactor_scheduler::post_deferred_completions(ready_queue& ops) const
766		{
767	15934x	if (ops.empty())
768	15934x	return;
769
770	2x	if (auto* ctx = reactor_find_context(this))
771		{
772	2x	ctx->private_queue.splice(ops);
773	2x	return;
774		}
775
776	✗	lock_type lock(mutex_);
777	✗	completed_ops_.splice(ops);
778	✗	wake_one_thread_and_unlock(lock);
779	✗	}
780
781		inline void
782	23032x	reactor_scheduler::retire_descriptor(
783		reactor_retirable& d, object_ref keep) const
784		{
785	23032x	lock_type lock(mutex_);
786	23032x	if (!task_running_.load(std::memory_order_relaxed))
787		{
788		// No pass can hold an event, but an earlier one may have
789		// queued d.
790	17769x	d.hand_off(keep);
791	17769x	return;
792		}
793		// Already parked for this pass: the held reference suffices.
794	5263x	if (d.retired_ref_)
795	117x	return;
796	5146x	d.retired_ref_ = std::move(keep);
797	5146x	d.retired_next_ = retired_;
798	5146x	retired_ = &d;
799	23032x	}
800
801		inline void
802	286761x	reactor_scheduler::settle_retired(lock_type& lock) const
803		{
804		// Handed off under the lock, since a later pass may retire the same
805		// descriptor again as soon as it is released. A reference that is
806		// not handed off may be the owner's last, and its release recycles
807		// through the pool mutex, so it is dropped with the lock released.
808	286761x	reactor_retirable* dropped = nullptr;
809	291907x	while (auto* d = retired_)
810		{
811	5146x	retired_ = d->retired_next_;
812	5146x	d->retired_next_ = nullptr;
813	5146x	auto keep = std::move(d->retired_ref_);
814	5146x	d->hand_off(keep);
815		// A drop already pending keeps the owner alive, so this one
816		// cannot be the last and is released here.
817	5146x	if (!keep \|\| d->dropped_ref_)
818	376x	continue;
819	4770x	d->dropped_ref_ = std::move(keep);
820	4770x	d->dropped_next_ = dropped;
821	4770x	dropped = d;
822	10292x	}
823	291531x	while (auto* d = dropped)
824		{
825	4770x	dropped = d->dropped_next_;
826	4770x	d->dropped_next_ = nullptr;
827	4770x	auto keep = std::move(d->dropped_ref_);
828	4770x	lock.unlock();
829	4770x	keep.reset();
830	4770x	lock.lock();
831	4770x	}
832	286761x	}
833
834		inline void
835	393x	reactor_scheduler::on_timer_service_attached(timer_service& svc)
836		{
837	393x	svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) {
838	3212x	static_cast<reactor_scheduler*>(p)->interrupt_reactor();
839	3212x	}));
840	393x	}
841
842		inline void
843	4510x	reactor_scheduler::shutdown_drain()
844		{
845	4510x	wait_cancels_posting();
846
847	4510x	lock_type lock(mutex_);
848	4510x	settle_retired(lock);
849	4510x	destroy_queued(lock);
850	4510x	signal_all(lock);
851	4510x	}
852
853	4526x	inline reactor_scheduler::~reactor_scheduler()
854		{
855		// Services give up their work in shutdown(), but anything posted
856		// after the drain, such as from a destructor, is still destroyed.
857	4526x	lock_type lock(mutex_);
858	4526x	destroy_queued(lock);
859	4526x	}
860
861		inline void
862	9036x	reactor_scheduler::destroy_queued(lock_type& lock) noexcept
863		{
864	22928x	while (auto e = completed_ops_.pop())
865		{
866	13892x	if (ready_is_continuation(e))
867		{
868	7843x	lock.unlock();
869	7843x	if (auto* c = ready_as_cont(e); c->h)
870	7843x	capy::abandon(*c);
871	7843x	lock.lock();
872		}
873		else
874		{
875	6049x	auto* op = ready_as_op(e);
876	6049x	if (op == &task_op_)
877	4507x	continue;
878	1542x	lock.unlock();
879	1542x	op->destroy();
880	1542x	lock.lock();
881		}
882	13892x	}
883	9036x	}
884
885		inline void
886	11842x	reactor_scheduler::signal_all(lock_type&) const
887		{
888	11842x	state_ \|= signaled_bit;
889	11842x	cond_.notify_all();
890	11842x	}
891
892		inline bool
893	42681x	reactor_scheduler::maybe_unlock_and_signal_one(lock_type& lock) const
894		{
895	42681x	state_ \|= signaled_bit;
896	42681x	if (state_ > signaled_bit)
897		{
898		// Signalled before unlocking: a post from another context's
899		// thread may be the last touch of this scheduler before the
900		// posted work runs, finishes and lets it be destroyed.
901	178x	cond_.notify_one();
902	178x	lock.unlock();
903	178x	return true;
904		}
905	42503x	return false;
906		}
907
908		inline bool
909	620857x	reactor_scheduler::unlock_and_signal_one(lock_type& lock) const
910		{
911	620857x	state_ \|= signaled_bit;
912	620857x	bool have_waiters = state_ > signaled_bit;
913	620857x	lock.unlock();
914	620857x	if (have_waiters)
915	4194x	cond_.notify_one();
916	620857x	return have_waiters;
917		}
918
919		inline void
920	791x	reactor_scheduler::clear_signal() const
921		{
922	791x	state_ &= ~signaled_bit;
923	791x	}
924
925		inline void
926	762x	reactor_scheduler::wait_for_signal(lock_type& lock) const
927		{
928	1590x	while ((state_ & signaled_bit) == 0)
929		{
930	828x	state_ += waiter_increment;
931	828x	cond_.wait(lock);
932	828x	state_ -= waiter_increment;
933		}
934	762x	}
935
936		inline void
937	29x	reactor_scheduler::wait_for_signal_for(lock_type& lock, long timeout_us) const
938		{
939	29x	if ((state_ & signaled_bit) == 0)
940		{
941	29x	state_ += waiter_increment;
942	29x	cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
943	29x	state_ -= waiter_increment;
944		}
945	29x	}
946
947		inline void
948	38858x	reactor_scheduler::wake_one_thread_and_unlock(lock_type& lock) const
949		{
950	38858x	if (maybe_unlock_and_signal_one(lock))
951	159x	return;
952
953	38699x	if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
954		{
955	1836x	task_interrupted_ = true;
956		// Under the lock, for the same reason as the signal above.
957	1836x	interrupt_reactor();
958	1836x	lock.unlock();
959		}
960		else
961		{
962	36863x	lock.unlock();
963		}
964		}
965
966	374354x	inline reactor_scheduler::work_cleanup::~work_cleanup()
967		{
968	374354x	std::int64_t produced = ctx.private_outstanding_work;
969	374354x	if (produced > 1)
970	1000x	sched->outstanding_work_.fetch_add(
971		produced - 1, std::memory_order_relaxed);
972	373354x	else if (produced < 1)
973	64946x	sched->work_finished();
974	374354x	ctx.private_outstanding_work = 0;
975
976	374354x	if (!ctx.private_queue.empty())
977		{
978	279555x	lock->lock();
979	279555x	sched->completed_ops_.splice(ctx.private_queue);
980		}
981	374354x	}
982
983	282254x	inline reactor_scheduler::task_cleanup::~task_cleanup()
984		{
985	282254x	if (ctx.private_outstanding_work > 0)
986		{
987	9598x	sched->outstanding_work_.fetch_add(
988	9598x	ctx.private_outstanding_work, std::memory_order_relaxed);
989	9598x	ctx.private_outstanding_work = 0;
990		}
991
992	282254x	if (!ctx.private_queue.empty())
993		{
994	9598x	if (!lock->owns_lock())
995	✗	lock->lock();
996	9598x	sched->completed_ops_.splice(ctx.private_queue);
997		}
998	282254x	}
999
1000		inline std::size_t
1001	384571x	reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx)
1002		{
1003		for (;;)
1004		{
1005	663788x	if (stopped_.load(std::memory_order_acquire))
1006	6380x	return 0;
1007
1008	657408x	auto const e = completed_ops_.pop();
1009	657408x	scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
1010
1011		// Handle reactor sentinel — time to poll for I/O
1012	657408x	if (op == &task_op_)
1013		{
1014	282260x	bool more_handlers = !completed_ops_.empty();
1015
1016	316150x	if (!more_handlers &&
1017	67780x	outstanding_work_.load(std::memory_order_acquire) == 0)
1018		{
1019	6x	completed_ops_.push(&task_op_);
1020	6x	return 0;
1021		}
1022
1023	282254x	long task_timeout_us = more_handlers ? 0 : timeout_us;
1024	282254x	task_interrupted_ = task_timeout_us == 0;
1025	282254x	task_running_.store(true, std::memory_order_release);
1026
1027		// Wake a peer to take the pending handlers while this thread
1028		// polls the reactor; skipped when one_thread_ (no peer exists).
1029	282254x	if (more_handlers && !one_thread_)
1030	248359x	unlock_and_signal_one(lock);
1031
1032		try
1033		{
1034	282254x	run_task(lock, ctx, task_timeout_us);
1035		}
1036	3x	catch (...)
1037		{
1038	3x	task_running_.store(false, std::memory_order_relaxed);
1039	3x	throw;
1040	3x	}
1041
1042	282251x	task_running_.store(false, std::memory_order_relaxed);
1043	282251x	bool const reaped = !completed_ops_.empty();
1044	282251x	completed_ops_.push(&task_op_);
1045	282251x	settle_retired(lock);
1046
1047		// wait_one() returns after one pass; poll() and poll_one()
1048		// return after a pass that reaps nothing. Either way, wake a
1049		// parked peer to take the sentinel.
1050	282251x	if (timeout_us > 0 \|\| (timeout_us == 0 && !reaped))
1051		{
1052	3825x	if (!one_thread_)
1053	3823x	maybe_unlock_and_signal_one(lock);
1054	3825x	return 0;
1055		}
1056	278426x	continue;
1057	278426x	}
1058
1059		// Handle ready entry (op or continuation)
1060	375148x	if (e != nullptr)
1061		{
1062	374354x	bool more = !completed_ops_.empty();
1063
1064	374354x	if (more && !one_thread_)
1065		{
1066		// Wake a peer for the remaining work; unassisted if none
1067		// was parked to take it.
1068	372498x	ctx.unassisted = !unlock_and_signal_one(lock);
1069		}
1070		else
1071		{
1072		// No peer to wake (one_thread_, or nothing more queued).
1073	1856x	ctx.unassisted = more;
1074	1856x	lock.unlock();
1075		}
1076
1077	374354x	[[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
1078
1079	374354x	if (ready_is_continuation(e))
1080	228951x	ready_as_cont(e)->h.resume();
1081		else
1082	145403x	(*op)();
1083	374354x	return 1;
1084	374354x	}
1085
1086	1588x	if (outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
1087		timeout_us == 0)
1088	3x	return 0;
1089
1090	791x	clear_signal();
1091	791x	if (timeout_us < 0)
1092	762x	wait_for_signal(lock);
1093		else
1094	29x	wait_for_signal_for(lock, timeout_us);
1095	279217x	}
1096		}
1097
1098		} // namespace boost::corosio::detail
1099
1100		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
1101