Index/src / corosio / src / local_connect_pair.cpp

src/corosio/src/local_connect_pair.cpp

100.0% Lines (42/42) 100.0% List of functions (5/5) 63.0% Branches (34/54)
[ ] prev/next file n p prev/next uncovered line
local_connect_pair.cpp
f(x) Functions (5)
Function Calls Lines Branches Blocks
boost::corosio::(anonymous namespace)::make_pair_fds(int, int&, int&) :49 142x 100.0% 68.2% 94.0% std::__1::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) :76 61x 100.0% 45.0% 40.0% std::__1::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) :76 73x 100.0% – 90.0% boost::corosio::connect_pair(boost::corosio::local_stream_socket&, boost::corosio::local_stream_socket&) :333 81x 100.0% 83.3% 100.0% boost::corosio::connect_pair(boost::corosio::local_datagram_socket&, boost::corosio::local_datagram_socket&) :356 65x 100.0% 83.3% 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 142x make_pair_fds(int type, int& a_fd, int& b_fd) noexcept
50 {
51 int fds[2];
52
3/4
✓ Branch 0 taken 142 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
✓ Branch 3 taken 138 times.
 142x if (::socketpair(AF_UNIX, type, 0, fds) != 0)
53
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4x 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
2/2
✓ Branch 0 taken 134 times.
✓ Branch 1 taken 272 times.
 406x for (int i = 0; i < 2; ++i)
58 {
59
1/2
✓ Branch 0 taken 272 times.
✗ Branch 1 not taken.
 272x int flags = ::fcntl(fds[i], F_GETFL, 0);
60
5/6
✓ Branch 0 taken 270 times.
✓ Branch 1 taken 2 times.
✓ Branch 2 taken 270 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 2 times.
✓ Branch 5 taken 268 times.
 272x if (flags < 0 || ::fcntl(fds[i], F_SETFL, flags | O_NONBLOCK) < 0)
61 {
62
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4x auto ec = detail::make_err(errno);
63
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4x ::close(fds[0]);
64
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4x ::close(fds[1]);
65 4x return ec;
66 }
67 268x }
68
69 134x a_fd = fds[0];
70 134x b_fd = fds[1];
71 134x return {};
72 142x }
73
74 template<class Socket>
75 std::error_code
76 134x assign_pair(Socket& a, Socket& b, int a_fd, int b_fd) noexcept
77 {
78
3/4
✓ Branch 0 taken 71 times.
✓ Branch 1 taken 2 times.
✓ Branch 2 taken 61 times.
✗ Branch 3 not taken.
 134x if (auto ec = a.assign(a_fd))
79 {
80
1/4
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 2x ::close(a_fd);
81
1/4
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 2x ::close(b_fd);
82 2x return ec;
83 }
84
85
3/4
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 69 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 61 times.
 132x if (auto ec = b.assign(b_fd))
86 {
87 2x a.close();
88
1/4
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 2x ::close(b_fd);
89 2x return ec;
90 }
91
92 130x return {};
93 134x }
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 81x connect_pair(local_stream_socket& a, local_stream_socket& b) noexcept
334 {
335
3/4
✓ Branch 0 taken 79 times.
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 79 times.
 81x 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 79x int a_fd = -1, b_fd = -1;
340
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 73 times.
 79x if (auto ec = make_pair_fds(SOCK_STREAM, a_fd, b_fd))
341 6x return ec;
342 73x 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 81x }
352
353 #if BOOST_COROSIO_POSIX
354
355 std::error_code
356 65x connect_pair(local_datagram_socket& a, local_datagram_socket& b) noexcept
357 {
358
3/4
✓ Branch 0 taken 65 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 63 times.
 65x if (a.is_open() || b.is_open())
359 2x return std::make_error_code(std::errc::already_connected);
360
361 63x int a_fd = -1, b_fd = -1;
362
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 61 times.
 63x if (auto ec = make_pair_fds(SOCK_DGRAM, a_fd, b_fd))
363 2x return ec;
364 61x return assign_pair(a, b, a_fd, b_fd);
365 65x }
366
367 #endif
368
369 } // namespace boost::corosio
370