Index/src / corosio / src / local_connect_pair.cpp

src/corosio/src/local_connect_pair.cpp

100.0% Lines (37 / 37) 100.0% Functions (5 / 5)
[ ] prev/next file
local_connect_pair.cpp
f(x) Functions (5)
Function Calls Lines Blocks
boost::corosio::(anonymous namespace)::make_pair_fds(int, int&, int&) :53 173x 100.0% 100.0% std::error_code boost::corosio::(anonymous namespace)::assign_pair<boost::corosio::local_datagram_socket>(boost::corosio::local_datagram_socket&, boost::corosio::local_datagram_socket&, int, int) :80 69x 40.0% 62.0% std::error_code boost::corosio::(anonymous namespace)::assign_pair<boost::corosio::local_stream_socket>(boost::corosio::local_stream_socket&, boost::corosio::local_stream_socket&, int, int) :80 84x 100.0% 100.0% boost::corosio::connect_pair(boost::corosio::local_stream_socket&, boost::corosio::local_stream_socket&) :415 101x 100.0% 100.0% boost::corosio::connect_pair(boost::corosio::local_datagram_socket&, boost::corosio::local_datagram_socket&) :438 76x 100.0% 100.0%
Line TLA Hits Source Code
1 //
2 // Copyright (c) 2026 Steve Gerbino
3 // Copyright (c) 2026 Michael Vandeberg
4 //
5 // Distributed under the Boost Software License, Version 1.0. (See accompanying
6 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7 //
8 // Official repository: https://github.com/cppalliance/corosio
9 //
10
11 #include <boost/corosio/local_connect_pair.hpp>
12 #include <boost/corosio/error.hpp>
13 #include <boost/corosio/detail/platform.hpp>
14 #include <boost/corosio/native/detail/make_err.hpp>
15
16 #include <system_error>
17
18 #if BOOST_COROSIO_POSIX
19 #include <fcntl.h>
20 #include <sys/socket.h>
21 #include <sys/un.h>
22 #include <unistd.h>
23 #elif BOOST_COROSIO_HAS_IOCP
24 #include <boost/corosio/native/detail/endpoint_convert.hpp>
25
26 #include <algorithm>
27 #include <atomic>
28 #include <cstdint>
29 #include <cstring>
30 #include <filesystem>
31 #include <new>
32 #include <random>
33 #include <string>
34 #include <thread>
35
36 #ifndef WIN32_LEAN_AND_MEAN
37 #define WIN32_LEAN_AND_MEAN
38 #endif
39 #include <winsock2.h>
40
41 #ifndef AF_UNIX
42 #define AF_UNIX 1
43 #endif
44 #endif
45
46 namespace boost::corosio {
47
48 namespace {
49
50 #if BOOST_COROSIO_POSIX
51
52 std::error_code
53 173x make_pair_fds(int type, int& a_fd, int& b_fd) noexcept
54 {
55 int fds[2];
56 173x if (::socketpair(AF_UNIX, type, 0, fds) != 0)
57 10x return detail::make_err(errno);
58
59 // assign() is documented "adopt-only" and will not mutate the fd;
60 // set O_NONBLOCK before transferring ownership.
61 469x for (int i = 0; i < 2; ++i)
62 {
63 316x int flags = ::fcntl(fds[i], F_GETFL, 0);
64 316x if (flags < 0 || ::fcntl(fds[i], F_SETFL, flags | O_NONBLOCK) < 0)
65 {
66 10x auto ec = detail::make_err(errno);
67 10x ::close(fds[0]);
68 10x ::close(fds[1]);
69 10x return ec;
70 }
71 }
72
73 153x a_fd = fds[0];
74 153x b_fd = fds[1];
75 153x return {};
76 }
77
78 template<class Socket>
79 std::error_code
80 153x assign_pair(Socket& a, Socket& b, int a_fd, int b_fd) noexcept
81 {
82 153x if (auto ec = a.assign(a_fd))
83 {
84 5x ::close(a_fd);
85 5x ::close(b_fd);
86 5x return ec;
87 }
88
89 148x if (auto ec = b.assign(b_fd))
90 {
91 5x a.close();
92 5x ::close(b_fd);
93 5x return ec;
94 }
95
96 143x return {};
97 }
98
99 #elif BOOST_COROSIO_HAS_IOCP
100
101 // random_device may throw, and a predictable seed only costs a retry
102 // on a name collision.
103 std::uint64_t
104 pair_seed() noexcept
105 {
106 try
107 {
108 return std::random_device{}();
109 }
110 catch (...)
111 {
112 LARGE_INTEGER t{};
113 ::QueryPerformanceCounter(&t);
114 return static_cast<std::uint64_t>(t.QuadPart) ^
115 (static_cast<std::uint64_t>(::GetCurrentProcessId()) << 32) ^
116 ::GetCurrentThreadId();
117 }
118 }
119
120 // Build a unique sub-directory under temp and the socket path inside
121 // it.
122 std::error_code
123 pick_pair_path(
124 std::filesystem::path& dir_out,
125 std::filesystem::path& file_out,
126 std::string& path_out) noexcept
127 {
128 namespace fs = std::filesystem;
129
130 thread_local std::mt19937_64 gen{pair_seed()};
131
132 try
133 {
134 std::error_code ec;
135 fs::path const tmp = fs::temp_directory_path(ec);
136 if (ec)
137 return ec;
138
139 for (int attempt = 0; attempt < 16; ++attempt)
140 {
141 fs::path dir = tmp / ("co_pair_" + std::to_string(gen()));
142 fs::path file = dir / "s";
143 // Converted before the directory exists, so a throw here
144 // leaves nothing on disk to remove.
145 std::string path = file.string();
146 if (fs::create_directory(dir, ec))
147 {
148 dir_out = std::move(dir);
149 file_out = std::move(file);
150 path_out = std::move(path);
151 return {};
152 }
153 }
154 if (ec)
155 return ec;
156 return detail::make_err(ERROR_PATH_NOT_FOUND);
157 }
158 catch (std::system_error const& e)
159 {
160 return e.code();
161 }
162 catch (std::bad_alloc const&)
163 {
164 return std::make_error_code(std::errc::not_enough_memory);
165 }
166 }
167
168 void
169 remove_pair_path(
170 std::filesystem::path const& dir,
171 std::filesystem::path const& file) noexcept
172 {
173 std::error_code ec;
174 std::filesystem::remove(file, ec);
175 std::filesystem::remove(dir, ec);
176 }
177
178 // Synchronously rendezvous two AF_UNIX SOCK_STREAM sockets. The
179 // listener and accept happen on the caller's thread; the connect
180 // runs on a short-lived worker to avoid a deadlock. The returned
181 // sockets are created with WSA_FLAG_OVERLAPPED so they can be
182 // registered with IOCP by assign_socket().
183 std::error_code
184 make_pair_sockets(SOCKET& a_sock, SOCKET& b_sock) noexcept
185 {
186 namespace fs = std::filesystem;
187
188 a_sock = INVALID_SOCKET;
189 b_sock = INVALID_SOCKET;
190
191 fs::path dir;
192 fs::path file;
193 std::string path;
194 if (auto ec = pick_pair_path(dir, file, path))
195 return ec;
196
197 SOCKET listen_sock =
198 ::WSASocketW(AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
199 if (listen_sock == INVALID_SOCKET)
200 {
201 auto ec = detail::make_err(::WSAGetLastError());
202 remove_pair_path(dir, file);
203 return ec;
204 }
205
206 detail::un_sa_t addr{};
207 addr.sun_family = AF_UNIX;
208 std::memcpy(
209 addr.sun_path, path.c_str(),
210 (std::min)(path.size(), sizeof(addr.sun_path) - 1));
211 int addr_len =
212 static_cast<int>(offsetof(detail::un_sa_t, sun_path) + path.size() + 1);
213
214 if (::bind(listen_sock, reinterpret_cast<sockaddr*>(&addr), addr_len) ==
215 SOCKET_ERROR)
216 {
217 auto ec = detail::make_err(::WSAGetLastError());
218 ::closesocket(listen_sock);
219 remove_pair_path(dir, file);
220 return ec;
221 }
222
223 if (::listen(listen_sock, 1) == SOCKET_ERROR)
224 {
225 auto ec = detail::make_err(::WSAGetLastError());
226 ::closesocket(listen_sock);
227 remove_pair_path(dir, file);
228 return ec;
229 }
230
231 // A worker that fails before connecting produces no connection at
232 // all, so the accept below must be able to give up. Poll the
233 // listener instead of blocking in accept() forever.
234 u_long non_blocking = 1;
235 if (::ioctlsocket(listen_sock, FIONBIO, &non_blocking) == SOCKET_ERROR)
236 {
237 auto ec = detail::make_err(::WSAGetLastError());
238 ::closesocket(listen_sock);
239 remove_pair_path(dir, file);
240 return ec;
241 }
242
243 SOCKET worker_sock = INVALID_SOCKET;
244 std::error_code worker_ec;
245 std::atomic<bool> worker_done{false};
246
247 // One exit, so worker_done is published on every path: the accept
248 // below waits on it, and a path that skipped it would hang.
249 auto connect_worker = [&] {
250 worker_sock = ::WSASocketW(
251 AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
252 if (worker_sock == INVALID_SOCKET)
253 {
254 worker_ec = detail::make_err(::WSAGetLastError());
255 }
256 else
257 {
258 detail::un_sa_t caddr{};
259 caddr.sun_family = AF_UNIX;
260 std::memcpy(
261 caddr.sun_path, path.c_str(),
262 (std::min)(path.size(), sizeof(caddr.sun_path) - 1));
263 int caddr_len = static_cast<int>(
264 offsetof(detail::un_sa_t, sun_path) + path.size() + 1);
265
266 if (::connect(
267 worker_sock, reinterpret_cast<sockaddr*>(&caddr),
268 caddr_len) == SOCKET_ERROR)
269 {
270 worker_ec = detail::make_err(::WSAGetLastError());
271 ::closesocket(worker_sock);
272 worker_sock = INVALID_SOCKET;
273 }
274 }
275 // Released last so a reader that sees it also sees worker_ec.
276 worker_done.store(true, std::memory_order_release);
277 };
278
279 // A thread that never starts publishes nothing, so this returns
280 // before the accept loop could wait on it.
281 std::thread worker;
282 std::error_code thread_ec;
283 try
284 {
285 worker = std::thread(connect_worker);
286 }
287 catch (std::system_error const& e)
288 {
289 thread_ec = e.code();
290 }
291 catch (std::bad_alloc const&)
292 {
293 thread_ec = std::make_error_code(std::errc::not_enough_memory);
294 }
295 catch (...)
296 {
297 }
298 if (!worker.joinable())
299 {
300 if (!thread_ec)
301 {
302 thread_ec =
303 std::make_error_code(std::errc::resource_unavailable_try_again);
304 }
305 ::closesocket(listen_sock);
306 remove_pair_path(dir, file);
307 return thread_ec;
308 }
309
310 SOCKET accept_sock = INVALID_SOCKET;
311 std::error_code accept_ec;
312 for (;;)
313 {
314 // A worker that succeeded has left a connection in the
315 // backlog, so only a failed one means nothing is coming.
316 if (worker_done.load(std::memory_order_acquire) && worker_ec)
317 break;
318
319 WSAPOLLFD pfd{listen_sock, POLLRDNORM, 0};
320 int const n = ::WSAPoll(&pfd, 1, 100);
321 if (n == SOCKET_ERROR)
322 {
323 accept_ec = detail::make_err(::WSAGetLastError());
324 break;
325 }
326 if (n == 0)
327 continue;
328
329 // Readiness that is not "a connection is waiting" is an error
330 // condition on the listener; accepting on it would spin. The
331 // condition carries no retrievable code, so this one is
332 // corosio's own and has to compare equal on every toolchain.
333 if ((pfd.revents & POLLRDNORM) == 0)
334 {
335 accept_ec = std::make_error_code(std::errc::connection_aborted);
336 break;
337 }
338
339 accept_sock = ::accept(listen_sock, nullptr, nullptr);
340 if (accept_sock != INVALID_SOCKET)
341 break;
342 DWORD const err = ::WSAGetLastError();
343 // A readiness report with nothing left to accept: keep
344 // waiting for the worker's connection.
345 if (err == WSAEWOULDBLOCK)
346 continue;
347 accept_ec = detail::make_err(err);
348 break;
349 }
350
351 worker.join();
352
353 ::closesocket(listen_sock);
354 remove_pair_path(dir, file);
355
356 if (accept_ec)
357 {
358 if (worker_sock != INVALID_SOCKET)
359 ::closesocket(worker_sock);
360 return accept_ec;
361 }
362 if (worker_ec)
363 {
364 if (accept_sock != INVALID_SOCKET)
365 ::closesocket(accept_sock);
366 return worker_ec;
367 }
368
369 // accept() inherits the listener's non-blocking mode; the rest of
370 // the IOCP backend hands out blocking sockets and drives them
371 // through overlapped I/O.
372 non_blocking = 0;
373 if (::ioctlsocket(accept_sock, FIONBIO, &non_blocking) == SOCKET_ERROR)
374 {
375 auto ec = detail::make_err(::WSAGetLastError());
376 ::closesocket(accept_sock);
377 ::closesocket(worker_sock);
378 return ec;
379 }
380
381 a_sock = accept_sock;
382 b_sock = worker_sock;
383 return {};
384 }
385
386 std::error_code
387 assign_pair(
388 local_stream_socket& a,
389 local_stream_socket& b,
390 SOCKET a_sock,
391 SOCKET b_sock) noexcept
392 {
393 if (auto ec = a.assign(static_cast<native_handle_type>(a_sock)))
394 {
395 ::closesocket(a_sock);
396 ::closesocket(b_sock);
397 return ec;
398 }
399
400 if (auto ec = b.assign(static_cast<native_handle_type>(b_sock)))
401 {
402 a.close();
403 ::closesocket(b_sock);
404 return ec;
405 }
406
407 return {};
408 }
409
410 #endif
411
412 } // namespace
413
414 std::error_code
415 101x connect_pair(local_stream_socket& a, local_stream_socket& b) noexcept
416 {
417 101x if (a.is_open() || b.is_open())
418 2x return make_error_code(error::already_open);
419
420 #if BOOST_COROSIO_POSIX
421 99x int a_fd = -1, b_fd = -1;
422 99x if (auto ec = make_pair_fds(SOCK_STREAM, a_fd, b_fd))
423 15x return ec;
424 84x return assign_pair(a, b, a_fd, b_fd);
425 #elif BOOST_COROSIO_HAS_IOCP
426 SOCKET a_sock = INVALID_SOCKET, b_sock = INVALID_SOCKET;
427 if (auto ec = make_pair_sockets(a_sock, b_sock))
428 return ec;
429 return assign_pair(a, b, a_sock, b_sock);
430 #else
431 return detail::make_err(ENOSYS);
432 #endif
433 }
434
435 #if BOOST_COROSIO_POSIX
436
437 std::error_code
438 76x connect_pair(local_datagram_socket& a, local_datagram_socket& b) noexcept
439 {
440 76x if (a.is_open() || b.is_open())
441 2x return make_error_code(error::already_open);
442
443 74x int a_fd = -1, b_fd = -1;
444 74x if (auto ec = make_pair_fds(SOCK_DGRAM, a_fd, b_fd))
445 5x return ec;
446 69x return assign_pair(a, b, a_fd, b_fd);
447 }
448
449 #endif
450
451 } // namespace boost::corosio
452