src/corosio/src/local_connect_pair.cpp

66.3% Lines (69/104) 100.0% List of functions (6/6) 61.0% Branches (36/59)
local_connect_pair.cpp
f(x) Functions (6)
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 <cstring>
26 #include <filesystem>
27 #include <random>
28 #include <string>
29 #include <thread>
30
31 #ifndef WIN32_LEAN_AND_MEAN
32 #define WIN32_LEAN_AND_MEAN
33 #endif
34 #include <WinSock2.h>
35
36 #ifndef AF_UNIX
37 #define AF_UNIX 1
38 #endif
39 #endif
40
41 namespace boost::corosio {
42
43 namespace {
44
45 #if BOOST_COROSIO_POSIX
46
47 std::error_code
48 make_pair_fds(int type, int& a_fd, int& b_fd) noexcept
49 {
50 int fds[2];
51 if (::socketpair(AF_UNIX, type, 0, fds) != 0)
52 return detail::make_err(errno);
53
54 // assign() is documented "adopt-only" and will not mutate the fd;
55 // set O_NONBLOCK before transferring ownership.
56 for (int i = 0; i < 2; ++i)
57 {
58 int flags = ::fcntl(fds[i], F_GETFL, 0);
59 if (flags < 0 || ::fcntl(fds[i], F_SETFL, flags | O_NONBLOCK) < 0)
60 {
61 auto ec = detail::make_err(errno);
62 ::close(fds[0]);
63 ::close(fds[1]);
64 return ec;
65 }
66 }
67
68 a_fd = fds[0];
69 b_fd = fds[1];
70 return {};
71 }
72
73 template<class Socket>
74 std::error_code
75 assign_pair(Socket& a, Socket& b, int a_fd, int b_fd) noexcept
76 {
77 try
78 {
79 a.assign(a_fd);
80 }
81 catch (std::system_error const& e)
82 {
83 ::close(a_fd);
84 ::close(b_fd);
85 return e.code();
86 }
87
88 try
89 {
90 b.assign(b_fd);
91 }
92 catch (std::system_error const& e)
93 {
94 a.close();
95 ::close(b_fd);
96 return e.code();
97 }
98
99 return {};
100 }
101
102 #elif BOOST_COROSIO_HAS_IOCP
103
104 // Build a unique sub-directory under temp and return the full socket
105 // path inside it. Empty string on failure.
106 std::string
107 1x pick_pair_path(std::filesystem::path& dir_out)
108 {
109 namespace fs = std::filesystem;
110
111
4/5
✓ Branch 2 → 3 taken 1 time.
✗ Branch 2 → 8 not taken.
✓ Branch 3 → 4 taken 1 time.
✓ Branch 4 → 5 taken 1 time.
✓ Branch 5 → 6 taken 1 time.
1x thread_local std::mt19937_64 gen{std::random_device{}()};
112
113
1/2
✓ Branch 36 → 9 taken 1 time.
✗ Branch 36 → 37 not taken.
1x for (int attempt = 0; attempt < 16; ++attempt)
114 {
115 auto candidate =
116
1/1
✓ Branch 13 → 14 taken 1 time.
2x fs::temp_directory_path() /
117
5/5
✓ Branch 9 → 10 taken 1 time.
✓ Branch 10 → 11 taken 1 time.
✓ Branch 11 → 12 taken 1 time.
✓ Branch 12 → 13 taken 1 time.
✓ Branch 14 → 15 taken 1 time.
3x ("co_pair_" + std::to_string(gen()));
118 1x std::error_code ec;
119
1/2
✓ Branch 21 → 22 taken 1 time.
✗ Branch 21 → 30 not taken.
1x if (fs::create_directory(candidate, ec))
120 {
121
1/1
✓ Branch 22 → 23 taken 1 time.
1x dir_out = candidate;
122
3/3
✓ Branch 23 → 24 taken 1 time.
✓ Branch 24 → 25 taken 1 time.
✓ Branch 25 → 26 taken 1 time.
1x return (candidate / "s").string();
123 }
124 1x }
125 return {};
126 }
127
128 void
129 1x remove_pair_path(std::filesystem::path const& dir, std::string const& path)
130 {
131 1x std::error_code ec;
132
1/1
✓ Branch 3 → 4 taken 1 time.
1x std::filesystem::remove(std::filesystem::path(path), ec);
133 1x std::filesystem::remove(dir, ec);
134 1x }
135
136 // Synchronously rendezvous two AF_UNIX SOCK_STREAM sockets. The
137 // listener and accept happen on the caller's thread; the connect
138 // runs on a short-lived worker to avoid a deadlock. The returned
139 // sockets are created with WSA_FLAG_OVERLAPPED so they can be
140 // registered with IOCP by assign_socket().
141 std::error_code
142 1x make_pair_sockets(SOCKET& a_sock, SOCKET& b_sock) noexcept
143 {
144 namespace fs = std::filesystem;
145
146 1x a_sock = INVALID_SOCKET;
147 1x b_sock = INVALID_SOCKET;
148
149 1x fs::path dir;
150 1x std::string path = pick_pair_path(dir);
151
1/2
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 7 taken 1 time.
1x if (path.empty())
152 return detail::make_err(ERROR_PATH_NOT_FOUND);
153
154 1x SOCKET listen_sock = ::WSASocketW(
155 AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
156
1/2
✗ Branch 8 → 9 not taken.
✓ Branch 8 → 13 taken 1 time.
1x if (listen_sock == INVALID_SOCKET)
157 {
158 auto ec = detail::make_err(::WSAGetLastError());
159 remove_pair_path(dir, path);
160 return ec;
161 }
162
163 1x detail::un_sa_t addr{};
164 1x addr.sun_family = AF_UNIX;
165 2x std::memcpy(
166 1x addr.sun_path, path.c_str(),
167 1x (std::min)(path.size(), sizeof(addr.sun_path) - 1));
168 int addr_len = static_cast<int>(
169 1x offsetof(detail::un_sa_t, sun_path) + path.size() + 1);
170
171 1x if (::bind(
172 listen_sock, reinterpret_cast<sockaddr*>(&addr), addr_len)
173
1/2
✗ Branch 18 → 19 not taken.
✓ Branch 18 → 24 taken 1 time.
1x == SOCKET_ERROR)
174 {
175 auto ec = detail::make_err(::WSAGetLastError());
176 ::closesocket(listen_sock);
177 remove_pair_path(dir, path);
178 return ec;
179 }
180
181
1/2
✗ Branch 25 → 26 not taken.
✓ Branch 25 → 31 taken 1 time.
1x if (::listen(listen_sock, 1) == SOCKET_ERROR)
182 {
183 auto ec = detail::make_err(::WSAGetLastError());
184 ::closesocket(listen_sock);
185 remove_pair_path(dir, path);
186 return ec;
187 }
188
189 1x SOCKET worker_sock = INVALID_SOCKET;
190 1x std::error_code worker_ec;
191
192 3x std::thread worker([&] {
193
1/1
✓ Branch 2 → 3 taken 1 time.
1x worker_sock = ::WSASocketW(
194 AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
195
1/2
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 7 taken 1 time.
1x if (worker_sock == INVALID_SOCKET)
196 {
197 worker_ec = detail::make_err(::WSAGetLastError());
198 return;
199 }
200
201 1x detail::un_sa_t caddr{};
202 1x caddr.sun_family = AF_UNIX;
203 2x std::memcpy(
204 1x caddr.sun_path, path.c_str(),
205 1x (std::min)(path.size(), sizeof(caddr.sun_path) - 1));
206 int caddr_len = static_cast<int>(
207 1x offsetof(detail::un_sa_t, sun_path) + path.size() + 1);
208
209
1/1
✓ Branch 11 → 12 taken 1 time.
1x if (::connect(
210 worker_sock, reinterpret_cast<sockaddr*>(&caddr), caddr_len)
211
1/2
✗ Branch 12 → 13 not taken.
✓ Branch 12 → 17 taken 1 time.
1x == SOCKET_ERROR)
212 {
213 worker_ec = detail::make_err(::WSAGetLastError());
214 ::closesocket(worker_sock);
215 worker_sock = INVALID_SOCKET;
216 }
217 1x });
218
219 1x SOCKET accept_sock = ::accept(listen_sock, nullptr, nullptr);
220 1x std::error_code accept_ec;
221
1/2
✗ Branch 35 → 36 not taken.
✓ Branch 35 → 38 taken 1 time.
1x if (accept_sock == INVALID_SOCKET)
222 accept_ec = detail::make_err(::WSAGetLastError());
223
224 1x worker.join();
225
226 1x ::closesocket(listen_sock);
227 1x remove_pair_path(dir, path);
228
229
1/2
✗ Branch 42 → 43 not taken.
✓ Branch 42 → 46 taken 1 time.
1x if (accept_ec)
230 {
231 if (worker_sock != INVALID_SOCKET)
232 ::closesocket(worker_sock);
233 return accept_ec;
234 }
235
1/2
✗ Branch 47 → 48 not taken.
✓ Branch 47 → 50 taken 1 time.
1x if (worker_ec)
236 {
237 ::closesocket(accept_sock);
238 return worker_ec;
239 }
240
241 1x a_sock = accept_sock;
242 1x b_sock = worker_sock;
243 1x return {};
244 1x }
245
246 std::error_code
247 1x assign_pair(
248 local_stream_socket& a,
249 local_stream_socket& b,
250 SOCKET a_sock,
251 SOCKET b_sock) noexcept
252 {
253 try
254 {
255
1/1
✓ Branch 2 → 3 taken 1 time.
1x a.assign(static_cast<native_handle_type>(a_sock));
256 }
257 catch (std::system_error const& e)
258 {
259 ::closesocket(a_sock);
260 ::closesocket(b_sock);
261 return e.code();
262 }
263
264 try
265 {
266
1/1
✓ Branch 3 → 4 taken 1 time.
1x b.assign(static_cast<native_handle_type>(b_sock));
267 }
268 catch (std::system_error const& e)
269 {
270 a.close();
271 ::closesocket(b_sock);
272 return e.code();
273 }
274
275 1x return {};
276 }
277
278 #endif
279
280 } // namespace
281
282 std::error_code
283 2x connect_pair(local_stream_socket& a, local_stream_socket& b) noexcept
284 {
285
5/6
✓ Branch 3 → 4 taken 1 time.
✓ Branch 3 → 6 taken 1 time.
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 7 taken 1 time.
✓ Branch 8 → 9 taken 1 time.
✓ Branch 8 → 11 taken 1 time.
2x if (a.is_open() || b.is_open())
286 1x return detail::make_err(
287 #if BOOST_COROSIO_POSIX
288 EISCONN
289 #else
290 WSAEISCONN
291 #endif
292 1x );
293
294 #if BOOST_COROSIO_POSIX
295 int a_fd = -1, b_fd = -1;
296 if (auto ec = make_pair_fds(SOCK_STREAM, a_fd, b_fd))
297 return ec;
298 return assign_pair(a, b, a_fd, b_fd);
299 #elif BOOST_COROSIO_HAS_IOCP
300 1x SOCKET a_sock = INVALID_SOCKET, b_sock = INVALID_SOCKET;
301
1/2
✗ Branch 13 → 14 not taken.
✓ Branch 13 → 15 taken 1 time.
1x if (auto ec = make_pair_sockets(a_sock, b_sock))
302 return ec;
303 1x return assign_pair(a, b, a_sock, b_sock);
304 #else
305 return detail::make_err(ENOSYS);
306 #endif
307 }
308
309 #if BOOST_COROSIO_POSIX
310
311 std::error_code
312 connect_pair(local_datagram_socket& a, local_datagram_socket& b) noexcept
313 {
314 if (a.is_open() || b.is_open())
315 return detail::make_err(EISCONN);
316
317 int a_fd = -1, b_fd = -1;
318 if (auto ec = make_pair_fds(SOCK_DGRAM, a_fd, b_fd))
319 return ec;
320 return assign_pair(a, b, a_fd, b_fd);
321 }
322
323 #endif
324
325 } // namespace boost::corosio
326