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

99.1% Lines (427 / 431, 2 excl) 80.3% Branches (188 / 234) 100.0% Functions (56 / 56, 2 excl)

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

reactor_scheduler.hpp

f(x) Functions (58)

Function Calls Lines Branches Blocks

Line	Branch	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		28662x	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		14331x	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		14331x	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		14331x	~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		2224000x	reactor_find_context(reactor_scheduler const* self) noexcept
111			{
112	2/2 ✓ Branch 0 taken 2141591 times. ✓ Branch 1 taken 82940 times.	2224531x	for (auto* c = reactor_context_stack.get(); c != nullptr; c = c->next)
113			{
114	2/2 ✓ Branch 0 taken 2141060 times. ✓ Branch 1 taken 531 times.	2141591x	if (c->key == self)
115		2141060x	return c;
116		531x	}
117		82940x	return nullptr;
118		2224000x	}
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		489x	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		20459x	unsigned inline_budget_initial() const noexcept
277			{
278		20459x	return inline_budget_initial_;
279			}
280
281			/// Return true when scheduler locking is disabled (fully-lockless tier).
282		1381x	bool scheduler_locking_disabled() const noexcept override
283			{
284		1381x	return scheduler_locking_disabled_;
285			}
286
287		4629x	void configure_threading(threading_config cfg) noexcept override
288			{
289		4629x	scheduler_locking_disabled_ = !cfg.scheduler_locking;
290			// reactor_io_locking takes effect at descriptor registration (see the
291			// register_descriptor overrides), not here.
292		4629x	reactor_io_locking_ = cfg.reactor_io_locking;
293		4629x	one_thread_ = cfg.one_thread;
294		4629x	mutex_.set_enabled(cfg.scheduler_locking);
295		4629x	cond_.set_enabled(cfg.scheduler_locking);
296		4629x	}
297
298			protected:
299		4639x	bool scheduler_locking_disabled_ = false;
300		4639x	bool reactor_io_locking_ = true;
301		4639x	bool one_thread_ = false;
302
303		13917x	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		4639x	mutable mutex_type mutex_{true};
330		4639x	mutable event_type cond_{true};
331			mutable ready_queue completed_ops_;
332		4639x	mutable std::atomic<std::int64_t> outstanding_work_{0};
333		4639x	std::atomic<bool> stopped_{false};
334		4639x	mutable std::atomic<bool> task_running_{false};
335		4639x	mutable bool task_interrupted_ = false;
336
337			// Descriptors retired during the running reactor pass (guarded by
338			// mutex_).
339		4639x	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		4639x	unsigned max_events_per_poll_ = 128;
348		4639x	unsigned inline_budget_initial_ = 2;
349		4639x	unsigned inline_budget_max_ = 16;
350		4639x	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		4639x	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		40914x	explicit reactor_thread_context_guard(
423			reactor_scheduler const* sched) noexcept
424	1/2 ✓ Branch 0 taken 20457 times. ✗ Branch 1 not taken.	20457x	: frame_(sched, reactor_context_stack.get())
425		20457x	, running_(sched)
426		20457x	{
427		20457x	reactor_context_stack.set(&frame_);
428		40914x	}
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		40913x	~reactor_thread_context_guard() noexcept
437		20457x	{
438		20456x	reactor_context_stack.set(frame_.next);
439		40913x	}
440			};
441
442			// ---- Inline implementations ------------------------------------------------
443
444		61370x	inline reactor_scheduler_context::reactor_scheduler_context(
445			reactor_scheduler const* k, reactor_scheduler_context* n)
446		20457x	: key(k)
447		20457x	, next(n)
448		20457x	, private_outstanding_work(0)
449		20457x	, inline_budget(0)
450		20457x	, inline_budget_max(static_cast<int>(k->inline_budget_initial()))
451		20457x	, unassisted(false)
452		20456x	{
453		40913x	}
454
455			inline void
456		49x	reactor_scheduler::configure_reactor(
457			unsigned max_events,
458			unsigned budget_init,
459			unsigned budget_max,
460			unsigned unassisted)
461			{
462	2/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 49 times. ✓ Branch 2 taken 49 times. ✗ Branch 3 not taken.	49x	BOOST_COROSIO_ASSERT(
463			max_events >= 1 &&
464			max_events <= static_cast<unsigned>((std::numeric_limits<int>::max)()));
465	1/2 ✓ Branch 0 taken 49 times. ✗ Branch 1 not taken.	49x	BOOST_COROSIO_ASSERT(
466			budget_max <= static_cast<unsigned>((std::numeric_limits<int>::max)()));
467
468			// Clamp initial and unassisted to budget_max.
469	2/2 ✓ Branch 0 taken 29 times. ✓ Branch 1 taken 20 times.	49x	if (budget_init > budget_max)
470		20x	budget_init = budget_max;
471	2/2 ✓ Branch 0 taken 29 times. ✓ Branch 1 taken 20 times.	49x	if (unassisted > budget_max)
472		20x	unassisted = budget_max;
473
474		49x	max_events_per_poll_ = max_events;
475		49x	inline_budget_initial_ = budget_init;
476		49x	inline_budget_max_ = budget_max;
477		49x	unassisted_budget_ = unassisted;
478		49x	}
479
480			inline void
481		365201x	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	2/2 ✓ Branch 0 taken 365143 times. ✓ Branch 1 taken 58 times.	365201x	if (inline_budget_max_ == 0)
486		58x	return;
487	2/2 ✓ Branch 0 taken 5 times. ✓ Branch 1 taken 365148 times.	365143x	if (auto* ctx = reactor_find_context(this))
488			{
489			// Cap when no other thread absorbed queued work
490	2/2 ✓ Branch 0 taken 355238 times. ✓ Branch 1 taken 9910 times.	365148x	if (ctx->unassisted)
491			{
492		355238x	ctx->inline_budget_max = static_cast<int>(unassisted_budget_);
493		355238x	ctx->inline_budget = static_cast<int>(unassisted_budget_);
494		355238x	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	2/2 ✓ Branch 0 taken 9109 times. ✓ Branch 1 taken 801 times.	9910x	if (ctx->inline_budget == 0)
500			{
501		801x	long long const doubled =
502	1/2 ✓ Branch 0 taken 801 times. ✗ Branch 1 not taken.	801x	2LL * (std::max)(1, ctx->inline_budget_max);
503		801x	long long const cap = inline_budget_max_;
504	1/2 ✓ Branch 0 taken 801 times. ✗ Branch 1 not taken.	801x	ctx->inline_budget_max = static_cast<int>((std::min)(doubled, cap));
505		801x	}
506	2/2 ✓ Branch 0 taken 7691 times. ✓ Branch 1 taken 1418 times.	9109x	else if (ctx->inline_budget < ctx->inline_budget_max)
507			{
508		1418x	ctx->inline_budget_max = static_cast<int>(inline_budget_initial_);
509		1418x	}
510		9910x	ctx->inline_budget = ctx->inline_budget_max;
511		9910x	}
512		365211x	}
513
514			inline bool
515		1295136x	reactor_scheduler::try_consume_inline_budget() const noexcept
516			{
517	2/2 ✓ Branch 0 taken 1295094 times. ✓ Branch 1 taken 42 times.	1295136x	if (inline_budget_max_ == 0)
518		42x	return false;
519	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 1295094 times.	1295094x	if (auto* ctx = reactor_find_context(this))
520			{
521	2/2 ✓ Branch 0 taken 1050786 times. ✓ Branch 1 taken 244308 times.	1295094x	if (ctx->inline_budget > 0)
522			{
523		1050786x	--ctx->inline_budget;
524		1050786x	return true;
525			}
526		244308x	}
527		244308x	return false;
528		1295136x	}
529
530			inline void
531		3820x	reactor_scheduler::post(std::coroutine_handle<> h) const
532			{
533			struct post_handler final : scheduler_op
534			{
535			std::coroutine_handle<> h_;
536
537		7626x	explicit post_handler(std::coroutine_handle<> h) : h_(h) {}
538		7615x	~post_handler() override = default;
539
540		3793x	void operator()() override
541			{
542		3793x	auto saved = h_;
543	2/2 ✓ Branch 0 taken 1 time. ✓ Branch 1 taken 3792 times.	3793x	delete this;
544		3793x	saved.resume();
545		3793x	}
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	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 16 times.	16x	delete this;
553		16x	saved.destroy();
554		16x	}
555			};
556
557		3820x	auto ph = std::make_unique<post_handler>(h);
558
559	2/2 ✓ Branch 0 taken 154 times. ✓ Branch 1 taken 3666 times.	3820x	if (auto* ctx = reactor_find_context(this))
560			{
561		154x	++ctx->private_outstanding_work;
562		154x	ctx->private_queue.push(ph.release());
563		154x	return;
564			}
565
566		3666x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
567
568	1/2 ✓ Branch 0 taken 3666 times. ✗ Branch 1 not taken.	3666x	lock_type lock(mutex_);
569		3666x	completed_ops_.push(ph.release());
570	1/2 ✓ Branch 0 taken 3666 times. ✗ Branch 1 not taken.	3666x	wake_one_thread_and_unlock(lock);
571		3820x	}
572
573			inline void
574		265181x	reactor_scheduler::post(scheduler_op* h) const
575			{
576	2/2 ✓ Branch 0 taken 263551 times. ✓ Branch 1 taken 1630 times.	265181x	if (auto* ctx = reactor_find_context(this))
577			{
578		263551x	++ctx->private_outstanding_work;
579		263551x	ctx->private_queue.push(h);
580		263551x	return;
581			}
582
583		1630x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
584
585		1630x	lock_type lock(mutex_);
586		1630x	completed_ops_.push(h);
587	1/2 ✓ Branch 0 taken 1630 times. ✗ Branch 1 not taken.	1630x	wake_one_thread_and_unlock(lock);
588		265181x	}
589
590			inline void
591		249434x	reactor_scheduler::post(capy::continuation& c) const
592			{
593	2/2 ✓ Branch 0 taken 207201 times. ✓ Branch 1 taken 42233 times.	249434x	if (auto* ctx = reactor_find_context(this))
594			{
595		207201x	++ctx->private_outstanding_work;
596		207201x	ctx->private_queue.push(c);
597		207201x	return;
598			}
599
600		42233x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
601
602		42233x	lock_type lock(mutex_);
603		42233x	completed_ops_.push(c);
604	1/2 ✓ Branch 0 taken 42233 times. ✗ Branch 1 not taken.	42233x	wake_one_thread_and_unlock(lock);
605		249434x	}
606
607			inline bool
608		34030x	reactor_scheduler::running_in_this_thread() const noexcept
609			{
610		34030x	return reactor_find_context(this) != nullptr;
611			}
612
613			inline void
614		15389x	reactor_scheduler::stop()
615			{
616		15389x	lock_type lock(mutex_);
617	2/2 ✓ Branch 0 taken 931 times. ✓ Branch 1 taken 14458 times.	15389x	if (!stopped_.load(std::memory_order_acquire))
618			{
619		14458x	stopped_.store(true, std::memory_order_release);
620	2/2 ✓ Branch 0 taken 14455 times. ✓ Branch 1 taken 3 times.	14458x	signal_all(lock);
621	2/2 ✓ Branch 0 taken 14458 times. ✓ Branch 1 taken 3 times.	14455x	interrupt_reactor();
622		14458x	}
623		15395x	}
624
625			inline bool
626		3686x	reactor_scheduler::stopped() const noexcept
627			{
628		3686x	return stopped_.load(std::memory_order_acquire);
629			}
630
631			inline void
632		12707x	reactor_scheduler::restart()
633			{
634		12707x	stopped_.store(false, std::memory_order_release);
635		12707x	}
636
637			inline std::size_t
638		13700x	reactor_scheduler::run()
639			{
640	2/2 ✓ Branch 0 taken 13576 times. ✓ Branch 1 taken 124 times.	13700x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
641			{
642		124x	stop();
643		124x	return 0;
644			}
645
646		13576x	reactor_thread_context_guard ctx(this);
647	1/2 ✓ Branch 0 taken 13576 times. ✗ Branch 1 not taken.	13576x	lock_type lock(mutex_);
648
649		13576x	std::size_t n = 0;
650		637520x	for (;;)
651			{
652	4/4 ✓ Branch 0 taken 637479 times. ✓ Branch 1 taken 41 times. ✓ Branch 2 taken 623981 times. ✓ Branch 3 taken 13498 times.	637520x	if (!do_one(lock, -1, ctx.frame_))
653		13498x	break;
654	2/2 ✓ Branch 0 taken 38 times. ✓ Branch 1 taken 623943 times.	623981x	if (n != (std::numeric_limits<std::size_t>::max)())
655		623943x	++n;
656	2/2 ✓ Branch 0 taken 163578 times. ✓ Branch 1 taken 460327 times.	623981x	if (!lock.owns_lock())
657	2/2 ✓ Branch 0 taken 163617 times. ✓ Branch 1 taken 39 times.	163578x	lock.lock();
658			}
659		13498x	return n;
660		13702x	}
661
662			inline std::size_t
663		128x	reactor_scheduler::run_one()
664			{
665	2/2 ✓ Branch 0 taken 125 times. ✓ Branch 1 taken 3 times.	128x	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	1/2 ✓ Branch 0 taken 125 times. ✗ Branch 1 not taken.	125x	lock_type lock(mutex_);
673	1/2 ✓ Branch 0 taken 125 times. ✗ Branch 1 not taken.	125x	return do_one(lock, -1, ctx.frame_);
674		128x	}
675
676			inline std::size_t
677		6301x	reactor_scheduler::wait_one(long usec)
678			{
679	2/2 ✓ Branch 0 taken 5470 times. ✓ Branch 1 taken 831 times.	6301x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
680			{
681		831x	stop();
682		831x	return 0;
683			}
684
685		5470x	reactor_thread_context_guard ctx(this);
686	1/2 ✓ Branch 0 taken 5470 times. ✗ Branch 1 not taken.	5470x	lock_type lock(mutex_);
687	1/2 ✓ Branch 0 taken 5470 times. ✗ Branch 1 not taken.	5470x	return do_one(lock, usec, ctx.frame_);
688		6301x	}
689
690			inline std::size_t
691		1347x	reactor_scheduler::poll()
692			{
693	2/2 ✓ Branch 0 taken 1318 times. ✓ Branch 1 taken 29 times.	1347x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
694			{
695		29x	stop();
696		29x	return 0;
697			}
698
699		1318x	reactor_thread_context_guard ctx(this);
700	1/2 ✓ Branch 0 taken 1318 times. ✗ Branch 1 not taken.	1318x	lock_type lock(mutex_);
701
702		1318x	std::size_t n = 0;
703		9309x	for (;;)
704			{
705	3/4 ✓ Branch 0 taken 9309 times. ✗ Branch 1 not taken. ✓ Branch 2 taken 7991 times. ✓ Branch 3 taken 1318 times.	9309x	if (!do_one(lock, 0, ctx.frame_))
706		1318x	break;
707	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 7991 times.	7991x	if (n != (std::numeric_limits<std::size_t>::max)())
708		7991x	++n;
709	2/2 ✓ Branch 0 taken 7982 times. ✓ Branch 1 taken 9 times.	7991x	if (!lock.owns_lock())
710	1/2 ✓ Branch 0 taken 7982 times. ✗ Branch 1 not taken.	7982x	lock.lock();
711			}
712		1318x	return n;
713		1347x	}
714
715			inline std::size_t
716		53x	reactor_scheduler::poll_one()
717			{
718	2/2 ✓ Branch 0 taken 48 times. ✓ Branch 1 taken 5 times.	53x	if (outstanding_work_.load(std::memory_order_acquire) == 0)
719			{
720		5x	stop();
721		5x	return 0;
722			}
723
724		48x	reactor_thread_context_guard ctx(this);
725	1/2 ✓ Branch 0 taken 48 times. ✗ Branch 1 not taken.	48x	lock_type lock(mutex_);
726	1/2 ✓ Branch 0 taken 48 times. ✗ Branch 1 not taken.	48x	return do_one(lock, 0, ctx.frame_);
727		53x	}
728
729			inline void
730		174816x	reactor_scheduler::work_started() noexcept
731			{
732		174816x	outstanding_work_.fetch_add(1, std::memory_order_relaxed);
733		174816x	}
734
735			inline void
736		221819x	reactor_scheduler::work_finished() noexcept
737			{
738	2/2 ✓ Branch 0 taken 207488 times. ✓ Branch 1 taken 14331 times.	221819x	if (outstanding_work_.fetch_sub(1, std::memory_order_acq_rel) == 1)
739	1/2 ✓ Branch 0 taken 14331 times. ✗ Branch 1 not taken.	14331x	stop();
740		221819x	}
741
742			inline void
743		8644x	reactor_scheduler::compensating_work_started() const noexcept
744			{
745		8644x	auto* ctx = reactor_find_context(this);
746	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 8642 times.	8644x	if (ctx)
747		8642x	++ctx->private_outstanding_work;
748		8644x	}
749
750			inline void
751		2691x	reactor_scheduler::post_counted(scheduler_op* op) const
752			{
753	2/2 ✓ Branch 0 taken 1309 times. ✓ Branch 1 taken 1382 times.	2691x	if (auto* ctx = reactor_find_context(this))
754			{
755		1309x	ctx->private_queue.push(op);
756		1309x	return;
757			}
758
759		1382x	lock_type lock(mutex_);
760		1382x	completed_ops_.push(op);
761	1/2 ✓ Branch 0 taken 1382 times. ✗ Branch 1 not taken.	1382x	wake_one_thread_and_unlock(lock);
762		2691x	}
763
764			inline void
765		115239x	reactor_scheduler::post_deferred_completions(ready_queue& ops) const
766			{
767	2/2 ✓ Branch 0 taken 115237 times. ✓ Branch 1 taken 2 times.	115239x	if (ops.empty())
768		115237x	return;
769
770	1/2 ✓ Branch 0 taken 2 times. ✗ Branch 1 not taken.	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	0/2 ✗ Branch 0 not taken. ✗ Branch 1 not taken.	✗	wake_one_thread_and_unlock(lock);
779		115239x	}
780
781			inline void
782		35829x	reactor_scheduler::retire_descriptor(
783			reactor_retirable& d, object_ref keep) const
784			{
785		35829x	lock_type lock(mutex_);
786	2/2 ✓ Branch 0 taken 5889 times. ✓ Branch 1 taken 29940 times.	35829x	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		29940x	d.hand_off(keep);
791		29940x	return;
792			}
793			// Already parked for this pass: the held reference suffices.
794	2/2 ✓ Branch 0 taken 499 times. ✓ Branch 1 taken 5390 times.	5889x	if (d.retired_ref_)
795		499x	return;
796		5390x	d.retired_ref_ = std::move(keep);
797		5390x	d.retired_next_ = retired_;
798		5390x	retired_ = &d;
799		35829x	}
800
801			inline void
802		405779x	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		405779x	reactor_retirable* dropped = nullptr;
809	2/2 ✓ Branch 0 taken 5390 times. ✓ Branch 1 taken 405779 times.	411169x	while (auto* d = retired_)
810			{
811		5390x	retired_ = d->retired_next_;
812		5390x	d->retired_next_ = nullptr;
813		5390x	auto keep = std::move(d->retired_ref_);
814		5390x	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	3/4 ✓ Branch 0 taken 5366 times. ✓ Branch 1 taken 24 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 5366 times.	5390x	if (!keep \|\| d->dropped_ref_)
818		24x	continue;
819		5366x	d->dropped_ref_ = std::move(keep);
820		5366x	d->dropped_next_ = dropped;
821		5366x	dropped = d;
822		5390x	}
823	2/2 ✓ Branch 0 taken 5366 times. ✓ Branch 1 taken 405779 times.	411145x	while (auto* d = dropped)
824			{
825		5366x	dropped = d->dropped_next_;
826		5366x	d->dropped_next_ = nullptr;
827		5366x	auto keep = std::move(d->dropped_ref_);
828	1/2 ✓ Branch 0 taken 5366 times. ✗ Branch 1 not taken.	5366x	lock.unlock();
829		5366x	keep.reset();
830	1/2 ✓ Branch 0 taken 5366 times. ✗ Branch 1 not taken.	5366x	lock.lock();
831		5366x	}
832		405779x	}
833
834			inline void
835		874x	reactor_scheduler::on_timer_service_attached(timer_service& svc)
836			{
837		9904x	svc.set_on_earliest_changed(timer_service::callback(this, [](void* p) {
838		9030x	static_cast<reactor_scheduler*>(p)->interrupt_reactor();
839		9030x	}));
840		874x	}
841
842			inline void
843		4629x	reactor_scheduler::shutdown_drain()
844			{
845		4629x	wait_cancels_posting();
846
847		4629x	lock_type lock(mutex_);
848	2/2 ✓ Branch 0 taken 4627 times. ✓ Branch 1 taken 2 times.	4629x	settle_retired(lock);
849		4627x	destroy_queued(lock);
850	2/2 ✓ Branch 0 taken 4625 times. ✓ Branch 1 taken 2 times.	4627x	signal_all(lock);
851		4629x	}
852
853		4638x	inline reactor_scheduler::~reactor_scheduler()
854		4638x	{
855			// Services give up their work in shutdown(), but anything posted
856			// after the drain, such as from a destructor, is still destroyed.
857	1/2 ✓ Branch 0 taken 4638 times. ✗ Branch 1 not taken.	4638x	lock_type lock(mutex_);
858		4638x	destroy_queued(lock);
859		4638x	}
860
861			inline void
862		9266x	reactor_scheduler::destroy_queued(lock_type& lock) noexcept
863			{
864	2/2 ✓ Branch 0 taken 13961 times. ✓ Branch 1 taken 9266 times.	23227x	while (auto e = completed_ops_.pop())
865			{
866	2/2 ✓ Branch 0 taken 7852 times. ✓ Branch 1 taken 6109 times.	13961x	if (ready_is_continuation(e))
867			{
868	1/2 ✓ Branch 0 taken 7852 times. ✗ Branch 1 not taken.	7852x	lock.unlock();
869	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 7852 times.	7852x	if (auto* c = ready_as_cont(e); c->h)
870		7852x	capy::abandon(*c);
871	1/2 ✓ Branch 0 taken 7852 times. ✗ Branch 1 not taken.	7852x	lock.lock();
872		7852x	}
873			else
874			{
875		6109x	auto* op = ready_as_op(e);
876	2/2 ✓ Branch 0 taken 1482 times. ✓ Branch 1 taken 4627 times.	6109x	if (op == &task_op_)
877		4627x	continue;
878	1/2 ✓ Branch 0 taken 1482 times. ✗ Branch 1 not taken.	1482x	lock.unlock();
879	1/2 ✓ Branch 0 taken 1482 times. ✗ Branch 1 not taken.	1482x	op->destroy();
880	1/2 ✓ Branch 0 taken 1482 times. ✗ Branch 1 not taken.	1482x	lock.lock();
881			}
882			}
883		9266x	}
884
885			inline void
886		19083x	reactor_scheduler::signal_all(lock_type&) const
887			{
888		19083x	state_ \|= signaled_bit;
889		19083x	cond_.notify_all();
890		19083x	}
891
892			inline bool
893		48911x	reactor_scheduler::maybe_unlock_and_signal_one(lock_type& lock) const
894			{
895		48911x	state_ \|= signaled_bit;
896	2/2 ✓ Branch 0 taken 277 times. ✓ Branch 1 taken 48634 times.	48911x	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		277x	cond_.notify_one();
902		277x	lock.unlock();
903		277x	return true;
904			}
905		48634x	return false;
906		48911x	}
907
908			inline bool
909		899407x	reactor_scheduler::unlock_and_signal_one(lock_type& lock) const
910			{
911		899407x	state_ \|= signaled_bit;
912		899407x	bool have_waiters = state_ > signaled_bit;
913		899407x	lock.unlock();
914	2/2 ✓ Branch 0 taken 882933 times. ✓ Branch 1 taken 16474 times.	899407x	if (have_waiters)
915		16474x	cond_.notify_one();
916		899407x	return have_waiters;
917			}
918
919			inline void
920		3801x	reactor_scheduler::clear_signal() const
921			{
922		3801x	state_ &= ~signaled_bit;
923		3801x	}
924
925			inline void
926		3795x	reactor_scheduler::wait_for_signal(lock_type& lock) const
927			{
928	2/2 ✓ Branch 0 taken 6406 times. ✓ Branch 1 taken 3795 times.	10201x	while ((state_ & signaled_bit) == 0)
929			{
930		6406x	state_ += waiter_increment;
931		6406x	cond_.wait(lock);
932		6406x	state_ -= waiter_increment;
933			}
934		3795x	}
935
936			inline void
937		6x	reactor_scheduler::wait_for_signal_for(lock_type& lock, long timeout_us) const
938			{
939	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 6 times.	6x	if ((state_ & signaled_bit) == 0)
940			{
941		6x	state_ += waiter_increment;
942		6x	cond_.wait_for(lock, std::chrono::microseconds(timeout_us));
943		6x	state_ -= waiter_increment;
944		6x	}
945		6x	}
946
947			inline void
948		48910x	reactor_scheduler::wake_one_thread_and_unlock(lock_type& lock) const
949			{
950	2/2 ✓ Branch 0 taken 277 times. ✓ Branch 1 taken 48633 times.	48910x	if (maybe_unlock_and_signal_one(lock))
951		277x	return;
952
953	4/4 ✓ Branch 0 taken 1458 times. ✓ Branch 1 taken 47175 times. ✓ Branch 2 taken 70 times. ✓ Branch 3 taken 1388 times.	48633x	if (task_running_.load(std::memory_order_relaxed) && !task_interrupted_)
954			{
955		1388x	task_interrupted_ = true;
956			// Under the lock, for the same reason as the signal above.
957		1388x	interrupt_reactor();
958		1388x	lock.unlock();
959		1388x	}
960			else
961			{
962		47245x	lock.unlock();
963			}
964		48910x	}
965
966		1269570x	inline reactor_scheduler::work_cleanup::~work_cleanup()
967		634757x	{
968		634813x	std::int64_t produced = ctx.private_outstanding_work;
969	2/2 ✓ Branch 0 taken 1862 times. ✓ Branch 1 taken 632951 times.	634813x	if (produced > 1)
970		3724x	sched->outstanding_work_.fetch_add(
971		1862x	produced - 1, std::memory_order_relaxed);
972	2/2 ✓ Branch 0 taken 465697 times. ✓ Branch 1 taken 167254 times.	632951x	else if (produced < 1)
973		167254x	sched->work_finished();
974		634813x	ctx.private_outstanding_work = 0;
975
976	2/2 ✓ Branch 0 taken 173900 times. ✓ Branch 1 taken 460913 times.	634813x	if (!ctx.private_queue.empty())
977			{
978	1/2 ✓ Branch 0 taken 460913 times. ✗ Branch 1 not taken.	460913x	lock->lock();
979		460913x	sched->completed_ops_.splice(ctx.private_queue);
980		460913x	}
981		1269570x	}
982
983		802287x	inline reactor_scheduler::task_cleanup::~task_cleanup()
984		401144x	{
985	2/2 ✓ Branch 0 taken 394083 times. ✓ Branch 1 taken 7060 times.	401143x	if (ctx.private_outstanding_work > 0)
986			{
987		14120x	sched->outstanding_work_.fetch_add(
988		7060x	ctx.private_outstanding_work, std::memory_order_relaxed);
989		7060x	ctx.private_outstanding_work = 0;
990		7060x	}
991
992	2/2 ✓ Branch 0 taken 394083 times. ✓ Branch 1 taken 7060 times.	401143x	if (!ctx.private_queue.empty())
993			{
994	1/2 ✓ Branch 0 taken 7060 times. ✗ Branch 1 not taken.	7060x	if (!lock->owns_lock())
995	0/2 ✗ Branch 0 not taken. ✗ Branch 1 not taken.	✗	lock->lock();
996		7060x	sched->completed_ops_.splice(ctx.private_queue);
997		7060x	}
998		802287x	}
999
1000			inline std::size_t
1001		652421x	reactor_scheduler::do_one(lock_type& lock, long timeout_us, context_type& ctx)
1002			{
1003		656222x	for (;;)
1004			{
1005	2/2 ✓ Branch 0 taken 1040928 times. ✓ Branch 1 taken 13675 times.	1054603x	if (stopped_.load(std::memory_order_acquire))
1006		13675x	return 0;
1007
1008		1040928x	auto const e = completed_ops_.pop();
1009	2/2 ✓ Branch 0 taken 241642 times. ✓ Branch 1 taken 799286 times.	1040928x	scheduler_op* op = ready_is_continuation(e) ? nullptr : ready_as_op(e);
1010
1011			// Handle reactor sentinel — time to poll for I/O
1012	2/2 ✓ Branch 0 taken 638619 times. ✓ Branch 1 taken 402309 times.	1040928x	if (op == &task_op_)
1013			{
1014		402309x	bool more_handlers = !completed_ops_.empty();
1015
1016	4/4 ✓ Branch 0 taken 133860 times. ✓ Branch 1 taken 268449 times. ✓ Branch 2 taken 132716 times. ✓ Branch 3 taken 1144 times.	536169x	if (!more_handlers &&
1017	1/2 ✓ Branch 0 taken 133860 times. ✗ Branch 1 not taken.	133860x	(outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
1018		133860x	timeout_us == 0))
1019			{
1020		1144x	completed_ops_.push(&task_op_);
1021		1144x	return 0;
1022			}
1023
1024	2/2 ✓ Branch 0 taken 268450 times. ✓ Branch 1 taken 132715 times.	401165x	long task_timeout_us = more_handlers ? 0 : timeout_us;
1025		401165x	task_interrupted_ = task_timeout_us == 0;
1026		401165x	task_running_.store(true, std::memory_order_release);
1027
1028			// Wake a peer to take the pending handlers while this thread
1029			// polls the reactor; skipped when one_thread_ (no peer exists).
1030	4/4 ✓ Branch 0 taken 268444 times. ✓ Branch 1 taken 132721 times. ✓ Branch 2 taken 11 times. ✓ Branch 3 taken 268433 times.	401165x	if (more_handlers && !one_thread_)
1031		268433x	unlock_and_signal_one(lock);
1032
1033			try
1034			{
1035	2/2 ✓ Branch 0 taken 2 times. ✓ Branch 1 taken 401163 times.	401165x	run_task(lock, ctx, task_timeout_us);
1036		401165x	}
1037			catch (...)
1038			{
1039		2x	task_running_.store(false, std::memory_order_relaxed);
1040	1/2 ✗ Branch 0 not taken. ✓ Branch 1 taken 2 times.	2x	throw;
1041		2x	}
1042
1043		401163x	task_running_.store(false, std::memory_order_relaxed);
1044		401163x	completed_ops_.push(&task_op_);
1045		401163x	settle_retired(lock);
1046	2/2 ✓ Branch 0 taken 2782 times. ✓ Branch 1 taken 398381 times.	401163x	if (timeout_us > 0)
1047		2782x	return 0;
1048		398381x	continue;
1049			}
1050
1051			// Handle ready entry (op or continuation)
1052	2/2 ✓ Branch 0 taken 3802 times. ✓ Branch 1 taken 634817 times.	638619x	if (e != nullptr)
1053			{
1054		634817x	bool more = !completed_ops_.empty();
1055
1056	4/4 ✓ Branch 0 taken 631073 times. ✓ Branch 1 taken 3744 times. ✓ Branch 2 taken 18 times. ✓ Branch 3 taken 631055 times.	634817x	if (more && !one_thread_)
1057			{
1058			// Wake a peer for the remaining work; unassisted if none
1059			// was parked to take it.
1060		631055x	ctx.unassisted = !unlock_and_signal_one(lock);
1061		631055x	}
1062			else
1063			{
1064			// No peer to wake (one_thread_, or nothing more queued).
1065		3762x	ctx.unassisted = more;
1066		3762x	lock.unlock();
1067			}
1068
1069		634817x	[[maybe_unused]] work_cleanup on_exit{this, &lock, ctx};
1070
1071	2/2 ✓ Branch 0 taken 393243 times. ✓ Branch 1 taken 241574 times.	634817x	if (ready_is_continuation(e))
1072	2/2 ✓ Branch 0 taken 241573 times. ✓ Branch 1 taken 1 time.	241574x	ready_as_cont(e)->h.resume();
1073			else
1074	2/2 ✓ Branch 0 taken 393244 times. ✓ Branch 1 taken 1 time.	393243x	(*op)();
1075		634817x	return 1;
1076		634819x	}
1077
1078	3/4 ✓ Branch 0 taken 3801 times. ✓ Branch 1 taken 1 time. ✗ Branch 2 not taken. ✓ Branch 3 taken 3801 times.	3802x	if (outstanding_work_.load(std::memory_order_acquire) == 0 \|\|
1079		3801x	timeout_us == 0)
1080		1x	return 0;
1081
1082		3801x	clear_signal();
1083	2/2 ✓ Branch 0 taken 6 times. ✓ Branch 1 taken 3795 times.	3801x	if (timeout_us < 0)
1084		3795x	wait_for_signal(lock);
1085			else
1086		6x	wait_for_signal_for(lock, timeout_us);
1087			}
1088		652425x	}
1089
1090			} // namespace boost::corosio::detail
1091
1092			#endif // BOOST_COROSIO_NATIVE_DETAIL_REACTOR_REACTOR_SCHEDULER_HPP
1093