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