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

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