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

91.1% Lines (163 / 179) 68.6% Branches (48 / 70) 90.6% Functions (29 / 32)

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

win_object_handle_service.hpp

f(x) Functions (32)

Function Calls Lines Branches Blocks

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2026 Michael Vandeberg
3			//
4			// Distributed under the Boost Software License, Version 1.0. (See accompanying
5			// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6			//
7			// Official repository: https://github.com/cppalliance/corosio
8			//
9
10			#ifndef BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_OBJECT_HANDLE_SERVICE_HPP
11			#define BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_OBJECT_HANDLE_SERVICE_HPP
12
13			#include <boost/corosio/detail/platform.hpp>
14
15			#if BOOST_COROSIO_HAS_IOCP
16
17			#include <boost/corosio/detail/config.hpp>
18			#include <boost/corosio/detail/dispatch_coro.hpp>
19			#include <boost/corosio/detail/object_pool.hpp>
20			#include <boost/corosio/detail/object_ref.hpp>
21			#include <boost/corosio/detail/win_handle_service.hpp>
22			#include <boost/corosio/native/detail/iocp/win_overlapped_handle.hpp>
23			#include <boost/corosio/native/detail/iocp/win_validate_handle.hpp>
24			#include <boost/capy/error.hpp>
25
26			#include <atomic>
27			#include <cstdint>
28			#include <mutex>
29
30			/* win_object_handle on the Windows thread pool.
31
32			One PTP_WAIT per state, created at the first wait() and re-armed per
33			wait. The claim word (phase in the low two bits, a generation above)
34			decides which of the pool callback and a cancellation delivers the
35			one op: whoever moves it from armed to claimed posts the op.
36
37			Cancellation never cancels a queued callback. By the time a callback
38			is queued the kernel has already satisfied the wait -- consumed an
39			auto-reset event's signal, a semaphore unit -- and the callback is
40			the only witness to that. So every rundown (cancel, stop token,
41			close, release, assign, destroy, shutdown) drains instead:
42			SetThreadpoolWait(nullptr) stops an unsatisfied wait, then
43			WaitForThreadpoolWaitCallbacks(FALSE) lets a queued callback finish
44			and claim, and only then does the rundown try to claim for
45			cancellation.
46
47			arm_mutex_ orders arming against the disarm step, and disarmed_gen_
48			records which generation a rundown disarmed, so a rundown that
49			disarmed before wait() armed is never followed by that arm. The
50			callback never takes the mutex, which keeps the drain deadlock-free.
51			*/
52
53			namespace boost::corosio::detail {
54
55			/** Object handle state, embedded in the implementation named by `io`.
56
57			The thread-pool wait survives recycling and is closed with the
58			implementation's storage.
59			*/
60			class win_object_handle_state
61			{
62			public:
63		2826x	win_object_handle_state(
64			win_scheduler& sched, io_object::implementation& io) noexcept
65		2826x	: sched_(sched)
66		2826x	, io_(io)
67			{
68		2826x	op_.self = this;
69		2826x	}
70
71		2826x	~win_object_handle_state()
72			{
73	2/2 ✓ Branch 2 → 3 taken 2818 times. ✓ Branch 2 → 6 taken 8 times.	2826x	if (tp_wait_)
74			{
75		2818x	::SetThreadpoolWait(tp_wait_, nullptr, nullptr);
76		2818x	::WaitForThreadpoolWaitCallbacks(tp_wait_, FALSE);
77		2818x	::CloseThreadpoolWait(tp_wait_);
78			}
79		2826x	}
80
81			win_object_handle_state(win_object_handle_state const&) = delete;
82			win_object_handle_state& operator=(win_object_handle_state const&) = delete;
83
84		8266x	HANDLE native_handle() const noexcept
85			{
86		8266x	return handle_;
87			}
88
89		✗	bool is_open() const noexcept
90			{
91		✗	return handle_ != INVALID_HANDLE_VALUE;
92			}
93
94		3222x	std::coroutine_handle<> wait(
95			capy::continuation& cont,
96			capy::executor_ref ex,
97			std::stop_token token,
98			std::error_code* ec)
99			{
100			// Reserved as claimed under the current generation until armed:
101			// a concurrent wait() sees it busy, nothing else claims it, and
102			// a rundown in the meantime records a generation this wait never
103			// uses.
104		3222x	std::uint64_t cur = state_.load(std::memory_order_acquire);
105	4/4 ✓ Branch 17 → 18 taken 2943 times. ✓ Branch 17 → 31 taken 279 times. ✓ Branch 33 → 34 taken 279 times. ✓ Branch 33 → 37 taken 2943 times.	6165x	if ((cur & phase_mask) != phase_idle \|\|
106	1/2 ✗ Branch 30 → 31 not taken. ✓ Branch 30 → 32 taken 2943 times.	5886x	!state_.compare_exchange_strong(
107			cur, cur \| phase_claimed, std::memory_order_acq_rel))
108			{
109			// Complete immediately: the returned handle is resumed by
110			// symmetric transfer on the caller's own executor.
111	1/2 ✓ Branch 34 → 35 taken 279 times. ✗ Branch 34 → 36 not taken.	279x	if (ec)
112		279x	*ec = std::make_error_code(std::errc::operation_in_progress);
113		279x	return cont.h;
114			}
115
116		2943x	std::uint64_t const gen = (cur >> 2) + 1;
117		2943x	std::uint64_t const armed = (gen << 2) \| phase_armed;
118
119		2943x	op_.reset();
120		2943x	op_.object_ref_ = detail::object_ref(&io_);
121			// The op may complete and the object be destroyed on another
122			// thread before wait() returns; this pin outlives both.
123		2943x	auto const pin = op_.object_ref_;
124		2943x	op_.cont = &cont;
125		2943x	op_.ex = ex;
126		2943x	op_.ec_out = ec;
127		2943x	op_.bytes_out = nullptr;
128		2943x	sched_.work_started();
129
130	2/2 ✓ Branch 43 → 44 taken 1 time. ✓ Branch 43 → 68 taken 2942 times.	2943x	if (handle_ == INVALID_HANDLE_VALUE)
131			{
132		1x	state_.store((gen << 2) \| phase_claimed, std::memory_order_release);
133	1/1 ✓ Branch 64 → 65 taken 1 time.	1x	sched_.on_completion(&op_, ERROR_INVALID_HANDLE, 0);
134		1x	return std::noop_coroutine();
135			}
136
137	2/2 ✓ Branch 68 → 69 taken 2818 times. ✓ Branch 68 → 96 taken 124 times.	2942x	if (!tp_wait_)
138			{
139	1/1 ✓ Branch 69 → 70 taken 2818 times.	2818x	tp_wait_ = ::CreateThreadpoolWait(&on_signaled, this, nullptr);
140	1/2 ✗ Branch 70 → 71 not taken. ✓ Branch 70 → 96 taken 2818 times.	2818x	if (!tp_wait_)
141			{
142		✗	state_.store(
143		✗	(gen << 2) \| phase_claimed, std::memory_order_release);
144	0/2 ✗ Branch 91 → 92 not taken. ✗ Branch 92 → 93 not taken.	✗	sched_.on_completion(&op_, ::GetLastError(), 0);
145		✗	return std::noop_coroutine();
146			}
147			}
148
149			// The stop callback is engaged before the wait is published, so
150			// a completion can never find it mid-construction. A stop that
151			// runs before publication finds nothing to claim; the check
152			// below claims for it.
153		2942x	op_.start(token);
154		2942x	state_.store(armed, std::memory_order_release);
155			{
156			// A rundown that disarmed this generation first has claimed,
157			// or will claim, the op; arming now would let a later signal
158			// be consumed with nobody left to report it.
159		2942x	std::lock_guard<win_mutex> lock(arm_mutex_);
160	2/4 ✓ Branch 133 → 134 taken 2942 times. ✗ Branch 133 → 136 not taken. ✓ Branch 137 → 138 taken 2942 times. ✗ Branch 137 → 144 not taken.	8826x	if (state_.load(std::memory_order_acquire) == armed &&
161	1/2 ✓ Branch 134 → 135 taken 2942 times. ✗ Branch 134 → 136 not taken.	2942x	disarmed_gen_ != gen)
162			{
163	1/2 ✓ Branch 139 → 140 taken 2942 times. ✗ Branch 139 → 141 not taken.	2942x	if (!op_.cancelled.load(std::memory_order_acquire))
164	1/1 ✓ Branch 140 → 144 taken 2942 times.	2942x	::SetThreadpoolWait(tp_wait_, handle_, nullptr);
165	0/2 ✗ Branch 142 → 143 not taken. ✗ Branch 142 → 144 not taken.	✗	else if (claim(armed))
166	0/1 ✗ Branch 143 → 144 not taken.	✗	sched_.on_completion(&op_, ERROR_OPERATION_ABORTED, 0);
167			}
168		2942x	}
169		2942x	return std::noop_coroutine();
170		2943x	}
171
172		2201x	void cancel() noexcept
173			{
174		2201x	rundown();
175		2201x	}
176
177		8271x	void close_handle() noexcept
178			{
179		8271x	rundown();
180	2/2 ✓ Branch 3 → 4 taken 2619 times. ✓ Branch 3 → 6 taken 5652 times.	8271x	if (handle_ != INVALID_HANDLE_VALUE)
181			{
182		2619x	::CloseHandle(handle_);
183		2619x	handle_ = INVALID_HANDLE_VALUE;
184			}
185		8271x	}
186
187		201x	native_handle_type release() noexcept
188			{
189		201x	rundown();
190		201x	HANDLE h = handle_;
191		201x	handle_ = INVALID_HANDLE_VALUE;
192		201x	return reinterpret_cast<native_handle_type>(h);
193			}
194
195			/// Return true if no wait is armed or being delivered.
196		✗	bool is_idle() const noexcept
197			{
198		✗	return (state_.load(std::memory_order_acquire) & phase_mask) ==
199		✗	phase_idle;
200			}
201
202		2824x	std::error_code assign(native_handle_type nh) noexcept
203			{
204		2824x	HANDLE h = reinterpret_cast<HANDLE>(nh);
205	2/2 ✓ Branch 4 → 5 taken 4 times. ✓ Branch 4 → 6 taken 2820 times.	2824x	if (auto ec = validate_object_handle(h))
206		4x	return ec;
207		2820x	handle_ = h;
208		2820x	return {};
209			}
210
211			private:
212			static constexpr std::uint64_t phase_idle = 0;
213			static constexpr std::uint64_t phase_armed = 1;
214			static constexpr std::uint64_t phase_claimed = 2;
215			static constexpr std::uint64_t phase_mask = 3;
216
217			struct wait_op : overlapped_op
218			{
219			win_object_handle_state* self = nullptr;
220
221		2826x	wait_op() noexcept : overlapped_op(&do_complete)
222			{
223		2826x	cancel_func_ = &do_cancel_impl;
224		2826x	}
225
226		2x	static void do_cancel_impl(overlapped_op* base) noexcept
227			{
228		2x	static_cast<wait_op*>(base)->self->rundown();
229		2x	}
230
231		2943x	static void do_complete(
232			void* owner,
233			scheduler_op* base,
234			std::uint32_t /bytes/,
235			std::uint32_t /error/)
236			{
237		2943x	auto* op = static_cast<wait_op*>(base);
238		2943x	auto* self = op->self;
239		2943x	auto prevent_premature_destruction = std::move(op->object_ref_);
240
241	2/2 ✓ Branch 4 → 5 taken 1 time. ✓ Branch 4 → 8 taken 2942 times.	2943x	if (!owner)
242			{
243		1x	self->set_idle();
244		1x	op->destroy_awaiting();
245		1x	return;
246			}
247
248		2942x	op->stop_cb.reset();
249			// A claimed signal is success even if a cancel request raced
250			// it and set the cancelled flag: completion wins.
251	1/2 ✓ Branch 9 → 10 taken 2942 times. ✗ Branch 9 → 17 not taken.	2942x	if (op->ec_out)
252			{
253	2/2 ✓ Branch 10 → 11 taken 1805 times. ✓ Branch 10 → 13 taken 1137 times.	2942x	if (op->dwError == 0)
254		1805x	*op->ec_out = {};
255	2/2 ✓ Branch 13 → 14 taken 1136 times. ✓ Branch 13 → 16 taken 1 time.	1137x	else if (op->dwError == ERROR_OPERATION_ABORTED)
256		1136x	*op->ec_out = capy::error::canceled;
257			else
258		1x	*op->ec_out =
259		1x	iocp_make_err(op->dwError, /accept_path=/false);
260			}
261
262		2942x	capy::continuation* cont = op->cont;
263		2942x	capy::executor_ref ex = op->ex;
264
265			// Last touch of op_: from here a new wait() may reset it.
266		2942x	self->set_idle();
267	1/1 ✓ Branch 19 → 20 taken 2942 times.	2942x	dispatch_coro(ex, *cont).resume();
268		2943x	}
269			};
270
271			// Back to idle, keeping the generation.
272		2943x	void set_idle() noexcept
273			{
274		5886x	state_.store(
275		5886x	state_.load(std::memory_order_relaxed) & ~phase_mask,
276			std::memory_order_release);
277		2943x	}
278
279		1805x	static void CALLBACK on_signaled(
280			PTP_CALLBACK_INSTANCE, void* ctx, PTP_WAIT, TP_WAIT_RESULT) noexcept
281			{
282		1805x	auto* self = static_cast<win_object_handle_state*>(ctx);
283		1805x	std::uint64_t cur = self->state_.load(std::memory_order_acquire);
284	3/6 ✓ Branch 17 → 18 taken 1805 times. ✗ Branch 17 → 21 not taken. ✓ Branch 19 → 20 taken 1805 times. ✗ Branch 19 → 21 not taken. ✓ Branch 22 → 23 taken 1805 times. ✗ Branch 22 → 24 not taken.	1805x	if ((cur & phase_mask) == phase_armed && self->claim(cur))
285		1805x	self->sched_.on_completion(&self->op_, 0, 0);
286		1805x	}
287
288		2942x	bool claim(std::uint64_t armed) noexcept
289			{
290		5884x	return state_.compare_exchange_strong(
291		2942x	armed, (armed & ~phase_mask) \| phase_claimed,
292		2942x	std::memory_order_acq_rel);
293			}
294
295			// Stop an unsatisfied wait, let a queued callback finish, then claim
296			// for cancellation only if nothing else did. Never called from the
297			// pool callback.
298		10675x	void rundown() noexcept
299			{
300	2/2 ✓ Branch 2 → 3 taken 18 times. ✓ Branch 2 → 4 taken 10657 times.	10675x	if (!tp_wait_)
301		18x	return;
302			{
303		10657x	std::lock_guard<win_mutex> lock(arm_mutex_);
304		10657x	::SetThreadpoolWait(tp_wait_, nullptr, nullptr);
305		10657x	disarmed_gen_ = state_.load(std::memory_order_acquire) >> 2;
306		10657x	}
307		10657x	::WaitForThreadpoolWaitCallbacks(tp_wait_, FALSE);
308
309			// Counted from before the claim: a claimed op is invisible to the
310			// scheduler's shutdown until on_completion queues it.
311		10657x	sched_.cancel_posting_begin();
312		10657x	std::uint64_t cur = state_.load(std::memory_order_acquire);
313	5/6 ✓ Branch 39 → 40 taken 1137 times. ✓ Branch 39 → 43 taken 9520 times. ✓ Branch 41 → 42 taken 1137 times. ✗ Branch 41 → 43 not taken. ✓ Branch 44 → 45 taken 1137 times. ✓ Branch 44 → 47 taken 9520 times.	10657x	if ((cur & phase_mask) == phase_armed && claim(cur))
314			{
315		1137x	op_.request_cancel();
316		1137x	sched_.on_completion(&op_, ERROR_OPERATION_ABORTED, 0);
317			}
318		10657x	sched_.cancel_posting_end();
319			}
320
321			win_scheduler& sched_;
322			io_object::implementation& io_;
323			HANDLE handle_ = INVALID_HANDLE_VALUE;
324			PTP_WAIT tp_wait_ = nullptr;
325			std::atomic<std::uint64_t> state_{phase_idle};
326			win_mutex arm_mutex_;
327			std::uint64_t disarmed_gen_ = 0; // guarded by arm_mutex_
328			wait_op op_;
329			};
330
331			class win_object_handle_service;
332
333			/** IOCP implementation of @ref win_object_handle. */
334			class win_object_handle_impl final
335			: public win_object_handle::implementation
336			, public intrusive_list<win_object_handle_impl>::node
337			{
338			win_object_handle_service& svc_;
339			win_object_handle_state internal_;
340
341			public:
342		2826x	win_object_handle_impl(
343			win_object_handle_service& svc, win_scheduler& sched) noexcept
344		5652x	: svc_(svc)
345		2826x	, internal_(sched, *this)
346			{
347		2826x	}
348
349			/// Recycle into the owning service's pool. Defined out-of-line
350			/// after the service is complete.
351			void retire() noexcept override;
352
353			/** Assert the closed state a recycled handle starts from.
354
355			@pre refs_ == 0, handle closed, no wait in flight.
356			*/
357		✗	void reuse() noexcept
358			{
359	0/2 ✗ Branch 3 → 4 not taken. ✗ Branch 3 → 5 not taken.	✗	BOOST_COROSIO_ASSERT(!internal_.is_open());
360	0/2 ✗ Branch 7 → 8 not taken. ✗ Branch 7 → 9 not taken.	✗	BOOST_COROSIO_ASSERT(internal_.is_idle());
361		✗	}
362
363		11095x	win_object_handle_state* get_internal() noexcept
364			{
365		11095x	return &internal_;
366			}
367
368		3222x	std::coroutine_handle<> wait(
369			capy::continuation& cont,
370			capy::executor_ref ex,
371			std::stop_token token,
372			std::error_code* ec) override
373			{
374	1/1 ✓ Branch 4 → 5 taken 3222 times.	3222x	return internal_.wait(cont, ex, std::move(token), ec);
375			}
376
377		8266x	native_handle_type native_handle() const noexcept override
378			{
379		8266x	return reinterpret_cast<native_handle_type>(internal_.native_handle());
380			}
381
382		201x	native_handle_type release_handle() noexcept override
383			{
384		201x	return internal_.release();
385			}
386
387		2201x	void cancel() noexcept override
388			{
389		2201x	internal_.cancel();
390		2201x	}
391			};
392
393			/** IOCP service that owns the @ref win_object_handle implementations. */
394			class BOOST_COROSIO_DECL win_object_handle_service final
395			: public object_handle_service
396			{
397			friend class win_object_handle_impl;
398
399			public:
400		2826x	explicit win_object_handle_service(capy::execution_context& ctx)
401	1/1 ✓ Branch 3 → 4 taken 2826 times.	2826x	: sched_(ctx.use_service<win_scheduler>())
402			{
403		2826x	}
404
405		2826x	io_object::implementation* construct() override
406			{
407		2826x	return pool_.acquire(*this, sched_);
408			}
409
410		2826x	void destroy(io_object::implementation* p) override
411			{
412	1/2 ✗ Branch 2 → 3 not taken. ✓ Branch 2 → 4 taken 2826 times.	2826x	if (!p)
413		✗	return;
414		2826x	auto* impl = static_cast<win_object_handle_impl*>(p);
415		2826x	impl->get_internal()->close_handle();
416		2826x	release(impl);
417			}
418
419		5444x	void close(io_object::handle& h) override
420			{
421		5444x	static_cast<win_object_handle_impl&>(*h.get())
422			.get_internal()
423		5444x	->close_handle();
424		5444x	}
425
426		2826x	void shutdown() override
427			{
428	1/1 ✓ Branch 2 → 3 taken 2826 times.	2826x	pool_.shutdown([](win_object_handle_impl* impl) {
429		1x	impl->get_internal()->close_handle();
430		1x	});
431		2826x	}
432
433		2825x	std::error_code assign_object_handle(
434			win_object_handle::implementation& impl,
435			native_handle_type h) override
436			{
437			// The thread-pool callback completes from a foreign thread.
438	2/2 ✓ Branch 3 → 4 taken 1 time. ✓ Branch 3 → 5 taken 2824 times.	2825x	if (sched_.scheduler_locking_disabled())
439		1x	return std::make_error_code(std::errc::operation_not_supported);
440			return static_cast<win_object_handle_impl&>(impl)
441			.get_internal()
442		2824x	->assign(h);
443			}
444
445			private:
446			win_scheduler& sched_;
447			BOOST_COROSIO_MSVC_WARNING_PUSH
448			BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) // detail:: members, dll-interface
449			object_pool<win_object_handle_impl> pool_;
450			BOOST_COROSIO_MSVC_WARNING_POP
451			};
452
453			inline void
454		2826x	win_object_handle_impl::retire() noexcept
455			{
456		2826x	svc_.pool_.recycle(this);
457		2826x	}
458
459			} // namespace boost::corosio::detail
460
461			#endif // BOOST_COROSIO_HAS_IOCP
462
463			#endif
464