include/boost/corosio/native/detail/iocp/win_scheduler.hpp

84.7% Lines (254/300) 97.1% List of functions (33/34) 70.4% Branches (119/169)

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

win_scheduler.hpp

f(x) Functions (34)

Function Calls Lines Branches Blocks

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2025 Vinnie Falco ([email protected])
3			// Copyright (c) 2026 Steve Gerbino
4			// Copyright (c) 2026 Michael Vandeberg
5			//
6			// Distributed under the Boost Software License, Version 1.0. (See accompanying
7			// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
8			//
9			// Official repository: https://github.com/cppalliance/corosio
10			//
11
12			#ifndef BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_SCHEDULER_HPP
13			#define BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_SCHEDULER_HPP
14
15			#include <boost/corosio/detail/platform.hpp>
16
17			#if BOOST_COROSIO_HAS_IOCP
18
19			#include <boost/corosio/detail/config.hpp>
20			#include <boost/capy/ex/execution_context.hpp>
21
22			#include <boost/corosio/detail/scheduler.hpp>
23			#include <system_error>
24
25			#include <boost/corosio/detail/scheduler_op.hpp>
26			#include <boost/capy/continuation.hpp>
27			#include <boost/corosio/native/detail/iocp/win_completion_key.hpp>
28			#include <boost/corosio/native/detail/iocp/win_mutex.hpp>
29
30			#include <boost/corosio/native/detail/iocp/win_overlapped_op.hpp>
31			#include <boost/corosio/native/detail/iocp/win_timers.hpp>
32			#include <boost/corosio/detail/timer_service.hpp>
33			#include <boost/corosio/native/detail/iocp/win_resolver_service.hpp>
34			#include <boost/corosio/native/detail/make_err.hpp>
35			#include <boost/corosio/detail/except.hpp>
36			#include <boost/corosio/detail/thread_local_ptr.hpp>
37
38			#include <atomic>
39			#include <chrono>
40			#include <cstdint>
41			#include <limits>
42			#include <memory>
43			#include <mutex>
44
45			#include <boost/corosio/native/detail/iocp/win_windows.hpp>
46
47			namespace boost::corosio::detail {
48
49			// Forward declarations
50			struct overlapped_op;
51			class win_timers;
52			class win_wait_reactor;
53
54			class BOOST_COROSIO_DECL win_scheduler final
55			: public scheduler
56			, public capy::execution_context::service
57			{
58			public:
59			using key_type = scheduler;
60
61			win_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
62			~win_scheduler();
63			win_scheduler(win_scheduler const&) = delete;
64			win_scheduler& operator=(win_scheduler const&) = delete;
65
66			void shutdown() override;
67			void post(std::coroutine_handle<> h) const override;
68			void post(scheduler_op* h) const override;
69			void post(capy::continuation&) const override;
70			bool running_in_this_thread() const noexcept override;
71			void stop() override;
72			bool stopped() const noexcept override;
73			void restart() override;
74			std::size_t run() override;
75			std::size_t run_one() override;
76			std::size_t wait_one(long usec) override;
77			std::size_t poll() override;
78			std::size_t poll_one() override;
79
80		3078x	void* native_handle() const noexcept
81			{
82		3078x	return iocp_;
83			}
84
85			void work_started() noexcept override;
86			void work_finished() noexcept override;
87
88			// IOCP has only the dispatch mutex; reactor_io_locking and one_thread do
89			// not apply (the completion port provides its own synchronization).
90		615x	void configure_threading(threading_config cfg) noexcept override
91			{
92		615x	scheduler_locking_disabled_ = !cfg.scheduler_locking;
93		615x	dispatch_mutex_.set_enabled(cfg.scheduler_locking);
94		615x	}
95
96			/// Return true when scheduler locking is disabled (fully-lockless tier).
97		✗	bool scheduler_locking_disabled() const noexcept override
98			{
99		✗	return scheduler_locking_disabled_;
100			}
101
102			/** Signal that an overlapped I/O operation is now pending.
103			Coordinates with do_one() via the ready_ CAS protocol. */
104			void on_pending(overlapped_op* op) const;
105
106			/** Post an immediate completion with pre-stored results.
107			Used for sync errors and noop paths. */
108			void on_completion(overlapped_op* op, DWORD error, DWORD bytes) const;
109
110			// Timer service integration
111			void set_timer_service(timer_service* svc);
112			void update_timeout();
113
114			private:
115			static void on_timer_changed(void* ctx);
116			void post_deferred_completions(op_queue& ops);
117			std::size_t do_one(unsigned long timeout_ms);
118
119			timer_service* timer_svc_ = nullptr;
120			void* iocp_;
121			mutable long outstanding_work_;
122			mutable long stopped_;
123			long stop_event_posted_;
124			mutable long dispatch_required_;
125			bool scheduler_locking_disabled_ = false;
126
127			BOOST_COROSIO_MSVC_WARNING_PUSH
128			BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) // std::/detail:: members, dll-interface
129			mutable win_mutex dispatch_mutex_;
130			mutable op_queue completed_ops_;
131			std::unique_ptr<win_timers> timers_;
132			std::unique_ptr<win_wait_reactor> wait_reactor_;
133			std::once_flag wait_reactor_once_;
134			std::atomic<bool> wait_reactor_ready_{false};
135			BOOST_COROSIO_MSVC_WARNING_POP
136
137			public:
138			/** Auxiliary select-based reactor for IOCP wait operations.
139
140			Lazily created on first access; lives for the lifetime of the
141			scheduler and is stopped+joined in ~win_scheduler. Used by
142			socket and acceptor wait() implementations whose readiness
143			cannot be expressed natively in IOCP (datagram-read,
144			acceptor-read, error-wait).
145			*/
146			win_wait_reactor& wait_reactor();
147
148			/** Cancel a parked wait op only if the reactor exists.
149
150			Safe to call from any thread. If no wait op has ever been
151			registered, the reactor was never constructed, so there is
152			nothing to cancel and we avoid spinning up a thread + wakeup
153			socketpair on the cancel path. Acquire/release pairs with the
154			store in wait_reactor() so reads see a fully-constructed
155			reactor when the flag is true.
156			*/
157			void cancel_wait_if_constructed(overlapped_op* op) noexcept;
158			};
159
160			/*
161			ARCHITECTURE NOTE: Function Pointer Dispatch
162
163			All I/O handles are registered with the IOCP using key_io (0).
164			Dispatch happens via the function pointer stored in each scheduler_op.
165
166			When GQCS returns with an OVERLAPPED, we cast it to scheduler_op
167			and call the function pointer directly - no virtual dispatch.
168
169			The completion_key enum values are used only for internal signals:
170			- key_io (0): Normal I/O completion, dispatch via func_
171			- key_wake_dispatch (1): Timer wakeup, check dispatch_required_
172			- key_shutdown (2): Stop signal
173			- key_result_stored (3): Results pre-stored in OVERLAPPED
174			- key_posted: Carries a scheduler_op* in the OVERLAPPED pointer
175			- key_continuation: Carries a capy::continuation* in the OVERLAPPED pointer
176			*/
177
178			namespace iocp {
179
180			// Poll interval (ms) for the one-time shutdown drain loop. Unlike the
181			// steady-state loop (which blocks indefinitely until a real wake-up),
182			// the drain must keep re-checking the work count and the deferred
183			// completion queue to make progress, so it uses a finite wait.
184			inline constexpr unsigned long shutdown_drain_timeout_ms = 500;
185
186			struct BOOST_COROSIO_SYMBOL_VISIBLE scheduler_context
187			{
188			win_scheduler const* key;
189			scheduler_context* next;
190			};
191
192			inline thread_local_ptr<scheduler_context> context_stack;
193
194			struct thread_context_guard
195			{
196			scheduler_context frame_;
197
198		2959x	explicit thread_context_guard(win_scheduler const* ctx) noexcept
199		2959x	: frame_{ctx, context_stack.get()}
200			{
201		2959x	context_stack.set(&frame_);
202		2959x	}
203
204		2959x	~thread_context_guard() noexcept
205			{
206		2959x	context_stack.set(frame_.next);
207		2959x	}
208			};
209
210			} // namespace iocp
211
212		615x	inline win_scheduler::win_scheduler(
213		615x	capy::execution_context& ctx, int concurrency_hint)
214		615x	: iocp_(nullptr)
215		615x	, outstanding_work_(0)
216		615x	, stopped_(0)
217		615x	, stop_event_posted_(0)
218	1/1 ✓ Branch 4 → 5 taken 615 times.	615x	, dispatch_required_(0)
219			{
220			// concurrency_hint < 0 means use system default (DWORD(~0) = max)
221	2/3 ✓ Branch 10 → 11 taken 615 times. ✗ Branch 10 → 12 not taken. ✓ Branch 13 → 14 taken 615 times.	615x	iocp_ = ::CreateIoCompletionPort(
222			INVALID_HANDLE_VALUE, nullptr, 0,
223			static_cast<DWORD>(
224			concurrency_hint >= 0 ? concurrency_hint : DWORD(~0)));
225
226	1/2 ✗ Branch 14 → 15 not taken. ✓ Branch 14 → 18 taken 615 times.	615x	if (iocp_ == nullptr)
227		✗	detail::throw_system_error(make_err(::GetLastError()));
228
229			// Create timer wakeup mechanism (tries NT native, falls back to thread)
230	1/1 ✓ Branch 18 → 19 taken 615 times.	615x	timers_ = make_win_timers(iocp_, &dispatch_required_);
231
232			// Connect timer service to scheduler
233	2/2 ✓ Branch 21 → 22 taken 615 times. ✓ Branch 22 → 23 taken 615 times.	615x	set_timer_service(&get_timer_service(ctx, *this));
234
235			// Initialize resolver service
236	1/1 ✓ Branch 23 → 24 taken 615 times.	615x	ctx.make_service<win_resolver_service>(*this);
237		615x	}
238
239			// ~win_scheduler() and shutdown() are defined at the bottom of this
240			// header so the unique_ptr<win_wait_reactor>'s deleter and
241			// wait_reactor_->stop() see the type complete.
242
243			inline void
244		1843x	win_scheduler::post(std::coroutine_handle<> h) const
245			{
246			struct post_handler final : scheduler_op
247			{
248			std::coroutine_handle<> h_;
249
250		1843x	static void do_complete(
251			void* owner, scheduler_op* base, std::uint32_t, std::uint32_t)
252			{
253		1843x	auto* self = static_cast<post_handler*>(base);
254	2/2 ✓ Branch 2 → 3 taken 6 times. ✓ Branch 2 → 10 taken 1837 times.	1843x	if (!owner)
255			{
256			// Shutdown path: destroy the coroutine frame synchronously.
257			//
258			// Bounded destruction invariant: the chain triggered by
259			// coro.destroy() is at most two levels deep:
260			// 1. task frame destroyed → ~io_awaitable_promise_base()
261			// destroys stored continuation (if != noop_coroutine)
262			// 2. continuation (trampoline) destroyed → final_suspend
263			// returns suspend_never, no further continuation
264			//
265			// If a future refactor adds deeper continuation chains,
266			// this would reintroduce re-entrant stack overflow risk.
267			#ifndef NDEBUG
268			static thread_local int destroy_depth = 0;
269		6x	++destroy_depth;
270	1/2 ✗ Branch 3 → 4 not taken. ✓ Branch 3 → 5 taken 6 times.	6x	BOOST_COROSIO_ASSERT(destroy_depth <= 2);
271			#endif
272		6x	auto coro = self->h_;
273	1/2 ✓ Branch 6 → 7 taken 6 times. ✗ Branch 6 → 8 not taken.	6x	delete self;
274	1/1 ✓ Branch 8 → 9 taken 6 times.	6x	coro.destroy();
275			#ifndef NDEBUG
276		6x	--destroy_depth;
277			#endif
278		6x	return;
279			}
280		1837x	auto coro = self->h_;
281	1/2 ✓ Branch 10 → 11 taken 1837 times. ✗ Branch 10 → 12 not taken.	1837x	delete self;
282	1/1 ✓ Branch 12 → 13 taken 1837 times.	1837x	coro.resume();
283			}
284
285		1843x	explicit post_handler(std::coroutine_handle<> coro)
286		1843x	: scheduler_op(&do_complete)
287		1843x	, h_(coro)
288			{
289		1843x	}
290			};
291
292		1843x	auto* ph = new post_handler(h);
293		1843x	::InterlockedIncrement(&outstanding_work_);
294
295	1/2 ✗ Branch 7 → 8 not taken. ✓ Branch 7 → 14 taken 1843 times.	1843x	if (!::PostQueuedCompletionStatus(
296		1843x	iocp_, 0, key_posted, reinterpret_cast<LPOVERLAPPED>(ph)))
297			{
298		✗	std::lock_guard<win_mutex> lock(dispatch_mutex_);
299		✗	completed_ops_.push(ph);
300		✗	::InterlockedExchange(&dispatch_required_, 1);
301		✗	}
302		1843x	}
303
304			inline void
305		2041x	win_scheduler::post(scheduler_op* h) const
306			{
307		2041x	::InterlockedIncrement(&outstanding_work_);
308
309	1/2 ✗ Branch 5 → 6 not taken. ✓ Branch 5 → 12 taken 2041 times.	2041x	if (!::PostQueuedCompletionStatus(
310		2041x	iocp_, 0, key_posted, reinterpret_cast<LPOVERLAPPED>(h)))
311			{
312		✗	std::lock_guard<win_mutex> lock(dispatch_mutex_);
313		✗	completed_ops_.push(h);
314		✗	::InterlockedExchange(&dispatch_required_, 1);
315		✗	}
316		2041x	}
317
318			inline void
319		11286x	win_scheduler::post(capy::continuation& c) const
320			{
321		11286x	::InterlockedIncrement(&outstanding_work_);
322
323	1/2 ✗ Branch 5 → 6 not taken. ✓ Branch 5 → 9 taken 11286 times.	11286x	if (!::PostQueuedCompletionStatus(
324		11286x	iocp_, 0, key_continuation, reinterpret_cast<LPOVERLAPPED>(&c)))
325			{
326			// completed_ops_ is an op_queue and cannot carry a raw continuation,
327			// so on the rare PQCS failure fall back to the allocating handle
328			// path. Drop the increment first; post(c.h) does its own accounting.
329		✗	::InterlockedDecrement(&outstanding_work_);
330		✗	post(c.h);
331			}
332		11286x	}
333
334			inline bool
335		11468x	win_scheduler::running_in_this_thread() const noexcept
336			{
337	2/2 ✓ Branch 6 → 3 taken 3358 times. ✓ Branch 6 → 7 taken 8110 times.	11468x	for (auto* c = iocp::context_stack.get(); c != nullptr; c = c->next)
338	1/2 ✓ Branch 3 → 4 taken 3358 times. ✗ Branch 3 → 5 not taken.	3358x	if (c->key == this)
339		3358x	return true;
340		8110x	return false;
341			}
342
343			inline void
344		510746x	win_scheduler::work_started() noexcept
345			{
346		510746x	::InterlockedIncrement(&outstanding_work_);
347		510746x	}
348
349			inline void
350		525903x	win_scheduler::work_finished() noexcept
351			{
352	2/2 ✓ Branch 4 → 5 taken 2915 times. ✓ Branch 4 → 6 taken 522988 times.	1051806x	if (::InterlockedDecrement(&outstanding_work_) == 0)
353		2915x	stop();
354		525903x	}
355
356			inline void
357		497347x	win_scheduler::on_pending(overlapped_op* op) const
358			{
359			// If the CAS fails (ready_ was already 1), the completer got here first
360			// and stored the results — re-post so do_one() can dispatch. The acquire
361			// on failure makes those payload writes visible, so the re-posted op
362			// carries valid dwError / bytes_transferred.
363		497347x	long expected = 0;
364	1/2 ✗ Branch 11 → 12 not taken. ✓ Branch 11 → 23 taken 497347 times.	994694x	if (!op->ready_.compare_exchange_strong(
365			expected, 1,
366			std::memory_order_acq_rel, std::memory_order_acquire))
367			{
368		✗	if (!::PostQueuedCompletionStatus(
369		✗	iocp_, 0, key_result_stored, static_cast<LPOVERLAPPED>(op)))
370			{
371		✗	std::lock_guard<win_mutex> lock(dispatch_mutex_);
372		✗	completed_ops_.push(op);
373		✗	::InterlockedExchange(&dispatch_required_, 1);
374		✗	}
375			}
376		497347x	}
377
378			inline void
379		55x	win_scheduler::on_completion(overlapped_op* op, DWORD error, DWORD bytes) const
380			{
381			// Synchronous-completion path. Write the payload before the release store
382			// to ready_ so the GQCS thread that dequeues the key_result_stored post
383			// sees both fields once it observes ready_ == 1.
384		55x	op->dwError = error;
385		55x	op->bytes_transferred = bytes;
386		55x	op->ready_.store(1, std::memory_order_release);
387
388	2/4 ✓ Branch 22 → 23 taken 55 times. ✗ Branch 22 → 24 not taken. ✗ Branch 26 → 27 not taken. ✓ Branch 26 → 33 taken 55 times.	110x	if (!::PostQueuedCompletionStatus(
389		55x	iocp_, 0, key_result_stored, static_cast<LPOVERLAPPED>(op)))
390			{
391		✗	std::lock_guard<win_mutex> lock(dispatch_mutex_);
392		✗	completed_ops_.push(op);
393		✗	::InterlockedExchange(&dispatch_required_, 1);
394		✗	}
395		55x	}
396
397			inline void
398		5850x	win_scheduler::stop()
399			{
400	2/2 ✓ Branch 4 → 5 taken 2927 times. ✓ Branch 4 → 13 taken 2923 times.	11700x	if (::InterlockedExchange(&stopped_, 1) == 0)
401			{
402	1/2 ✓ Branch 7 → 8 taken 2927 times. ✗ Branch 7 → 13 not taken.	5854x	if (::InterlockedExchange(&stop_event_posted_, 1) == 0)
403			{
404	1/2 ✗ Branch 9 → 10 not taken. ✓ Branch 9 → 13 taken 2927 times.	2927x	if (!::PostQueuedCompletionStatus(iocp_, 0, key_shutdown, nullptr))
405			{
406			// The shutdown post is the only thing that wakes a
407			// run()/run_one() thread blocked indefinitely in GQCS.
408			// With no periodic timeout there is no fallback, so a
409			// failed post is fatal (matches Asio). It can only fail
410			// under resource exhaustion (ERROR_NO_SYSTEM_RESOURCES).
411		✗	detail::throw_system_error(make_err(::GetLastError()));
412			}
413			}
414			}
415		5850x	}
416
417			inline bool
418		2597x	win_scheduler::stopped() const noexcept
419			{
420			// equivalent to atomic read
421		5194x	return ::InterlockedExchangeAdd(&stopped_, 0) != 0;
422			}
423
424			inline void
425		2553x	win_scheduler::restart()
426			{
427		2553x	::InterlockedExchange(&stopped_, 0);
428		2553x	::InterlockedExchange(&stop_event_posted_, 0);
429		2553x	}
430
431			inline std::size_t
432		2948x	win_scheduler::run()
433			{
434	2/2 ✓ Branch 4 → 5 taken 21 times. ✓ Branch 4 → 7 taken 2927 times.	5896x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
435			{
436	1/1 ✓ Branch 5 → 6 taken 21 times.	21x	stop();
437		21x	return 0;
438			}
439
440		2927x	iocp::thread_context_guard ctx(this);
441
442		2927x	std::size_t n = 0;
443			for (;;)
444			{
445	3/3 ✓ Branch 9 → 10 taken 512557 times. ✓ Branch 10 → 11 taken 30 times. ✓ Branch 10 → 12 taken 512527 times.	512557x	if (!do_one(INFINITE))
446		30x	break;
447	1/2 ✓ Branch 13 → 14 taken 512527 times. ✗ Branch 13 → 15 not taken.	512527x	if (n != (std::numeric_limits<std::size_t>::max)())
448		512527x	++n;
449	2/2 ✓ Branch 17 → 18 taken 2897 times. ✓ Branch 17 → 20 taken 509630 times.	1025054x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
450			{
451	1/1 ✓ Branch 18 → 19 taken 2897 times.	2897x	stop();
452		2897x	break;
453			}
454			}
455		2927x	return n;
456		2927x	}
457
458			inline std::size_t
459		5x	win_scheduler::run_one()
460			{
461	1/2 ✗ Branch 4 → 5 not taken. ✓ Branch 4 → 7 taken 5 times.	10x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
462			{
463		✗	stop();
464		✗	return 0;
465			}
466
467		5x	iocp::thread_context_guard ctx(this);
468	1/1 ✓ Branch 8 → 9 taken 5 times.	5x	return do_one(INFINITE);
469		5x	}
470
471			inline std::size_t
472		28x	win_scheduler::wait_one(long usec)
473			{
474	2/2 ✓ Branch 4 → 5 taken 11 times. ✓ Branch 4 → 7 taken 17 times.	56x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
475			{
476	1/1 ✓ Branch 5 → 6 taken 11 times.	11x	stop();
477		11x	return 0;
478			}
479
480		17x	iocp::thread_context_guard ctx(this);
481		17x	unsigned long timeout_ms = INFINITE;
482	1/2 ✓ Branch 8 → 9 taken 17 times. ✗ Branch 8 → 13 not taken.	17x	if (usec >= 0)
483			{
484		17x	auto ms = (static_cast<long long>(usec) + 999) / 1000;
485	1/2 ✓ Branch 9 → 10 taken 17 times. ✗ Branch 9 → 11 not taken.	17x	timeout_ms = ms >= 0xFFFFFFFELL ? static_cast<unsigned long>(0xFFFFFFFE)
486			: static_cast<unsigned long>(ms);
487			}
488	1/1 ✓ Branch 13 → 14 taken 17 times.	17x	return do_one(timeout_ms);
489		17x	}
490
491			inline std::size_t
492		9x	win_scheduler::poll()
493			{
494	2/2 ✓ Branch 4 → 5 taken 1 time. ✓ Branch 4 → 7 taken 8 times.	18x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
495			{
496	1/1 ✓ Branch 5 → 6 taken 1 time.	1x	stop();
497		1x	return 0;
498			}
499
500		8x	iocp::thread_context_guard ctx(this);
501
502		8x	std::size_t n = 0;
503	3/3 ✓ Branch 12 → 13 taken 25 times. ✓ Branch 13 → 9 taken 17 times. ✓ Branch 13 → 14 taken 8 times.	25x	while (do_one(0))
504	1/2 ✓ Branch 10 → 11 taken 17 times. ✗ Branch 10 → 12 not taken.	17x	if (n != (std::numeric_limits<std::size_t>::max)())
505		17x	++n;
506		8x	return n;
507		8x	}
508
509			inline std::size_t
510		4x	win_scheduler::poll_one()
511			{
512	2/2 ✓ Branch 4 → 5 taken 2 times. ✓ Branch 4 → 7 taken 2 times.	8x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
513			{
514	1/1 ✓ Branch 5 → 6 taken 2 times.	2x	stop();
515		2x	return 0;
516			}
517
518		2x	iocp::thread_context_guard ctx(this);
519	1/1 ✓ Branch 8 → 9 taken 2 times.	2x	return do_one(0);
520		2x	}
521
522			inline void
523		446x	win_scheduler::post_deferred_completions(op_queue& ops)
524			{
525	1/2 ✗ Branch 3 → 4 not taken. ✓ Branch 3 → 15 taken 446 times.	446x	while (auto h = ops.pop())
526			{
527		✗	if (::PostQueuedCompletionStatus(
528			iocp_, 0, key_posted, reinterpret_cast<LPOVERLAPPED>(h)))
529		✗	continue;
530
531			// Out of resources, put the failed op and remaining ops back
532		✗	ops.push(h);
533		✗	std::lock_guard<win_mutex> lock(dispatch_mutex_);
534		✗	completed_ops_.splice(ops);
535		✗	::InterlockedExchange(&dispatch_required_, 1);
536		✗	return;
537		✗	}
538			}
539
540			inline std::size_t
541		515575x	win_scheduler::do_one(unsigned long timeout_ms)
542			{
543			for (;;)
544			{
545			// Check if we need to process timers or deferred ops
546	2/2 ✓ Branch 4 → 5 taken 446 times. ✓ Branch 4 → 13 taken 515129 times.	1031150x	if (::InterlockedCompareExchange(&dispatch_required_, 0, 1) == 1)
547			{
548		446x	op_queue local_ops;
549			{
550		446x	std::lock_guard<win_mutex> lock(dispatch_mutex_);
551		446x	local_ops.splice(completed_ops_);
552		446x	}
553	1/1 ✓ Branch 8 → 9 taken 446 times.	446x	post_deferred_completions(local_ops);
554
555	1/2 ✓ Branch 9 → 10 taken 446 times. ✗ Branch 9 → 11 not taken.	446x	if (timer_svc_)
556	1/1 ✓ Branch 10 → 11 taken 446 times.	446x	timer_svc_->process_expired();
557
558	1/1 ✓ Branch 11 → 12 taken 446 times.	446x	update_timeout();
559			}
560
561		515575x	DWORD bytes = 0;
562		515575x	ULONG_PTR key = 0;
563		515575x	LPOVERLAPPED overlapped = nullptr;
564	1/1 ✓ Branch 13 → 14 taken 515575 times.	515575x	::SetLastError(0);
565
566	1/1 ✓ Branch 14 → 15 taken 515575 times.	515575x	BOOL result = ::GetQueuedCompletionStatus(
567			iocp_, &bytes, &key, &overlapped, timeout_ms);
568	1/1 ✓ Branch 15 → 16 taken 515575 times.	515575x	DWORD dwError = ::GetLastError();
569
570			// Handle based on completion key
571	2/2 ✓ Branch 16 → 17 taken 512563 times. ✓ Branch 16 → 45 taken 3012 times.	515575x	if (overlapped)
572			{
573	2/2 ✓ Branch 17 → 18 taken 1896 times. ✓ Branch 17 → 19 taken 510667 times.	512563x	DWORD err = result ? 0 : dwError;
574
575	3/4 ✓ Branch 20 → 21 taken 497402 times. ✓ Branch 20 → 38 taken 3877 times. ✓ Branch 20 → 41 taken 11284 times. ✗ Branch 20 → 44 not taken.	512563x	switch (key)
576			{
577		497402x	case key_io:
578			case key_result_stored:
579			{
580		497402x	auto* ov_op = overlapped_to_op(overlapped);
581
582			// key_io carries fresh kernel results — publish them before
583			// the CAS so that losing the race (old value 0) still leaves
584			// valid data for the on_pending() re-post. For key_result_stored
585			// the payload was already written and released (by the completer
586			// in on_pending, or by on_completion), so we must not overwrite
587			// it.
588	2/2 ✓ Branch 22 → 23 taken 497347 times. ✓ Branch 22 → 24 taken 55 times.	497402x	if (key == key_io)
589		497347x	ov_op->store_result(bytes, err);
590
591			// If old value was 1 the initiator already returned — dispatch.
592			// The acquire pairs with the publisher's release so the payload
593			// reads in complete() are ordered after the store that produced
594			// them. If old value was 0 the initiator hasn't returned yet;
595			// skip and let on_pending() re-post.
596		497402x	long expected = 0;
597	1/2 ✓ Branch 33 → 34 taken 497402 times. ✗ Branch 33 → 37 not taken.	994804x	if (!ov_op->ready_.compare_exchange_strong(
598			expected, 1,
599			std::memory_order_acq_rel,
600			std::memory_order_acquire))
601			{
602		497402x	ov_op->complete(
603	1/1 ✓ Branch 34 → 35 taken 497402 times.	497402x	this, ov_op->bytes_transferred, ov_op->dwError);
604		497402x	work_finished();
605		497402x	return 1;
606			}
607		✗	continue;
608		✗	}
609
610		3877x	case key_posted:
611			{
612			// Posted scheduler_op: overlapped is actually a scheduler_op
613		3877x	auto* op = reinterpret_cast<scheduler_op*>(overlapped);
614	1/1 ✓ Branch 38 → 39 taken 3877 times.	3877x	op->complete(this, bytes, err);
615		3877x	work_finished();
616		3877x	return 1;
617			}
618
619		11284x	case key_continuation:
620			{
621			// Posted continuation: overlapped is actually a continuation*
622		11284x	auto* c = reinterpret_cast<capy::continuation*>(overlapped);
623	1/1 ✓ Branch 41 → 42 taken 11284 times.	11284x	c->h.resume();
624		11284x	work_finished();
625		11284x	return 1;
626			}
627
628		✗	default:
629		✗	continue;
630		✗	}
631			}
632
633			// Signal completions (no OVERLAPPED)
634	2/2 ✓ Branch 45 → 46 taken 3001 times. ✓ Branch 45 → 59 taken 11 times.	3012x	if (result)
635			{
636	2/3 ✓ Branch 46 → 47 taken 446 times. ✓ Branch 46 → 48 taken 2555 times. ✗ Branch 46 → 58 not taken.	3001x	switch (key)
637			{
638		446x	case key_wake_dispatch:
639			// Timer wakeup - loop to check dispatch_required_
640		446x	continue;
641
642		2555x	case key_shutdown:
643		2555x	::InterlockedExchange(&stop_event_posted_, 0);
644	2/2 ✓ Branch 51 → 52 taken 32 times. ✓ Branch 51 → 57 taken 2523 times.	2555x	if (stopped())
645			{
646			// Re-post for other waiting threads
647	1/2 ✓ Branch 54 → 55 taken 32 times. ✗ Branch 54 → 56 not taken.	64x	if (::InterlockedExchange(&stop_event_posted_, 1) == 0)
648			{
649	1/1 ✓ Branch 55 → 56 taken 32 times.	32x	::PostQueuedCompletionStatus(
650			iocp_, 0, key_shutdown, nullptr);
651			}
652		32x	return 0;
653			}
654		2523x	continue;
655
656		✗	default:
657		✗	continue;
658			}
659			}
660
661			// Timeout or error. INFINITE never times out, so a WAIT_TIMEOUT
662			// can only be a finite caller timeout (wait_one/poll) elapsing
663			// with no work. run()/run_one() pass INFINITE and block until a
664			// real wake-up; they exit only when stop() posts key_shutdown
665			// (a failed post is fatal in stop()).
666	1/2 ✗ Branch 59 → 60 not taken. ✓ Branch 59 → 62 taken 11 times.	11x	if (dwError != WAIT_TIMEOUT)
667		✗	detail::throw_system_error(make_err(dwError));
668		11x	return 0;
669		2969x	}
670			}
671
672			inline void
673		1808x	win_scheduler::on_timer_changed(void* ctx)
674			{
675		1808x	static_cast<win_scheduler*>(ctx)->update_timeout();
676		1808x	}
677
678			inline void
679		615x	win_scheduler::set_timer_service(timer_service* svc)
680			{
681		615x	timer_svc_ = svc;
682			// Pass 'this' as context - callback routes to correct instance
683		615x	svc->set_on_earliest_changed(
684		615x	timer_service::callback{this, &on_timer_changed});
685	1/2 ✓ Branch 5 → 6 taken 615 times. ✗ Branch 5 → 8 not taken.	615x	if (timers_)
686		615x	timers_->start();
687		615x	}
688
689			inline void
690		2254x	win_scheduler::update_timeout()
691			{
692	3/6 ✓ Branch 2 → 3 taken 2254 times. ✗ Branch 2 → 6 not taken. ✓ Branch 4 → 5 taken 2254 times. ✗ Branch 4 → 6 not taken. ✓ Branch 7 → 8 taken 2254 times. ✗ Branch 7 → 11 not taken.	2254x	if (timer_svc_ && timers_)
693		2254x	timers_->update_timeout(timer_svc_->nearest_expiry());
694		2254x	}
695
696			} // namespace boost::corosio::detail
697
698			// Defer including the auxiliary wait reactor until the scheduler is
699			// fully defined, since the reactor's inline methods call back into
700			// win_scheduler. This also gives the dtor and wait_reactor() below a
701			// complete win_wait_reactor type for unique_ptr destruction and
702			// lazy construction.
703			//
704			// The macro lets win_wait_reactor.hpp diagnose direct inclusion
705			// (which would land it here with win_scheduler still incomplete).
706			#define BOOST_COROSIO_DETAIL_IOCP_WIN_SCHEDULER_BODY_DONE
707			#include <boost/corosio/native/detail/iocp/win_wait_reactor.hpp>
708
709			namespace boost::corosio::detail {
710
711			inline void
712		615x	win_scheduler::shutdown()
713			{
714	1/2 ✓ Branch 3 → 4 taken 615 times. ✗ Branch 3 → 6 not taken.	615x	if (timers_)
715		615x	timers_->stop();
716
717			// Drain timer heap before the work-counting loop. The timer_service
718			// was registered after this scheduler (nested make_service from our
719			// constructor), so execution_context::shutdown() calls us first.
720			// Asio avoids this by owning timer queues directly inside the
721			// scheduler; we bridge the gap by shutting down the timer service
722			// early. The subsequent call from execution_context is a no-op.
723	1/2 ✓ Branch 6 → 7 taken 615 times. ✗ Branch 6 → 8 not taken.	615x	if (timer_svc_)
724		615x	timer_svc_->shutdown();
725
726			// Same problem for the auxiliary wait reactor: ops parked in it
727			// hold work_started credit. Stop the reactor early so its loop
728			// drains them as cancelled and the work counter can reach zero.
729	2/2 ✓ Branch 9 → 10 taken 9 times. ✓ Branch 9 → 12 taken 606 times.	615x	if (wait_reactor_ready_.load(std::memory_order_acquire))
730		9x	wait_reactor_->stop();
731
732	2/2 ✓ Branch 40 → 13 taken 13 times. ✓ Branch 40 → 41 taken 615 times.	1256x	while (::InterlockedExchangeAdd(&outstanding_work_, 0) > 0)
733			{
734		13x	op_queue ops;
735			{
736		13x	std::lock_guard<win_mutex> lock(dispatch_mutex_);
737		13x	ops.splice(completed_ops_);
738		13x	}
739
740	1/2 ✗ Branch 17 → 18 not taken. ✓ Branch 17 → 24 taken 13 times.	13x	if (!ops.empty())
741			{
742		✗	while (auto* h = ops.pop())
743			{
744		✗	::InterlockedDecrement(&outstanding_work_);
745		✗	h->destroy();
746		✗	}
747			}
748			else
749			{
750			DWORD bytes;
751			ULONG_PTR key;
752			LPOVERLAPPED overlapped;
753	1/1 ✓ Branch 24 → 25 taken 13 times.	13x	::GetQueuedCompletionStatus(
754			iocp_, &bytes, &key, &overlapped,
755			iocp::shutdown_drain_timeout_ms);
756	1/2 ✓ Branch 25 → 26 taken 13 times. ✗ Branch 25 → 36 not taken.	13x	if (overlapped)
757			{
758		13x	::InterlockedDecrement(&outstanding_work_);
759	2/2 ✓ Branch 28 → 29 taken 7 times. ✓ Branch 28 → 30 taken 6 times.	13x	if (key == key_posted)
760			{
761		7x	auto* op = reinterpret_cast<scheduler_op*>(overlapped);
762	1/1 ✓ Branch 29 → 36 taken 7 times.	7x	op->destroy();
763			}
764	2/2 ✓ Branch 30 → 31 taken 2 times. ✓ Branch 30 → 34 taken 4 times.	6x	else if (key == key_continuation)
765			{
766			// Drain without resuming: destroy the parked frame.
767		2x	auto* c = reinterpret_cast<capy::continuation*>(overlapped);
768	1/2 ✓ Branch 32 → 33 taken 2 times. ✗ Branch 32 → 36 not taken.	2x	if (c->h)
769	1/1 ✓ Branch 33 → 36 taken 2 times.	2x	c->h.destroy();
770			}
771			else
772			{
773		4x	auto* op = overlapped_to_op(overlapped);
774	1/1 ✓ Branch 35 → 36 taken 4 times.	4x	op->destroy();
775			}
776			}
777			}
778			}
779		615x	}
780
781		1230x	inline win_scheduler::~win_scheduler()
782			{
783	2/2 ✓ Branch 3 → 4 taken 9 times. ✓ Branch 3 → 6 taken 606 times.	615x	if (wait_reactor_)
784		9x	wait_reactor_->stop();
785		615x	wait_reactor_.reset();
786
787	1/2 ✓ Branch 7 → 8 taken 615 times. ✗ Branch 7 → 9 not taken.	615x	if (iocp_ != nullptr)
788		615x	::CloseHandle(iocp_);
789		1230x	}
790
791			inline win_wait_reactor&
792		9x	win_scheduler::wait_reactor()
793			{
794			// Lazy thread-safe init: multiple IOCP workers may race the first
795			// wait() call. wait_reactor_ready_ is set with release ordering
796			// after construction so cancel_wait_if_constructed can safely
797			// observe the reactor without forcing construction itself.
798	1/1 ✓ Branch 2 → 3 taken 9 times.	9x	std::call_once(wait_reactor_once_, [this] {
799	1/1 ✓ Branch 2 → 3 taken 9 times.	9x	wait_reactor_ = std::make_unique<win_wait_reactor>(*this);
800		9x	wait_reactor_ready_.store(true, std::memory_order_release);
801		9x	});
802		9x	return *wait_reactor_;
803			}
804
805			inline void
806		19193x	win_scheduler::cancel_wait_if_constructed(overlapped_op* op) noexcept
807			{
808	2/2 ✓ Branch 3 → 4 taken 54 times. ✓ Branch 3 → 6 taken 19139 times.	19193x	if (wait_reactor_ready_.load(std::memory_order_acquire))
809		54x	wait_reactor_->cancel_wait(op);
810		19193x	}
811
812			} // namespace boost::corosio::detail
813
814			#endif // BOOST_COROSIO_HAS_IOCP
815
816			#endif // BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_SCHEDULER_HPP
817