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

94.2% Lines (295/318) 97.0% List of functions (32/33) 74.0% Branches (162/219)

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

win_scheduler.hpp

f(x) Functions (33)

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