src/corosio/src/local_connect_pair.cpp

68.7% Lines (68/99) 100.0% List of functions (6/6) 64.9% Branches (37/57)
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 if (auto ec = a.assign(a_fd))
78 {
79 ::close(a_fd);
80 ::close(b_fd);
81 return ec;
82 }
83
84 if (auto ec = b.assign(b_fd))
85 {
86 a.close();
87 ::close(b_fd);
88 return ec;
89 }
90
91 return {};
92 }
93
94 #elif BOOST_COROSIO_HAS_IOCP
95
96 // Build a unique sub-directory under temp and return the full socket
97 // path inside it. Empty string on failure.
98 std::string
99 15x pick_pair_path(std::filesystem::path& dir_out)
100 {
101 namespace fs = std::filesystem;
102
103
5/5
✓ Branch 2 → 3 taken 3 times.
✓ Branch 2 → 8 taken 12 times.
✓ Branch 3 → 4 taken 3 times.
✓ Branch 4 → 5 taken 3 times.
✓ Branch 5 → 6 taken 3 times.
15x thread_local std::mt19937_64 gen{std::random_device{}()};
104
105
1/2
✓ Branch 36 → 9 taken 15 times.
✗ Branch 36 → 37 not taken.
15x for (int attempt = 0; attempt < 16; ++attempt)
106 {
107 auto candidate =
108
1/1
✓ Branch 13 → 14 taken 15 times.
30x fs::temp_directory_path() /
109
5/5
✓ Branch 9 → 10 taken 15 times.
✓ Branch 10 → 11 taken 15 times.
✓ Branch 11 → 12 taken 15 times.
✓ Branch 12 → 13 taken 15 times.
✓ Branch 14 → 15 taken 15 times.
45x ("co_pair_" + std::to_string(gen()));
110 15x std::error_code ec;
111
1/2
✓ Branch 21 → 22 taken 15 times.
✗ Branch 21 → 30 not taken.
15x if (fs::create_directory(candidate, ec))
112 {
113
1/1
✓ Branch 22 → 23 taken 15 times.
15x dir_out = candidate;
114
3/3
✓ Branch 23 → 24 taken 15 times.
✓ Branch 24 → 25 taken 15 times.
✓ Branch 25 → 26 taken 15 times.
15x return (candidate / "s").string();
115 }
116 15x }
117 return {};
118 }
119
120 void
121 15x remove_pair_path(std::filesystem::path const& dir, std::string const& path)
122 {
123 15x std::error_code ec;
124
1/1
✓ Branch 3 → 4 taken 15 times.
15x std::filesystem::remove(std::filesystem::path(path), ec);
125 15x std::filesystem::remove(dir, ec);
126 15x }
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 15x make_pair_sockets(SOCKET& a_sock, SOCKET& b_sock) noexcept
135 {
136 namespace fs = std::filesystem;
137
138 15x a_sock = INVALID_SOCKET;
139 15x b_sock = INVALID_SOCKET;
140
141 15x fs::path dir;
142 15x std::string path = pick_pair_path(dir);
143
1/2
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 7 taken 15 times.
15x if (path.empty())
144 return detail::make_err(ERROR_PATH_NOT_FOUND);
145
146 15x SOCKET listen_sock = ::WSASocketW(
147 AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
148
1/2
✗ Branch 8 → 9 not taken.
✓ Branch 8 → 13 taken 15 times.
15x 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 15x detail::un_sa_t addr{};
156 15x addr.sun_family = AF_UNIX;
157 30x std::memcpy(
158 15x addr.sun_path, path.c_str(),
159 15x (std::min)(path.size(), sizeof(addr.sun_path) - 1));
160 int addr_len = static_cast<int>(
161 15x offsetof(detail::un_sa_t, sun_path) + path.size() + 1);
162
163 15x if (::bind(
164 listen_sock, reinterpret_cast<sockaddr*>(&addr), addr_len)
165
1/2
✗ Branch 18 → 19 not taken.
✓ Branch 18 → 24 taken 15 times.
15x == SOCKET_ERROR)
166 {
167 auto ec = detail::make_err(::WSAGetLastError());
168 ::closesocket(listen_sock);
169 remove_pair_path(dir, path);
170 return ec;
171 }
172
173
1/2
✗ Branch 25 → 26 not taken.
✓ Branch 25 → 31 taken 15 times.
15x if (::listen(listen_sock, 1) == 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 15x SOCKET worker_sock = INVALID_SOCKET;
182 15x std::error_code worker_ec;
183
184 45x std::thread worker([&] {
185
1/1
✓ Branch 2 → 3 taken 15 times.
15x worker_sock = ::WSASocketW(
186 AF_UNIX, SOCK_STREAM, 0, nullptr, 0, WSA_FLAG_OVERLAPPED);
187
1/2
✗ Branch 3 → 4 not taken.
✓ Branch 3 → 7 taken 15 times.
15x if (worker_sock == INVALID_SOCKET)
188 {
189 worker_ec = detail::make_err(::WSAGetLastError());
190 return;
191 }
192
193 15x detail::un_sa_t caddr{};
194 15x caddr.sun_family = AF_UNIX;
195 30x std::memcpy(
196 15x caddr.sun_path, path.c_str(),
197 15x (std::min)(path.size(), sizeof(caddr.sun_path) - 1));
198 int caddr_len = static_cast<int>(
199 15x offsetof(detail::un_sa_t, sun_path) + path.size() + 1);
200
201
1/1
✓ Branch 11 → 12 taken 15 times.
15x if (::connect(
202 worker_sock, reinterpret_cast<sockaddr*>(&caddr), caddr_len)
203
1/2
✗ Branch 12 → 13 not taken.
✓ Branch 12 → 17 taken 15 times.
15x == SOCKET_ERROR)
204 {
205 worker_ec = detail::make_err(::WSAGetLastError());
206 ::closesocket(worker_sock);
207 worker_sock = INVALID_SOCKET;
208 }
209 15x });
210
211 15x SOCKET accept_sock = ::accept(listen_sock, nullptr, nullptr);
212 15x std::error_code accept_ec;
213
1/2
✗ Branch 35 → 36 not taken.
✓ Branch 35 → 38 taken 15 times.
15x if (accept_sock == INVALID_SOCKET)
214 accept_ec = detail::make_err(::WSAGetLastError());
215
216 15x worker.join();
217
218 15x ::closesocket(listen_sock);
219 15x remove_pair_path(dir, path);
220
221
1/2
✗ Branch 42 → 43 not taken.
✓ Branch 42 → 46 taken 15 times.
15x if (accept_ec)
222 {
223 if (worker_sock != INVALID_SOCKET)
224 ::closesocket(worker_sock);
225 return accept_ec;
226 }
227
1/2
✗ Branch 47 → 48 not taken.
✓ Branch 47 → 50 taken 15 times.
15x if (worker_ec)
228 {
229 ::closesocket(accept_sock);
230 return worker_ec;
231 }
232
233 15x a_sock = accept_sock;
234 15x b_sock = worker_sock;
235 15x return {};
236 15x }
237
238 std::error_code
239 15x assign_pair(
240 local_stream_socket& a,
241 local_stream_socket& b,
242 SOCKET a_sock,
243 SOCKET b_sock) noexcept
244 {
245
1/2
✗ Branch 4 → 5 not taken.
✓ Branch 4 → 8 taken 15 times.
15x if (auto ec = a.assign(static_cast<native_handle_type>(a_sock)))
246 {
247 ::closesocket(a_sock);
248 ::closesocket(b_sock);
249 return ec;
250 }
251
252
1/2
✗ Branch 10 → 11 not taken.
✓ Branch 10 → 14 taken 15 times.
15x if (auto ec = b.assign(static_cast<native_handle_type>(b_sock)))
253 {
254 a.close();
255 ::closesocket(b_sock);
256 return ec;
257 }
258
259 15x return {};
260 }
261
262 #endif
263
264 } // namespace
265
266 std::error_code
267 16x connect_pair(local_stream_socket& a, local_stream_socket& b) noexcept
268 {
269
5/6
✓ Branch 3 → 4 taken 15 times.
✓ Branch 3 → 6 taken 1 time.
✗ Branch 5 → 6 not taken.
✓ Branch 5 → 7 taken 15 times.
✓ Branch 8 → 9 taken 1 time.
✓ Branch 8 → 10 taken 15 times.
16x if (a.is_open() || b.is_open())
270 1x return std::make_error_code(std::errc::already_connected);
271
272 #if BOOST_COROSIO_POSIX
273 int a_fd = -1, b_fd = -1;
274 if (auto ec = make_pair_fds(SOCK_STREAM, a_fd, b_fd))
275 return ec;
276 return assign_pair(a, b, a_fd, b_fd);
277 #elif BOOST_COROSIO_HAS_IOCP
278 15x SOCKET a_sock = INVALID_SOCKET, b_sock = INVALID_SOCKET;
279
1/2
✗ Branch 12 → 13 not taken.
✓ Branch 12 → 14 taken 15 times.
15x if (auto ec = make_pair_sockets(a_sock, b_sock))
280 return ec;
281 15x return assign_pair(a, b, a_sock, b_sock);
282 #else
283 return detail::make_err(ENOSYS);
284 #endif
285 }
286
287 #if BOOST_COROSIO_POSIX
288
289 std::error_code
290 connect_pair(local_datagram_socket& a, local_datagram_socket& b) noexcept
291 {
292 if (a.is_open() || b.is_open())
293 return std::make_error_code(std::errc::already_connected);
294
295 int a_fd = -1, b_fd = -1;
296 if (auto ec = make_pair_fds(SOCK_DGRAM, a_fd, b_fd))
297 return ec;
298 return assign_pair(a, b, a_fd, b_fd);
299 }
300
301 #endif
302
303 } // namespace boost::corosio
304