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

84.9% Lines (270/318) 97.1% List of functions (33/34) 70.3% Branches (128/182)

[ ] 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		5608x	void* native_handle() const noexcept
81			{
82		5608x	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		1121x	void configure_threading(threading_config cfg) noexcept override
91			{
92		1121x	scheduler_locking_disabled_ = !cfg.scheduler_locking;
93		1121x	dispatch_mutex_.set_enabled(cfg.scheduler_locking);
94		1121x	}
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
123			// Packets in flight to the completion port that reference
124			// overlapped-op memory: kernel completions owed after a pending
125			// submission, plus successful stored-result posts. Shutdown reaps
126			// until this is zero before the services free op storage. The
127			// run-loop counter cannot serve that role: frames abandoned at
128			// teardown never return their work-guard credits.
129			mutable long pending_io_ = 0;
130			mutable long stopped_;
131			long stop_event_posted_;
132			mutable long dispatch_required_;
133			bool scheduler_locking_disabled_ = false;
134
135			BOOST_COROSIO_MSVC_WARNING_PUSH
136			BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) // std::/detail:: members, dll-interface
137			mutable win_mutex dispatch_mutex_;
138			mutable op_queue completed_ops_;
139			std::unique_ptr<win_timers> timers_;
140			std::unique_ptr<win_wait_reactor> wait_reactor_;
141			std::once_flag wait_reactor_once_;
142			std::atomic<bool> wait_reactor_ready_{false};
143			BOOST_COROSIO_MSVC_WARNING_POP
144
145			public:
146			/** Auxiliary select-based reactor for IOCP wait operations.
147
148			Lazily created on first access; lives for the lifetime of the
149			scheduler and is stopped+joined in ~win_scheduler. Used by
150			socket and acceptor wait() implementations whose readiness
151			cannot be expressed natively in IOCP (datagram-read,
152			acceptor-read, error-wait).
153			*/
154			win_wait_reactor& wait_reactor();
155
156			/** Cancel a parked wait op only if the reactor exists.
157
158			Safe to call from any thread. If no wait op has ever been
159			registered, the reactor was never constructed, so there is
160			nothing to cancel and we avoid spinning up a thread + wakeup
161			socketpair on the cancel path. Acquire/release pairs with the
162			store in wait_reactor() so reads see a fully-constructed
163			reactor when the flag is true.
164			*/
165			void cancel_wait_if_constructed(overlapped_op* op) noexcept;
166			};
167
168			/*
169			ARCHITECTURE NOTE: Function Pointer Dispatch
170
171			All I/O handles are registered with the IOCP using key_io (0).
172			Dispatch happens via the function pointer stored in each scheduler_op.
173
174			When GQCS returns with an OVERLAPPED, we cast it to scheduler_op
175			and call the function pointer directly - no virtual dispatch.
176
177			The completion_key enum values are used only for internal signals:
178			- key_io (0): Normal I/O completion, dispatch via func_
179			- key_wake_dispatch (1): Timer wakeup, check dispatch_required_
180			- key_shutdown (2): Stop signal
181			- key_result_stored (3): Results pre-stored in OVERLAPPED
182			- key_posted: Carries a scheduler_op* in the OVERLAPPED pointer
183			- key_continuation: Carries a capy::continuation* in the OVERLAPPED pointer
184			*/
185
186			namespace iocp {
187
188			// Poll interval (ms) for the one-time shutdown drain loop. Unlike the
189			// steady-state loop (which blocks indefinitely until a real wake-up),
190			// the drain must keep re-checking the work count and the deferred
191			// completion queue to make progress, so it uses a finite wait.
192			inline constexpr unsigned long shutdown_drain_timeout_ms = 500;
193
194			struct BOOST_COROSIO_SYMBOL_VISIBLE scheduler_context
195			{
196			win_scheduler const* key;
197			scheduler_context* next;
198			};
199
200			inline thread_local_ptr<scheduler_context> context_stack;
201
202			struct thread_context_guard
203			{
204			scheduler_context frame_;
205
206		5431x	explicit thread_context_guard(win_scheduler const* ctx) noexcept
207		5431x	: frame_{ctx, context_stack.get()}
208			{
209		5431x	context_stack.set(&frame_);
210		5431x	}
211
212		5431x	~thread_context_guard() noexcept
213			{
214		5431x	context_stack.set(frame_.next);
215		5431x	}
216			};
217
218			} // namespace iocp
219
220			// The constructor, ~win_scheduler() and shutdown() are defined at the
221			// bottom of this header so the unique_ptr<win_wait_reactor>'s deleter
222			// and wait_reactor_->stop() see the type complete. The constructor
223			// needs it too: its unwind path destroys wait_reactor_.
224
225			inline void
226		1843x	win_scheduler::post(std::coroutine_handle<> h) const
227			{
228			struct post_handler final : scheduler_op
229			{
230			std::coroutine_handle<> h_;
231
232		1843x	static void do_complete(
233			void* owner, scheduler_op* base, std::uint32_t, std::uint32_t)
234			{
235		1843x	auto* self = static_cast<post_handler*>(base);
236	2/2 ✓ Branch 2 → 3 taken 6 times. ✓ Branch 2 → 10 taken 1837 times.	1843x	if (!owner)
237			{
238			// Shutdown path: destroy the coroutine frame synchronously.
239			//
240			// Bounded destruction invariant: the chain triggered by
241			// coro.destroy() is at most two levels deep:
242			// 1. task frame destroyed → ~io_awaitable_promise_base()
243			// destroys stored continuation (if != noop_coroutine)
244			// 2. continuation (trampoline) destroyed → final_suspend
245			// returns suspend_never, no further continuation
246			//
247			// If a future refactor adds deeper continuation chains,
248			// this would reintroduce re-entrant stack overflow risk.
249			#ifndef NDEBUG
250			static thread_local int destroy_depth = 0;
251		6x	++destroy_depth;
252	1/2 ✗ Branch 3 → 4 not taken. ✓ Branch 3 → 5 taken 6 times.	6x	BOOST_COROSIO_ASSERT(destroy_depth <= 2);
253			#endif
254		6x	auto coro = self->h_;
255	1/2 ✓ Branch 6 → 7 taken 6 times. ✗ Branch 6 → 8 not taken.	6x	delete self;
256	1/1 ✓ Branch 8 → 9 taken 6 times.	6x	coro.destroy();
257			#ifndef NDEBUG
258		6x	--destroy_depth;
259			#endif
260		6x	return;
261			}
262		1837x	auto coro = self->h_;
263	1/2 ✓ Branch 10 → 11 taken 1837 times. ✗ Branch 10 → 12 not taken.	1837x	delete self;
264	1/1 ✓ Branch 12 → 13 taken 1837 times.	1837x	coro.resume();
265			}
266
267		1843x	explicit post_handler(std::coroutine_handle<> coro)
268		1843x	: scheduler_op(&do_complete)
269		1843x	, h_(coro)
270			{
271		1843x	}
272			};
273
274		1843x	auto* ph = new post_handler(h);
275		1843x	::InterlockedIncrement(&outstanding_work_);
276
277	1/2 ✗ Branch 7 → 8 not taken. ✓ Branch 7 → 14 taken 1843 times.	1843x	if (!::PostQueuedCompletionStatus(
278		1843x	iocp_, 0, key_posted, reinterpret_cast<LPOVERLAPPED>(ph)))
279			{
280		✗	std::lock_guard<win_mutex> lock(dispatch_mutex_);
281		✗	completed_ops_.push(ph);
282		✗	::InterlockedExchange(&dispatch_required_, 1);
283		✗	}
284		1843x	}
285
286			inline void
287		3593x	win_scheduler::post(scheduler_op* h) const
288			{
289		3593x	::InterlockedIncrement(&outstanding_work_);
290
291	1/2 ✗ Branch 5 → 6 not taken. ✓ Branch 5 → 12 taken 3593 times.	3593x	if (!::PostQueuedCompletionStatus(
292		3593x	iocp_, 0, key_posted, reinterpret_cast<LPOVERLAPPED>(h)))
293			{
294		✗	std::lock_guard<win_mutex> lock(dispatch_mutex_);
295		✗	completed_ops_.push(h);
296		✗	::InterlockedExchange(&dispatch_required_, 1);
297		✗	}
298		3593x	}
299
300			inline void
301		17818x	win_scheduler::post(capy::continuation& c) const
302			{
303		17818x	::InterlockedIncrement(&outstanding_work_);
304
305	1/2 ✗ Branch 5 → 6 not taken. ✓ Branch 5 → 9 taken 17818 times.	17818x	if (!::PostQueuedCompletionStatus(
306		17818x	iocp_, 0, key_continuation, reinterpret_cast<LPOVERLAPPED>(&c)))
307			{
308			// completed_ops_ is an op_queue and cannot carry a raw continuation,
309			// so on the rare PQCS failure fall back to the allocating handle
310			// path. Drop the increment first; post(c.h) does its own accounting.
311		✗	::InterlockedDecrement(&outstanding_work_);
312		✗	post(c.h);
313			}
314		17818x	}
315
316			inline bool
317		17300x	win_scheduler::running_in_this_thread() const noexcept
318			{
319	2/2 ✓ Branch 6 → 3 taken 3371 times. ✓ Branch 6 → 7 taken 13929 times.	17300x	for (auto* c = iocp::context_stack.get(); c != nullptr; c = c->next)
320	1/2 ✓ Branch 3 → 4 taken 3371 times. ✗ Branch 3 → 5 not taken.	3371x	if (c->key == this)
321		3371x	return true;
322		13929x	return false;
323			}
324
325			inline void
326		648517x	win_scheduler::work_started() noexcept
327			{
328		648517x	::InterlockedIncrement(&outstanding_work_);
329		648517x	}
330
331			inline void
332		671750x	win_scheduler::work_finished() noexcept
333			{
334	2/2 ✓ Branch 4 → 5 taken 5376 times. ✓ Branch 4 → 6 taken 666374 times.	1343500x	if (::InterlockedDecrement(&outstanding_work_) == 0)
335		5376x	stop();
336		671750x	}
337
338			inline void
339		627557x	win_scheduler::on_pending(overlapped_op* op) const
340			{
341			// If the CAS fails (ready_ was already 1), the completer got here first
342			// and stored the results — re-post so do_one() can dispatch. The acquire
343			// on failure makes those payload writes visible, so the re-posted op
344			// carries valid dwError / bytes_transferred.
345			//
346			// pending_io_ counts the packet that will dispatch this op: on CAS
347			// success the kernel's own completion will find ready_ == 1 and
348			// dispatch; on CAS failure the kernel's packet was consumed as a
349			// skip (uncounted) and the re-post is the dispatching packet. A
350			// failed re-post falls back to the deferred queue, which holds the
351			// op memory itself — no packet, no count.
352		627557x	long expected = 0;
353	1/2 ✓ Branch 11 → 12 taken 627557 times. ✗ Branch 11 → 15 not taken.	1255114x	if (op->ready_.compare_exchange_strong(
354			expected, 1,
355			std::memory_order_acq_rel, std::memory_order_acquire))
356			{
357		627557x	::InterlockedIncrement(&pending_io_);
358			}
359			else
360			{
361		✗	if (::PostQueuedCompletionStatus(
362		✗	iocp_, 0, key_result_stored, static_cast<LPOVERLAPPED>(op)))
363			{
364		✗	::InterlockedIncrement(&pending_io_);
365			}
366			else
367			{
368		✗	std::lock_guard<win_mutex> lock(dispatch_mutex_);
369		✗	completed_ops_.push(op);
370		✗	::InterlockedExchange(&dispatch_required_, 1);
371		✗	}
372			}
373		627557x	}
374
375			inline void
376		243x	win_scheduler::on_completion(overlapped_op* op, DWORD error, DWORD bytes) const
377			{
378			// Synchronous-completion path. Write the payload before the release store
379			// to ready_ so the GQCS thread that dequeues the key_result_stored post
380			// sees both fields once it observes ready_ == 1.
381		243x	op->dwError = error;
382		243x	op->bytes_transferred = bytes;
383		243x	op->ready_.store(1, std::memory_order_release);
384
385	2/4 ✓ Branch 22 → 23 taken 243 times. ✗ Branch 22 → 24 not taken. ✓ Branch 26 → 27 taken 243 times. ✗ Branch 26 → 30 not taken.	486x	if (::PostQueuedCompletionStatus(
386		243x	iocp_, 0, key_result_stored, static_cast<LPOVERLAPPED>(op)))
387			{
388		243x	::InterlockedIncrement(&pending_io_);
389			}
390			else
391			{
392		✗	std::lock_guard<win_mutex> lock(dispatch_mutex_);
393		✗	completed_ops_.push(op);
394		✗	::InterlockedExchange(&dispatch_required_, 1);
395		✗	}
396		243x	}
397
398			inline void
399		10809x	win_scheduler::stop()
400			{
401	2/2 ✓ Branch 4 → 5 taken 5414 times. ✓ Branch 4 → 13 taken 5395 times.	21618x	if (::InterlockedExchange(&stopped_, 1) == 0)
402			{
403	1/2 ✓ Branch 7 → 8 taken 5414 times. ✗ Branch 7 → 13 not taken.	10828x	if (::InterlockedExchange(&stop_event_posted_, 1) == 0)
404			{
405	1/2 ✗ Branch 9 → 10 not taken. ✓ Branch 9 → 13 taken 5414 times.	5414x	if (!::PostQueuedCompletionStatus(iocp_, 0, key_shutdown, nullptr))
406			{
407			// The shutdown post is the only thing that wakes a
408			// run()/run_one() thread blocked indefinitely in GQCS.
409			// With no periodic timeout there is no fallback, so a
410			// failed post is fatal (matches Asio). It can only fail
411			// under resource exhaustion (ERROR_NO_SYSTEM_RESOURCES).
412		✗	detail::throw_system_error(make_err(::GetLastError()));
413			}
414			}
415			}
416		10809x	}
417
418			inline bool
419		4635x	win_scheduler::stopped() const noexcept
420			{
421			// equivalent to atomic read
422		9270x	return ::InterlockedExchangeAdd(&stopped_, 0) != 0;
423			}
424
425			inline void
426		4668x	win_scheduler::restart()
427			{
428		4668x	::InterlockedExchange(&stopped_, 0);
429		4668x	::InterlockedExchange(&stop_event_posted_, 0);
430		4668x	}
431
432			inline std::size_t
433		5438x	win_scheduler::run()
434			{
435	2/2 ✓ Branch 4 → 5 taken 45 times. ✓ Branch 4 → 7 taken 5393 times.	10876x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
436			{
437	1/1 ✓ Branch 5 → 6 taken 45 times.	45x	stop();
438		45x	return 0;
439			}
440
441		5393x	iocp::thread_context_guard ctx(this);
442
443		5393x	std::size_t n = 0;
444			for (;;)
445			{
446	3/3 ✓ Branch 9 → 10 taken 651035 times. ✓ Branch 10 → 11 taken 36 times. ✓ Branch 10 → 12 taken 650999 times.	651035x	if (!do_one(INFINITE))
447		36x	break;
448	1/2 ✓ Branch 13 → 14 taken 650999 times. ✗ Branch 13 → 15 not taken.	650999x	if (n != (std::numeric_limits<std::size_t>::max)())
449		650999x	++n;
450	2/2 ✓ Branch 17 → 18 taken 5357 times. ✓ Branch 17 → 20 taken 645642 times.	1301998x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
451			{
452	1/1 ✓ Branch 18 → 19 taken 5357 times.	5357x	stop();
453		5357x	break;
454			}
455			}
456		5393x	return n;
457		5393x	}
458
459			inline std::size_t
460		11x	win_scheduler::run_one()
461			{
462	2/2 ✓ Branch 4 → 5 taken 1 time. ✓ Branch 4 → 7 taken 10 times.	22x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
463			{
464	1/1 ✓ Branch 5 → 6 taken 1 time.	1x	stop();
465		1x	return 0;
466			}
467
468		10x	iocp::thread_context_guard ctx(this);
469	1/1 ✓ Branch 8 → 9 taken 10 times.	10x	return do_one(INFINITE);
470		10x	}
471
472			inline std::size_t
473		30x	win_scheduler::wait_one(long usec)
474			{
475	2/2 ✓ Branch 4 → 5 taken 13 times. ✓ Branch 4 → 7 taken 17 times.	60x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
476			{
477	1/1 ✓ Branch 5 → 6 taken 13 times.	13x	stop();
478		13x	return 0;
479			}
480
481		17x	iocp::thread_context_guard ctx(this);
482		17x	unsigned long timeout_ms = INFINITE;
483	1/2 ✓ Branch 8 → 9 taken 17 times. ✗ Branch 8 → 13 not taken.	17x	if (usec >= 0)
484			{
485		17x	auto ms = (static_cast<long long>(usec) + 999) / 1000;
486	1/2 ✓ Branch 9 → 10 taken 17 times. ✗ Branch 9 → 11 not taken.	17x	timeout_ms = ms >= 0xFFFFFFFELL ? static_cast<unsigned long>(0xFFFFFFFE)
487			: static_cast<unsigned long>(ms);
488			}
489	1/1 ✓ Branch 13 → 14 taken 17 times.	17x	return do_one(timeout_ms);
490		17x	}
491
492			inline std::size_t
493		17x	win_scheduler::poll()
494			{
495	2/2 ✓ Branch 4 → 5 taken 8 times. ✓ Branch 4 → 7 taken 9 times.	34x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
496			{
497	1/1 ✓ Branch 5 → 6 taken 8 times.	8x	stop();
498		8x	return 0;
499			}
500
501		9x	iocp::thread_context_guard ctx(this);
502
503		9x	std::size_t n = 0;
504	3/3 ✓ Branch 12 → 13 taken 27 times. ✓ Branch 13 → 9 taken 18 times. ✓ Branch 13 → 14 taken 9 times.	27x	while (do_one(0))
505	1/2 ✓ Branch 10 → 11 taken 18 times. ✗ Branch 10 → 12 not taken.	18x	if (n != (std::numeric_limits<std::size_t>::max)())
506		18x	++n;
507		9x	return n;
508		9x	}
509
510			inline std::size_t
511		5x	win_scheduler::poll_one()
512			{
513	2/2 ✓ Branch 4 → 5 taken 3 times. ✓ Branch 4 → 7 taken 2 times.	10x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
514			{
515	1/1 ✓ Branch 5 → 6 taken 3 times.	3x	stop();
516		3x	return 0;
517			}
518
519		2x	iocp::thread_context_guard ctx(this);
520	1/1 ✓ Branch 8 → 9 taken 2 times.	2x	return do_one(0);
521		2x	}
522
523			inline void
524		1350x	win_scheduler::post_deferred_completions(op_queue& ops)
525			{
526	1/2 ✗ Branch 3 → 4 not taken. ✓ Branch 3 → 15 taken 1350 times.	1350x	while (auto h = ops.pop())
527			{
528		✗	if (::PostQueuedCompletionStatus(
529			iocp_, 0, key_posted, reinterpret_cast<LPOVERLAPPED>(h)))
530		✗	continue;
531
532			// Out of resources, put the failed op and remaining ops back
533		✗	ops.push(h);
534		✗	std::lock_guard<win_mutex> lock(dispatch_mutex_);
535		✗	completed_ops_.splice(ops);
536		✗	::InterlockedExchange(&dispatch_required_, 1);
537		✗	return;
538		✗	}
539			}
540
541			inline std::size_t
542		656990x	win_scheduler::do_one(unsigned long timeout_ms)
543			{
544			for (;;)
545			{
546			// Check if we need to process timers or deferred ops
547	2/2 ✓ Branch 4 → 5 taken 1350 times. ✓ Branch 4 → 13 taken 655640 times.	1313980x	if (::InterlockedCompareExchange(&dispatch_required_, 0, 1) == 1)
548			{
549		1350x	op_queue local_ops;
550			{
551		1350x	std::lock_guard<win_mutex> lock(dispatch_mutex_);
552		1350x	local_ops.splice(completed_ops_);
553		1350x	}
554	1/1 ✓ Branch 8 → 9 taken 1350 times.	1350x	post_deferred_completions(local_ops);
555
556	1/2 ✓ Branch 9 → 10 taken 1350 times. ✗ Branch 9 → 11 not taken.	1350x	if (timer_svc_)
557	1/1 ✓ Branch 10 → 11 taken 1350 times.	1350x	timer_svc_->process_expired();
558
559	1/1 ✓ Branch 11 → 12 taken 1350 times.	1350x	update_timeout();
560			}
561
562		656990x	DWORD bytes = 0;
563		656990x	ULONG_PTR key = 0;
564		656990x	LPOVERLAPPED overlapped = nullptr;
565	1/1 ✓ Branch 13 → 14 taken 656990 times.	656990x	::SetLastError(0);
566
567	1/1 ✓ Branch 14 → 15 taken 656990 times.	656990x	BOOL result = ::GetQueuedCompletionStatus(
568			iocp_, &bytes, &key, &overlapped, timeout_ms);
569	1/1 ✓ Branch 15 → 16 taken 656990 times.	656990x	DWORD dwError = ::GetLastError();
570
571			// Handle based on completion key
572	2/2 ✓ Branch 16 → 17 taken 651041 times. ✓ Branch 16 → 47 taken 5949 times.	656990x	if (overlapped)
573			{
574	2/2 ✓ Branch 17 → 18 taken 2552 times. ✓ Branch 17 → 19 taken 648489 times.	651041x	DWORD err = result ? 0 : dwError;
575
576	3/4 ✓ Branch 20 → 21 taken 627796 times. ✓ Branch 20 → 40 taken 5429 times. ✓ Branch 20 → 43 taken 17816 times. ✗ Branch 20 → 46 not taken.	651041x	switch (key)
577			{
578		627796x	case key_io:
579			case key_result_stored:
580			{
581		627796x	auto* ov_op = overlapped_to_op(overlapped);
582
583			// key_io carries fresh kernel results — publish them before
584			// the CAS so that losing the race (old value 0) still leaves
585			// valid data for the on_pending() re-post. For key_result_stored
586			// the payload was already written and released (by the completer
587			// in on_pending, or by on_completion), so we must not overwrite
588			// it.
589	2/2 ✓ Branch 22 → 23 taken 627553 times. ✓ Branch 22 → 24 taken 243 times.	627796x	if (key == key_io)
590		627553x	ov_op->store_result(bytes, err);
591
592			// If old value was 1 the initiator already returned — dispatch.
593			// The acquire pairs with the publisher's release so the payload
594			// reads in complete() are ordered after the store that produced
595			// them. If old value was 0 the initiator hasn't returned yet;
596			// skip and let on_pending() re-post.
597		627796x	long expected = 0;
598	1/2 ✓ Branch 33 → 34 taken 627796 times. ✗ Branch 33 → 39 not taken.	1255592x	if (!ov_op->ready_.compare_exchange_strong(
599			expected, 1,
600			std::memory_order_acq_rel,
601			std::memory_order_acquire))
602			{
603		627796x	::InterlockedDecrement(&pending_io_);
604		627796x	ov_op->complete(
605	1/1 ✓ Branch 36 → 37 taken 627796 times.	627796x	this, ov_op->bytes_transferred, ov_op->dwError);
606		627796x	work_finished();
607		627796x	return 1;
608			}
609		✗	continue;
610		✗	}
611
612		5429x	case key_posted:
613			{
614			// Posted scheduler_op: overlapped is actually a scheduler_op
615		5429x	auto* op = reinterpret_cast<scheduler_op*>(overlapped);
616	1/1 ✓ Branch 40 → 41 taken 5429 times.	5429x	op->complete(this, bytes, err);
617		5429x	work_finished();
618		5429x	return 1;
619			}
620
621		17816x	case key_continuation:
622			{
623			// Posted continuation: overlapped is actually a continuation*
624		17816x	auto* c = reinterpret_cast<capy::continuation*>(overlapped);
625	1/1 ✓ Branch 43 → 44 taken 17816 times.	17816x	c->h.resume();
626		17816x	work_finished();
627		17816x	return 1;
628			}
629
630		✗	default:
631		✗	continue;
632		✗	}
633			}
634
635			// Signal completions (no OVERLAPPED)
636	2/2 ✓ Branch 47 → 48 taken 5937 times. ✓ Branch 47 → 61 taken 12 times.	5949x	if (result)
637			{
638	2/3 ✓ Branch 48 → 49 taken 1350 times. ✓ Branch 48 → 50 taken 4587 times. ✗ Branch 48 → 60 not taken.	5937x	switch (key)
639			{
640		1350x	case key_wake_dispatch:
641			// Timer wakeup - loop to check dispatch_required_
642		1350x	continue;
643
644		4587x	case key_shutdown:
645		4587x	::InterlockedExchange(&stop_event_posted_, 0);
646	2/2 ✓ Branch 53 → 54 taken 38 times. ✓ Branch 53 → 59 taken 4549 times.	4587x	if (stopped())
647			{
648			// Re-post for other waiting threads
649	1/2 ✓ Branch 56 → 57 taken 38 times. ✗ Branch 56 → 58 not taken.	76x	if (::InterlockedExchange(&stop_event_posted_, 1) == 0)
650			{
651	1/1 ✓ Branch 57 → 58 taken 38 times.	38x	::PostQueuedCompletionStatus(
652			iocp_, 0, key_shutdown, nullptr);
653			}
654		38x	return 0;
655			}
656		4549x	continue;
657
658		✗	default:
659		✗	continue;
660			}
661			}
662
663			// Timeout or error. INFINITE never times out, so a WAIT_TIMEOUT
664			// can only be a finite caller timeout (wait_one/poll) elapsing
665			// with no work. run()/run_one() pass INFINITE and block until a
666			// real wake-up; they exit only when stop() posts key_shutdown
667			// (a failed post is fatal in stop()).
668	1/2 ✗ Branch 61 → 62 not taken. ✓ Branch 61 → 64 taken 12 times.	12x	if (dwError != WAIT_TIMEOUT)
669		✗	detail::throw_system_error(make_err(dwError));
670		12x	return 0;
671		5899x	}
672			}
673
674			inline void
675		3346x	win_scheduler::on_timer_changed(void* ctx)
676			{
677		3346x	static_cast<win_scheduler*>(ctx)->update_timeout();
678		3346x	}
679
680			inline void
681		1121x	win_scheduler::set_timer_service(timer_service* svc)
682			{
683		1121x	timer_svc_ = svc;
684			// Pass 'this' as context - callback routes to correct instance
685		1121x	svc->set_on_earliest_changed(
686		1121x	timer_service::callback{this, &on_timer_changed});
687	1/2 ✓ Branch 5 → 6 taken 1121 times. ✗ Branch 5 → 8 not taken.	1121x	if (timers_)
688		1121x	timers_->start();
689		1121x	}
690
691			inline void
692		4696x	win_scheduler::update_timeout()
693			{
694	3/6 ✓ Branch 2 → 3 taken 4696 times. ✗ Branch 2 → 6 not taken. ✓ Branch 4 → 5 taken 4696 times. ✗ Branch 4 → 6 not taken. ✓ Branch 7 → 8 taken 4696 times. ✗ Branch 7 → 11 not taken.	4696x	if (timer_svc_ && timers_)
695		4696x	timers_->update_timeout(timer_svc_->nearest_expiry());
696		4696x	}
697
698			} // namespace boost::corosio::detail
699
700			// Defer including the auxiliary wait reactor until the scheduler is
701			// fully defined, since the reactor's inline methods call back into
702			// win_scheduler. This also gives the ctor, dtor and wait_reactor()
703			// below a complete win_wait_reactor type for unique_ptr destruction
704			// and lazy construction.
705			//
706			// The macro lets win_wait_reactor.hpp diagnose direct inclusion
707			// (which would land it here with win_scheduler still incomplete).
708			#define BOOST_COROSIO_DETAIL_IOCP_WIN_SCHEDULER_BODY_DONE
709			#include <boost/corosio/native/detail/iocp/win_wait_reactor.hpp>
710
711			namespace boost::corosio::detail {
712
713		1121x	inline win_scheduler::win_scheduler(
714		1121x	capy::execution_context& ctx, int concurrency_hint)
715		1121x	: iocp_(nullptr)
716		1121x	, outstanding_work_(0)
717		1121x	, stopped_(0)
718		1121x	, stop_event_posted_(0)
719	1/1 ✓ Branch 4 → 5 taken 1121 times.	1121x	, dispatch_required_(0)
720			{
721			// concurrency_hint < 0 means use system default (DWORD(~0) = max)
722	2/3 ✓ Branch 10 → 11 taken 1121 times. ✗ Branch 10 → 12 not taken. ✓ Branch 13 → 14 taken 1121 times.	1121x	iocp_ = ::CreateIoCompletionPort(
723			INVALID_HANDLE_VALUE, nullptr, 0,
724			static_cast<DWORD>(
725			concurrency_hint >= 0 ? concurrency_hint : DWORD(~0)));
726
727	1/2 ✗ Branch 14 → 15 not taken. ✓ Branch 14 → 18 taken 1121 times.	1121x	if (iocp_ == nullptr)
728		✗	detail::throw_system_error(make_err(::GetLastError()));
729
730			// Create timer wakeup mechanism (tries NT native, falls back to thread)
731	1/1 ✓ Branch 18 → 19 taken 1121 times.	1121x	timers_ = make_win_timers(iocp_, &dispatch_required_);
732
733			// Connect timer service to scheduler
734	2/2 ✓ Branch 21 → 22 taken 1121 times. ✓ Branch 22 → 23 taken 1121 times.	1121x	set_timer_service(&get_timer_service(ctx, *this));
735
736			// Initialize resolver service
737	1/1 ✓ Branch 23 → 24 taken 1121 times.	1121x	ctx.make_service<win_resolver_service>(*this);
738		1121x	}
739
740			inline void
741		1121x	win_scheduler::shutdown()
742			{
743	1/2 ✓ Branch 3 → 4 taken 1121 times. ✗ Branch 3 → 6 not taken.	1121x	if (timers_)
744		1121x	timers_->stop();
745
746			// Drain timer heap before the work-counting loop. The timer_service
747			// was registered after this scheduler (nested make_service from our
748			// constructor), so execution_context::shutdown() calls us first.
749			// Asio avoids this by owning timer queues directly inside the
750			// scheduler; we bridge the gap by shutting down the timer service
751			// early. The subsequent call from execution_context is a no-op.
752	1/2 ✓ Branch 6 → 7 taken 1121 times. ✗ Branch 6 → 8 not taken.	1121x	if (timer_svc_)
753		1121x	timer_svc_->shutdown();
754
755			// Same problem for the auxiliary wait reactor: ops parked in it
756			// owe completion packets. Stop the reactor early so its loop
757			// posts them as cancelled and the pending count can reach zero.
758	2/2 ✓ Branch 9 → 10 taken 24 times. ✓ Branch 9 → 12 taken 1097 times.	1121x	if (wait_reactor_ready_.load(std::memory_order_acquire))
759		24x	wait_reactor_->stop();
760
761			// Reap every packet still owed to the port before the services
762			// free the op memory those packets reference. Work-guard credits,
763			// posted handlers, and queued continuations have no bearing here.
764	2/2 ✓ Branch 36 → 13 taken 4 times. ✓ Branch 36 → 37 taken 1121 times.	2250x	while (::InterlockedExchangeAdd(&pending_io_, 0) > 0)
765			{
766		4x	op_queue ops;
767			{
768		4x	std::lock_guard<win_mutex> lock(dispatch_mutex_);
769		4x	ops.splice(completed_ops_);
770		4x	}
771
772			// Deferred-queue entries are process-owned (failed-post
773			// fallbacks and posted handlers); no packet references them.
774	1/2 ✗ Branch 18 → 19 not taken. ✓ Branch 18 → 21 taken 4 times.	4x	while (auto* h = ops.pop())
775		✗	h->destroy();
776
777			DWORD bytes;
778			ULONG_PTR key;
779			LPOVERLAPPED overlapped;
780	1/1 ✓ Branch 21 → 22 taken 4 times.	4x	::GetQueuedCompletionStatus(
781			iocp_, &bytes, &key, &overlapped,
782			iocp::shutdown_drain_timeout_ms);
783	1/2 ✓ Branch 22 → 23 taken 4 times. ✗ Branch 22 → 33 not taken.	4x	if (overlapped)
784			{
785	1/2 ✗ Branch 23 → 24 not taken. ✓ Branch 23 → 25 taken 4 times.	4x	if (key == key_posted)
786			{
787		✗	auto* op = reinterpret_cast<scheduler_op*>(overlapped);
788		✗	op->destroy();
789			}
790	1/2 ✗ Branch 25 → 26 not taken. ✓ Branch 25 → 29 taken 4 times.	4x	else if (key == key_continuation)
791			{
792			// Drain without resuming: destroy the parked frame.
793		✗	auto* c = reinterpret_cast<capy::continuation*>(overlapped);
794		✗	if (c->h)
795		✗	c->h.destroy();
796			}
797			else
798			{
799		4x	::InterlockedDecrement(&pending_io_);
800		4x	auto* op = overlapped_to_op(overlapped);
801	1/1 ✓ Branch 32 → 33 taken 4 times.	4x	op->destroy();
802			}
803			}
804			}
805
806			// Final sweep: packets can sit in the port or deferred queue
807			// without a pending_io_ count (posted directly against the
808			// handle). Destroy them so service teardown does not free state
809			// they still reference.
810			for (;;)
811			{
812		1134x	op_queue ops;
813			{
814		1134x	std::lock_guard<win_mutex> lock(dispatch_mutex_);
815		1134x	ops.splice(completed_ops_);
816		1134x	}
817	1/2 ✗ Branch 42 → 43 not taken. ✓ Branch 42 → 45 taken 1134 times.	1134x	while (auto* h = ops.pop())
818		✗	h->destroy();
819
820			DWORD bytes;
821			ULONG_PTR key;
822			LPOVERLAPPED overlapped;
823	1/1 ✓ Branch 45 → 46 taken 1134 times.	1134x	::GetQueuedCompletionStatus(iocp_, &bytes, &key, &overlapped, 0);
824	2/2 ✓ Branch 46 → 47 taken 1121 times. ✓ Branch 46 → 48 taken 13 times.	1134x	if (!overlapped)
825		1121x	break;
826	2/2 ✓ Branch 48 → 49 taken 7 times. ✓ Branch 48 → 50 taken 6 times.	13x	if (key == key_posted)
827			{
828	1/1 ✓ Branch 49 → 56 taken 7 times.	7x	reinterpret_cast<scheduler_op*>(overlapped)->destroy();
829			}
830	2/2 ✓ Branch 50 → 51 taken 2 times. ✓ Branch 50 → 54 taken 4 times.	6x	else if (key == key_continuation)
831			{
832		2x	auto* c = reinterpret_cast<capy::continuation*>(overlapped);
833	1/2 ✓ Branch 52 → 53 taken 2 times. ✗ Branch 52 → 56 not taken.	2x	if (c->h)
834	1/1 ✓ Branch 53 → 56 taken 2 times.	2x	c->h.destroy();
835			}
836			else
837			{
838	1/1 ✓ Branch 55 → 56 taken 4 times.	4x	overlapped_to_op(overlapped)->destroy();
839			}
840		13x	}
841		1121x	}
842
843		2242x	inline win_scheduler::~win_scheduler()
844			{
845	2/2 ✓ Branch 3 → 4 taken 24 times. ✓ Branch 3 → 6 taken 1097 times.	1121x	if (wait_reactor_)
846		24x	wait_reactor_->stop();
847		1121x	wait_reactor_.reset();
848
849	1/2 ✓ Branch 7 → 8 taken 1121 times. ✗ Branch 7 → 9 not taken.	1121x	if (iocp_ != nullptr)
850		1121x	::CloseHandle(iocp_);
851		2242x	}
852
853			inline win_wait_reactor&
854		25x	win_scheduler::wait_reactor()
855			{
856			// Lazy thread-safe init: multiple IOCP workers may race the first
857			// wait() call. wait_reactor_ready_ is set with release ordering
858			// after construction so cancel_wait_if_constructed can safely
859			// observe the reactor without forcing construction itself.
860	1/1 ✓ Branch 2 → 3 taken 25 times.	25x	std::call_once(wait_reactor_once_, [this] {
861	1/1 ✓ Branch 2 → 3 taken 24 times.	24x	wait_reactor_ = std::make_unique<win_wait_reactor>(*this);
862		24x	wait_reactor_ready_.store(true, std::memory_order_release);
863		24x	});
864		25x	return *wait_reactor_;
865			}
866
867			inline void
868		34426x	win_scheduler::cancel_wait_if_constructed(overlapped_op* op) noexcept
869			{
870	2/2 ✓ Branch 3 → 4 taken 159 times. ✓ Branch 3 → 6 taken 34267 times.	34426x	if (wait_reactor_ready_.load(std::memory_order_acquire))
871		159x	wait_reactor_->cancel_wait(op);
872		34426x	}
873
874			} // namespace boost::corosio::detail
875
876			#endif // BOOST_COROSIO_HAS_IOCP
877
878			#endif // BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_SCHEDULER_HPP
879