Index/src / corosio / src / local_connect_pair.cpp

src/corosio/src/local_connect_pair.cpp

96.2% Lines (125/130) 100.0% List of functions (6/6) 89.3% Branches (67/75)
[ ] prev/next file n p prev/next uncovered line
local_connect_pair.cpp
f(x) Functions (6)
Function Calls Lines Branches Blocks
boost::corosio::(anonymous namespace)::pick_pair_path(std::filesystem::__cxx11::path&) :100 83x 90.0% 89.5% 55.7% boost::corosio::(anonymous namespace)::remove_pair_path(std::filesystem::__cxx11::path const&, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) :121 83x 100.0% 100.0% 87.5% boost::corosio::(anonymous namespace)::make_pair_sockets(unsigned long long&, unsigned long long&) :134 83x 95.1% 86.1% 95.7% boost::corosio::(anonymous namespace)::make_pair_sockets(unsigned long long&, unsigned long long&)::{lambda()#1}::operator()() const :198 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) :305 42x 100.0% 100.0% 100.0% boost::corosio::connect_pair(boost::corosio::local_stream_socket&, boost::corosio::local_stream_socket&) :333 84x 100.0% 87.5% 100.0%
Line Branch TLA 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 make_pair_fds(int type, int& a_fd, int& b_fd) noexcept
50 {
51 int fds[2];
52 if (::socketpair(AF_UNIX, type, 0, fds) != 0)
53 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 for (int i = 0; i < 2; ++i)
58 {
59 int flags = ::fcntl(fds[i], F_GETFL, 0);
60 if (flags < 0 || ::fcntl(fds[i], F_SETFL, flags | O_NONBLOCK) < 0)
61 {
62 auto ec = detail::make_err(errno);
63 ::close(fds[0]);
64 ::close(fds[1]);
65 return ec;
66 }
67 }
68
69 a_fd = fds[0];
70 b_fd = fds[1];
71 return {};
72 }
73
74 template<class Socket>
75 std::error_code
76 assign_pair(Socket& a, Socket& b, int a_fd, int b_fd) noexcept
77 {
78 if (auto ec = a.assign(a_fd))
79 {
80 ::close(a_fd);
81 ::close(b_fd);
82 return ec;
83 }
84
85 if (auto ec = b.assign(b_fd))
86 {
87 a.close();
88 ::close(b_fd);
89 return ec;
90 }
91
92 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 83x pick_pair_path(std::filesystem::path& dir_out)
101 {
102 namespace fs = std::filesystem;
103
104
5/5
✓ Branch 2 → 3 taken 32 times.
✓ Branch 2 → 8 taken 51 times.
✓ Branch 3 → 4 taken 32 times.
✓ Branch 4 → 5 taken 32 times.
✓ Branch 5 → 6 taken 32 times.
 83x thread_local std::mt19937_64 gen{std::random_device{}()};
105
106
1/2
✓ Branch 36 → 9 taken 83 times.
✗ Branch 36 → 37 not taken.
 83x for (int attempt = 0; attempt < 16; ++attempt)
107 {
108 auto candidate =
109
6/6
✓ Branch 9 → 10 taken 83 times.
✓ Branch 10 → 11 taken 83 times.
✓ Branch 11 → 12 taken 83 times.
✓ Branch 12 → 13 taken 83 times.
✓ Branch 13 → 14 taken 83 times.
✓ Branch 14 → 15 taken 83 times.
 83x fs::temp_directory_path() / ("co_pair_" + std::to_string(gen()));
110 83x std::error_code ec;
111
1/2
✓ Branch 21 → 22 taken 83 times.
✗ Branch 21 → 30 not taken.
 83x if (fs::create_directory(candidate, ec))
112 {
113
1/1
✓ Branch 22 → 23 taken 83 times.
 83x dir_out = candidate;
114
3/3
✓ Branch 23 → 24 taken 83 times.
✓ Branch 24 → 25 taken 83 times.
✓ Branch 25 → 26 taken 83 times.
 83x return (candidate / "s").string();
115 }
116 83x }
117 ✗ return {};
118 }
119
120 void
121 83x remove_pair_path(std::filesystem::path const& dir, std::string const& path)
122 {
123 83x std::error_code ec;
124
1/1
✓ Branch 3 → 4 taken 83 times.
 83x std::filesystem::remove(std::filesystem::path(path), ec);
125 83x std::filesystem::remove(dir, ec);
126 83x }
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 83x make_pair_sockets(SOCKET& a_sock, SOCKET& b_sock) noexcept
135 {
136 namespace fs = std::filesystem;
137
138 83x a_sock = INVALID_SOCKET;
139 83x b_sock = INVALID_SOCKET;
140
141 83x fs::path dir;
142 83x std::string path = pick_pair_path(dir);
143
1/2
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 7 taken 83 times.
 83x if (path.empty())
144 ✗ return detail::make_err(ERROR_PATH_NOT_FOUND);
145
146 SOCKET listen_sock =
147 83x ::WSASocketW(AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
148
2/2
✓ Branch 8 → 9 taken 1 time.
✓ Branch 8 → 13 taken 82 times.
 83x if (listen_sock == INVALID_SOCKET)
149 {
150 1x auto ec = detail::make_err(::WSAGetLastError());
151 1x remove_pair_path(dir, path);
152 1x return ec;
153 }
154
155 82x detail::un_sa_t addr{};
156 82x addr.sun_family = AF_UNIX;
157 164x std::memcpy(
158 82x addr.sun_path, path.c_str(),
159 82x (std::min)(path.size(), sizeof(addr.sun_path) - 1));
160 int addr_len =
161 82x static_cast<int>(offsetof(detail::un_sa_t, sun_path) + path.size() + 1);
162
163
2/2
✓ Branch 18 → 19 taken 1 time.
✓ Branch 18 → 24 taken 81 times.
 82x if (::bind(listen_sock, reinterpret_cast<sockaddr*>(&addr), addr_len) ==
164 SOCKET_ERROR)
165 {
166 1x auto ec = detail::make_err(::WSAGetLastError());
167 1x ::closesocket(listen_sock);
168 1x remove_pair_path(dir, path);
169 1x return ec;
170 }
171
172
2/2
✓ Branch 25 → 26 taken 1 time.
✓ Branch 25 → 31 taken 80 times.
 81x if (::listen(listen_sock, 1) == SOCKET_ERROR)
173 {
174 1x auto ec = detail::make_err(::WSAGetLastError());
175 1x ::closesocket(listen_sock);
176 1x remove_pair_path(dir, path);
177 1x 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 80x u_long non_blocking = 1;
184
2/2
✓ Branch 32 → 33 taken 1 time.
✓ Branch 32 → 38 taken 79 times.
 80x if (::ioctlsocket(listen_sock, FIONBIO, &non_blocking) == SOCKET_ERROR)
185 {
186 1x auto ec = detail::make_err(::WSAGetLastError());
187 1x ::closesocket(listen_sock);
188 1x remove_pair_path(dir, path);
189 1x return ec;
190 }
191
192 79x SOCKET worker_sock = INVALID_SOCKET;
193 79x std::error_code worker_ec;
194 79x 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 237x std::thread worker([&] {
199 79x worker_sock = ::WSASocketW(
200 AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
201
2/2
✓ Branch 3 → 4 taken 9 times.
✓ Branch 3 → 6 taken 70 times.
 79x if (worker_sock == INVALID_SOCKET)
202 {
203 9x worker_ec = detail::make_err(::WSAGetLastError());
204 }
205 else
206 {
207 70x detail::un_sa_t caddr{};
208 70x caddr.sun_family = AF_UNIX;
209 140x std::memcpy(
210 70x caddr.sun_path, path.c_str(),
211 70x (std::min)(path.size(), sizeof(caddr.sun_path) - 1));
212 int caddr_len = static_cast<int>(
213 70x offsetof(detail::un_sa_t, sun_path) + path.size() + 1);
214
215
1/1
✓ Branch 10 → 11 taken 70 times.
 70x if (::connect(
216 worker_sock, reinterpret_cast<sockaddr*>(&caddr),
217
2/2
✓ Branch 11 → 12 taken 9 times.
✓ Branch 11 → 16 taken 61 times.
 70x caddr_len) == SOCKET_ERROR)
218 {
219
1/1
✓ Branch 12 → 13 taken 9 times.
 9x worker_ec = detail::make_err(::WSAGetLastError());
220
1/1
✓ Branch 14 → 15 taken 9 times.
 9x ::closesocket(worker_sock);
221 9x worker_sock = INVALID_SOCKET;
222 }
223 }
224 // Released last so a reader that sees it also sees worker_ec.
225 79x worker_done.store(true, std::memory_order_release);
226 158x });
227
228 79x SOCKET accept_sock = INVALID_SOCKET;
229 79x 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
5/6
✓ Branch 42 → 43 taken 18 times.
✓ Branch 42 → 46 taken 80 times.
✓ Branch 44 → 45 taken 18 times.
✗ Branch 44 → 46 not taken.
✓ Branch 47 → 48 taken 18 times.
✓ Branch 47 → 49 taken 80 times.
 98x if (worker_done.load(std::memory_order_acquire) && worker_ec)
235 18x break;
236
237 80x WSAPOLLFD pfd{listen_sock, POLLRDNORM, 0};
238 80x int const n = ::WSAPoll(&pfd, 1, 100);
239
2/2
✓ Branch 50 → 51 taken 9 times.
✓ Branch 50 → 54 taken 71 times.
 80x if (n == SOCKET_ERROR)
240 {
241 9x accept_ec = detail::make_err(::WSAGetLastError());
242 9x break;
243 }
244
2/2
✓ Branch 54 → 55 taken 18 times.
✓ Branch 54 → 56 taken 53 times.
 71x if (n == 0)
245 18x 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
1/2
✗ Branch 56 → 57 not taken.
✓ Branch 56 → 59 taken 53 times.
 53x if ((pfd.revents & POLLRDNORM) == 0)
252 {
253 ✗ accept_ec = std::make_error_code(std::errc::connection_aborted);
254 ✗ break;
255 }
256
257 53x accept_sock = ::accept(listen_sock, nullptr, nullptr);
258
2/2
✓ Branch 60 → 61 taken 51 times.
✓ Branch 60 → 62 taken 2 times.
 53x if (accept_sock != INVALID_SOCKET)
259 51x break;
260 2x DWORD const err = ::WSAGetLastError();
261 // A readiness report with nothing left to accept: keep
262 // waiting for the worker's connection.
263
2/2
✓ Branch 63 → 64 taken 1 time.
✓ Branch 63 → 65 taken 1 time.
 2x if (err == WSAEWOULDBLOCK)
264 1x continue;
265 1x accept_ec = detail::make_err(err);
266 1x break;
267 19x }
268
269 79x worker.join();
270
271 79x ::closesocket(listen_sock);
272 79x remove_pair_path(dir, path);
273
274
2/2
✓ Branch 71 → 73 taken 10 times.
✓ Branch 71 → 76 taken 69 times.
 79x if (accept_ec)
275 {
276
1/2
✓ Branch 73 → 74 taken 10 times.
✗ Branch 73 → 75 not taken.
 10x if (worker_sock != INVALID_SOCKET)
277 10x ::closesocket(worker_sock);
278 10x return accept_ec;
279 }
280
2/2
✓ Branch 77 → 78 taken 18 times.
✓ Branch 77 → 81 taken 51 times.
 69x if (worker_ec)
281 {
282
1/2
✗ Branch 78 → 79 not taken.
✓ Branch 78 → 80 taken 18 times.
 18x if (accept_sock != INVALID_SOCKET)
283 ✗ ::closesocket(accept_sock);
284 18x 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 51x non_blocking = 0;
291
2/2
✓ Branch 82 → 83 taken 9 times.
✓ Branch 82 → 88 taken 42 times.
 51x if (::ioctlsocket(accept_sock, FIONBIO, &non_blocking) == SOCKET_ERROR)
292 {
293 9x auto ec = detail::make_err(::WSAGetLastError());
294 9x ::closesocket(accept_sock);
295 9x ::closesocket(worker_sock);
296 9x return ec;
297 }
298
299 42x a_sock = accept_sock;
300 42x b_sock = worker_sock;
301 42x return {};
302 83x }
303
304 std::error_code
305 42x assign_pair(
306 local_stream_socket& a,
307 local_stream_socket& b,
308 SOCKET a_sock,
309 SOCKET b_sock) noexcept
310 {
311
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)))
312 {
313 9x ::closesocket(a_sock);
314 9x ::closesocket(b_sock);
315 9x return ec;
316 }
317
318
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)))
319 {
320 9x a.close();
321 9x ::closesocket(b_sock);
322 9x return ec;
323 }
324
325 24x return {};
326 }
327
328 #endif
329
330 } // namespace
331
332 std::error_code
333 84x connect_pair(local_stream_socket& a, local_stream_socket& b) noexcept
334 {
335
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())
336 1x return std::make_error_code(std::errc::already_connected);
337
338 #if BOOST_COROSIO_POSIX
339 int a_fd = -1, b_fd = -1;
340 if (auto ec = make_pair_fds(SOCK_STREAM, a_fd, b_fd))
341 return ec;
342 return assign_pair(a, b, a_fd, b_fd);
343 #elif BOOST_COROSIO_HAS_IOCP
344 83x SOCKET a_sock = INVALID_SOCKET, b_sock = INVALID_SOCKET;
345
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))
346 41x return ec;
347 42x 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 connect_pair(local_datagram_socket& a, local_datagram_socket& b) noexcept
357 {
358 if (a.is_open() || b.is_open())
359 return std::make_error_code(std::errc::already_connected);
360
361 int a_fd = -1, b_fd = -1;
362 if (auto ec = make_pair_fds(SOCK_DGRAM, a_fd, b_fd))
363 return ec;
364 return assign_pair(a, b, a_fd, b_fd);
365 }
366
367 #endif
368
369 } // namespace boost::corosio
370