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

98.5% Lines (386 / 392, 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	12211x	~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	1006427x	reactor_find_context(reactor_scheduler const* self) noexcept
111		{
112	1006966x	for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
113		{
114	944142x	if (c->key == self)
115	943603x	return c;
116		}
117	62824x	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	369x	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	11962x	unsigned inline_budget_initial() const noexcept
277		{
278	11962x	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	4508x	void configure_threading(threading_config cfg) noexcept override
288		{
289	4508x	scheduler_locking_disabled_ = !cfg.scheduler_locking;
290		// reactor_io_locking takes effect at descriptor registration (see the
291		// register_descriptor overrides), not here.
292	4508x	reactor_io_locking_ = cfg.reactor_io_locking;
293	4508x	one_thread_ = cfg.one_thread;
294	4508x	mutex_.set_enabled(cfg.scheduler_locking);
295	4508x	cond_.set_enabled(cfg.scheduler_locking);
296	4508x	}
297
298		protected:
299		bool scheduler_locking_disabled_ = false;
300		bool reactor_io_locking_ = true;
301		bool one_thread_ = false;
302
303	4524x	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		// Set while the sentinel waits behind queued handlers (guarded by
337		// mutex_).
338		bool task_deferred_ = false;
339
340		// Descriptors retired during the running reactor pass (guarded by
341		// mutex_).
342		mutable reactor_retirable* retired_ = nullptr;
343
344		/// Settle every descriptor retired before the current pass ended.
345		/// @pre `lock` holds `mutex_`; it is held again on return.
346		void settle_retired(lock_type& lock) const;
347
348		// Runtime-configurable reactor tuning parameters.
349		// Defaults match the library's built-in values.
350		unsigned max_events_per_poll_ = 128;
351		unsigned inline_budget_initial_ = 2;
352		unsigned inline_budget_max_ = 16;
353		unsigned unassisted_budget_ = 4;
354
355		/// Bit 0 of `state_`: set when the condvar should be signaled.
356		static constexpr std::size_t signaled_bit = 1;
357
358		/// Increment per waiting thread in `state_`.
359		static constexpr std::size_t waiter_increment = 2;
360		mutable std::size_t state_ = 0;
361
362		/// Sentinel op that triggers a reactor poll when dequeued.
363		struct task_op final : scheduler_op
364		{
365		// LCOV_EXCL_START: the sentinel is intercepted by pointer
366		// identity; its virtuals exist for vtable completeness.
367	−	void operator()() override {}
368	−	void destroy() override {}
369		// LCOV_EXCL_STOP
370		};
371		task_op task_op_;
372
373		/** Run the platform-specific reactor poll.
374
375		@par Postconditions
376		`lock` is owned on return, however the poll ended. An
377		implementation that unlocks around the blocking call owes the
378		caller a matching re-acquire on every path out, including the
379		errors it retries rather than reports.
380		*/
381		virtual void
382		run_task(lock_type& lock, context_type& ctx, long timeout_us) = 0;
383
384		/// Wake a blocked reactor (e.g. write to eventfd or pipe).
385		virtual void interrupt_reactor() const = 0;
386
387		/// Wake the reactor whenever the earliest timer changes.
388		void on_timer_service_attached(timer_service& svc) override;
389
390		private:
391		struct work_cleanup
392		{
393		reactor_scheduler* sched;
394		lock_type* lock;
395		context_type& ctx;
396		~work_cleanup();
397		};
398
399		std::size_t do_one(lock_type& lock, long timeout_us, context_type& ctx);
400
401		void signal_all(lock_type& lock) const;
402		bool maybe_unlock_and_signal_one(lock_type& lock) const;
403		bool unlock_and_signal_one(lock_type& lock) const;
404		void clear_signal() const;
405		void wait_for_signal(lock_type& lock) const;
406		void wait_for_signal_for(lock_type& lock, long timeout_us) const;
407		void wake_one_thread_and_unlock(lock_type& lock) const;
408		};
409
410		/** RAII guard that pushes/pops a scheduler context frame.
411
412		On construction, pushes a new context frame onto the
413		thread-local stack. On destruction, drains any remaining
414		private queue items to the global queue and pops the frame.
415		*/
416		struct reactor_thread_context_guard
417		{
418		/// The context frame managed by this guard.
419		reactor_scheduler_context frame_;
420
421		/// Marks the scheduler innermost for the executor's fast check.
422		running_scheduler_guard running_;
423
424		/// Construct the guard, pushing a frame for @a sched.
425	11962x	explicit reactor_thread_context_guard(
426		reactor_scheduler const* sched) noexcept
427	11962x	: frame_(sched, reactor_context_stack.get())
428	11962x	, running_(sched)
429		{
430	11962x	reactor_context_stack.set(&frame_);
431	11962x	}
432
433		/** Destroy the guard, popping the frame.
434
435		The private queue is empty here by invariant: work_cleanup and
436		task_cleanup splice it to the global queue after every handler
437		and every reactor pass.
438		*/
439	11962x	~reactor_thread_context_guard() noexcept
440		{
441	11962x	reactor_context_stack.set(frame_.next);
442	11962x	}
443		};
444
445		// ---- Inline implementations ------------------------------------------------
446
447	11962x	inline reactor_scheduler_context::reactor_scheduler_context(
448	11962x	reactor_scheduler const* k, reactor_scheduler_context* n)
449	11962x	: key(k)
450	11962x	, next(n)
451	11962x	, private_outstanding_work(0)
452	11962x	, inline_budget(0)
453	11962x	, inline_budget_max(static_cast<int>(k->inline_budget_initial()))
454	11962x	, unassisted(false)
455		{
456	11962x	}
457
458		inline void
459	50x	reactor_scheduler::configure_reactor(
460		unsigned max_events,
461		unsigned budget_init,
462		unsigned budget_max,
463		unsigned unassisted)
464		{
465	50x	BOOST_COROSIO_ASSERT(
466		max_events >= 1 &&
467		max_events <= static_cast<unsigned>((std::numeric_limits<int>::max)()));
468	50x	BOOST_COROSIO_ASSERT(
469		budget_max <= static_cast<unsigned>((std::numeric_limits<int>::max)()));
470
471		// Clamp initial and unassisted to budget_max.
472	50x	if (budget_init > budget_max)
473	20x	budget_init = budget_max;
474	50x	if (unassisted > budget_max)
475	20x	unassisted = budget_max;
476
477	50x	max_events_per_poll_ = max_events;
478	50x	inline_budget_initial_ = budget_init;
479	50x	inline_budget_max_ = budget_max;
480	50x	unassisted_budget_ = unassisted;
481	50x	}
482
483		inline void
484	112730x	reactor_scheduler::reset_inline_budget() const noexcept
485		{
486		// When budget is disabled (max==0), all paths below would no-op
487		// (inline_budget stays 0). Skip the TLS lookup entirely.
488	112730x	if (inline_budget_max_ == 0)
489	58x	return;
490	112672x	if (auto* ctx = reactor_find_context(this))
491		{
492		// Cap when no other thread absorbed queued work
493	112672x	if (ctx->unassisted)
494		{
495	111859x	ctx->inline_budget_max = static_cast<int>(unassisted_budget_);
496	111859x	ctx->inline_budget = static_cast<int>(unassisted_budget_);
497	111859x	return;
498		}
499		// Ramp up when previous cycle fully consumed budget.
500		// max(1, ...) ensures the doubling escapes zero.
501		// Doubled in a wider type: the cap may be as large as INT_MAX.
502	813x	if (ctx->inline_budget == 0)
503		{
504		long long const doubled =
505	262x	2LL * (std::max)(1, ctx->inline_budget_max);
506	262x	long long const cap = inline_budget_max_;
507	262x	ctx->inline_budget_max = static_cast<int>((std::min)(doubled, cap));
508		}
509	551x	else if (ctx->inline_budget < ctx->inline_budget_max)
510		{
511	199x	ctx->inline_budget_max = static_cast<int>(inline_budget_initial_);
512		}
513	813x	ctx->inline_budget = ctx->inline_budget_max;
514		}
515		}
516
517		inline bool
518	482508x	reactor_scheduler::try_consume_inline_budget() const noexcept
519		{
520	482508x	if (inline_budget_max_ == 0)
521	44x	return false;
522	482464x	if (auto* ctx = reactor_find_context(this))
523		{
524	482463x	if (ctx->inline_budget > 0)
525		{
526	385728x	--ctx->inline_budget;
527	385728x	return true;
528		}
529		}
530	96736x	return false;
531		}
532
533		inline void
534	3678x	reactor_scheduler::post(std::coroutine_handle<> h) const
535		{
536		struct post_handler final : scheduler_op
537		{
538		std::coroutine_handle<> h_;
539
540	3678x	explicit post_handler(std::coroutine_handle<> h) : h_(h) {}
541	7356x	~post_handler() override = default;
542
543	3662x	void operator()() override
544		{
545	3662x	auto saved = h_;
546	3662x	delete this;
547	3662x	saved.resume();
548	3662x	}
549
550	16x	void destroy() override
551		{
552		// A bare handle carries no owner to unwind from: it is
553		// destroyed directly.
554	16x	auto saved = h_;
555	16x	delete this;
556	16x	saved.destroy();
557	16x	}
558		};
559
560	3678x	auto ph = std::make_unique<post_handler>(h);
561
562	3678x	if (auto* ctx = reactor_find_context(this))
563		{
564	12x	++ctx->private_outstanding_work;
565	12x	ctx->private_queue.push(ph.release());
566	12x	return;
567		}
568
569	3666x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
570
571	3666x	lock_type lock(mutex_);
572	3666x	completed_ops_.push(ph.release());
573	3666x	wake_one_thread_and_unlock(lock);
574	3678x	}
575
576		inline void
577	122550x	reactor_scheduler::post(scheduler_op* h) const
578		{
579	122550x	if (auto* ctx = reactor_find_context(this))
580		{
581	120270x	++ctx->private_outstanding_work;
582	120270x	ctx->private_queue.push(h);
583	120270x	return;
584		}
585
586	2280x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
587
588	2280x	lock_type lock(mutex_);
589	2280x	completed_ops_.push(h);
590	2280x	wake_one_thread_and_unlock(lock);
591	2280x	}
592
593		inline void
594	236393x	reactor_scheduler::post(capy::continuation& c) const
595		{
596	236393x	if (auto* ctx = reactor_find_context(this))
597		{
598	204420x	++ctx->private_outstanding_work;
599	204420x	ctx->private_queue.push(c);
600	204420x	return;
601		}
602
603	31973x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
604
605	31973x	lock_type lock(mutex_);
606	31973x	completed_ops_.push(c);
607	31973x	wake_one_thread_and_unlock(lock);
608	31973x	}
609
610		inline bool
611	23514x	reactor_scheduler::running_in_this_thread() const noexcept
612		{
613	23514x	return reactor_find_context(this) != nullptr;
614		}
615
616		inline void
617	8106x	reactor_scheduler::stop()
618		{
619	8106x	lock_type lock(mutex_);
620	8106x	if (!stopped_.load(std::memory_order_acquire))
621		{
622	7059x	stopped_.store(true, std::memory_order_release);
623	7059x	signal_all(lock);
624	7059x	interrupt_reactor();
625		}
626	8106x	}
627
628		inline bool
629	2726x	reactor_scheduler::stopped() const noexcept
630		{
631	2726x	return stopped_.load(std::memory_order_acquire);
632		}
633
634		inline void
635	5405x	reactor_scheduler::restart()
636		{
637		// A run can end with the sentinel marked deferred; clearing the
638		// mark lets the next run defer it again.
639	5405x	lock_type lock(mutex_);
640	5405x	task_deferred_ = false;
641	5405x	stopped_.store(false, std::memory_order_release);
642	5405x	}
643
644		inline std::size_t
645	6229x	reactor_scheduler::run()
646		{
647	12458x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
648		{
649	248x	stop();
650	248x	return 0;
651		}
652
653	5981x	reactor_thread_context_guard ctx(this);
654	5981x	lock_type lock(mutex_);
655
656	5981x	std::size_t n = 0;
657		for (;;)
658		{
659	388578x	if (!do_one(lock, -1, ctx.frame_))
660	5978x	break;
661	382597x	if (n != (std::numeric_limits<std::size_t>::max)())
662	382597x	++n;
663	382597x	if (!lock.owns_lock())
664	77600x	lock.lock();
665		}
666	5978x	return n;
667	5984x	}
668
669		inline std::size_t
670	128x	reactor_scheduler::run_one()
671		{
672	256x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
673		{
674	3x	stop();
675	3x	return 0;
676		}
677
678	125x	reactor_thread_context_guard ctx(this);
679	125x	lock_type lock(mutex_);
680	125x	return do_one(lock, -1, ctx.frame_);
681	125x	}
682
683		inline std::size_t
684	5322x	reactor_scheduler::wait_one(long usec)
685		{
686	10644x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
687		{
688	823x	stop();
689	823x	return 0;
690		}
691
692	4499x	reactor_thread_context_guard ctx(this);
693	4499x	lock_type lock(mutex_);
694	4499x	return do_one(lock, usec, ctx.frame_);
695	4499x	}
696
697		inline std::size_t
698	1338x	reactor_scheduler::poll()
699		{
700	2676x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
701		{
702	29x	stop();
703	29x	return 0;
704		}
705
706	1309x	reactor_thread_context_guard ctx(this);
707	1309x	lock_type lock(mutex_);
708
709	1309x	std::size_t n = 0;
710		for (;;)
711		{
712	9315x	if (!do_one(lock, 0, ctx.frame_))
713	1309x	break;
714	8006x	if (n != (std::numeric_limits<std::size_t>::max)())
715	8006x	++n;
716	8006x	if (!lock.owns_lock())
717	8004x	lock.lock();
718		}
719	1309x	return n;
720	1309x	}
721
722		inline std::size_t
723	53x	reactor_scheduler::poll_one()
724		{
725	106x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
726		{
727	5x	stop();
728	5x	return 0;
729		}
730
731	48x	reactor_thread_context_guard ctx(this);
732	48x	lock_type lock(mutex_);
733	48x	return do_one(lock, 0, ctx.frame_);
734	48x	}
735
736		inline void
737	66879x	reactor_scheduler::work_started() noexcept
738		{
739	66879x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
740	66879x	}
741
742		inline void
743	113131x	reactor_scheduler::work_finished() noexcept
744		{
745	226262x	if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
746	6931x	stop();
747	113131x	}
748
749		inline void
750	23165x	reactor_scheduler::compensating_work_started() const noexcept
751		{
752	23165x	auto* ctx = reactor_find_context(this);
753	23165x	if (ctx)
754	23163x	++ctx->private_outstanding_work;
755	23165x	}
756
757		inline void
758	1989x	reactor_scheduler::post_counted(scheduler_op* op) const
759		{
760	1989x	if (auto* ctx = reactor_find_context(this))
761		{
762	601x	ctx->private_queue.push(op);
763	601x	return;
764		}
765
766	1388x	lock_type lock(mutex_);
767	1388x	completed_ops_.push(op);
768	1388x	wake_one_thread_and_unlock(lock);
769	1388x	}
770
771		inline void
772	15681x	reactor_scheduler::post_deferred_completions(ready_queue& ops) const
773		{
774	15681x	if (ops.empty())
775	15681x	return;
776
777	2x	if (auto* ctx = reactor_find_context(this))
778		{
779	2x	ctx->private_queue.splice(ops);
780	2x	return;
781		}
782
783	✗	lock_type lock(mutex_);
784	✗	completed_ops_.splice(ops);
785	✗	wake_one_thread_and_unlock(lock);
786	✗	}
787
788		inline void
789	22571x	reactor_scheduler::retire_descriptor(
790		reactor_retirable& d, object_ref keep) const
791		{
792	22571x	lock_type lock(mutex_);
793	22571x	if (!task_running_.load(std::memory_order_relaxed))
794		{
795		// No pass can hold an event, but an earlier one may have
796		// queued d.
797	17484x	d.hand_off(keep);
798	17484x	return;
799		}
800		// Already parked for this pass: the held reference suffices.
801	5087x	if (d.retired_ref_)
802	81x	return;
803	5006x	d.retired_ref_ = std::move(keep);
804	5006x	d.retired_next_ = retired_;
805	5006x	retired_ = &d;
806	22571x	}
807
808		inline void
809	168199x	reactor_scheduler::settle_retired(lock_type& lock) const
810		{
811		// Handed off under the lock, since a later pass may retire the same
812		// descriptor again as soon as it is released. A reference that is
813		// not handed off may be the owner's last, and its release recycles
814		// through the pool mutex, so it is dropped with the lock released.
815	168199x	reactor_retirable* dropped = nullptr;
816	173205x	while (auto* d = retired_)
817		{
818	5006x	retired_ = d->retired_next_;
819	5006x	d->retired_next_ = nullptr;
820	5006x	auto keep = std::move(d->retired_ref_);
821	5006x	d->hand_off(keep);
822		// A drop already pending keeps the owner alive, so this one
823		// cannot be the last and is released here.
824	5006x	if (!keep \|\| d->dropped_ref_)
825	416x	continue;
826	4590x	d->dropped_ref_ = std::move(keep);
827	4590x	d->dropped_next_ = dropped;
828	4590x	dropped = d;
829	10012x	}
830	172789x	while (auto* d = dropped)
831		{
832	4590x	dropped = d->dropped_next_;
833	4590x	d->dropped_next_ = nullptr;
834	4590x	auto keep = std::move(d->dropped_ref_);
835	4590x	lock.unlock();
836	4590x	keep.reset();
837	4590x	lock.lock();
838	4590x	}
839	168199x	}
840
841		inline void
842	395x	reactor_scheduler::on_timer_service_attached(timer_service& svc)
843		{
844	395x	svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) {
845	2946x	static_cast<reactor_scheduler*>(p)->interrupt_reactor();
846	2946x	}));
847	395x	}
848
849		inline void
850	4508x	reactor_scheduler::shutdown_drain()
851		{
852	4508x	wait_cancels_posting();
853
854	4508x	lock_type lock(mutex_);
855	4508x	settle_retired(lock);
856	4508x	destroy_queued(lock);
857	4508x	signal_all(lock);
858	4508x	}
859
860	4524x	inline reactor_scheduler::~reactor_scheduler()
861		{
862		// Services give up their work in shutdown(), but anything posted
863		// after the drain, such as from a destructor, is still destroyed.
864	4524x	lock_type lock(mutex_);
865	4524x	destroy_queued(lock);
866	4524x	}
867
868		inline void
869	9032x	reactor_scheduler::destroy_queued(lock_type& lock) noexcept
870		{
871	22964x	while (auto e = completed_ops_.pop())
872		{
873	13932x	if (ready_is_continuation(e))
874		{
875	7836x	lock.unlock();
876	7836x	if (auto* c = ready_as_cont(e); c->h)
877	7836x	capy::abandon(*c);
878	7836x	lock.lock();
879		}
880		else
881		{
882	6096x	auto* op = ready_as_op(e);
883	6096x	if (op == &task_op_)
884	4505x	continue;
885	1591x	lock.unlock();
886	1591x	op->destroy();
887	1591x	lock.lock();
888		}
889	13932x	}
890	9032x	}
891
892		inline void
893	11567x	reactor_scheduler::signal_all(lock_type&) const
894		{
895	11567x	state_ \|= signaled_bit;
896	11567x	cond_.notify_all();
897	11567x	}
898
899		inline bool
900	39307x	reactor_scheduler::maybe_unlock_and_signal_one(lock_type& lock) const
901		{
902	39307x	state_ \|= signaled_bit;
903	39307x	if (state_ > signaled_bit)
904		{
905		// Signalled before unlocking: a post from another context's
906		// thread may be the last touch of this scheduler before the
907		// posted work runs, finishes and lets it be destroyed.
908	323x	cond_.notify_one();
909	323x	lock.unlock();
910	323x	return true;
911		}
912	38984x	return false;
913		}
914
915		inline bool
916	519409x	reactor_scheduler::unlock_and_signal_one(lock_type& lock) const
917		{
918	519409x	state_ \|= signaled_bit;
919	519409x	bool have_waiters = state_ > signaled_bit;
920	519409x	lock.unlock();
921	519409x	if (have_waiters)
922	3691x	cond_.notify_one();
923	519409x	return have_waiters;
924		}
925
926		inline void
927	751x	reactor_scheduler::clear_signal() const
928		{
929	751x	state_ &= ~signaled_bit;
930	751x	}
931
932		inline void
933	739x	reactor_scheduler::wait_for_signal(lock_type& lock) const
934		{
935	1539x	while ((state_ & signaled_bit) == 0)
936		{
937	800x	state_ += waiter_increment;
938	800x	cond_.wait(lock);
939	800x	state_ -= waiter_increment;
940		}
941	739x	}
942
943		inline void
944	12x	reactor_scheduler::wait_for_signal_for(lock_type& lock, long timeout_us) const
945		{
946	12x	if ((state_ & signaled_bit) == 0)
947		{
948	12x	state_ += waiter_increment;
949	12x	cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
950	12x	state_ -= waiter_increment;
951		}
952	12x	}
953
954		inline void
955	39307x	reactor_scheduler::wake_one_thread_and_unlock(lock_type& lock) const
956		{
957	39307x	if (maybe_unlock_and_signal_one(lock))
958	323x	return;
959
960	38984x	if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
961		{
962	1727x	task_interrupted_ = true;
963		// Under the lock, for the same reason as the signal above.
964	1727x	interrupt_reactor();
965	1727x	lock.unlock();
966		}
967		else
968		{
969	37257x	lock.unlock();
970		}
971		}
972
973	393444x	inline reactor_scheduler::work_cleanup::~work_cleanup()
974		{
975	393444x	std::int64_t produced = ctx.private_outstanding_work;
976	393444x	if (produced > 1)
977	936x	sched->outstanding_work_.fetch_add(
978		produced - 1, std::memory_order_relaxed);
979	392508x	else if (produced < 1)
980	66269x	sched->work_finished();
981	393444x	ctx.private_outstanding_work = 0;
982
983	393444x	if (!ctx.private_queue.empty())
984		{
985	305328x	lock->lock();
986	305328x	sched->completed_ops_.splice(ctx.private_queue);
987		}
988	393444x	}
989
990	163694x	inline reactor_scheduler::task_cleanup::~task_cleanup()
991		{
992	163694x	if (ctx.private_outstanding_work > 0)
993		{
994	9613x	sched->outstanding_work_.fetch_add(
995	9613x	ctx.private_outstanding_work, std::memory_order_relaxed);
996	9613x	ctx.private_outstanding_work = 0;
997		}
998
999	163694x	if (!ctx.private_queue.empty())
1000		{
1001	9613x	if (!lock->owns_lock())
1002	✗	lock->lock();
1003	9613x	sched->completed_ops_.splice(ctx.private_queue);
1004		}
1005	163694x	}
1006
1007		inline std::size_t
1008	402565x	reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx)
1009		{
1010		for (;;)
1011		{
1012	697962x	if (stopped_.load(std::memory_order_acquire))
1013	6153x	return 0;
1014
1015	691809x	auto const e = completed_ops_.pop();
1016	691809x	scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
1017
1018		// Handle reactor sentinel — time to poll for I/O
1019	691809x	if (op == &task_op_)
1020		{
1021	297614x	bool more_handlers = !completed_ops_.empty();
1022
1023		// Handlers queued here were mostly posted by the pass that
1024		// just ran, so polling ahead of them usually finds nothing.
1025		// Defer at most once in a row so reposting work cannot
1026		// starve I/O. poll() and poll_one() don't defer: the reactor
1027		// turn they would skip may be their last before returning.
1028	297614x	if (more_handlers && !task_deferred_ && timeout_us != 0)
1029		{
1030	132784x	task_deferred_ = true;
1031	132784x	completed_ops_.push(&task_op_);
1032	132784x	continue;
1033		}
1034	164830x	task_deferred_ = false;
1035
1036	202033x	if (!more_handlers &&
1037	74406x	(outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
1038		timeout_us == 0))
1039		{
1040	1136x	completed_ops_.push(&task_op_);
1041	1136x	return 0;
1042		}
1043
1044	163694x	long task_timeout_us = more_handlers ? 0 : timeout_us;
1045	163694x	task_interrupted_ = task_timeout_us == 0;
1046	163694x	task_running_.store(true, std::memory_order_release);
1047
1048		// Wake a peer to take the pending handlers while this thread
1049		// polls the reactor; skipped when one_thread_ (no peer exists).
1050	163694x	if (more_handlers && !one_thread_)
1051	127625x	unlock_and_signal_one(lock);
1052
1053		try
1054		{
1055	163694x	run_task(lock, ctx, task_timeout_us);
1056		}
1057	3x	catch (...)
1058		{
1059	3x	task_running_.store(false, std::memory_order_relaxed);
1060	3x	throw;
1061	3x	}
1062
1063	163691x	task_running_.store(false, std::memory_order_relaxed);
1064	163691x	completed_ops_.push(&task_op_);
1065	163691x	settle_retired(lock);
1066	163691x	if (timeout_us > 0)
1067	1829x	return 0;
1068	161862x	continue;
1069	161862x	}
1070
1071		// Handle ready entry (op or continuation)
1072	394195x	if (e != nullptr)
1073		{
1074	393444x	bool more = !completed_ops_.empty();
1075
1076	393444x	if (more && !one_thread_)
1077		{
1078		// Wake a peer for the remaining work; unassisted if none
1079		// was parked to take it.
1080	391784x	ctx.unassisted = !unlock_and_signal_one(lock);
1081		}
1082		else
1083		{
1084		// No peer to wake (one_thread_, or nothing more queued).
1085	1660x	ctx.unassisted = more;
1086	1660x	lock.unlock();
1087		}
1088
1089	393444x	[[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
1090
1091	393444x	if (ready_is_continuation(e))
1092	228557x	ready_as_cont(e)->h.resume();
1093		else
1094	164887x	(*op)();
1095	393444x	return 1;
1096	393444x	}
1097
1098	1502x	if (outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
1099		timeout_us == 0)
1100	✗	return 0;
1101
1102	751x	clear_signal();
1103	751x	if (timeout_us < 0)
1104	739x	wait_for_signal(lock);
1105		else
1106	12x	wait_for_signal_for(lock, timeout_us);
1107	295397x	}
1108		}
1109
1110		} // namespace boost::corosio::detail
1111
1112		#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
1113