Index/src / corosio / src / local_connect_pair.cpp

src/corosio/src/local_connect_pair.cpp

80.3% Lines (139 / 173) 77.0% Branches (67 / 87) 100.0% Functions (7 / 7)
[ ] prev/next file n p prev/next uncovered line
local_connect_pair.cpp
f(x) Functions (7)
Function Calls Lines Branches Blocks
boost::corosio::(anonymous namespace)::pair_seed() :104 32x 22.2% 100.0% 43.8% boost::corosio::(anonymous namespace)::pick_pair_path(std::filesystem::__cxx11::path&, std::filesystem::__cxx11::path&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >&) :123 83x 61.5% 63.6% 54.2% boost::corosio::(anonymous namespace)::remove_pair_path(std::filesystem::__cxx11::path const&, std::filesystem::__cxx11::path const&) :169 83x 100.0% – 100.0% boost::corosio::(anonymous namespace)::make_pair_sockets(unsigned long long&, unsigned long long&) :184 83x 82.8% 75.0% 83.9% boost::corosio::(anonymous namespace)::make_pair_sockets(unsigned long long&, unsigned long long&)::{lambda()#1}::operator()() const :249 79x 100.0% 100.0% 100.0% boost::corosio::(anonymous namespace)::assign_pair(boost::corosio::local_stream_socket&, boost::corosio::local_stream_socket&, unsigned long long, unsigned long long) :387 42x 100.0% 100.0% 100.0% boost::corosio::connect_pair(boost::corosio::local_stream_socket&, boost::corosio::local_stream_socket&) :415 84x 100.0% 87.5% 100.0%
Line Branch 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 make_pair_fds(int type, int& a_fd, int& b_fd) noexcept
54 {
55 int fds[2];
56 if (::socketpair(AF_UNIX, type, 0, fds) != 0)
57 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 for (int i = 0; i < 2; ++i)
62 {
63 int flags = ::fcntl(fds[i], F_GETFL, 0);
64 if (flags < 0 || ::fcntl(fds[i], F_SETFL, flags | O_NONBLOCK) < 0)
65 {
66 auto ec = detail::make_err(errno);
67 ::close(fds[0]);
68 ::close(fds[1]);
69 return ec;
70 }
71 }
72
73 a_fd = fds[0];
74 b_fd = fds[1];
75 return {};
76 }
77
78 template<class Socket>
79 std::error_code
80 assign_pair(Socket& a, Socket& b, int a_fd, int b_fd) noexcept
81 {
82 if (auto ec = a.assign(a_fd))
83 {
84 ::close(a_fd);
85 ::close(b_fd);
86 return ec;
87 }
88
89 if (auto ec = b.assign(b_fd))
90 {
91 a.close();
92 ::close(b_fd);
93 return ec;
94 }
95
96 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 32x pair_seed() noexcept
105 {
106 try
107 {
108
2/2
✓ Branch 2 → 3 taken 32 times.
✓ Branch 3 → 4 taken 32 times.
 32x 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 83x 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
2/2
✓ Branch 2 → 3 taken 32 times.
✓ Branch 2 → 6 taken 51 times.
 83x thread_local std::mt19937_64 gen{pair_seed()};
131
132 try
133 {
134 83x std::error_code ec;
135
1/1
✓ Branch 7 → 8 taken 83 times.
 83x fs::path const tmp = fs::temp_directory_path(ec);
136
1/2
✗ Branch 9 → 10 not taken.
✓ Branch 9 → 11 taken 83 times.
 83x if (ec)
137 ✗ return ec;
138
139
1/2
✓ Branch 48 → 12 taken 83 times.
✗ Branch 48 → 49 not taken.
 83x for (int attempt = 0; attempt < 16; ++attempt)
140 {
141
5/5
✓ Branch 12 → 13 taken 83 times.
✓ Branch 13 → 14 taken 83 times.
✓ Branch 14 → 15 taken 83 times.
✓ Branch 15 → 16 taken 83 times.
✓ Branch 16 → 17 taken 83 times.
 83x fs::path dir = tmp / ("co_pair_" + std::to_string(gen()));
142
2/2
✓ Branch 20 → 21 taken 83 times.
✓ Branch 21 → 22 taken 83 times.
 83x fs::path file = dir / "s";
143 // Converted before the directory exists, so a throw here
144 // leaves nothing on disk to remove.
145
1/1
✓ Branch 23 → 24 taken 83 times.
 83x std::string path = file.string();
146
1/2
✓ Branch 25 → 26 taken 83 times.
✗ Branch 25 → 34 not taken.
 83x if (fs::create_directory(dir, ec))
147 {
148 83x dir_out = std::move(dir);
149 83x file_out = std::move(file);
150 83x path_out = std::move(path);
151 83x return {};
152 }
153 249x }
154
0/2
✗ Branch 50 → 51 not taken.
✗ Branch 50 → 52 not taken.
 ✗ if (ec)
155 ✗ return ec;
156 ✗ return detail::make_err(ERROR_PATH_NOT_FOUND);
157 83x }
158
0/3
✗ Branch 76 → 77 not taken.
✗ Branch 76 → 78 not taken.
✗ Branch 76 → 82 not taken.
 ✗ 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 83x remove_pair_path(
170 std::filesystem::path const& dir,
171 std::filesystem::path const& file) noexcept
172 {
173 83x std::error_code ec;
174 83x std::filesystem::remove(file, ec);
175 83x std::filesystem::remove(dir, ec);
176 83x }
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 83x make_pair_sockets(SOCKET& a_sock, SOCKET& b_sock) noexcept
185 {
186 namespace fs = std::filesystem;
187
188 83x a_sock = INVALID_SOCKET;
189 83x b_sock = INVALID_SOCKET;
190
191 83x fs::path dir;
192 83x fs::path file;
193 83x std::string path;
194
1/2
✗ Branch 7 → 8 not taken.
✓ Branch 7 → 9 taken 83 times.
 83x if (auto ec = pick_pair_path(dir, file, path))
195 ✗ return ec;
196
197 SOCKET listen_sock =
198 83x ::WSASocketW(AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
199
2/2
✓ Branch 10 → 11 taken 1 time.
✓ Branch 10 → 15 taken 82 times.
 83x if (listen_sock == INVALID_SOCKET)
200 {
201 1x auto ec = detail::make_err(::WSAGetLastError());
202 1x remove_pair_path(dir, file);
203 1x return ec;
204 }
205
206 82x detail::un_sa_t addr{};
207 82x addr.sun_family = AF_UNIX;
208 164x std::memcpy(
209 82x addr.sun_path, path.c_str(),
210 82x (std::min)(path.size(), sizeof(addr.sun_path) - 1));
211 int addr_len =
212 82x static_cast<int>(offsetof(detail::un_sa_t, sun_path) + path.size() + 1);
213
214
2/2
✓ Branch 20 → 21 taken 1 time.
✓ Branch 20 → 26 taken 81 times.
 82x if (::bind(listen_sock, reinterpret_cast<sockaddr*>(&addr), addr_len) ==
215 SOCKET_ERROR)
216 {
217 1x auto ec = detail::make_err(::WSAGetLastError());
218 1x ::closesocket(listen_sock);
219 1x remove_pair_path(dir, file);
220 1x return ec;
221 }
222
223
2/2
✓ Branch 27 → 28 taken 1 time.
✓ Branch 27 → 33 taken 80 times.
 81x if (::listen(listen_sock, 1) == SOCKET_ERROR)
224 {
225 1x auto ec = detail::make_err(::WSAGetLastError());
226 1x ::closesocket(listen_sock);
227 1x remove_pair_path(dir, file);
228 1x 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 80x u_long non_blocking = 1;
235
2/2
✓ Branch 34 → 35 taken 1 time.
✓ Branch 34 → 40 taken 79 times.
 80x if (::ioctlsocket(listen_sock, FIONBIO, &non_blocking) == SOCKET_ERROR)
236 {
237 1x auto ec = detail::make_err(::WSAGetLastError());
238 1x ::closesocket(listen_sock);
239 1x remove_pair_path(dir, file);
240 1x return ec;
241 }
242
243 79x SOCKET worker_sock = INVALID_SOCKET;
244 79x std::error_code worker_ec;
245 79x 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 79x auto connect_worker = [&] {
250 79x worker_sock = ::WSASocketW(
251 AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
252
2/2
✓ Branch 3 → 4 taken 9 times.
✓ Branch 3 → 6 taken 70 times.
 79x if (worker_sock == INVALID_SOCKET)
253 {
254 9x worker_ec = detail::make_err(::WSAGetLastError());
255 }
256 else
257 {
258 70x detail::un_sa_t caddr{};
259 70x caddr.sun_family = AF_UNIX;
260 140x std::memcpy(
261 70x caddr.sun_path, path.c_str(),
262 70x (std::min)(path.size(), sizeof(caddr.sun_path) - 1));
263 int caddr_len = static_cast<int>(
264 70x offsetof(detail::un_sa_t, sun_path) + path.size() + 1);
265
266
1/1
✓ Branch 10 → 11 taken 70 times.
 70x if (::connect(
267 worker_sock, reinterpret_cast<sockaddr*>(&caddr),
268
2/2
✓ Branch 11 → 12 taken 9 times.
✓ Branch 11 → 16 taken 61 times.
 70x caddr_len) == SOCKET_ERROR)
269 {
270
1/1
✓ Branch 12 → 13 taken 9 times.
 9x worker_ec = detail::make_err(::WSAGetLastError());
271
1/1
✓ Branch 14 → 15 taken 9 times.
 9x ::closesocket(worker_sock);
272 9x worker_sock = INVALID_SOCKET;
273 }
274 }
275 // Released last so a reader that sees it also sees worker_ec.
276 79x worker_done.store(true, std::memory_order_release);
277 79x };
278
279 // A thread that never starts publishes nothing, so this returns
280 // before the accept loop could wait on it.
281 79x std::thread worker;
282 79x std::error_code thread_ec;
283 try
284 {
285
1/1
✓ Branch 43 → 44 taken 79 times.
 79x worker = std::thread(connect_worker);
286 }
287
0/3
✗ Branch 111 → 112 not taken.
✗ Branch 111 → 115 not taken.
✗ Branch 111 → 118 not taken.
 ✗ 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
1/2
✗ Branch 48 → 49 not taken.
✓ Branch 48 → 55 taken 79 times.
 79x if (!worker.joinable())
299 {
300
0/2
✗ Branch 50 → 51 not taken.
✗ Branch 50 → 52 not taken.
 ✗ 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 79x SOCKET accept_sock = INVALID_SOCKET;
311 79x 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
5/6
✓ Branch 57 → 58 taken 18 times.
✓ Branch 57 → 61 taken 80 times.
✓ Branch 59 → 60 taken 18 times.
✗ Branch 59 → 61 not taken.
✓ Branch 62 → 63 taken 18 times.
✓ Branch 62 → 64 taken 80 times.
 98x if (worker_done.load(std::memory_order_acquire) && worker_ec)
317 18x break;
318
319 80x WSAPOLLFD pfd{listen_sock, POLLRDNORM, 0};
320 80x int const n = ::WSAPoll(&pfd, 1, 100);
321
2/2
✓ Branch 65 → 66 taken 9 times.
✓ Branch 65 → 69 taken 71 times.
 80x if (n == SOCKET_ERROR)
322 {
323 9x accept_ec = detail::make_err(::WSAGetLastError());
324 9x break;
325 }
326
2/2
✓ Branch 69 → 70 taken 18 times.
✓ Branch 69 → 71 taken 53 times.
 71x if (n == 0)
327 18x 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
1/2
✗ Branch 71 → 72 not taken.
✓ Branch 71 → 74 taken 53 times.
 53x if ((pfd.revents & POLLRDNORM) == 0)
334 {
335 ✗ accept_ec = std::make_error_code(std::errc::connection_aborted);
336 ✗ break;
337 }
338
339 53x accept_sock = ::accept(listen_sock, nullptr, nullptr);
340
2/2
✓ Branch 75 → 76 taken 51 times.
✓ Branch 75 → 77 taken 2 times.
 53x if (accept_sock != INVALID_SOCKET)
341 51x break;
342 2x DWORD const err = ::WSAGetLastError();
343 // A readiness report with nothing left to accept: keep
344 // waiting for the worker's connection.
345
2/2
✓ Branch 78 → 79 taken 1 time.
✓ Branch 78 → 80 taken 1 time.
 2x if (err == WSAEWOULDBLOCK)
346 1x continue;
347 1x accept_ec = detail::make_err(err);
348 1x break;
349 19x }
350
351 79x worker.join();
352
353 79x ::closesocket(listen_sock);
354 79x remove_pair_path(dir, file);
355
356
2/2
✓ Branch 86 → 88 taken 10 times.
✓ Branch 86 → 91 taken 69 times.
 79x if (accept_ec)
357 {
358
1/2
✓ Branch 88 → 89 taken 10 times.
✗ Branch 88 → 90 not taken.
 10x if (worker_sock != INVALID_SOCKET)
359 10x ::closesocket(worker_sock);
360 10x return accept_ec;
361 }
362
2/2
✓ Branch 92 → 93 taken 18 times.
✓ Branch 92 → 96 taken 51 times.
 69x if (worker_ec)
363 {
364
1/2
✗ Branch 93 → 94 not taken.
✓ Branch 93 → 95 taken 18 times.
 18x if (accept_sock != INVALID_SOCKET)
365 ✗ ::closesocket(accept_sock);
366 18x 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 51x non_blocking = 0;
373
2/2
✓ Branch 97 → 98 taken 9 times.
✓ Branch 97 → 103 taken 42 times.
 51x if (::ioctlsocket(accept_sock, FIONBIO, &non_blocking) == SOCKET_ERROR)
374 {
375 9x auto ec = detail::make_err(::WSAGetLastError());
376 9x ::closesocket(accept_sock);
377 9x ::closesocket(worker_sock);
378 9x return ec;
379 }
380
381 42x a_sock = accept_sock;
382 42x b_sock = worker_sock;
383 42x return {};
384 83x }
385
386 std::error_code
387 42x assign_pair(
388 local_stream_socket& a,
389 local_stream_socket& b,
390 SOCKET a_sock,
391 SOCKET b_sock) noexcept
392 {
393
2/2
✓ Branch 4 → 5 taken 9 times.
✓ Branch 4 → 8 taken 33 times.
 42x if (auto ec = a.assign(static_cast<native_handle_type>(a_sock)))
394 {
395 9x ::closesocket(a_sock);
396 9x ::closesocket(b_sock);
397 9x return ec;
398 }
399
400
2/2
✓ Branch 10 → 11 taken 9 times.
✓ Branch 10 → 14 taken 24 times.
 33x if (auto ec = b.assign(static_cast<native_handle_type>(b_sock)))
401 {
402 9x a.close();
403 9x ::closesocket(b_sock);
404 9x return ec;
405 }
406
407 24x return {};
408 }
409
410 #endif
411
412 } // namespace
413
414 std::error_code
415 84x connect_pair(local_stream_socket& a, local_stream_socket& b) noexcept
416 {
417
5/6
✓ Branch 3 → 4 taken 83 times.
✓ Branch 3 → 6 taken 1 time.
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 7 taken 83 times.
✓ Branch 8 → 9 taken 1 time.
✓ Branch 8 → 10 taken 83 times.
 84x if (a.is_open() || b.is_open())
418 1x return make_error_code(error::already_open);
419
420 #if BOOST_COROSIO_POSIX
421 int a_fd = -1, b_fd = -1;
422 if (auto ec = make_pair_fds(SOCK_STREAM, a_fd, b_fd))
423 return ec;
424 return assign_pair(a, b, a_fd, b_fd);
425 #elif BOOST_COROSIO_HAS_IOCP
426 83x SOCKET a_sock = INVALID_SOCKET, b_sock = INVALID_SOCKET;
427
2/2
✓ Branch 12 → 13 taken 41 times.
✓ Branch 12 → 14 taken 42 times.
 83x if (auto ec = make_pair_sockets(a_sock, b_sock))
428 41x return ec;
429 42x 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 connect_pair(local_datagram_socket& a, local_datagram_socket& b) noexcept
439 {
440 if (a.is_open() || b.is_open())
441 return make_error_code(error::already_open);
442
443 int a_fd = -1, b_fd = -1;
444 if (auto ec = make_pair_fds(SOCK_DGRAM, a_fd, b_fd))
445 return ec;
446 return assign_pair(a, b, a_fd, b_fd);
447 }
448
449 #endif
450
451 } // namespace boost::corosio
452