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

96.1% Lines (446 / 464) 78.7% Branches (148 / 188) 98.1% Functions (51 / 52)

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

win_local_stream_service.hpp

f(x) Functions (52)

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_LOCAL_STREAM_SERVICE_HPP
11			#define BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_LOCAL_STREAM_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/object_pool.hpp>
19			#include <boost/corosio/detail/local_stream_service.hpp>
20
21			#include <boost/corosio/native/detail/iocp/win_dissociate.hpp>
22			#include <boost/corosio/native/detail/iocp/win_validate_handle.hpp>
23			#include <boost/corosio/native/detail/iocp/win_local_stream_acceptor.hpp>
24			#include <boost/corosio/native/detail/iocp/win_local_stream_socket.hpp>
25			#include <boost/corosio/native/detail/iocp/win_tcp_acceptor_service.hpp>
26			#include <boost/corosio/native/detail/iocp/win_scheduler.hpp>
27			#include <boost/corosio/native/detail/iocp/win_completion_key.hpp>
28			#include <boost/corosio/native/detail/iocp/win_wsa_init.hpp>
29
30			#include <boost/corosio/native/detail/endpoint_convert.hpp>
31			#include <boost/corosio/native/detail/make_err.hpp>
32			#include <boost/corosio/detail/dispatch_coro.hpp>
33
34			#include <ws2tcpip.h>
35
36			namespace boost::corosio::detail {
37
38			/* IOCP local stream socket service.
39
40			Inherits from local_stream_service to enable runtime polymorphism
41			via use_service<local_stream_service>(). Reuses the ConnectEx /
42			AcceptEx function pointers already loaded by win_tcp_service.
43			*/
44			class BOOST_COROSIO_DECL win_local_stream_service final
45			: private win_wsa_init
46			, public local_stream_service
47			{
48			friend class win_local_stream_socket;
49			friend class win_local_stream_acceptor;
50
51			public:
52			io_object::implementation* construct() override;
53
54			void destroy(io_object::implementation* p) override;
55
56			void close(io_object::handle& h) override;
57
58			explicit win_local_stream_service(capy::execution_context& ctx);
59
60			win_local_stream_service(win_local_stream_service const&) = delete;
61			win_local_stream_service&
62			operator=(win_local_stream_service const&) = delete;
63
64			void shutdown() override;
65
66			std::error_code open_socket(
67			local_stream_socket::implementation& impl,
68			int family,
69			int type,
70			int protocol) override;
71
72			std::error_code assign_socket(
73			local_stream_socket::implementation& impl,
74			native_handle_type fd) override;
75
76			std::error_code open_socket_internal(
77			win_local_stream_socket& impl, int family, int type, int protocol);
78
79			win_local_stream_acceptor* acquire_acceptor_impl();
80
81			std::error_code open_acceptor_socket(
82			win_local_stream_acceptor& impl, int family, int type, int protocol);
83
84			std::error_code assign_acceptor_socket(
85			win_local_stream_acceptor& impl, native_handle_type fd);
86
87			std::error_code
88			bind_acceptor(win_local_stream_acceptor& impl, corosio::local_endpoint ep);
89
90			std::error_code
91			listen_acceptor(win_local_stream_acceptor& impl, int backlog);
92
93			void* native_handle() const noexcept;
94			LPFN_CONNECTEX connect_ex() const noexcept;
95			LPFN_ACCEPTEX accept_ex() const noexcept;
96
97			void post(overlapped_op* op);
98			void on_pending(overlapped_op* op) noexcept;
99			void on_completion(overlapped_op* op, DWORD error, DWORD bytes) noexcept;
100			void work_started() noexcept;
101			void work_finished() noexcept;
102
103			/** Return the owning IOCP scheduler. */
104		1079x	win_scheduler& scheduler() noexcept
105			{
106		1079x	return sched_;
107			}
108
109			private:
110			win_tcp_service& tcp_svc_;
111			win_scheduler& sched_;
112			void* iocp_;
113			BOOST_COROSIO_MSVC_WARNING_PUSH
114			BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) // detail:: members, dll-interface
115			object_pool<win_local_stream_socket> pool_;
116			object_pool<win_local_stream_acceptor> acceptor_pool_;
117			BOOST_COROSIO_MSVC_WARNING_POP
118			};
119
120			// ============================================================
121			// Operation constructors
122			// ============================================================
123
124		140x	inline local_stream_connect_op::local_stream_connect_op(
125		140x	win_local_stream_socket& internal_) noexcept
126			: overlapped_op(&do_complete)
127		140x	, internal(internal_)
128			{
129		140x	cancel_func_ = &do_cancel_impl;
130		140x	}
131
132		140x	inline local_stream_read_op::local_stream_read_op(
133		140x	win_local_stream_socket& internal_) noexcept
134			: overlapped_op(&do_complete)
135		140x	, internal(internal_)
136			{
137		140x	cancel_func_ = &do_cancel_impl;
138		140x	}
139
140		140x	inline local_stream_write_op::local_stream_write_op(
141		140x	win_local_stream_socket& internal_) noexcept
142			: overlapped_op(&do_complete)
143		140x	, internal(internal_)
144			{
145		140x	cancel_func_ = &do_cancel_impl;
146		140x	}
147
148		140x	inline local_stream_wait_op::local_stream_wait_op(
149		140x	win_local_stream_socket& internal_) noexcept
150			: overlapped_op(&do_complete)
151		140x	, internal(internal_)
152			{
153		140x	cancel_func_ = &do_cancel_impl;
154		140x	}
155
156			// ============================================================
157			// Cancellation functions
158			// ============================================================
159
160			inline void
161		✗	local_stream_connect_op::do_cancel_impl(overlapped_op* base) noexcept
162			{
163		✗	auto* op = static_cast<local_stream_connect_op*>(base);
164		✗	op->cancelled.store(true, std::memory_order_release);
165	0/2 ✗ Branch 3 → 4 not taken. ✗ Branch 3 → 8 not taken.	✗	if (op->internal.socket_ != INVALID_SOCKET)
166			{
167	0/2 ✗ Branch 4 → 5 not taken. ✗ Branch 4 → 6 not taken.	✗	::CancelIoEx(reinterpret_cast<HANDLE>(op->internal.socket_), op);
168			}
169		✗	}
170
171			inline void
172		2x	local_stream_read_op::do_cancel_impl(overlapped_op* base) noexcept
173			{
174		2x	auto* op = static_cast<local_stream_read_op*>(base);
175		2x	op->cancelled.store(true, std::memory_order_release);
176	1/2 ✓ Branch 3 → 4 taken 2 times. ✗ Branch 3 → 8 not taken.	2x	if (op->internal.socket_ != INVALID_SOCKET)
177			{
178	1/2 ✓ Branch 4 → 5 taken 2 times. ✗ Branch 4 → 6 not taken.	2x	::CancelIoEx(reinterpret_cast<HANDLE>(op->internal.socket_), op);
179			}
180		2x	}
181
182			inline void
183		1x	local_stream_write_op::do_cancel_impl(overlapped_op* base) noexcept
184			{
185		1x	auto* op = static_cast<local_stream_write_op*>(base);
186		1x	op->cancelled.store(true, std::memory_order_release);
187	1/2 ✓ Branch 3 → 4 taken 1 time. ✗ Branch 3 → 8 not taken.	1x	if (op->internal.socket_ != INVALID_SOCKET)
188			{
189	1/2 ✓ Branch 4 → 5 taken 1 time. ✗ Branch 4 → 6 not taken.	1x	::CancelIoEx(reinterpret_cast<HANDLE>(op->internal.socket_), op);
190			}
191		1x	}
192
193			inline void
194		1x	local_stream_wait_op::do_cancel_impl(overlapped_op* base) noexcept
195			{
196		1x	auto* op = static_cast<local_stream_wait_op*>(base);
197		1x	op->cancelled.store(true, std::memory_order_release);
198	1/2 ✓ Branch 3 → 4 taken 1 time. ✗ Branch 3 → 8 not taken.	1x	if (op->internal.socket_ != INVALID_SOCKET)
199			{
200	1/2 ✓ Branch 4 → 5 taken 1 time. ✗ Branch 4 → 6 not taken.	1x	::CancelIoEx(reinterpret_cast<HANDLE>(op->internal.socket_), op);
201			}
202		1x	op->internal.svc_.scheduler().cancel_wait(op);
203		1x	}
204
205			// ============================================================
206			// connect_op completion handler
207			// ============================================================
208
209			inline void
210		27x	local_stream_connect_op::do_complete(
211			void* owner,
212			scheduler_op* base,
213			std::uint32_t /bytes/,
214			std::uint32_t /error/)
215			{
216		27x	auto* op = static_cast<local_stream_connect_op*>(base);
217
218	2/2 ✓ Branch 2 → 3 taken 1 time. ✓ Branch 2 → 5 taken 26 times.	27x	if (!owner)
219			{
220		1x	op->destroy_awaiting();
221		1x	return;
222			}
223
224			bool success =
225	3/4 ✓ Branch 5 → 6 taken 22 times. ✓ Branch 5 → 9 taken 4 times. ✓ Branch 7 → 8 taken 22 times. ✗ Branch 7 → 9 not taken.	26x	(op->dwError == 0 && !op->cancelled.load(std::memory_order_acquire));
226	3/4 ✓ Branch 10 → 11 taken 22 times. ✓ Branch 10 → 18 taken 4 times. ✓ Branch 11 → 12 taken 22 times. ✗ Branch 11 → 18 not taken.	26x	if (success && op->internal.socket_ != INVALID_SOCKET)
227			{
228			// Required after ConnectEx
229		22x	::setsockopt(
230	1/1 ✓ Branch 12 → 13 taken 22 times.	22x	op->internal.socket_, SOL_SOCKET, SO_UPDATE_CONNECT_CONTEXT,
231			nullptr, 0);
232
233		22x	corosio::local_endpoint local_ep;
234		22x	sockaddr_storage local_storage{};
235		22x	int local_len = sizeof(local_storage);
236		44x	if (::getsockname(
237	1/1 ✓ Branch 13 → 14 taken 22 times.	22x	op->internal.socket_,
238	1/2 ✓ Branch 14 → 15 taken 22 times. ✗ Branch 14 → 16 not taken.	22x	reinterpret_cast<sockaddr*>(&local_storage), &local_len) == 0)
239		22x	local_ep = from_sockaddr_local(
240			local_storage, static_cast<socklen_t>(local_len));
241		22x	op->internal.set_endpoints(local_ep, op->target_endpoint);
242			}
243
244		26x	auto prevent_premature_destruction = std::move(op->object_ref_);
245	1/1 ✓ Branch 20 → 21 taken 26 times.	26x	op->invoke_handler();
246		26x	}
247
248			// ============================================================
249			// read_op completion handler
250			// ============================================================
251
252			inline void
253		22x	local_stream_read_op::do_complete(
254			void* owner,
255			scheduler_op* base,
256			std::uint32_t /bytes/,
257			std::uint32_t /error/)
258			{
259		22x	auto* op = static_cast<local_stream_read_op*>(base);
260
261	2/2 ✓ Branch 2 → 3 taken 3 times. ✓ Branch 2 → 5 taken 19 times.	22x	if (!owner)
262			{
263		3x	op->destroy_awaiting();
264		3x	return;
265			}
266
267		19x	auto prevent_premature_destruction = std::move(op->object_ref_);
268	1/1 ✓ Branch 7 → 8 taken 19 times.	19x	op->invoke_handler();
269		19x	}
270
271			// ============================================================
272			// write_op completion handler
273			// ============================================================
274
275			inline void
276		18x	local_stream_write_op::do_complete(
277			void* owner,
278			scheduler_op* base,
279			std::uint32_t /bytes/,
280			std::uint32_t /error/)
281			{
282		18x	auto* op = static_cast<local_stream_write_op*>(base);
283
284	2/2 ✓ Branch 2 → 3 taken 1 time. ✓ Branch 2 → 5 taken 17 times.	18x	if (!owner)
285			{
286		1x	op->destroy_awaiting();
287		1x	return;
288			}
289
290		17x	auto prevent_premature_destruction = std::move(op->object_ref_);
291	1/1 ✓ Branch 7 → 8 taken 17 times.	17x	op->invoke_handler();
292		17x	}
293
294			// ============================================================
295			// wait_op completion handler
296			// ============================================================
297
298			inline void
299		9x	local_stream_wait_op::do_complete(
300			void* owner,
301			scheduler_op* base,
302			std::uint32_t /bytes/,
303			std::uint32_t /error/)
304			{
305		9x	auto* op = static_cast<local_stream_wait_op*>(base);
306
307	2/2 ✓ Branch 2 → 3 taken 1 time. ✓ Branch 2 → 5 taken 8 times.	9x	if (!owner)
308			{
309		1x	op->destroy_awaiting();
310		1x	return;
311			}
312
313			// Only the zero-byte WSARecv read wait; write and error waits come
314			// from the poll reactor through this op too.
315	2/2 ✓ Branch 5 → 6 taken 6 times. ✓ Branch 5 → 8 taken 2 times.	8x	if (op->w == wait_type::read)
316		6x	op->dwError = normalize_read_wait_error(op->dwError);
317
318		8x	auto prevent_premature_destruction = std::move(op->object_ref_);
319	1/1 ✓ Branch 10 → 11 taken 8 times.	8x	op->invoke_handler();
320		8x	}
321
322			// ============================================================
323			// win_local_stream_socket
324			// ============================================================
325
326		140x	inline win_local_stream_socket::win_local_stream_socket(
327		140x	win_local_stream_service& svc) noexcept
328		140x	: svc_(svc)
329		140x	, conn_(*this)
330		140x	, rd_(*this)
331		140x	, wr_(*this)
332		280x	, wt_(*this)
333			{
334		140x	}
335
336			inline void
337		156x	win_local_stream_socket::reuse() noexcept
338			{
339	1/2 ✗ Branch 2 → 3 not taken. ✓ Branch 2 → 4 taken 156 times.	156x	BOOST_COROSIO_ASSERT(socket_ == INVALID_SOCKET);
340			// A connect completing on another thread while the socket closed
341			// can still have written these.
342		156x	local_endpoint_ = corosio::local_endpoint{};
343		156x	remote_endpoint_ = corosio::local_endpoint{};
344	1/2 ✗ Branch 8 → 9 not taken. ✓ Branch 8 → 10 taken 156 times.	156x	BOOST_COROSIO_ASSERT(!conn_.stop_cb);
345	1/2 ✗ Branch 12 → 13 not taken. ✓ Branch 12 → 14 taken 156 times.	156x	BOOST_COROSIO_ASSERT(!rd_.stop_cb);
346	1/2 ✗ Branch 16 → 17 not taken. ✓ Branch 16 → 18 taken 156 times.	156x	BOOST_COROSIO_ASSERT(!wr_.stop_cb);
347	1/2 ✗ Branch 20 → 21 not taken. ✓ Branch 20 → 22 taken 156 times.	156x	BOOST_COROSIO_ASSERT(!wt_.stop_cb);
348		156x	}
349
350			inline void
351		16x	win_local_stream_socket::set_socket(SOCKET s) noexcept
352			{
353		16x	socket_ = s;
354		16x	}
355
356			inline void
357		38x	win_local_stream_socket::set_endpoints(
358			corosio::local_endpoint local, corosio::local_endpoint remote) noexcept
359			{
360		38x	local_endpoint_ = local;
361		38x	remote_endpoint_ = remote;
362		38x	}
363
364			inline std::coroutine_handle<>
365		27x	win_local_stream_socket::connect(
366			capy::continuation& cont,
367			capy::executor_ref d,
368			corosio::local_endpoint ep,
369			std::stop_token token,
370			std::error_code* ec)
371			{
372		27x	conn_.object_ref_ = detail::object_ref(this);
373
374		27x	auto& op = conn_;
375		27x	op.reset();
376		27x	op.cont = &cont;
377		27x	op.ex = d;
378		27x	op.ec_out = ec;
379		27x	op.target_endpoint = ep;
380		27x	op.start(token);
381
382		27x	svc_.work_started();
383
384			// ConnectEx requires the socket to be bound. For AF_UNIX,
385			// bind to a family-only sockaddr_un (empty path).
386	1/2 ✓ Branch 9 → 10 taken 27 times. ✗ Branch 9 → 18 not taken.	27x	if (local_endpoint_.empty())
387			{
388		27x	un_sa_t bind_sa{};
389		27x	bind_sa.sun_family = AF_UNIX;
390		27x	socklen_t bind_len =
391			static_cast<socklen_t>(offsetof(un_sa_t, sun_path));
392
393	3/3 ✓ Branch 10 → 11 taken 27 times. ✓ Branch 11 → 12 taken 1 time. ✓ Branch 11 → 17 taken 26 times.	27x	if (::bind(socket_, reinterpret_cast<sockaddr*>(&bind_sa), bind_len) ==
394			SOCKET_ERROR)
395			{
396	1/1 ✓ Branch 12 → 13 taken 1 time.	1x	svc_.on_completion(&op, ::WSAGetLastError(), 0);
397		1x	return std::noop_coroutine();
398			}
399			}
400
401		26x	auto connect_ex = svc_.connect_ex();
402	2/2 ✓ Branch 19 → 20 taken 1 time. ✓ Branch 19 → 24 taken 25 times.	26x	if (!connect_ex)
403			{
404		1x	svc_.on_completion(&op, WSAEOPNOTSUPP, 0);
405		1x	return std::noop_coroutine();
406			}
407
408		25x	sockaddr_storage storage{};
409		25x	socklen_t addrlen = detail::to_sockaddr(ep, storage);
410
411	1/1 ✓ Branch 25 → 26 taken 25 times.	25x	BOOL result = connect_ex(
412			socket_, reinterpret_cast<sockaddr*>(&storage),
413			static_cast<int>(addrlen), nullptr, 0, nullptr, &op);
414
415	1/2 ✓ Branch 26 → 27 taken 25 times. ✗ Branch 26 → 33 not taken.	25x	if (!result)
416			{
417	1/1 ✓ Branch 27 → 28 taken 25 times.	25x	DWORD err = ::WSAGetLastError();
418	2/2 ✓ Branch 28 → 29 taken 2 times. ✓ Branch 28 → 33 taken 23 times.	25x	if (err != ERROR_IO_PENDING)
419			{
420		2x	svc_.on_completion(&op, err, 0);
421		2x	return std::noop_coroutine();
422			}
423			}
424
425			// Must precede on_pending: once it runs another thread may
426			// complete the op and recycle this impl.
427	1/2 ✗ Branch 34 → 35 not taken. ✓ Branch 34 → 36 taken 23 times.	23x	if (op.cancelled.load(std::memory_order_acquire))
428	0/1 ✗ Branch 35 → 36 not taken.	✗	::CancelIoEx(reinterpret_cast<HANDLE>(socket_), &op);
429
430		23x	svc_.on_pending(&op);
431
432		23x	return std::noop_coroutine();
433			}
434
435			inline std::coroutine_handle<>
436		22x	win_local_stream_socket::read_some(
437			capy::continuation& cont,
438			capy::executor_ref d,
439			buffer_param param,
440			std::stop_token token,
441			std::error_code* ec,
442			std::size_t* bytes_out)
443			{
444		22x	rd_.object_ref_ = detail::object_ref(this);
445
446		22x	auto& op = rd_;
447		22x	op.reset();
448		22x	op.is_read = true;
449		22x	op.cont = &cont;
450		22x	op.ex = d;
451		22x	op.ec_out = ec;
452		22x	op.bytes_out = bytes_out;
453		22x	op.start(token);
454
455		22x	svc_.work_started();
456
457			// Closed-object contract: complete with bad_file_descriptor
458			// without touching the kernel.
459	2/2 ✓ Branch 8 → 9 taken 2 times. ✓ Branch 8 → 13 taken 20 times.	22x	if (socket_ == INVALID_SOCKET)
460			{
461		2x	svc_.on_completion(&op, WSAEBADF, 0);
462		2x	return std::noop_coroutine();
463			}
464
465		20x	capy::mutable_buffer bufs[local_stream_read_op::max_buffers];
466		20x	op.wsabuf_count = static_cast<DWORD>(
467		20x	param.copy_to(bufs, local_stream_read_op::max_buffers));
468
469	2/2 ✓ Branch 14 → 15 taken 1 time. ✓ Branch 14 → 19 taken 19 times.	20x	if (op.wsabuf_count == 0)
470			{
471		1x	op.empty_buffer = true;
472		1x	svc_.on_completion(&op, 0, 0);
473		1x	return std::noop_coroutine();
474			}
475
476	2/2 ✓ Branch 24 → 20 taken 19 times. ✓ Branch 24 → 25 taken 19 times.	38x	for (DWORD i = 0; i < op.wsabuf_count; ++i)
477			{
478		19x	op.wsabufs[i].buf = static_cast<char*>(bufs[i].data());
479		19x	op.wsabufs[i].len = static_cast<ULONG>(
480		19x	(std::min)(bufs[i].size(), std::size_t(0x7fffffff)));
481			}
482
483		19x	op.flags = 0;
484
485		38x	int result = ::WSARecv(
486	1/1 ✓ Branch 25 → 26 taken 19 times.	19x	socket_, op.wsabufs, op.wsabuf_count, nullptr, &op.flags, &op, nullptr);
487
488	2/2 ✓ Branch 26 → 27 taken 16 times. ✓ Branch 26 → 33 taken 3 times.	19x	if (result == SOCKET_ERROR)
489			{
490	1/1 ✓ Branch 27 → 28 taken 16 times.	16x	DWORD err = ::WSAGetLastError();
491	2/2 ✓ Branch 28 → 29 taken 1 time. ✓ Branch 28 → 33 taken 15 times.	16x	if (err != WSA_IO_PENDING)
492			{
493		1x	svc_.on_completion(&op, err, 0);
494		1x	return std::noop_coroutine();
495			}
496			}
497
498			// Must precede on_pending: once it runs another thread may
499			// complete the op and recycle this impl.
500	1/2 ✗ Branch 34 → 35 not taken. ✓ Branch 34 → 36 taken 18 times.	18x	if (op.cancelled.load(std::memory_order_acquire))
501	0/1 ✗ Branch 35 → 36 not taken.	✗	::CancelIoEx(reinterpret_cast<HANDLE>(socket_), &op);
502
503		18x	svc_.on_pending(&op);
504
505		18x	return std::noop_coroutine();
506			}
507
508			inline std::coroutine_handle<>
509		18x	win_local_stream_socket::write_some(
510			capy::continuation& cont,
511			capy::executor_ref d,
512			buffer_param param,
513			std::stop_token token,
514			std::error_code* ec,
515			std::size_t* bytes_out)
516			{
517		18x	wr_.object_ref_ = detail::object_ref(this);
518
519		18x	auto& op = wr_;
520		18x	op.reset();
521		18x	op.cont = &cont;
522		18x	op.ex = d;
523		18x	op.ec_out = ec;
524		18x	op.bytes_out = bytes_out;
525		18x	op.start(token);
526
527		18x	svc_.work_started();
528
529			// Closed-object contract: complete with bad_file_descriptor
530			// without touching the kernel.
531	2/2 ✓ Branch 8 → 9 taken 2 times. ✓ Branch 8 → 13 taken 16 times.	18x	if (socket_ == INVALID_SOCKET)
532			{
533		2x	svc_.on_completion(&op, WSAEBADF, 0);
534		2x	return std::noop_coroutine();
535			}
536
537		16x	capy::mutable_buffer bufs[local_stream_write_op::max_buffers];
538		16x	op.wsabuf_count = static_cast<DWORD>(
539		16x	param.copy_to(bufs, local_stream_write_op::max_buffers));
540
541	2/2 ✓ Branch 14 → 15 taken 1 time. ✓ Branch 14 → 19 taken 15 times.	16x	if (op.wsabuf_count == 0)
542			{
543		1x	svc_.on_completion(&op, 0, 0);
544		1x	return std::noop_coroutine();
545			}
546
547	2/2 ✓ Branch 24 → 20 taken 15 times. ✓ Branch 24 → 25 taken 15 times.	30x	for (DWORD i = 0; i < op.wsabuf_count; ++i)
548			{
549		15x	op.wsabufs[i].buf = static_cast<char*>(bufs[i].data());
550		15x	op.wsabufs[i].len = static_cast<ULONG>(
551		15x	(std::min)(bufs[i].size(), std::size_t(0x7fffffff)));
552			}
553
554		30x	int result = ::WSASend(
555	1/1 ✓ Branch 25 → 26 taken 15 times.	15x	socket_, op.wsabufs, op.wsabuf_count, nullptr, 0, &op, nullptr);
556
557	2/2 ✓ Branch 26 → 27 taken 2 times. ✓ Branch 26 → 33 taken 13 times.	15x	if (result == SOCKET_ERROR)
558			{
559	1/1 ✓ Branch 27 → 28 taken 2 times.	2x	DWORD err = ::WSAGetLastError();
560	2/2 ✓ Branch 28 → 29 taken 1 time. ✓ Branch 28 → 33 taken 1 time.	2x	if (err != WSA_IO_PENDING)
561			{
562		1x	svc_.on_completion(&op, err, 0);
563		1x	return std::noop_coroutine();
564			}
565			}
566
567			// Must precede on_pending: once it runs another thread may
568			// complete the op and recycle this impl.
569	1/2 ✗ Branch 34 → 35 not taken. ✓ Branch 34 → 36 taken 14 times.	14x	if (op.cancelled.load(std::memory_order_acquire))
570	0/1 ✗ Branch 35 → 36 not taken.	✗	::CancelIoEx(reinterpret_cast<HANDLE>(socket_), &op);
571
572		14x	svc_.on_pending(&op);
573
574		14x	return std::noop_coroutine();
575			}
576
577			inline std::coroutine_handle<>
578		9x	win_local_stream_socket::wait(
579			capy::continuation& cont,
580			capy::executor_ref d,
581			wait_type w,
582			std::stop_token token,
583			std::error_code* ec)
584			{
585		9x	wt_.object_ref_ = detail::object_ref(this);
586
587		9x	auto& op = wt_;
588		9x	op.reset();
589		9x	op.cont = &cont;
590		9x	op.ex = d;
591		9x	op.ec_out = ec;
592		9x	op.bytes_out = nullptr;
593		9x	op.empty_buffer = true;
594		9x	op.w = w;
595		9x	op.start(token);
596
597		9x	svc_.work_started();
598
599			// Closed-object contract: complete with bad_file_descriptor
600			// without touching the kernel or the wait reactor.
601	2/2 ✓ Branch 8 → 9 taken 2 times. ✓ Branch 8 → 13 taken 7 times.	9x	if (socket_ == INVALID_SOCKET)
602			{
603		2x	svc_.on_completion(&op, WSAEBADF, 0);
604		2x	return std::noop_coroutine();
605			}
606
607	2/2 ✓ Branch 13 → 14 taken 5 times. ✓ Branch 13 → 29 taken 2 times.	7x	if (w == wait_type::read)
608			{
609			// Zero-byte WSARecv — completes when data is available
610			// without consuming any bytes.
611		5x	op.wsabuf = WSABUF{0, nullptr};
612		5x	op.flags = 0;
613
614			int result =
615		5x	::WSARecv(socket_, &op.wsabuf, 1, nullptr, &op.flags, &op, nullptr);
616
617	1/2 ✓ Branch 15 → 16 taken 5 times. ✗ Branch 15 → 22 not taken.	5x	if (result == SOCKET_ERROR)
618			{
619		5x	DWORD err = ::WSAGetLastError();
620	2/2 ✓ Branch 17 → 18 taken 1 time. ✓ Branch 17 → 22 taken 4 times.	5x	if (err != WSA_IO_PENDING)
621			{
622		1x	svc_.on_completion(&op, err, 0);
623		1x	return std::noop_coroutine();
624			}
625			}
626
627			// Must precede on_pending: once it runs another thread may
628			// complete the op and recycle this impl.
629	1/2 ✗ Branch 23 → 24 not taken. ✓ Branch 23 → 25 taken 4 times.	4x	if (op.cancelled.load(std::memory_order_acquire))
630		✗	::CancelIoEx(reinterpret_cast<HANDLE>(socket_), &op);
631
632		4x	svc_.on_pending(&op);
633
634		4x	return std::noop_coroutine();
635			}
636
637			// wait_type::write and wait_type::error: route through the
638			// auxiliary poll reactor. There is no overlapped primitive for
639			// "the send buffer has room" that does not also transfer bytes,
640			// and a write wait must report real writability.
641		2x	svc_.scheduler().wait_reactor().register_wait(socket_, w, &op);
642		2x	return std::noop_coroutine();
643			}
644
645			inline void
646		11x	win_local_stream_socket::cancel() noexcept
647			{
648	1/2 ✓ Branch 2 → 3 taken 11 times. ✗ Branch 2 → 4 not taken.	11x	if (socket_ != INVALID_SOCKET)
649			{
650		11x	::CancelIoEx(reinterpret_cast<HANDLE>(socket_), nullptr);
651			}
652
653		11x	conn_.request_cancel();
654		11x	rd_.request_cancel();
655		11x	wr_.request_cancel();
656		11x	wt_.request_cancel();
657		11x	svc_.scheduler().cancel_wait(&wt_);
658		11x	}
659
660			inline void
661		737x	win_local_stream_socket::close_socket() noexcept
662			{
663			// Flag every op cancelled before closing so a closesocket-delivered
664			// ERROR_NETNAME_DELETED is short-circuited to canceled rather than mapped
665			// to connection_reset by iocp_make_err (see win_tcp_socket close_socket).
666		737x	conn_.request_cancel();
667		737x	rd_.request_cancel();
668		737x	wr_.request_cancel();
669		737x	wt_.request_cancel();
670		737x	svc_.scheduler().cancel_wait(&wt_);
671
672	2/2 ✓ Branch 8 → 9 taken 105 times. ✓ Branch 8 → 12 taken 632 times.	737x	if (socket_ != INVALID_SOCKET)
673			{
674		105x	::CancelIoEx(reinterpret_cast<HANDLE>(socket_), nullptr);
675		105x	::closesocket(socket_);
676		105x	socket_ = INVALID_SOCKET;
677			}
678
679		737x	local_endpoint_ = corosio::local_endpoint{};
680		737x	remote_endpoint_ = corosio::local_endpoint{};
681		737x	}
682
683			inline std::error_code
684		4x	win_local_stream_socket::shutdown(
685			local_stream_socket::shutdown_type what) noexcept
686			{
687			int how;
688	3/4 ✓ Branch 2 → 3 taken 1 time. ✓ Branch 2 → 4 taken 2 times. ✓ Branch 2 → 5 taken 1 time. ✗ Branch 2 → 6 not taken.	4x	switch (what)
689			{
690		1x	case local_stream_socket::shutdown_receive:
691		1x	how = SD_RECEIVE;
692		1x	break;
693		2x	case local_stream_socket::shutdown_send:
694		2x	how = SD_SEND;
695		2x	break;
696		1x	case local_stream_socket::shutdown_both:
697		1x	how = SD_BOTH;
698		1x	break;
699		✗	default:
700		✗	return make_err(WSAEINVAL);
701			}
702	2/2 ✓ Branch 8 → 9 taken 1 time. ✓ Branch 8 → 11 taken 3 times.	4x	if (::shutdown(socket_, how) != 0)
703		1x	return make_err(WSAGetLastError());
704		3x	return {};
705			}
706
707			inline native_handle_type
708		864x	win_local_stream_socket::native_handle() const noexcept
709			{
710		864x	return static_cast<native_handle_type>(socket_);
711			}
712
713			inline native_handle_type
714		3x	win_local_stream_socket::release_socket() noexcept
715			{
716		3x	SOCKET s = socket_;
717	1/2 ✓ Branch 2 → 3 taken 3 times. ✗ Branch 2 → 8 not taken.	3x	if (s != INVALID_SOCKET)
718			{
719		3x	cancel();
720			// Sever the port association so the descriptor can be
721			// adopted again; best-effort, the caller keeps a working
722			// socket either way.
723		3x	dissociate_from_iocp(s);
724		3x	socket_ = INVALID_SOCKET;
725		3x	local_endpoint_ = corosio::local_endpoint{};
726		3x	remote_endpoint_ = corosio::local_endpoint{};
727			}
728		3x	return static_cast<native_handle_type>(s);
729			}
730
731			inline std::error_code
732		3x	win_local_stream_socket::set_option(
733			int level, int optname, void const* data, std::size_t size) noexcept
734			{
735		3x	if (::setsockopt(
736			socket_, level, optname, reinterpret_cast<char const*>(data),
737	2/2 ✓ Branch 3 → 4 taken 1 time. ✓ Branch 3 → 6 taken 2 times.	3x	static_cast<int>(size)) != 0)
738		1x	return make_err(WSAGetLastError());
739		2x	return {};
740			}
741
742			inline std::error_code
743		2x	win_local_stream_socket::get_option(
744			int level, int optname, void* data, std::size_t* size) const noexcept
745			{
746		2x	int len = static_cast<int>(*size);
747		4x	if (::getsockopt(
748	2/2 ✓ Branch 3 → 4 taken 1 time. ✓ Branch 3 → 6 taken 1 time.	2x	socket_, level, optname, reinterpret_cast<char*>(data), &len) != 0)
749		1x	return make_err(WSAGetLastError());
750		1x	*size = static_cast<std::size_t>(len);
751		1x	return {};
752			}
753
754			inline corosio::local_endpoint
755		2x	win_local_stream_socket::local_endpoint() const noexcept
756			{
757		2x	return local_endpoint_;
758			}
759
760			inline corosio::local_endpoint
761		2x	win_local_stream_socket::remote_endpoint() const noexcept
762			{
763		2x	return remote_endpoint_;
764			}
765
766			inline void
767		296x	win_local_stream_socket::retire() noexcept
768			{
769		296x	svc_.pool_.recycle(this);
770		296x	}
771
772			// ============================================================
773			// win_local_stream_service
774			// ============================================================
775
776		94x	inline win_local_stream_service::win_local_stream_service(
777		94x	capy::execution_context& ctx)
778		188x	: tcp_svc_(ctx.use_service<win_tcp_service>())
779	1/1 ✓ Branch 5 → 6 taken 94 times.	94x	, sched_(ctx.use_service<win_scheduler>())
780	1/1 ✓ Branch 4 → 5 taken 94 times.	188x	, iocp_(sched_.native_handle())
781			{
782		94x	}
783
784			inline void
785		94x	win_local_stream_service::shutdown()
786			{
787	1/1 ✓ Branch 2 → 3 taken 94 times.	94x	pool_.shutdown(
788		7x	[](win_local_stream_socket* impl) { impl->close_socket(); });
789
790	1/1 ✓ Branch 3 → 4 taken 94 times.	94x	acceptor_pool_.shutdown(
791		2x	[](win_local_stream_acceptor* impl) { impl->close_socket(); });
792		94x	}
793
794			inline io_object::implementation*
795		296x	win_local_stream_service::construct()
796			{
797		296x	return pool_.acquire(*this);
798			}
799
800			inline void
801		296x	win_local_stream_service::destroy(io_object::implementation* p)
802			{
803	1/2 ✗ Branch 2 → 3 not taken. ✓ Branch 2 → 4 taken 296 times.	296x	if (!p)
804		✗	return;
805		296x	auto* s = static_cast<win_local_stream_socket*>(p);
806		296x	s->close_socket();
807		296x	release(s);
808			}
809
810			inline void
811		389x	win_local_stream_service::close(io_object::handle& h)
812			{
813		389x	static_cast<win_local_stream_socket*>(h.get())->close_socket();
814		389x	}
815
816			inline std::error_code
817		45x	win_local_stream_service::open_socket(
818			local_stream_socket::implementation& impl,
819			int family,
820			int type,
821			int protocol)
822			{
823		45x	auto& sock = static_cast<win_local_stream_socket&>(impl);
824		45x	return open_socket_internal(sock, family, type, protocol);
825			}
826
827			inline std::error_code
828		94x	win_local_stream_service::assign_socket(
829			local_stream_socket::implementation& impl, native_handle_type fd)
830			{
831		94x	auto& sock = static_cast<win_local_stream_socket&>(impl);
832
833		94x	SOCKET s = static_cast<SOCKET>(fd);
834	1/2 ✗ Branch 2 → 3 not taken. ✓ Branch 2 → 4 taken 94 times.	94x	if (s == INVALID_SOCKET)
835		✗	return make_err(WSAENOTSOCK);
836
837			// SO_PROTOCOL_INFOW works on an unbound socket, unlike getsockname
838			// (WSAEINVAL until bind/connect names it) -- connect_pair hands in
839			// a socket that reached connected state without an explicit bind.
840		94x	WSAPROTOCOL_INFOW proto_info{};
841		94x	int proto_len = sizeof(proto_info);
842	1/1 ✓ Branch 4 → 5 taken 94 times.	94x	if (::getsockopt(
843			s, SOL_SOCKET, SO_PROTOCOL_INFOW,
844	2/2 ✓ Branch 5 → 6 taken 27 times. ✓ Branch 5 → 8 taken 67 times.	94x	reinterpret_cast<char*>(&proto_info), &proto_len) != 0)
845	1/1 ✓ Branch 6 → 7 taken 27 times.	27x	return make_err(::WSAGetLastError());
846	1/2 ✗ Branch 8 → 9 not taken. ✓ Branch 8 → 10 taken 67 times.	67x	if (proto_info.iAddressFamily != AF_UNIX)
847		✗	return make_err(WSAEAFNOSUPPORT);
848	1/2 ✗ Branch 10 → 11 not taken. ✓ Branch 10 → 12 taken 67 times.	67x	if (proto_info.iSocketType != SOCK_STREAM)
849		✗	return make_err(WSAEPROTOTYPE);
850
851	2/2 ✓ Branch 13 → 14 taken 1 time. ✓ Branch 13 → 15 taken 66 times.	67x	if (has_skip_on_success(reinterpret_cast<HANDLE>(s)))
852		1x	return std::make_error_code(std::errc::operation_not_supported);
853
854		132x	HANDLE result = ::CreateIoCompletionPort(
855	1/1 ✓ Branch 15 → 16 taken 66 times.	66x	reinterpret_cast<HANDLE>(s), static_cast<HANDLE>(iocp_), key_io, 0);
856	2/2 ✓ Branch 16 → 17 taken 9 times. ✓ Branch 16 → 19 taken 57 times.	66x	if (result == nullptr)
857	1/1 ✓ Branch 17 → 18 taken 9 times.	9x	return make_err(::GetLastError());
858
859		57x	sock.socket_ = s;
860
861		57x	sockaddr_storage local{};
862		57x	int local_len = sizeof(local);
863		57x	corosio::local_endpoint lep{}, rep{};
864	2/3 ✓ Branch 19 → 20 taken 57 times. ✓ Branch 20 → 21 taken 57 times. ✗ Branch 20 → 22 not taken.	57x	if (::getsockname(s, reinterpret_cast<sockaddr*>(&local), &local_len) == 0)
865		57x	lep = from_sockaddr_local(local, static_cast<socklen_t>(local_len));
866		57x	sockaddr_storage remote{};
867		57x	int remote_len = sizeof(remote);
868	2/3 ✓ Branch 22 → 23 taken 57 times. ✓ Branch 23 → 24 taken 57 times. ✗ Branch 23 → 25 not taken.	57x	if (::getpeername(s, reinterpret_cast<sockaddr*>(&remote), &remote_len) ==
869			0)
870		57x	rep = from_sockaddr_local(remote, static_cast<socklen_t>(remote_len));
871		57x	sock.local_endpoint_ = lep;
872		57x	sock.remote_endpoint_ = rep;
873		57x	return {};
874			}
875
876			inline std::error_code
877		45x	win_local_stream_service::open_socket_internal(
878			win_local_stream_socket& impl, int family, int type, int protocol)
879			{
880		45x	impl.close_socket();
881
882			SOCKET sock =
883		45x	::WSASocketW(family, type, protocol, nullptr, 0, WSA_FLAG_OVERLAPPED);
884
885	2/2 ✓ Branch 4 → 5 taken 1 time. ✓ Branch 4 → 7 taken 44 times.	45x	if (sock == INVALID_SOCKET)
886		1x	return make_err(::WSAGetLastError());
887
888			// No IPV6_V6ONLY for AF_UNIX
889
890		88x	HANDLE result = ::CreateIoCompletionPort(
891		44x	reinterpret_cast<HANDLE>(sock), static_cast<HANDLE>(iocp_), key_io, 0);
892
893	2/2 ✓ Branch 8 → 9 taken 9 times. ✓ Branch 8 → 12 taken 35 times.	44x	if (result == nullptr)
894			{
895		9x	DWORD dwError = ::GetLastError();
896		9x	::closesocket(sock);
897		9x	return make_err(dwError);
898			}
899
900		35x	impl.socket_ = sock;
901		35x	return {};
902			}
903
904			inline void*
905		26x	win_local_stream_service::native_handle() const noexcept
906			{
907		26x	return iocp_;
908			}
909
910			inline LPFN_CONNECTEX
911		26x	win_local_stream_service::connect_ex() const noexcept
912			{
913		26x	return tcp_svc_.connect_ex();
914			}
915
916			inline LPFN_ACCEPTEX
917		25x	win_local_stream_service::accept_ex() const noexcept
918			{
919		25x	return tcp_svc_.accept_ex();
920			}
921
922			inline void
923			win_local_stream_service::post(overlapped_op* op)
924			{
925			sched_.post(op);
926			}
927
928			inline void
929		81x	win_local_stream_service::on_pending(overlapped_op* op) noexcept
930			{
931		81x	sched_.on_pending(op);
932		81x	}
933
934			inline void
935		22x	win_local_stream_service::on_completion(
936			overlapped_op* op, DWORD error, DWORD bytes) noexcept
937			{
938		22x	sched_.on_completion(op, error, bytes);
939		22x	}
940
941			inline void
942		110x	win_local_stream_service::work_started() noexcept
943			{
944		110x	sched_.work_started();
945		110x	}
946
947			inline void
948			win_local_stream_service::work_finished() noexcept
949			{
950			sched_.work_finished();
951			}
952
953			inline win_local_stream_acceptor*
954		97x	win_local_stream_service::acquire_acceptor_impl()
955			{
956		97x	return acceptor_pool_.acquire(*this);
957			}
958
959			inline std::error_code
960		66x	win_local_stream_service::open_acceptor_socket(
961			win_local_stream_acceptor& impl, int family, int type, int protocol)
962			{
963		66x	impl.close_socket();
964
965			SOCKET sock =
966		66x	::WSASocketW(family, type, protocol, nullptr, 0, WSA_FLAG_OVERLAPPED);
967
968	2/2 ✓ Branch 4 → 5 taken 2 times. ✓ Branch 4 → 7 taken 64 times.	66x	if (sock == INVALID_SOCKET)
969		2x	return make_err(::WSAGetLastError());
970
971		128x	HANDLE result = ::CreateIoCompletionPort(
972		64x	reinterpret_cast<HANDLE>(sock), static_cast<HANDLE>(iocp_), key_io, 0);
973
974	2/2 ✓ Branch 8 → 9 taken 9 times. ✓ Branch 8 → 12 taken 55 times.	64x	if (result == nullptr)
975			{
976		9x	DWORD dwError = ::GetLastError();
977		9x	::closesocket(sock);
978		9x	return make_err(dwError);
979			}
980
981		55x	impl.socket_ = sock;
982		55x	return {};
983			}
984
985			inline std::error_code
986		22x	win_local_stream_service::assign_acceptor_socket(
987			win_local_stream_acceptor& impl, native_handle_type fd)
988			{
989		22x	SOCKET sock = static_cast<SOCKET>(fd);
990	1/2 ✗ Branch 2 → 3 not taken. ✓ Branch 2 → 4 taken 22 times.	22x	if (sock == INVALID_SOCKET)
991		✗	return make_err(WSAENOTSOCK);
992
993			// SO_PROTOCOL_INFOW works on an unbound socket, unlike getsockname
994			// (WSAEINVAL until bind names it).
995		22x	WSAPROTOCOL_INFOW proto_info{};
996		22x	int proto_len = sizeof(proto_info);
997	1/1 ✓ Branch 4 → 5 taken 22 times.	22x	if (::getsockopt(
998			sock, SOL_SOCKET, SO_PROTOCOL_INFOW,
999	2/2 ✓ Branch 5 → 6 taken 9 times. ✓ Branch 5 → 8 taken 13 times.	22x	reinterpret_cast<char*>(&proto_info), &proto_len) != 0)
1000	1/1 ✓ Branch 6 → 7 taken 9 times.	9x	return make_err(::WSAGetLastError());
1001	2/2 ✓ Branch 8 → 9 taken 1 time. ✓ Branch 8 → 10 taken 12 times.	13x	if (proto_info.iAddressFamily != AF_UNIX)
1002		1x	return make_err(WSAEAFNOSUPPORT);
1003	1/2 ✗ Branch 10 → 11 not taken. ✓ Branch 10 → 12 taken 12 times.	12x	if (proto_info.iSocketType != SOCK_STREAM)
1004		✗	return make_err(WSAEPROTOTYPE);
1005
1006	2/2 ✓ Branch 13 → 14 taken 1 time. ✓ Branch 13 → 15 taken 11 times.	12x	if (has_skip_on_success(reinterpret_cast<HANDLE>(sock)))
1007		1x	return std::make_error_code(std::errc::operation_not_supported);
1008
1009		22x	HANDLE result = ::CreateIoCompletionPort(
1010	1/1 ✓ Branch 15 → 16 taken 11 times.	11x	reinterpret_cast<HANDLE>(sock), static_cast<HANDLE>(iocp_), key_io, 0);
1011	2/2 ✓ Branch 16 → 17 taken 9 times. ✓ Branch 16 → 19 taken 2 times.	11x	if (result == nullptr)
1012	1/1 ✓ Branch 17 → 18 taken 9 times.	9x	return make_err(::GetLastError());
1013
1014		2x	impl.socket_ = sock;
1015
1016		2x	sockaddr_storage local{};
1017		2x	int local_len = sizeof(local);
1018		2x	corosio::local_endpoint lep{};
1019	2/3 ✓ Branch 19 → 20 taken 2 times. ✓ Branch 20 → 21 taken 2 times. ✗ Branch 20 → 22 not taken.	2x	if (::getsockname(sock, reinterpret_cast<sockaddr*>(&local), &local_len) ==
1020			0)
1021		2x	lep = from_sockaddr_local(local, static_cast<socklen_t>(local_len));
1022		2x	impl.set_local_endpoint(lep);
1023
1024		2x	return {};
1025			}
1026
1027			inline std::error_code
1028		49x	win_local_stream_service::bind_acceptor(
1029			win_local_stream_acceptor& impl, corosio::local_endpoint ep)
1030			{
1031			// Reject abstract sockets on Windows
1032	2/2 ✓ Branch 3 → 4 taken 1 time. ✓ Branch 3 → 5 taken 48 times.	49x	if (ep.is_abstract())
1033		1x	return std::make_error_code(std::errc::operation_not_supported);
1034
1035		48x	SOCKET sock = impl.socket_;
1036
1037		48x	sockaddr_storage storage{};
1038		48x	socklen_t addrlen = detail::to_sockaddr(ep, storage);
1039	1/1 ✓ Branch 6 → 7 taken 48 times.	48x	if (::bind(
1040			sock, reinterpret_cast<sockaddr*>(&storage),
1041	2/2 ✓ Branch 7 → 8 taken 4 times. ✓ Branch 7 → 10 taken 44 times.	48x	static_cast<int>(addrlen)) == SOCKET_ERROR)
1042	1/1 ✓ Branch 8 → 9 taken 4 times.	4x	return make_err(::WSAGetLastError());
1043
1044		44x	impl.set_local_endpoint(ep);
1045		44x	return {};
1046			}
1047
1048			inline std::error_code
1049		39x	win_local_stream_service::listen_acceptor(
1050			win_local_stream_acceptor& impl, int backlog)
1051			{
1052		39x	SOCKET sock = impl.socket_;
1053
1054	2/2 ✓ Branch 3 → 4 taken 2 times. ✓ Branch 3 → 6 taken 37 times.	39x	if (::listen(sock, backlog) == SOCKET_ERROR)
1055		2x	return make_err(::WSAGetLastError());
1056
1057		37x	return {};
1058			}
1059
1060			} // namespace boost::corosio::detail
1061
1062			#endif // BOOST_COROSIO_HAS_IOCP
1063
1064			#endif // BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_LOCAL_STREAM_SERVICE_HPP
1065