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

96.1% Lines (371 / 386, 5 excl) 78.6% Branches (221 / 281) 100.0% Functions (41 / 41)

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

win_scheduler.hpp

f(x) Functions (41)

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/capy/ex/abandon.hpp>
28			#include <boost/corosio/native/detail/iocp/win_completion_key.hpp>
29			#include <boost/corosio/native/detail/iocp/win_mutex.hpp>
30
31			#include <boost/corosio/native/detail/iocp/win_overlapped_op.hpp>
32			#include <boost/corosio/native/detail/iocp/win_timers.hpp>
33			#include <boost/corosio/detail/timer_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 <exception>
42			#include <limits>
43			#include <memory>
44			#include <mutex>
45
46			#include <boost/corosio/native/detail/iocp/win_windows.hpp>
47
48			namespace boost::corosio::detail {
49
50			// Forward declarations
51			struct overlapped_op;
52			class win_timers;
53			class win_wait_reactor;
54
55			class BOOST_COROSIO_DECL win_scheduler final
56			: public scheduler
57			{
58			public:
59
60			win_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
61			~win_scheduler();
62			win_scheduler(win_scheduler const&) = delete;
63			win_scheduler& operator=(win_scheduler const&) = delete;
64
65			void shutdown() override;
66			void post(std::coroutine_handle<> h) const override;
67			void post(scheduler_op* h) const override;
68			void post(capy::continuation&) const override;
69			bool running_in_this_thread() const noexcept override;
70			void stop() override;
71			bool stopped() const noexcept override;
72			void restart() override;
73			std::size_t run() override;
74			std::size_t run_one() override;
75			std::size_t wait_one(long usec) override;
76			std::size_t poll() override;
77			std::size_t poll_one() override;
78
79		1853x	void* native_handle() const noexcept
80			{
81		1853x	return iocp_;
82			}
83
84			void work_started() noexcept override;
85			void work_finished() noexcept override;
86
87			// IOCP has only the dispatch mutex; reactor_io_locking and one_thread do
88			// not apply (the completion port provides its own synchronization).
89		5224x	void configure_threading(threading_config cfg) noexcept override
90			{
91		5224x	scheduler_locking_disabled_ = !cfg.scheduler_locking;
92		5224x	dispatch_mutex_.set_enabled(cfg.scheduler_locking);
93		5224x	}
94
95			/// Return true when scheduler locking is disabled (fully-lockless tier).
96		2825x	bool scheduler_locking_disabled() const noexcept override
97			{
98			// LCOV_EXCL_START: consulted only by the POSIX pool-backed
99			// services; the IOCP services do not read it yet.
100		−	return scheduler_locking_disabled_;
101			// LCOV_EXCL_STOP
102			}
103
104			/** Signal that an overlapped I/O operation is now pending.
105			Coordinates with do_one() via the ready_ CAS protocol. */
106			void on_pending(overlapped_op* op) const;
107
108			/** Post an immediate completion with pre-stored results.
109			Used for sync errors and noop paths. */
110			void on_completion(overlapped_op* op, DWORD error, DWORD bytes) const;
111
112			// Timer service integration
113			void update_timeout();
114
115			private:
116			void on_timer_service_attached(timer_service& svc) override;
117
118			/// Destroy every packet and deferred entry still queued, without
119			/// running them; stops once the port is empty.
120			void sweep_port() noexcept;
121
122			// Posts an allocated copy, which survives the caller and keeps the
123			// owner for teardown.
124			void post_copy(capy::continuation const& c) const;
125
126			static void on_timer_changed(void* ctx);
127			static unsigned long
128			timer_timeout_ms(timer_service& svc, unsigned long timeout_ms) noexcept;
129			void post_deferred_completions(op_queue& ops);
130
131			// Set stopped_ and post the shutdown packet; returns the Win32
132			// error of a failed post, or 0.
133			unsigned long post_stop() noexcept;
134
135			// Post the shutdown packet unless one is in flight; returns the
136			// Win32 error of a failed post, or 0.
137			unsigned long post_wakeup() noexcept;
138
139			// Retry a recorded wakeup that could not be queued, throwing if
140			// the retry fails too.
141			void report_stop_failure();
142
143			// Record a failed wakeup for the next run or poll call to retry,
144			// then throw it.
145			[[noreturn]] void throw_wakeup_failure(unsigned long err);
146			std::size_t do_one(unsigned long timeout_ms);
147			std::size_t do_one_wait(unsigned long timeout_ms, bool& took_wake);
148
149			void* iocp_;
150			mutable long outstanding_work_;
151
152			// Packets in flight to the completion port that reference
153			// overlapped-op memory: kernel completions owed after a pending
154			// submission, plus successful stored-result posts. Shutdown reaps
155			// until this is zero before the services free op storage. The
156			// run-loop counter cannot serve that role: frames destroyed at
157			// teardown never return their work-guard credits.
158			mutable long pending_io_ = 0;
159			mutable long stopped_;
160			long stop_event_posted_;
161			long stop_failure_ = 0;
162			mutable long dispatch_required_;
163			bool scheduler_locking_disabled_ = false;
164
165			BOOST_COROSIO_MSVC_WARNING_PUSH
166			BOOST_COROSIO_MSVC_WARNING_DISABLE(
167			4251) // std::/detail:: members, dll-interface
168			mutable win_mutex dispatch_mutex_;
169			mutable op_queue completed_ops_;
170			std::unique_ptr<win_timers> timers_;
171			std::unique_ptr<win_wait_reactor> wait_reactor_;
172			BOOST_COROSIO_MSVC_WARNING_POP
173
174			public:
175			/** Return the auxiliary select-based reactor for wait operations.
176
177			Built with the scheduler and stopped+joined in ~win_scheduler,
178			so it is always there to hand a wait to. Used by socket and
179			acceptor wait() implementations whose readiness cannot be
180			expressed natively in IOCP (datagram-read, acceptor-read,
181			error-wait).
182			*/
183			win_wait_reactor& wait_reactor();
184
185			/// Cancel a parked wait op. Safe to call from any thread.
186			void cancel_wait(overlapped_op* op) noexcept;
187			};
188
189			/*
190			ARCHITECTURE NOTE: Function Pointer Dispatch
191
192			All I/O handles are registered with the IOCP using key_io (0).
193			Dispatch happens via the function pointer stored in each scheduler_op.
194
195			When GQCS returns with an OVERLAPPED, we cast it to scheduler_op
196			and call the function pointer directly - no virtual dispatch.
197
198			The completion_key enum values are used only for internal signals:
199			- key_io (0): Normal I/O completion, dispatch via func_
200			- key_wake_dispatch (1): Timer wakeup, check dispatch_required_
201			- key_shutdown (2): Stop signal
202			- key_result_stored (3): Results pre-stored in OVERLAPPED
203			- key_posted: Carries a scheduler_op* in the OVERLAPPED pointer
204			- key_continuation: Carries a capy::continuation* in the OVERLAPPED pointer
205			*/
206
207			namespace iocp {
208
209			// Poll interval (ms) for the one-time shutdown drain loop. Unlike the
210			// steady-state loop (which blocks indefinitely until a real wake-up),
211			// the drain must keep re-checking the work count and the deferred
212			// completion queue to make progress, so it uses a finite wait.
213			inline constexpr unsigned long shutdown_drain_timeout_ms = 500;
214
215			struct BOOST_COROSIO_SYMBOL_VISIBLE scheduler_context
216			{
217			win_scheduler const* key;
218			scheduler_context* next;
219			};
220
221			inline thread_local_ptr<scheduler_context> context_stack;
222
223			struct thread_context_guard
224			{
225			scheduler_context frame_;
226			running_scheduler_guard running_;
227
228		17270x	explicit thread_context_guard(win_scheduler const* ctx) noexcept
229		17270x	: frame_{ctx, context_stack.get()}
230		17270x	, running_(ctx)
231			{
232		17270x	context_stack.set(&frame_);
233		17270x	}
234
235		17270x	~thread_context_guard() noexcept
236			{
237		17270x	context_stack.set(frame_.next);
238		17270x	}
239			};
240
241			} // namespace iocp
242
243			// The constructor, ~win_scheduler(), shutdown(), wait_reactor() and
244			// cancel_wait() are defined at the bottom of this header so the
245			// unique_ptr<win_wait_reactor>'s deleter, its operator* and
246			// wait_reactor_->stop() see the type complete. The constructor needs
247			// it to build the reactor, and its unwind path to destroy it.
248
249			inline void
250		1839x	win_scheduler::post(std::coroutine_handle<> h) const
251			{
252	1/1 ✓ Branch 2 → 3 taken 1839 times.	1839x	post_copy(capy::continuation{h});
253		1839x	}
254
255			inline void
256		1841x	win_scheduler::post_copy(capy::continuation const& c) const
257			{
258			struct post_handler final : scheduler_op
259			{
260			capy::continuation c_;
261
262		1841x	static void do_complete(
263			void* owner, scheduler_op* base, std::uint32_t, std::uint32_t)
264			{
265		1841x	auto* self = static_cast<post_handler*>(base);
266		1841x	auto c = self->c_;
267	1/2 ✓ Branch 2 → 3 taken 1841 times. ✗ Branch 2 → 4 not taken.	1841x	delete self;
268	2/2 ✓ Branch 4 → 5 taken 8 times. ✓ Branch 4 → 7 taken 1833 times.	1841x	if (!owner)
269			{
270		8x	capy::abandon(c);
271		8x	return;
272			}
273	1/1 ✓ Branch 7 → 8 taken 1833 times.	1833x	c.h.resume();
274			}
275
276		1841x	explicit post_handler(capy::continuation const& c)
277		1841x	: scheduler_op(&do_complete)
278		1841x	, c_(c)
279			{
280		1841x	}
281			};
282
283		1841x	auto* ph = new post_handler(c);
284		1841x	::InterlockedIncrement(&outstanding_work_);
285
286	2/2 ✓ Branch 7 → 8 taken 1 time. ✓ Branch 7 → 14 taken 1840 times.	1841x	if (!::PostQueuedCompletionStatus(
287		1841x	iocp_, 0, key_posted, reinterpret_cast<LPOVERLAPPED>(ph)))
288			{
289		1x	std::lock_guard<win_mutex> lock(dispatch_mutex_);
290		1x	completed_ops_.push(ph);
291		1x	::InterlockedExchange(&dispatch_required_, 1);
292		1x	}
293		1841x	}
294
295			inline void
296		2544x	win_scheduler::post(scheduler_op* h) const
297			{
298		2544x	::InterlockedIncrement(&outstanding_work_);
299
300	2/2 ✓ Branch 5 → 6 taken 1 time. ✓ Branch 5 → 12 taken 2543 times.	2544x	if (!::PostQueuedCompletionStatus(
301		2544x	iocp_, 0, key_posted, reinterpret_cast<LPOVERLAPPED>(h)))
302			{
303		1x	std::lock_guard<win_mutex> lock(dispatch_mutex_);
304		1x	completed_ops_.push(h);
305		1x	::InterlockedExchange(&dispatch_required_, 1);
306		1x	}
307		2544x	}
308
309			inline void
310		130853x	win_scheduler::post(capy::continuation& c) const
311			{
312		130853x	::InterlockedIncrement(&outstanding_work_);
313
314	2/2 ✓ Branch 5 → 6 taken 2 times. ✓ Branch 5 → 9 taken 130851 times.	130853x	if (!::PostQueuedCompletionStatus(
315		130853x	iocp_, 0, key_continuation, reinterpret_cast<LPOVERLAPPED>(&c)))
316			{
317			// completed_ops_ is an op_queue and cannot carry a raw continuation,
318			// so on the rare PQCS failure fall back to an allocated copy, which
319			// keeps the owner. Drop the increment first; post_copy counts its own.
320		2x	::InterlockedDecrement(&outstanding_work_);
321		2x	post_copy(c);
322			}
323		130853x	}
324
325			inline bool
326		24784x	win_scheduler::running_in_this_thread() const noexcept
327			{
328	2/2 ✓ Branch 6 → 3 taken 118 times. ✓ Branch 6 → 7 taken 24779 times.	24897x	for (auto* c = iocp::context_stack.get(); c != nullptr; c = c->next)
329	2/2 ✓ Branch 3 → 4 taken 5 times. ✓ Branch 3 → 5 taken 113 times.	118x	if (c->key == this)
330		5x	return true;
331		24779x	return false;
332			}
333
334			inline void
335		466300x	win_scheduler::work_started() noexcept
336			{
337		466300x	::InterlockedIncrement(&outstanding_work_);
338		466300x	}
339
340			inline void
341		595336x	win_scheduler::work_finished() noexcept
342			{
343	2/2 ✓ Branch 4 → 5 taken 585568 times. ✓ Branch 4 → 6 taken 9768 times.	1190672x	if (::InterlockedDecrement(&outstanding_work_) != 0)
344		585568x	return;
345		9768x	auto err = post_stop();
346	2/2 ✓ Branch 7 → 8 taken 9763 times. ✓ Branch 7 → 9 taken 5 times.	9768x	if (!err)
347		9763x	return;
348			// This path cannot throw. A run call on this thread retries the
349			// wakeup on its way out; on any other thread nothing would ever
350			// wake the threads blocked without a timeout.
351	1/2 ✗ Branch 10 → 11 not taken. ✓ Branch 10 → 12 taken 5 times.	5x	if (!running_in_this_thread())
352		✗	std::terminate();
353		5x	::InterlockedExchange(&stop_failure_, static_cast<long>(err));
354			}
355
356			inline void
357		428764x	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			//
364			// pending_io_ counts the packet that will dispatch this op: on CAS
365			// success the kernel's own completion will find ready_ == 1 and
366			// dispatch; on CAS failure the kernel's packet was consumed as a
367			// skip (uncounted) and the re-post is the dispatching packet. A
368			// failed re-post falls back to the deferred queue, which holds the
369			// op memory itself — no packet, no count.
370			//
371			// Counted before the packet can be consumed: once it can, a context
372			// whose thread runs it may finish and be destroyed, so this call
373			// touches nothing of the scheduler afterwards.
374		428764x	::InterlockedIncrement(&pending_io_);
375		428764x	long expected = 0;
376	2/2 ✓ Branch 13 → 14 taken 428595 times. ✓ Branch 13 → 15 taken 169 times.	857528x	if (op->ready_.compare_exchange_strong(
377			expected, 1, std::memory_order_acq_rel, std::memory_order_acquire))
378		428764x	return;
379	2/4 ✓ Branch 15 → 16 taken 169 times. ✗ Branch 15 → 17 not taken. ✓ Branch 19 → 20 taken 169 times. ✗ Branch 19 → 21 not taken.	338x	if (::PostQueuedCompletionStatus(
380	1/1 ✓ Branch 18 → 19 taken 169 times.	169x	iocp_, 0, key_result_stored, static_cast<LPOVERLAPPED>(op)))
381		169x	return;
382		✗	std::lock_guard<win_mutex> lock(dispatch_mutex_);
383		✗	completed_ops_.push(op);
384		✗	::InterlockedExchange(&dispatch_required_, 1);
385			// Uncounted only once queued, so a shutdown that sees no packet
386			// left finds the op in the queue.
387		✗	::InterlockedDecrement(&pending_io_);
388		✗	}
389
390			inline void
391		3615x	win_scheduler::on_completion(overlapped_op* op, DWORD error, DWORD bytes) const
392			{
393			// Synchronous-completion path. Write the payload before the release store
394			// to ready_ so the GQCS thread that dequeues the key_result_stored post
395			// sees both fields once it observes ready_ == 1.
396		3615x	op->dwError = error;
397		3615x	op->bytes_transferred = bytes;
398		3615x	op->ready_.store(1, std::memory_order_release);
399
400			// Counted before the packet can be consumed, as in on_pending.
401		3615x	::InterlockedIncrement(&pending_io_);
402	3/4 ✓ Branch 24 → 25 taken 3615 times. ✗ Branch 24 → 26 not taken. ✓ Branch 28 → 29 taken 3614 times. ✓ Branch 28 → 30 taken 1 time.	7230x	if (::PostQueuedCompletionStatus(
403	1/1 ✓ Branch 27 → 28 taken 3615 times.	3615x	iocp_, 0, key_result_stored, static_cast<LPOVERLAPPED>(op)))
404		3614x	return;
405		1x	std::lock_guard<win_mutex> lock(dispatch_mutex_);
406		1x	completed_ops_.push(op);
407		1x	::InterlockedExchange(&dispatch_required_, 1);
408			// Uncounted only once queued, so a shutdown that sees no packet
409			// left finds the op in the queue.
410		1x	::InterlockedDecrement(&pending_io_);
411		1x	}
412
413			inline unsigned long
414		11483097x	win_scheduler::post_stop() noexcept
415			{
416		11483097x	::InterlockedExchange(&stopped_, 1);
417		11483097x	return post_wakeup();
418			}
419
420			inline unsigned long
421		11489217x	win_scheduler::post_wakeup() noexcept
422			{
423	2/2 ✓ Branch 4 → 5 taken 11473738 times. ✓ Branch 4 → 6 taken 15479 times.	22978434x	if (::InterlockedExchange(&stop_event_posted_, 1) != 0)
424		11473738x	return 0;
425	2/2 ✓ Branch 7 → 8 taken 15464 times. ✓ Branch 7 → 9 taken 15 times.	15479x	if (::PostQueuedCompletionStatus(iocp_, 0, key_shutdown, nullptr))
426		15464x	return 0;
427			// A zero last error would read as a wakeup that was queued.
428		15x	auto err = ::GetLastError();
429	2/2 ✓ Branch 10 → 11 taken 1 time. ✓ Branch 10 → 12 taken 14 times.	15x	if (err == 0)
430		1x	err = ERROR_INVALID_PARAMETER;
431			// No packet is in flight, so a later stop retries the wakeup
432			// rather than leaving blocked threads asleep for good.
433		15x	::InterlockedExchange(&stop_event_posted_, 0);
434		15x	return err;
435			}
436
437			inline void
438		11473329x	win_scheduler::stop()
439			{
440			// The shutdown post is the only thing that wakes a run()/run_one()
441			// thread blocked indefinitely in GQCS. With no periodic timeout
442			// there is no fallback, so a failed post is reported rather than
443			// ignored. It can only fail under resource exhaustion.
444	2/2 ✓ Branch 3 → 4 taken 7 times. ✓ Branch 3 → 5 taken 11473322 times.	11473329x	if (auto err = post_stop())
445		7x	throw_wakeup_failure(err);
446		11473322x	}
447
448			inline void
449		10x	win_scheduler::throw_wakeup_failure(unsigned long err)
450			{
451			// Kept recorded, so a later run or poll call on the stopped context
452			// retries the wakeup instead of blocking while work remains.
453		10x	::InterlockedExchange(&stop_failure_, static_cast<long>(err));
454		10x	detail::throw_system_error(make_err(err));
455			}
456
457			inline void
458		11499109x	win_scheduler::report_stop_failure()
459			{
460	5/6 ✓ Branch 4 → 5 taken 7 times. ✓ Branch 4 → 7 taken 11499102 times. ✗ Branch 6 → 7 not taken. ✓ Branch 6 → 8 taken 7 times. ✓ Branch 9 → 10 taken 11499102 times. ✓ Branch 9 → 11 taken 7 times.	22998218x	if (::InterlockedExchange(&stop_failure_, 0) == 0 \|\| !stopped())
461		11499102x	return;
462			// Other threads may still be blocked waiting for the lost wakeup,
463			// so retry it and report only a retry that fails too.
464	2/2 ✓ Branch 12 → 13 taken 3 times. ✓ Branch 12 → 14 taken 4 times.	7x	if (auto err = post_wakeup())
465		3x	throw_wakeup_failure(err);
466			}
467
468			inline bool
469		11481514x	win_scheduler::stopped() const noexcept
470			{
471			// equivalent to atomic read
472		22963028x	return ::InterlockedExchangeAdd(&stopped_, 0) != 0;
473			}
474
475			inline void
476		5575x	win_scheduler::restart()
477			{
478		5575x	::InterlockedExchange(&stopped_, 0);
479		5575x	::InterlockedExchange(&stop_event_posted_, 0);
480		5575x	::InterlockedExchange(&stop_failure_, 0);
481		5575x	}
482
483			inline std::size_t
484		9000x	win_scheduler::run()
485			{
486	1/1 ✓ Branch 2 → 3 taken 8999 times.	9000x	report_stop_failure();
487	2/2 ✓ Branch 5 → 6 taken 135 times. ✓ Branch 5 → 8 taken 8864 times.	17998x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
488			{
489	1/1 ✓ Branch 6 → 7 taken 135 times.	135x	stop();
490		135x	return 0;
491			}
492
493		8864x	iocp::thread_context_guard ctx(this);
494
495		8864x	std::size_t n = 0;
496			for (;;)
497			{
498	3/3 ✓ Branch 10 → 11 taken 557368 times. ✓ Branch 11 → 12 taken 160 times. ✓ Branch 11 → 13 taken 557208 times.	557369x	if (!do_one(INFINITE))
499		160x	break;
500	1/2 ✓ Branch 14 → 15 taken 557208 times. ✗ Branch 14 → 16 not taken.	557208x	if (n != (std::numeric_limits<std::size_t>::max)())
501		557208x	++n;
502	2/2 ✓ Branch 18 → 19 taken 8703 times. ✓ Branch 18 → 21 taken 548505 times.	1114416x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
503			{
504	1/1 ✓ Branch 19 → 20 taken 8702 times.	8703x	stop();
505		8702x	break;
506			}
507			}
508	1/1 ✓ Branch 22 → 23 taken 8862 times.	8862x	report_stop_failure();
509		8862x	return n;
510		8864x	}
511
512			inline std::size_t
513		74x	win_scheduler::run_one()
514			{
515	1/1 ✓ Branch 2 → 3 taken 74 times.	74x	report_stop_failure();
516	2/2 ✓ Branch 5 → 6 taken 1 time. ✓ Branch 5 → 8 taken 73 times.	148x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
517			{
518	1/1 ✓ Branch 6 → 7 taken 1 time.	1x	stop();
519		1x	return 0;
520			}
521
522		73x	iocp::thread_context_guard ctx(this);
523	1/1 ✓ Branch 9 → 10 taken 73 times.	73x	auto n = do_one(INFINITE);
524	1/1 ✓ Branch 10 → 11 taken 72 times.	73x	report_stop_failure();
525		72x	return n;
526		73x	}
527
528			inline std::size_t
529		11472182x	win_scheduler::wait_one(long usec)
530			{
531	1/1 ✓ Branch 2 → 3 taken 11472182 times.	11472182x	report_stop_failure();
532	2/2 ✓ Branch 5 → 6 taken 11464421 times. ✓ Branch 5 → 8 taken 7761 times.	22944364x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
533			{
534	1/1 ✓ Branch 6 → 7 taken 11464421 times.	11464421x	stop();
535		11464421x	return 0;
536			}
537
538		7761x	iocp::thread_context_guard ctx(this);
539		7761x	unsigned long timeout_ms = INFINITE;
540	1/2 ✓ Branch 9 → 10 taken 7761 times. ✗ Branch 9 → 14 not taken.	7761x	if (usec >= 0)
541			{
542		7761x	auto ms = (static_cast<long long>(usec) + 999) / 1000;
543	1/2 ✓ Branch 10 → 11 taken 7761 times. ✗ Branch 10 → 12 not taken.	7761x	timeout_ms = ms >= 0xFFFFFFFELL ? static_cast<unsigned long>(0xFFFFFFFE)
544			: static_cast<unsigned long>(ms);
545			}
546	1/1 ✓ Branch 14 → 15 taken 7761 times.	7761x	auto n = do_one(timeout_ms);
547	1/1 ✓ Branch 15 → 16 taken 7761 times.	7761x	report_stop_failure();
548		7761x	return n;
549		7761x	}
550
551			inline std::size_t
552		557x	win_scheduler::poll()
553			{
554	1/1 ✓ Branch 2 → 3 taken 557 times.	557x	report_stop_failure();
555	2/2 ✓ Branch 5 → 6 taken 10 times. ✓ Branch 5 → 8 taken 547 times.	1114x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
556			{
557	1/1 ✓ Branch 6 → 7 taken 10 times.	10x	stop();
558		10x	return 0;
559			}
560
561		547x	iocp::thread_context_guard ctx(this);
562
563		547x	std::size_t n = 0;
564	3/5 ✓ Branch 13 → 14 taken 4443 times. ✓ Branch 14 → 10 taken 3896 times. ✓ Branch 14 → 15 taken 547 times. ✗ Branch 15 → 10 not taken. ✗ Branch 15 → 16 not taken.	4443x	while (do_one(0))
565	1/2 ✓ Branch 11 → 12 taken 3896 times. ✗ Branch 11 → 13 not taken.	3896x	if (n != (std::numeric_limits<std::size_t>::max)())
566		3896x	++n;
567	1/2 ✓ Branch 15 → 16 taken 547 times. ✗ Branch 16 → 17 not taken.	547x	report_stop_failure();
568		547x	return n;
569		547x	}
570
571			inline std::size_t
572		28x	win_scheduler::poll_one()
573			{
574	1/1 ✓ Branch 2 → 3 taken 28 times.	28x	report_stop_failure();
575	2/2 ✓ Branch 5 → 6 taken 3 times. ✓ Branch 5 → 8 taken 25 times.	56x	if (::InterlockedExchangeAdd(&outstanding_work_, 0) == 0)
576			{
577	1/1 ✓ Branch 6 → 7 taken 3 times.	3x	stop();
578		3x	return 0;
579			}
580
581		25x	iocp::thread_context_guard ctx(this);
582	1/1 ✓ Branch 9 → 10 taken 25 times.	25x	auto n = do_one(0);
583	1/1 ✓ Branch 10 → 11 taken 24 times.	25x	report_stop_failure();
584		24x	return n;
585		25x	}
586
587			inline void
588		411x	win_scheduler::post_deferred_completions(op_queue& ops)
589			{
590	3/4 ✓ Branch 3 → 4 taken 4 times. ✓ Branch 3 → 15 taken 390 times. ✗ Branch 4 → 5 not taken. ✓ Branch 4 → 16 taken 17 times.	411x	while (auto h = ops.pop())
591			{
592	3/5 ✓ Branch 4 → 5 taken 4 times. ✓ Branch 5 → 6 taken 3 times. ✓ Branch 5 → 7 taken 1 time. ✗ Branch 6 → 7 not taken. ✗ Branch 6 → 8 not taken.	4x	if (::PostQueuedCompletionStatus(
593			iocp_, 0, key_posted, reinterpret_cast<LPOVERLAPPED>(h)))
594		3x	continue;
595
596			// Out of resources, put the failed op and remaining ops back
597		1x	ops.push(h);
598		1x	std::lock_guard<win_mutex> lock(dispatch_mutex_);
599		1x	completed_ops_.splice(ops);
600		1x	::InterlockedExchange(&dispatch_required_, 1);
601		1x	return;
602		4x	}
603			}
604
605			inline std::size_t
606		569671x	win_scheduler::do_one(unsigned long timeout_ms)
607			{
608		569671x	bool took_wake = false;
609	1/1 ✓ Branch 2 → 3 taken 569670 times.	569671x	std::size_t n = do_one_wait(timeout_ms, took_wake);
610			// Without the timer thread, the wake is all that tells a thread
611			// blocked with no timeout about a new earliest timer; leaving
612			// with it would strand those threads.
613	6/6 ✓ Branch 3 → 4 taken 407 times. ✓ Branch 3 → 7 taken 569263 times. ✓ Branch 5 → 6 taken 17 times. ✓ Branch 5 → 7 taken 390 times. ✓ Branch 8 → 9 taken 17 times. ✓ Branch 8 → 19 taken 569653 times.	569670x	if (took_wake && !timers_)
614			{
615		17x	auto* tsvc = timers();
616	1/2 ✓ Branch 10 → 11 taken 17 times. ✗ Branch 10 → 16 not taken.	34x	if (tsvc &&
617	3/5 ✓ Branch 13 → 14 taken 17 times. ✗ Branch 14 → 15 not taken. ✓ Branch 14 → 16 taken 17 times. ✗ Branch 17 → 18 not taken. ✓ Branch 17 → 19 taken 17 times.	34x	tsvc->nearest_expiry() != (timer_service::time_point::max)())
618	0/1 ✗ Branch 18 → 19 not taken.	✗	::PostQueuedCompletionStatus(iocp_, 0, key_wake_dispatch, nullptr);
619			}
620		569670x	return n;
621			}
622
623			inline std::size_t
624		575197x	win_scheduler::do_one_wait(unsigned long timeout_ms, bool& took_wake)
625			{
626			for (;;)
627			{
628			// Check if we need to process timers or deferred ops
629	4/4 ✓ Branch 4 → 5 taken 391 times. ✓ Branch 4 → 14 taken 574307 times. ✓ Branch 5 → 6 taken 17 times. ✓ Branch 5 → 15 taken 551 times.	1150532x	if (::InterlockedCompareExchange(&dispatch_required_, 0, 1) == 1)
630			{
631		408x	op_queue local_ops;
632			{
633		408x	std::lock_guard<win_mutex> lock(dispatch_mutex_);
634		408x	local_ops.splice(completed_ops_);
635		408x	}
636	2/2 ✓ Branch 8 → 9 taken 391 times. ✓ Branch 9 → 10 taken 17 times.	408x	post_deferred_completions(local_ops);
637
638	3/4 ✓ Branch 10 → 11 taken 388 times. ✓ Branch 10 → 12 taken 3 times. ✓ Branch 11 → 12 taken 17 times. ✗ Branch 11 → 13 not taken.	408x	if (auto* tsvc = timers())
639	2/2 ✓ Branch 11 → 12 taken 388 times. ✓ Branch 12 → 13 taken 17 times.	405x	tsvc->process_expired();
640
641	2/2 ✓ Branch 12 → 13 taken 391 times. ✓ Branch 13 → 14 taken 17 times.	408x	update_timeout();
642			}
643
644		575266x	unsigned long wait_ms = timeout_ms;
645	10/12 ✓ Branch 15 → 16 taken 431230 times. ✓ Branch 15 → 19 taken 143468 times. ✓ Branch 16 → 17 taken 35 times. ✓ Branch 16 → 20 taken 533 times. ✓ Branch 17 → 18 taken 51 times. ✓ Branch 17 → 19 taken 431179 times. ✗ Branch 18 → 19 not taken. ✓ Branch 18 → 20 taken 35 times. ✓ Branch 20 → 21 taken 51 times. ✓ Branch 20 → 23 taken 574647 times. ✗ Branch 21 → 22 not taken. ✓ Branch 21 → 24 taken 568 times.	575266x	if (auto* tsvc = timers(); tsvc && !timers_)
646			{
647	1/2 ✓ Branch 21 → 22 taken 51 times. ✗ Branch 22 → 23 not taken.	51x	tsvc->process_expired();
648		51x	wait_ms = timer_timeout_ms(*tsvc, timeout_ms);
649			}
650
651		575266x	DWORD bytes = 0;
652		575266x	ULONG_PTR key = 0;
653		575266x	LPOVERLAPPED overlapped = nullptr;
654	2/2 ✓ Branch 23 → 24 taken 574698 times. ✓ Branch 24 → 25 taken 568 times.	575266x	::SetLastError(0);
655
656	2/2 ✓ Branch 24 → 25 taken 574698 times. ✓ Branch 25 → 26 taken 568 times.	575266x	BOOL result = ::GetQueuedCompletionStatus(
657			iocp_, &bytes, &key, &overlapped, wait_ms);
658	2/2 ✓ Branch 25 → 26 taken 574698 times. ✓ Branch 26 → 27 taken 568 times.	575266x	DWORD dwError = ::GetLastError();
659
660			// Handle based on completion key
661	4/4 ✓ Branch 26 → 27 taken 562694 times. ✓ Branch 26 → 57 taken 12004 times. ✓ Branch 27 → 28 taken 499 times. ✓ Branch 27 → 58 taken 69 times.	575266x	if (overlapped)
662			{
663	4/4 ✓ Branch 27 → 28 taken 2553 times. ✓ Branch 27 → 29 taken 560141 times. ✓ Branch 28 → 29 taken 12 times. ✓ Branch 28 → 30 taken 487 times.	563193x	DWORD err = result ? 0 : dwError;
664
665	6/8 ✓ Branch 30 → 31 taken 431663 times. ✓ Branch 30 → 50 taken 4352 times. ✓ Branch 30 → 53 taken 126679 times. ✗ Branch 30 → 56 not taken. ✓ Branch 31 → 32 taken 235 times. ✓ Branch 31 → 51 taken 20 times. ✓ Branch 31 → 54 taken 244 times. ✗ Branch 31 → 57 not taken.	563193x	switch (key)
666			{
667		431898x	case key_io:
668			case key_result_stored:
669			{
670		431898x	auto* ov_op = overlapped_to_op(overlapped);
671
672			// key_io carries fresh kernel results — publish them before
673			// the CAS so that losing the race (old value 0) still leaves
674			// valid data for the on_pending() re-post. For key_result_stored
675			// the payload was already written and released (by the completer
676			// in on_pending, or by on_completion), so we must not overwrite
677			// it.
678	4/4 ✓ Branch 32 → 33 taken 427892 times. ✓ Branch 32 → 34 taken 3771 times. ✓ Branch 33 → 34 taken 230 times. ✓ Branch 33 → 35 taken 5 times.	431898x	if (key == key_io)
679		428122x	ov_op->store_result(bytes, err);
680
681			// If old value was 1 the initiator already returned — dispatch.
682			// The acquire pairs with the publisher's release so the payload
683			// reads in complete() are ordered after the store that produced
684			// them. If old value was 0 the initiator hasn't returned yet;
685			// skip and let on_pending() re-post.
686		431898x	long expected = 0;
687	3/4 ✓ Branch 43 → 44 taken 431494 times. ✓ Branch 43 → 49 taken 169 times. ✓ Branch 44 → 45 taken 235 times. ✗ Branch 44 → 50 not taken.	863796x	if (!ov_op->ready_.compare_exchange_strong(
688			expected, 1, std::memory_order_acq_rel,
689			std::memory_order_acquire))
690			{
691		431729x	::InterlockedDecrement(&pending_io_);
692		431729x	ov_op->complete(
693	2/2 ✓ Branch 46 → 47 taken 431494 times. ✓ Branch 47 → 48 taken 235 times.	431729x	this, ov_op->bytes_transferred, ov_op->dwError);
694		431729x	work_finished();
695		431729x	return 1;
696			}
697		169x	continue;
698		169x	}
699
700		4372x	case key_posted:
701			{
702			// Posted scheduler_op: overlapped is actually a scheduler_op
703		4372x	auto* op = reinterpret_cast<scheduler_op*>(overlapped);
704	2/2 ✓ Branch 50 → 51 taken 4352 times. ✓ Branch 51 → 52 taken 20 times.	4372x	op->complete(this, bytes, err);
705		4372x	work_finished();
706		4372x	return 1;
707			}
708
709		126923x	case key_continuation:
710			{
711			// Posted continuation: overlapped is actually a continuation*
712		126923x	auto* c = reinterpret_cast<capy::continuation*>(overlapped);
713	2/2 ✓ Branch 53 → 54 taken 126679 times. ✓ Branch 54 → 55 taken 244 times.	126923x	c->h.resume();
714		126923x	work_finished();
715		126923x	return 1;
716			}
717
718		−	default: // LCOV_EXCL_LINE unreachable: closed key set
719		−	continue; // LCOV_EXCL_LINE unreachable: closed key set
720		✗	}
721			}
722
723			// Signal completions (no OVERLAPPED)
724	3/4 ✓ Branch 57 → 58 taken 11453 times. ✓ Branch 57 → 72 taken 551 times. ✓ Branch 58 → 59 taken 69 times. ✗ Branch 58 → 73 not taken.	12073x	if (result)
725			{
726	4/6 ✓ Branch 58 → 59 taken 404 times. ✓ Branch 58 → 60 taken 11049 times. ✗ Branch 58 → 71 not taken. ✓ Branch 59 → 60 taken 17 times. ✓ Branch 59 → 61 taken 52 times. ✗ Branch 59 → 72 not taken.	11522x	switch (key)
727			{
728		421x	case key_wake_dispatch:
729			// Timer wakeup - loop to check dispatch_required_
730		421x	took_wake = true;
731		421x	continue;
732
733		11101x	case key_shutdown:
734		11101x	::InterlockedExchange(&stop_event_posted_, 0);
735	3/4 ✓ Branch 63 → 64 taken 6113 times. ✓ Branch 63 → 70 taken 4936 times. ✗ Branch 64 → 65 not taken. ✓ Branch 64 → 71 taken 52 times.	11101x	if (stopped())
736			{
737			// Re-post for other waiting threads
738	1/4 ✗ Branch 65 → 66 not taken. ✓ Branch 65 → 69 taken 6113 times. ✗ Branch 66 → 67 not taken. ✗ Branch 66 → 70 not taken.	6113x	if (auto err = post_wakeup())
739		✗	::InterlockedExchange(
740		✗	&stop_failure_, static_cast<long>(err));
741		6113x	return 0;
742			}
743		4988x	continue;
744
745			// A key outside the closed set reaches here only if a
746			// third party posts to the port.
747		−	default: // LCOV_EXCL_LINE unreachable: closed key set
748		−	continue; // LCOV_EXCL_LINE unreachable: closed key set
749		5409x	}
750			}
751
752			// Timeout or error. Unless a timer cut the wait short, a
753			// WAIT_TIMEOUT is a finite caller timeout (wait_one/poll)
754			// elapsing with no work. run()/run_one() pass INFINITE and
755			// block until a real wake-up; they exit only once a
756			// key_shutdown is posted.
757	2/4 ✓ Branch 72 → 73 taken 1 time. ✓ Branch 72 → 75 taken 550 times. ✗ Branch 73 → 74 not taken. ✗ Branch 73 → 76 not taken.	551x	if (dwError != WAIT_TIMEOUT)
758		1x	detail::throw_system_error(make_err(dwError));
759	2/4 ✓ Branch 75 → 76 taken 533 times. ✓ Branch 75 → 77 taken 17 times. ✗ Branch 76 → 77 not taken. ✗ Branch 76 → 78 not taken.	550x	if (wait_ms == timeout_ms)
760		533x	return 0;
761	1/4 ✓ Branch 77 → 78 taken 17 times. ✗ Branch 77 → 79 not taken. ✗ Branch 78 → 79 not taken. ✗ Branch 78 → 80 not taken.	17x	if (timeout_ms != INFINITE)
762		17x	timeout_ms -= wait_ms;
763		5595x	}
764			}
765
766			inline void
767		2285x	win_scheduler::on_timer_changed(void* ctx)
768			{
769		2285x	static_cast<win_scheduler*>(ctx)->update_timeout();
770		2285x	}
771
772			// Rounds up, so a timer is never seen early and polled for again.
773			inline unsigned long
774		51x	win_scheduler::timer_timeout_ms(
775			timer_service& svc, unsigned long timeout_ms) noexcept
776			{
777		51x	auto nearest = svc.nearest_expiry();
778	2/2 ✓ Branch 5 → 6 taken 17 times. ✓ Branch 5 → 7 taken 34 times.	51x	if (nearest == (timer_service::time_point::max)())
779		17x	return timeout_ms;
780		34x	auto now = std::chrono::steady_clock::now();
781	1/2 ✗ Branch 10 → 11 not taken. ✓ Branch 10 → 12 taken 34 times.	34x	if (nearest <= now)
782		✗	return 0;
783			auto ms =
784		34x	std::chrono::ceil<std::chrono::milliseconds>(nearest - now).count();
785			// Kept below INFINITE, which would never time out.
786		34x	ms = (std::min)(ms, static_cast<decltype(ms)>(INFINITE - 1));
787		34x	return (std::min)(timeout_ms, static_cast<unsigned long>(ms));
788			}
789
790			inline void
791		420x	win_scheduler::on_timer_service_attached(timer_service& svc)
792			{
793			// Without the timer thread the completion-port wait timeout drives
794			// timers instead. The first timer is often made inside a delay's
795			// await_suspend, where a throw could only terminate.
796	1/1 ✓ Branch 2 → 3 taken 420 times.	420x	auto timers = make_win_timers(iocp_, &dispatch_required_);
797	3/3 ✓ Branch 4 → 5 taken 420 times. ✓ Branch 5 → 6 taken 403 times. ✓ Branch 5 → 8 taken 17 times.	420x	if (timers->start())
798		403x	timers_ = std::move(timers);
799		420x	svc.set_on_earliest_changed(
800		420x	timer_service::callback{this, &on_timer_changed});
801		420x	}
802
803			inline void
804		2693x	win_scheduler::update_timeout()
805			{
806		2693x	auto* tsvc = timers();
807	2/2 ✓ Branch 3 → 4 taken 3 times. ✓ Branch 3 → 5 taken 2690 times.	2693x	if (!tsvc)
808		3x	return;
809	2/2 ✓ Branch 6 → 7 taken 2673 times. ✓ Branch 6 → 10 taken 17 times.	2690x	if (timers_)
810			{
811		2673x	timers_->update_timeout(tsvc->nearest_expiry());
812			}
813			else
814			{
815			// A run loop may be waiting past the new earliest expiry.
816		17x	::PostQueuedCompletionStatus(iocp_, 0, key_wake_dispatch, nullptr);
817			}
818			}
819
820			} // namespace boost::corosio::detail
821
822			// Defer including the auxiliary wait reactor until the scheduler is
823			// fully defined, since the reactor's inline methods call back into
824			// win_scheduler. This also gives the ctor, dtor and wait_reactor()
825			// below a complete win_wait_reactor type: the constructor builds it,
826			// the destructor and unique_ptr's deleter tear it down.
827			//
828			// The macro lets win_wait_reactor.hpp diagnose direct inclusion
829			// (which would land it here with win_scheduler still incomplete).
830			#define BOOST_COROSIO_DETAIL_IOCP_WIN_SCHEDULER_BODY_DONE
831			#include <boost/corosio/native/detail/iocp/win_wait_reactor.hpp>
832
833			namespace boost::corosio::detail {
834
835		5378x	inline win_scheduler::win_scheduler(
836		5378x	capy::execution_context&, int concurrency_hint)
837		5378x	: iocp_(nullptr)
838		5378x	, outstanding_work_(0)
839		5378x	, stopped_(0)
840		5378x	, stop_event_posted_(0)
841	1/1 ✓ Branch 3 → 4 taken 5378 times.	5378x	, dispatch_required_(0)
842			{
843			// concurrency_hint < 0 means use system default (DWORD(~0) = max)
844	2/3 ✓ Branch 7 → 8 taken 5378 times. ✗ Branch 7 → 9 not taken. ✓ Branch 10 → 11 taken 5378 times.	5378x	iocp_ = ::CreateIoCompletionPort(
845			INVALID_HANDLE_VALUE, nullptr, 0,
846			static_cast<DWORD>(
847			concurrency_hint >= 0 ? concurrency_hint : DWORD(~0)));
848
849	2/2 ✓ Branch 11 → 12 taken 1 time. ✓ Branch 11 → 15 taken 5377 times.	5378x	if (iocp_ == nullptr)
850	1/1 ✓ Branch 12 → 13 taken 1 time.	1x	detail::throw_system_error(make_err(::GetLastError()));
851
852			try
853			{
854			// A scheduler whose wait reactor could not be built would
855			// answer every wait with a parked op, so it refuses to exist
856			// instead.
857	1/1 ✓ Branch 15 → 16 taken 5224 times.	5377x	wait_reactor_ = std::make_unique<win_wait_reactor>(*this);
858			}
859		153x	catch (...)
860			{
861			// ~win_scheduler never runs for a constructor that throws, and
862			// the port is a raw handle nothing else owns.
863	1/1 ✓ Branch 23 → 24 taken 153 times.	153x	::CloseHandle(iocp_);
864		153x	iocp_ = nullptr;
865		153x	throw;
866		153x	}
867		5840x	}
868
869			inline void
870		5224x	win_scheduler::shutdown()
871			{
872	2/2 ✓ Branch 3 → 4 taken 403 times. ✓ Branch 3 → 6 taken 4821 times.	5224x	if (timers_)
873		403x	timers_->stop();
874
875			// Ops parked in the auxiliary wait reactor owe completion packets.
876			// Stop the reactor early so its loop posts them as cancelled and
877			// the pending count can reach zero.
878		5224x	wait_reactor_->stop();
879
880		5224x	wait_cancels_posting();
881
882			// Reap every packet still owed to the port before the services
883			// free the op memory those packets reference. Work-guard credits,
884			// posted handlers, and queued continuations have no bearing here.
885	2/2 ✓ Branch 33 → 10 taken 654 times. ✓ Branch 33 → 34 taken 5224 times.	11756x	while (::InterlockedExchangeAdd(&pending_io_, 0) > 0)
886			{
887		654x	op_queue ops;
888			{
889		654x	std::lock_guard<win_mutex> lock(dispatch_mutex_);
890		654x	ops.splice(completed_ops_);
891		654x	}
892
893			// Deferred-queue entries are process-owned (failed-post
894			// fallbacks and posted handlers); no packet references them.
895	1/2 ✗ Branch 15 → 16 not taken. ✓ Branch 15 → 18 taken 654 times.	654x	while (auto* h = ops.pop())
896	0/1 ✗ Branch 16 → 17 not taken.	✗	h->destroy();
897
898			DWORD bytes;
899			ULONG_PTR key;
900			LPOVERLAPPED overlapped;
901	1/1 ✓ Branch 18 → 19 taken 654 times.	654x	::GetQueuedCompletionStatus(
902			iocp_, &bytes, &key, &overlapped, iocp::shutdown_drain_timeout_ms);
903	1/2 ✓ Branch 19 → 20 taken 654 times. ✗ Branch 19 → 30 not taken.	654x	if (overlapped)
904			{
905	1/2 ✗ Branch 20 → 21 not taken. ✓ Branch 20 → 22 taken 654 times.	654x	if (key == key_posted)
906			{
907		✗	auto* op = reinterpret_cast<scheduler_op*>(overlapped);
908	0/1 ✗ Branch 21 → 30 not taken.	✗	op->destroy();
909			}
910	2/2 ✓ Branch 22 → 23 taken 5 times. ✓ Branch 22 → 26 taken 649 times.	654x	else if (key == key_continuation)
911			{
912			// Drain without resuming: abandon the parked coroutine.
913		5x	auto* c = reinterpret_cast<capy::continuation*>(overlapped);
914	1/2 ✓ Branch 24 → 25 taken 5 times. ✗ Branch 24 → 30 not taken.	5x	if (c->h)
915		5x	capy::abandon(*c);
916			}
917			else
918			{
919		649x	::InterlockedDecrement(&pending_io_);
920		649x	auto* op = overlapped_to_op(overlapped);
921	1/1 ✓ Branch 29 → 30 taken 649 times.	649x	op->destroy();
922			}
923			}
924			}
925
926			// Final sweep: packets can sit in the port or deferred queue
927			// without a pending_io_ count (posted directly against the
928			// handle). Destroy them so service teardown does not free state
929			// they still reference.
930		5224x	sweep_port();
931		5224x	}
932
933			inline void
934		18637x	win_scheduler::sweep_port() noexcept
935			{
936			for (;;)
937			{
938		18752x	op_queue ops;
939			{
940		18752x	std::lock_guard<win_mutex> lock(dispatch_mutex_);
941		18752x	ops.splice(completed_ops_);
942		18752x	}
943	2/4 ✗ Branch 7 → 8 not taken. ✓ Branch 7 → 9 taken 18367 times. ✗ Branch 8 → 9 not taken. ✓ Branch 8 → 10 taken 385 times.	18752x	while (auto* h = ops.pop())
944		✗	h->destroy();
945
946			DWORD bytes;
947			ULONG_PTR key;
948			LPOVERLAPPED overlapped;
949			BOOL const dequeued =
950		18752x	::GetQueuedCompletionStatus(iocp_, &bytes, &key, &overlapped, 0);
951	3/4 ✓ Branch 10 → 11 taken 14426 times. ✓ Branch 10 → 14 taken 3941 times. ✓ Branch 11 → 12 taken 385 times. ✗ Branch 11 → 15 not taken.	18752x	if (!overlapped)
952			{
953			// A signal such as stop()'s carries no operation, and
954			// packets posted after it still need sweeping; only an
955			// empty port ends the sweep.
956	4/4 ✓ Branch 11 → 12 taken 10178 times. ✓ Branch 11 → 13 taken 4248 times. ✓ Branch 12 → 13 taken 270 times. ✓ Branch 12 → 14 taken 115 times.	14811x	if (!dequeued)
957		10448x	break;
958		4363x	continue;
959			}
960	2/4 ✓ Branch 14 → 15 taken 14 times. ✓ Branch 14 → 16 taken 3927 times. ✗ Branch 15 → 16 not taken. ✗ Branch 15 → 17 not taken.	3941x	if (key == key_posted)
961			{
962		14x	reinterpret_cast<scheduler_op*>(overlapped)->destroy();
963			}
964	2/4 ✓ Branch 16 → 17 taken 3923 times. ✓ Branch 16 → 20 taken 4 times. ✗ Branch 17 → 18 not taken. ✗ Branch 17 → 21 not taken.	3927x	else if (key == key_continuation)
965			{
966		3923x	auto* c = reinterpret_cast<capy::continuation*>(overlapped);
967	1/4 ✓ Branch 18 → 19 taken 3923 times. ✗ Branch 18 → 22 not taken. ✗ Branch 19 → 20 not taken. ✗ Branch 19 → 23 not taken.	3923x	if (c->h)
968		3923x	capy::abandon(*c);
969			}
970			else
971			{
972		4x	overlapped_to_op(overlapped)->destroy();
973			}
974		8304x	}
975		10448x	}
976
977		10448x	inline win_scheduler::~win_scheduler()
978			{
979			// A timer made after shutdown starts a thread that posts to the
980			// port, so it is joined before the port closes.
981		5224x	timers_.reset();
982		5224x	wait_reactor_->stop();
983		5224x	wait_reactor_.reset();
984
985	1/2 ✓ Branch 6 → 7 taken 5224 times. ✗ Branch 6 → 9 not taken.	5224x	if (iocp_ != nullptr)
986			{
987			// Services give up their work in shutdown(), but anything posted
988			// after the drain, such as from a destructor, is still destroyed.
989		5224x	sweep_port();
990		5224x	::CloseHandle(iocp_);
991			}
992		10448x	}
993
994			inline win_wait_reactor&
995		56x	win_scheduler::wait_reactor()
996			{
997		56x	return *wait_reactor_;
998			}
999
1000			inline void
1001		42456x	win_scheduler::cancel_wait(overlapped_op* op) noexcept
1002			{
1003		42456x	wait_reactor_->cancel_wait(op);
1004		42456x	}
1005
1006			} // namespace boost::corosio::detail
1007
1008			#endif // BOOST_COROSIO_HAS_IOCP
1009
1010			#endif // BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_SCHEDULER_HPP
1011