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

2.6% Lines (6/231) 6.5% List of functions (2/31) 0.0% Branches (0/113)

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

win_local_stream_acceptor_service.hpp

f(x) Functions (31)

Function Calls Lines Blocks

Line	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_ACCEPTOR_SERVICE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_LOCAL_STREAM_ACCEPTOR_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/except.hpp>
19		#include <boost/corosio/detail/local_stream_acceptor_service.hpp>
20		#include <boost/capy/ex/execution_context.hpp>
21
22		#include <boost/corosio/native/detail/iocp/win_local_stream_acceptor.hpp>
23		#include <boost/corosio/native/detail/iocp/win_local_stream_service.hpp>
24
25		#include <boost/corosio/native/detail/iocp/win_scheduler.hpp>
26		#include <boost/corosio/native/detail/iocp/win_completion_key.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		/* IOCP local stream acceptor service.
37
38		Delegates acceptor lifecycle management to win_local_stream_service
39		and provides the local_stream_acceptor_service virtual interface.
40		*/
41		class BOOST_COROSIO_DECL win_local_stream_acceptor_service final
42		: public local_stream_acceptor_service
43		{
44		public:
45		win_local_stream_acceptor_service(
46		capy::execution_context& ctx, win_local_stream_service& svc);
47
48		io_object::implementation* construct() override;
49
50		void destroy(io_object::implementation* p) override;
51
52		void close(io_object::handle& h) override;
53
54		std::error_code open_acceptor_socket(
55		local_stream_acceptor::implementation& impl,
56		int family, int type, int protocol) override;
57
58		std::error_code
59		bind_acceptor(
60		local_stream_acceptor::implementation& impl,
61		corosio::local_endpoint ep) override;
62
63		std::error_code
64		listen_acceptor(
65		local_stream_acceptor::implementation& impl, int backlog) override;
66
67		void shutdown() override;
68
69		private:
70		win_local_stream_service& svc_;
71		};
72
73		// ============================================================
74		// local_stream_accept_op
75		// ============================================================
76
77	✗	inline local_stream_accept_op::local_stream_accept_op() noexcept
78	✗	: overlapped_op(&do_complete)
79		{
80	✗	cancel_func_ = &do_cancel_impl;
81	✗	}
82
83		inline void
84	✗	local_stream_accept_op::do_cancel_impl(overlapped_op* base) noexcept
85		{
86	✗	auto* op = static_cast<local_stream_accept_op*>(base);
87	✗	if (op->listen_socket != INVALID_SOCKET)
88		{
89	✗	::CancelIoEx(reinterpret_cast<HANDLE>(op->listen_socket), op);
90		}
91	✗	}
92
93		// accept_op completion handler
94
95		inline void
96	✗	local_stream_accept_op::do_complete(
97		void* owner,
98		scheduler_op* base,
99		std::uint32_t /bytes/,
100		std::uint32_t /error/)
101		{
102	✗	auto* op = static_cast<local_stream_accept_op*>(base);
103
104	✗	if (!owner)
105		{
106	✗	if (op->accepted_socket != INVALID_SOCKET)
107		{
108	✗	::closesocket(op->accepted_socket);
109	✗	op->accepted_socket = INVALID_SOCKET;
110		}
111
112	✗	if (op->peer_wrapper)
113		{
114	✗	op->peer_wrapper->close_internal();
115	✗	op->peer_wrapper = nullptr;
116		}
117
118	✗	op->cleanup_only();
119	✗	op->acceptor_ptr.reset();
120	✗	return;
121		}
122
123	✗	op->stop_cb.reset();
124
125		bool success =
126	✗	(op->dwError == 0 && !op->cancelled.load(std::memory_order_acquire));
127
128	✗	if (op->ec_out)
129		{
130	✗	if (op->cancelled.load(std::memory_order_acquire))
131	✗	*op->ec_out = capy::error::canceled;
132	✗	else if (op->dwError != 0)
133	✗	*op->ec_out = make_err(op->dwError);
134		else
135	✗	*op->ec_out = {};
136		}
137
138	✗	if (success && op->accepted_socket != INVALID_SOCKET && op->peer_wrapper)
139		{
140	✗	::setsockopt(
141		op->accepted_socket, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT,
142	✗	reinterpret_cast<char*>(&op->listen_socket), sizeof(SOCKET));
143
144	✗	op->peer_wrapper->get_internal()->set_socket(op->accepted_socket);
145
146	✗	sockaddr_storage local_storage{};
147	✗	int local_len = sizeof(local_storage);
148	✗	sockaddr_storage remote_storage{};
149	✗	int remote_len = sizeof(remote_storage);
150
151	✗	corosio::local_endpoint local_ep, remote_ep;
152	✗	if (::getsockname(
153		op->accepted_socket,
154	✗	reinterpret_cast<sockaddr*>(&local_storage), &local_len) == 0)
155	✗	local_ep = from_sockaddr_local(
156		local_storage, static_cast<socklen_t>(local_len));
157	✗	if (::getpeername(
158		op->accepted_socket,
159	✗	reinterpret_cast<sockaddr*>(&remote_storage), &remote_len) == 0)
160	✗	remote_ep = from_sockaddr_local(
161		remote_storage, static_cast<socklen_t>(remote_len));
162
163	✗	op->peer_wrapper->get_internal()->set_endpoints(local_ep, remote_ep);
164	✗	op->accepted_socket = INVALID_SOCKET;
165
166	✗	if (op->impl_out)
167	✗	*op->impl_out = op->peer_wrapper;
168	✗	}
169		else
170		{
171	✗	if (op->accepted_socket != INVALID_SOCKET)
172		{
173	✗	::closesocket(op->accepted_socket);
174	✗	op->accepted_socket = INVALID_SOCKET;
175		}
176
177	✗	if (op->peer_wrapper)
178		{
179	✗	op->acceptor_ptr->socket_service().destroy(op->peer_wrapper);
180	✗	op->peer_wrapper = nullptr;
181		}
182
183	✗	if (op->impl_out)
184	✗	*op->impl_out = nullptr;
185		}
186
187	✗	op->cont_op.cont.h = op->h;
188	✗	auto saved_ex = op->ex;
189	✗	auto prevent_premature_destruction = std::move(op->acceptor_ptr);
190
191	✗	dispatch_coro(saved_ex, op->cont_op.cont).resume();
192	✗	}
193
194		// ============================================================
195		// win_local_stream_acceptor_internal
196		// ============================================================
197
198	✗	inline win_local_stream_acceptor_internal::win_local_stream_acceptor_internal(
199	✗	win_local_stream_service& svc) noexcept
200	✗	: svc_(svc)
201		{
202	✗	}
203
204	✗	inline win_local_stream_acceptor_internal::~win_local_stream_acceptor_internal()
205		{
206	✗	svc_.unregister_acceptor_impl(*this);
207	✗	}
208
209		inline win_local_stream_service&
210	✗	win_local_stream_acceptor_internal::socket_service() noexcept
211		{
212	✗	return svc_;
213		}
214
215		inline SOCKET
216	✗	win_local_stream_acceptor_internal::native_handle() const noexcept
217		{
218	✗	return socket_;
219		}
220
221		inline corosio::local_endpoint
222	✗	win_local_stream_acceptor_internal::local_endpoint() const noexcept
223		{
224	✗	return local_endpoint_;
225		}
226
227		inline bool
228	✗	win_local_stream_acceptor_internal::is_open() const noexcept
229		{
230	✗	return socket_ != INVALID_SOCKET;
231		}
232
233		inline void
234	✗	win_local_stream_acceptor_internal::set_local_endpoint(
235		corosio::local_endpoint ep) noexcept
236		{
237	✗	local_endpoint_ = ep;
238	✗	}
239
240		inline void
241	✗	win_local_stream_acceptor_internal::cancel() noexcept
242		{
243	✗	if (socket_ != INVALID_SOCKET)
244	✗	::CancelIoEx(reinterpret_cast<HANDLE>(socket_), nullptr);
245	✗	acc_.request_cancel();
246	✗	}
247
248		inline void
249	✗	win_local_stream_acceptor_internal::close_socket() noexcept
250		{
251	✗	if (socket_ != INVALID_SOCKET)
252		{
253	✗	::CancelIoEx(reinterpret_cast<HANDLE>(socket_), nullptr);
254	✗	::closesocket(socket_);
255	✗	socket_ = INVALID_SOCKET;
256		}
257	✗	local_endpoint_ = corosio::local_endpoint{};
258	✗	}
259
260		inline std::coroutine_handle<>
261	✗	win_local_stream_acceptor_internal::accept(
262		std::coroutine_handle<> h,
263		capy::executor_ref d,
264		std::stop_token token,
265		std::error_code* ec,
266		io_object::implementation** impl_out)
267		{
268	✗	acc_.acceptor_ptr = shared_from_this();
269
270	✗	auto& op = acc_;
271	✗	op.reset();
272	✗	op.h = h;
273	✗	op.ex = d;
274	✗	op.ec_out = ec;
275	✗	op.impl_out = impl_out;
276	✗	op.start(token);
277
278	✗	svc_.work_started();
279
280		// Create wrapper for the peer socket
281	✗	auto* peer_ptr = svc_.construct();
282	✗	if (!peer_ptr)
283		{
284	✗	svc_.on_completion(&op, ERROR_OUTOFMEMORY, 0);
285	✗	return std::noop_coroutine();
286		}
287	✗	auto& peer_wrapper = static_cast<win_local_stream_socket&>(*peer_ptr);
288
289		// Always AF_UNIX for local sockets
290	✗	SOCKET accepted = ::WSASocketW(
291		AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
292
293	✗	if (accepted == INVALID_SOCKET)
294		{
295	✗	svc_.destroy(&peer_wrapper);
296	✗	svc_.on_completion(&op, ::WSAGetLastError(), 0);
297	✗	return std::noop_coroutine();
298		}
299
300	✗	HANDLE result = ::CreateIoCompletionPort(
301	✗	reinterpret_cast<HANDLE>(accepted), svc_.native_handle(), key_io, 0);
302
303	✗	if (result == nullptr)
304		{
305	✗	DWORD err = ::GetLastError();
306	✗	::closesocket(accepted);
307	✗	svc_.destroy(&peer_wrapper);
308	✗	svc_.on_completion(&op, err, 0);
309	✗	return std::noop_coroutine();
310		}
311
312	✗	op.accepted_socket = accepted;
313	✗	op.peer_wrapper = &peer_wrapper;
314	✗	op.listen_socket = socket_;
315
316	✗	auto accept_ex = svc_.accept_ex();
317	✗	if (!accept_ex)
318		{
319	✗	::closesocket(accepted);
320	✗	svc_.destroy(&peer_wrapper);
321	✗	op.peer_wrapper = nullptr;
322	✗	op.accepted_socket = INVALID_SOCKET;
323	✗	svc_.on_completion(&op, WSAEOPNOTSUPP, 0);
324	✗	return std::noop_coroutine();
325		}
326
327		// AcceptEx address buffer sized for sockaddr_un
328	✗	DWORD addr_size =
329		static_cast<DWORD>(sizeof(un_sa_t) + 16);
330	✗	DWORD bytes_received = 0;
331
332	✗	BOOL ok = accept_ex(
333	✗	socket_, accepted, op.addr_buf, 0, addr_size, addr_size,
334		&bytes_received, &op);
335
336	✗	if (!ok)
337		{
338	✗	DWORD err = ::WSAGetLastError();
339	✗	if (err != ERROR_IO_PENDING)
340		{
341	✗	::closesocket(accepted);
342	✗	svc_.destroy(&peer_wrapper);
343	✗	op.peer_wrapper = nullptr;
344	✗	op.accepted_socket = INVALID_SOCKET;
345	✗	svc_.on_completion(&op, err, 0);
346	✗	return std::noop_coroutine();
347		}
348		}
349
350	✗	svc_.on_pending(&op);
351
352		// Re-check cancellation after I/O is pending
353	✗	if (op.cancelled.load(std::memory_order_acquire))
354	✗	::CancelIoEx(reinterpret_cast<HANDLE>(socket_), &op);
355
356	✗	return std::noop_coroutine();
357		}
358
359		// ============================================================
360		// win_local_stream_acceptor (wrapper)
361		// ============================================================
362
363	✗	inline win_local_stream_acceptor::win_local_stream_acceptor(
364	✗	std::shared_ptr<win_local_stream_acceptor_internal> internal) noexcept
365	✗	: internal_(std::move(internal))
366		{
367	✗	}
368
369		inline void
370	✗	win_local_stream_acceptor::close_internal() noexcept
371		{
372	✗	if (internal_)
373		{
374	✗	internal_->close_socket();
375	✗	internal_.reset();
376		}
377	✗	}
378
379		inline std::coroutine_handle<>
380	✗	win_local_stream_acceptor::accept(
381		std::coroutine_handle<> h,
382		capy::executor_ref d,
383		std::stop_token token,
384		std::error_code* ec,
385		io_object::implementation** impl_out)
386		{
387	✗	return internal_->accept(h, d, token, ec, impl_out);
388		}
389
390		inline corosio::local_endpoint
391	✗	win_local_stream_acceptor::local_endpoint() const noexcept
392		{
393	✗	return internal_->local_endpoint();
394		}
395
396		inline bool
397	✗	win_local_stream_acceptor::is_open() const noexcept
398		{
399	✗	return internal_ && internal_->is_open();
400		}
401
402		inline void
403	✗	win_local_stream_acceptor::cancel() noexcept
404		{
405	✗	if (internal_)
406	✗	internal_->cancel();
407	✗	}
408
409		inline native_handle_type
410	✗	win_local_stream_acceptor::release_socket() noexcept
411		{
412	✗	if (!internal_)
413	✗	return static_cast<native_handle_type>(INVALID_SOCKET);
414	✗	SOCKET s = internal_->socket_;
415	✗	if (s != INVALID_SOCKET)
416		{
417	✗	internal_->cancel();
418	✗	internal_->socket_ = INVALID_SOCKET;
419	✗	internal_->local_endpoint_ = corosio::local_endpoint{};
420		}
421	✗	return static_cast<native_handle_type>(s);
422		}
423
424		inline std::error_code
425	✗	win_local_stream_acceptor::set_option(
426		int level, int optname, void const* data, std::size_t size) noexcept
427		{
428	✗	if (!internal_ \|\| !internal_->is_open())
429	✗	return make_err(WSAENOTSOCK);
430	✗	if (::setsockopt(
431	✗	internal_->native_handle(), level, optname,
432	✗	reinterpret_cast<char const*>(data), static_cast<int>(size)) != 0)
433	✗	return make_err(WSAGetLastError());
434	✗	return {};
435		}
436
437		inline std::error_code
438	✗	win_local_stream_acceptor::get_option(
439		int level, int optname, void* data, std::size_t* size) const noexcept
440		{
441	✗	if (!internal_ \|\| !internal_->is_open())
442	✗	return make_err(WSAENOTSOCK);
443	✗	int len = static_cast<int>(*size);
444	✗	if (::getsockopt(
445	✗	internal_->native_handle(), level, optname,
446	✗	reinterpret_cast<char*>(data), &len) != 0)
447	✗	return make_err(WSAGetLastError());
448	✗	*size = static_cast<std::size_t>(len);
449	✗	return {};
450		}
451
452		inline win_local_stream_acceptor_internal*
453	✗	win_local_stream_acceptor::get_internal() const noexcept
454		{
455	✗	return internal_.get();
456		}
457
458		// ============================================================
459		// win_local_stream_acceptor_service
460		// ============================================================
461
462	442x	inline win_local_stream_acceptor_service::win_local_stream_acceptor_service(
463	442x	capy::execution_context& /ctx/, win_local_stream_service& svc)
464	442x	: svc_(svc)
465		{
466	442x	}
467
468		inline io_object::implementation*
469	✗	win_local_stream_acceptor_service::construct()
470		{
471		auto internal =
472	✗	std::make_shared<win_local_stream_acceptor_internal>(svc_);
473
474		// Allocate wrapper before mutating lists so a throw from
475		// new doesn't leave a dangling pointer in acceptor_list_.
476	✗	auto* raw = internal.get();
477	✗	auto* wrapper = new win_local_stream_acceptor(std::move(internal));
478
479		{
480	✗	std::lock_guard<win_mutex> lock(svc_.mutex_);
481	✗	svc_.acceptor_list_.push_back(raw);
482	✗	svc_.acceptor_wrapper_list_.push_back(wrapper);
483	✗	}
484
485	✗	return wrapper;
486	✗	}
487
488		inline void
489	✗	win_local_stream_acceptor_service::destroy(io_object::implementation* p)
490		{
491	✗	if (p)
492		{
493	✗	auto& wrapper = static_cast<win_local_stream_acceptor&>(*p);
494	✗	wrapper.close_internal();
495	✗	svc_.destroy_acceptor_impl(wrapper);
496		}
497	✗	}
498
499		inline void
500	✗	win_local_stream_acceptor_service::close(io_object::handle& h)
501		{
502	✗	auto& wrapper = static_cast<win_local_stream_acceptor&>(*h.get());
503	✗	wrapper.get_internal()->close_socket();
504	✗	}
505
506		inline std::error_code
507	✗	win_local_stream_acceptor_service::open_acceptor_socket(
508		local_stream_acceptor::implementation& impl,
509		int family, int type, int protocol)
510		{
511		auto* internal =
512	✗	static_cast<win_local_stream_acceptor&>(impl).get_internal();
513	✗	return svc_.open_acceptor_socket(*internal, family, type, protocol);
514		}
515
516		inline std::error_code
517	✗	win_local_stream_acceptor_service::bind_acceptor(
518		local_stream_acceptor::implementation& impl,
519		corosio::local_endpoint ep)
520		{
521		auto* internal =
522	✗	static_cast<win_local_stream_acceptor&>(impl).get_internal();
523	✗	return svc_.bind_acceptor(*internal, ep);
524		}
525
526		inline std::error_code
527	✗	win_local_stream_acceptor_service::listen_acceptor(
528		local_stream_acceptor::implementation& impl, int backlog)
529		{
530		auto* internal =
531	✗	static_cast<win_local_stream_acceptor&>(impl).get_internal();
532	✗	return svc_.listen_acceptor(*internal, backlog);
533		}
534
535		inline void
536	442x	win_local_stream_acceptor_service::shutdown()
537		{
538		// Socket shutdown is handled by win_local_stream_service::shutdown()
539	442x	}
540
541		} // namespace boost::corosio::detail
542
543		#endif // BOOST_COROSIO_HAS_IOCP
544
545		#endif // BOOST_COROSIO_NATIVE_DETAIL_IOCP_WIN_LOCAL_STREAM_ACCEPTOR_SERVICE_HPP
546