src/corosio/src/local_connect_pair.cpp

100.0% Lines (37/0/37) 100.0% List of functions (5/0/5)

[ ] prev/next file

local_connect_pair.cpp

f(x) Functions (5)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Steve Gerbino
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		#include <boost/corosio/local_connect_pair.hpp>
11		#include <boost/corosio/detail/platform.hpp>
12		#include <boost/corosio/native/detail/make_err.hpp>
13
14		#include <system_error>
15
16		#if BOOST_COROSIO_POSIX
17		#include <fcntl.h>
18		#include <sys/socket.h>
19		#include <sys/un.h>
20		#include <unistd.h>
21		#elif BOOST_COROSIO_HAS_IOCP
22		#include <boost/corosio/native/detail/endpoint_convert.hpp>
23
24		#include <algorithm>
25		#include <atomic>
26		#include <cstring>
27		#include <filesystem>
28		#include <random>
29		#include <string>
30		#include <thread>
31
32		#ifndef WIN32_LEAN_AND_MEAN
33		#define WIN32_LEAN_AND_MEAN
34		#endif
35		#include <WinSock2.h>
36
37		#ifndef AF_UNIX
38		#define AF_UNIX 1
39		#endif
40		#endif
41
42		namespace boost::corosio {
43
44		namespace {
45
46		#if BOOST_COROSIO_POSIX
47
48		std::error_code
49	161x	make_pair_fds(int type, int& a_fd, int& b_fd) noexcept
50		{
51		int fds[2];
52	161x	if (::socketpair(AF_UNIX, type, 0, fds) != 0)
53	10x	return detail::make_err(errno);
54
55		// assign() is documented "adopt-only" and will not mutate the fd;
56		// set O_NONBLOCK before transferring ownership.
57	433x	for (int i = 0; i < 2; ++i)
58		{
59	292x	int flags = ::fcntl(fds[i], F_GETFL, 0);
60	292x	if (flags < 0 \|\| ::fcntl(fds[i], F_SETFL, flags \| O_NONBLOCK) < 0)
61		{
62	10x	auto ec = detail::make_err(errno);
63	10x	::close(fds[0]);
64	10x	::close(fds[1]);
65	10x	return ec;
66		}
67		}
68
69	141x	a_fd = fds[0];
70	141x	b_fd = fds[1];
71	141x	return {};
72		}
73
74		template<class Socket>
75		std::error_code
76	141x	assign_pair(Socket& a, Socket& b, int a_fd, int b_fd) noexcept
77		{
78	141x	if (auto ec = a.assign(a_fd))
79		{
80	5x	::close(a_fd);
81	5x	::close(b_fd);
82	5x	return ec;
83		}
84
85	136x	if (auto ec = b.assign(b_fd))
86		{
87	5x	a.close();
88	5x	::close(b_fd);
89	5x	return ec;
90		}
91
92	131x	return {};
93		}
94
95		#elif BOOST_COROSIO_HAS_IOCP
96
97		// Build a unique sub-directory under temp and return the full socket
98		// path inside it. Empty string on failure.
99		std::string
100		pick_pair_path(std::filesystem::path& dir_out)
101		{
102		namespace fs = std::filesystem;
103
104		thread_local std::mt19937_64 gen{std::random_device{}()};
105
106		for (int attempt = 0; attempt < 16; ++attempt)
107		{
108		auto candidate =
109		fs::temp_directory_path() /
110		("co_pair_" + std::to_string(gen()));
111		std::error_code ec;
112		if (fs::create_directory(candidate, ec))
113		{
114		dir_out = candidate;
115		return (candidate / "s").string();
116		}
117		}
118		return {};
119		}
120
121		void
122		remove_pair_path(std::filesystem::path const& dir, std::string const& path)
123		{
124		std::error_code ec;
125		std::filesystem::remove(std::filesystem::path(path), ec);
126		std::filesystem::remove(dir, ec);
127		}
128
129		// Synchronously rendezvous two AF_UNIX SOCK_STREAM sockets. The
130		// listener and accept happen on the caller's thread; the connect
131		// runs on a short-lived worker to avoid a deadlock. The returned
132		// sockets are created with WSA_FLAG_OVERLAPPED so they can be
133		// registered with IOCP by assign_socket().
134		std::error_code
135		make_pair_sockets(SOCKET& a_sock, SOCKET& b_sock) noexcept
136		{
137		namespace fs = std::filesystem;
138
139		a_sock = INVALID_SOCKET;
140		b_sock = INVALID_SOCKET;
141
142		fs::path dir;
143		std::string path = pick_pair_path(dir);
144		if (path.empty())
145		return detail::make_err(ERROR_PATH_NOT_FOUND);
146
147		SOCKET listen_sock = ::WSASocketW(
148		AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
149		if (listen_sock == INVALID_SOCKET)
150		{
151		auto ec = detail::make_err(::WSAGetLastError());
152		remove_pair_path(dir, path);
153		return ec;
154		}
155
156		detail::un_sa_t addr{};
157		addr.sun_family = AF_UNIX;
158		std::memcpy(
159		addr.sun_path, path.c_str(),
160		(std::min)(path.size(), sizeof(addr.sun_path) - 1));
161		int addr_len = static_cast<int>(
162		offsetof(detail::un_sa_t, sun_path) + path.size() + 1);
163
164		if (::bind(
165		listen_sock, reinterpret_cast<sockaddr*>(&addr), addr_len)
166		== SOCKET_ERROR)
167		{
168		auto ec = detail::make_err(::WSAGetLastError());
169		::closesocket(listen_sock);
170		remove_pair_path(dir, path);
171		return ec;
172		}
173
174		if (::listen(listen_sock, 1) == SOCKET_ERROR)
175		{
176		auto ec = detail::make_err(::WSAGetLastError());
177		::closesocket(listen_sock);
178		remove_pair_path(dir, path);
179		return ec;
180		}
181
182		// A worker that fails before connecting produces no connection at
183		// all, so the accept below must be able to give up. Poll the
184		// listener instead of blocking in accept() forever.
185		u_long non_blocking = 1;
186		if (::ioctlsocket(listen_sock, FIONBIO, &non_blocking) == SOCKET_ERROR)
187		{
188		auto ec = detail::make_err(::WSAGetLastError());
189		::closesocket(listen_sock);
190		remove_pair_path(dir, path);
191		return ec;
192		}
193
194		SOCKET worker_sock = INVALID_SOCKET;
195		std::error_code worker_ec;
196		std::atomic<bool> worker_done{false};
197
198		// One exit, so worker_done is published on every path: the accept
199		// below waits on it, and a path that skipped it would hang.
200		std::thread worker([&] {
201		worker_sock = ::WSASocketW(
202		AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
203		if (worker_sock == INVALID_SOCKET)
204		{
205		worker_ec = detail::make_err(::WSAGetLastError());
206		}
207		else
208		{
209		detail::un_sa_t caddr{};
210		caddr.sun_family = AF_UNIX;
211		std::memcpy(
212		caddr.sun_path, path.c_str(),
213		(std::min)(path.size(), sizeof(caddr.sun_path) - 1));
214		int caddr_len = static_cast<int>(
215		offsetof(detail::un_sa_t, sun_path) + path.size() + 1);
216
217		if (::connect(
218		worker_sock,
219		reinterpret_cast<sockaddr*>(&caddr), caddr_len)
220		== SOCKET_ERROR)
221		{
222		worker_ec = detail::make_err(::WSAGetLastError());
223		::closesocket(worker_sock);
224		worker_sock = INVALID_SOCKET;
225		}
226		}
227		// Released last so a reader that sees it also sees worker_ec.
228		worker_done.store(true, std::memory_order_release);
229		});
230
231		SOCKET accept_sock = INVALID_SOCKET;
232		std::error_code accept_ec;
233		for (;;)
234		{
235		// A worker that succeeded has left a connection in the
236		// backlog, so only a failed one means nothing is coming.
237		if (worker_done.load(std::memory_order_acquire) && worker_ec)
238		break;
239
240		WSAPOLLFD pfd{listen_sock, POLLRDNORM, 0};
241		int const n = ::WSAPoll(&pfd, 1, 100);
242		if (n == SOCKET_ERROR)
243		{
244		accept_ec = detail::make_err(::WSAGetLastError());
245		break;
246		}
247		if (n == 0)
248		continue;
249
250		// Readiness that is not "a connection is waiting" is an error
251		// condition on the listener; accepting on it would spin. The
252		// condition carries no retrievable code, so this one is
253		// corosio's own and has to compare equal on every toolchain.
254		if ((pfd.revents & POLLRDNORM) == 0)
255		{
256		accept_ec =
257		std::make_error_code(std::errc::connection_aborted);
258		break;
259		}
260
261		accept_sock = ::accept(listen_sock, nullptr, nullptr);
262		if (accept_sock != INVALID_SOCKET)
263		break;
264		DWORD const err = ::WSAGetLastError();
265		// A readiness report with nothing left to accept: keep
266		// waiting for the worker's connection.
267		if (err == WSAEWOULDBLOCK)
268		continue;
269		accept_ec = detail::make_err(err);
270		break;
271		}
272
273		worker.join();
274
275		::closesocket(listen_sock);
276		remove_pair_path(dir, path);
277
278		if (accept_ec)
279		{
280		if (worker_sock != INVALID_SOCKET)
281		::closesocket(worker_sock);
282		return accept_ec;
283		}
284		if (worker_ec)
285		{
286		if (accept_sock != INVALID_SOCKET)
287		::closesocket(accept_sock);
288		return worker_ec;
289		}
290
291		// accept() inherits the listener's non-blocking mode; the rest of
292		// the IOCP backend hands out blocking sockets and drives them
293		// through overlapped I/O.
294		non_blocking = 0;
295		if (::ioctlsocket(accept_sock, FIONBIO, &non_blocking) == SOCKET_ERROR)
296		{
297		auto ec = detail::make_err(::WSAGetLastError());
298		::closesocket(accept_sock);
299		::closesocket(worker_sock);
300		return ec;
301		}
302
303		a_sock = accept_sock;
304		b_sock = worker_sock;
305		return {};
306		}
307
308		std::error_code
309		assign_pair(
310		local_stream_socket& a,
311		local_stream_socket& b,
312		SOCKET a_sock,
313		SOCKET b_sock) noexcept
314		{
315		if (auto ec = a.assign(static_cast<native_handle_type>(a_sock)))
316		{
317		::closesocket(a_sock);
318		::closesocket(b_sock);
319		return ec;
320		}
321
322		if (auto ec = b.assign(static_cast<native_handle_type>(b_sock)))
323		{
324		a.close();
325		::closesocket(b_sock);
326		return ec;
327		}
328
329		return {};
330		}
331
332		#endif
333
334		} // namespace
335
336		std::error_code
337	95x	connect_pair(local_stream_socket& a, local_stream_socket& b) noexcept
338		{
339	95x	if (a.is_open() \|\| b.is_open())
340	2x	return std::make_error_code(std::errc::already_connected);
341
342		#if BOOST_COROSIO_POSIX
343	93x	int a_fd = -1, b_fd = -1;
344	93x	if (auto ec = make_pair_fds(SOCK_STREAM, a_fd, b_fd))
345	15x	return ec;
346	78x	return assign_pair(a, b, a_fd, b_fd);
347		#elif BOOST_COROSIO_HAS_IOCP
348		SOCKET a_sock = INVALID_SOCKET, b_sock = INVALID_SOCKET;
349		if (auto ec = make_pair_sockets(a_sock, b_sock))
350		return ec;
351		return assign_pair(a, b, a_sock, b_sock);
352		#else
353		return detail::make_err(ENOSYS);
354		#endif
355		}
356
357		#if BOOST_COROSIO_POSIX
358
359		std::error_code
360	70x	connect_pair(local_datagram_socket& a, local_datagram_socket& b) noexcept
361		{
362	70x	if (a.is_open() \|\| b.is_open())
363	2x	return std::make_error_code(std::errc::already_connected);
364
365	68x	int a_fd = -1, b_fd = -1;
366	68x	if (auto ec = make_pair_fds(SOCK_DGRAM, a_fd, b_fd))
367	5x	return ec;
368	63x	return assign_pair(a, b, a_fd, b_fd);
369		}
370
371		#endif
372
373		} // namespace boost::corosio
374