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

69.9% Lines (332/475) 78.1% List of functions (50/64) 50.3% Branches (81/161)

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

win_local_stream_service.hpp

f(x) Functions (64)

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