1//===-- llvm/Support/raw_socket_stream.cpp - Socket streams --*- C++ -*-===//
2//
3// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
4// See https://llvm.org/LICENSE.txt for license information.
5// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
6//
7//===----------------------------------------------------------------------===//
8//
9// This file contains raw_ostream implementations for streams to communicate
10// via UNIX sockets
11//
12//===----------------------------------------------------------------------===//
13
14#include "llvm/Support/raw_socket_stream.h"
15#include "llvm/Config/config.h"
16#include "llvm/Support/Error.h"
17#include "llvm/Support/FileSystem.h"
18
19#include <atomic>
20#include <fcntl.h>
21#include <functional>
22
23#ifndef _WIN32
24#include <poll.h>
25#include <sys/socket.h>
26#include <sys/un.h>
27#else
28#include "llvm/Support/Windows/WindowsSupport.h"
29// winsock2.h must be included before afunix.h. Briefly turn off clang-format to
30// avoid error.
31// clang-format off
32#include <winsock2.h>
33#include <afunix.h>
34// clang-format on
35#include <io.h>
36#endif // _WIN32
37
38#if defined(HAVE_UNISTD_H)
39#include <unistd.h>
40#endif
41
42using namespace llvm;
43
44#ifdef _WIN32
45WSABalancer::WSABalancer() {
46 WSADATA WsaData;
47 ::memset(&WsaData, 0, sizeof(WsaData));
48 if (WSAStartup(MAKEWORD(2, 2), &WsaData) != 0) {
49 llvm::report_fatal_error("WSAStartup failed");
50 }
51}
52
53WSABalancer::~WSABalancer() { WSACleanup(); }
54#endif // _WIN32
55
56static std::error_code getLastSocketErrorCode() {
57#ifdef _WIN32
58 return std::error_code(::WSAGetLastError(), std::system_category());
59#else
60 return errnoAsErrorCode();
61#endif
62}
63
64static Expected<sockaddr_un> setSocketAddr(StringRef SocketPath) {
65 struct sockaddr_un Addr;
66 memset(s: &Addr, c: 0, n: sizeof(Addr));
67 Addr.sun_family = AF_UNIX;
68
69 if (sizeof(sockaddr_un::sun_path) <= SocketPath.size())
70 return make_error<StringError>(
71 Args: std::make_error_code(e: std::errc::filename_too_long),
72 Args: "Socket path exceeds sockaddr_un::sun_path size limit");
73
74 strncpy(dest: Addr.sun_path, src: SocketPath.str().c_str(), n: sizeof(Addr.sun_path) - 1);
75 return Addr;
76}
77
78static Expected<int> getSocketFD(StringRef SocketPath) {
79#ifdef _WIN32
80 SOCKET Socket = socket(AF_UNIX, SOCK_STREAM, 0);
81 if (Socket == INVALID_SOCKET) {
82#else
83 int Socket = socket(AF_UNIX, SOCK_STREAM, protocol: 0);
84 if (Socket == -1) {
85#endif // _WIN32
86 return llvm::make_error<StringError>(Args: getLastSocketErrorCode(),
87 Args: "Create socket failed");
88 }
89
90#ifdef __CYGWIN__
91 // On Cygwin, UNIX sockets involve a handshake between connect and accept
92 // to enable SO_PEERCRED/getpeereid handling. This necessitates accept being
93 // called before connect can return, but at least the tests in
94 // llvm/unittests/Support/raw_socket_stream_test do both on the same thread
95 // (first connect and then accept), resulting in a deadlock. This call turns
96 // off the handshake (and SO_PEERCRED/getpeereid support).
97 setsockopt(Socket, SOL_SOCKET, SO_PEERCRED, NULL, 0);
98#endif
99 Expected<struct sockaddr_un> Addr = setSocketAddr(SocketPath);
100 if (!Addr)
101 return Addr.takeError();
102
103 if (::connect(fd: Socket, addr: (struct sockaddr *)&*Addr, len: sizeof(*Addr)) == -1) {
104 ::close(fd: Socket);
105 return llvm::make_error<StringError>(Args: getLastSocketErrorCode(),
106 Args: "Connect socket failed");
107 }
108
109#ifdef _WIN32
110 return _open_osfhandle(Socket, 0);
111#else
112 return Socket;
113#endif // _WIN32
114}
115
116ListeningSocket::ListeningSocket(int SocketFD, StringRef SocketPath,
117 int PipeFD[2])
118 : FD(SocketFD), SocketPath(SocketPath), PipeFD{PipeFD[0], PipeFD[1]} {}
119
120ListeningSocket::ListeningSocket(ListeningSocket &&LS)
121 : FD(LS.FD.load()), SocketPath(LS.SocketPath),
122 PipeFD{LS.PipeFD[0], LS.PipeFD[1]} {
123
124 LS.FD = -1;
125 LS.SocketPath.clear();
126 LS.PipeFD[0] = -1;
127 LS.PipeFD[1] = -1;
128}
129
130Expected<ListeningSocket> ListeningSocket::createUnix(StringRef SocketPath,
131 int MaxBacklog) {
132
133 // Handle instances where the target socket address already exists and
134 // differentiate between a preexisting file with and without a bound socket
135 //
136 // ::bind will return std::errc:address_in_use if a file at the socket address
137 // already exists (e.g., the file was not properly unlinked due to a crash)
138 // even if another socket has not yet binded to that address
139 if (llvm::sys::fs::exists(Path: SocketPath)) {
140 Expected<int> MaybeFD = getSocketFD(SocketPath);
141 if (!MaybeFD) {
142
143 // Regardless of the error, notify the caller that a file already exists
144 // at the desired socket address and that there is no bound socket at that
145 // address. The file must be removed before ::bind can use the address
146 consumeError(Err: MaybeFD.takeError());
147 return llvm::make_error<StringError>(
148 Args: std::make_error_code(e: std::errc::file_exists),
149 Args: "Socket address unavailable");
150 }
151 ::close(fd: std::move(*MaybeFD));
152
153 // Notify caller that the provided socket address already has a bound socket
154 return llvm::make_error<StringError>(
155 Args: std::make_error_code(e: std::errc::address_in_use),
156 Args: "Socket address unavailable");
157 }
158
159#ifdef _WIN32
160 WSABalancer _;
161 SOCKET Socket = socket(AF_UNIX, SOCK_STREAM, 0);
162 if (Socket == INVALID_SOCKET)
163#else
164 int Socket = socket(AF_UNIX, SOCK_STREAM, protocol: 0);
165 if (Socket == -1)
166#endif
167 return llvm::make_error<StringError>(Args: getLastSocketErrorCode(),
168 Args: "socket create failed");
169
170#ifdef __CYGWIN__
171 // On Cygwin, UNIX sockets involve a handshake between connect and accept
172 // to enable SO_PEERCRED/getpeereid handling. This necessitates accept being
173 // called before connect can return, but at least the tests in
174 // llvm/unittests/Support/raw_socket_stream_test do both on the same thread
175 // (first connect and then accept), resulting in a deadlock. This call turns
176 // off the handshake (and SO_PEERCRED/getpeereid support).
177 setsockopt(Socket, SOL_SOCKET, SO_PEERCRED, NULL, 0);
178#endif
179 Expected<struct sockaddr_un> Addr = setSocketAddr(SocketPath);
180 if (!Addr)
181 return Addr.takeError();
182
183 if (::bind(fd: Socket, addr: (struct sockaddr *)&*Addr, len: sizeof(*Addr)) == -1) {
184 // Grab error code from call to ::bind before calling ::close
185 std::error_code EC = getLastSocketErrorCode();
186 ::close(fd: Socket);
187 return llvm::make_error<StringError>(Args&: EC, Args: "Bind error");
188 }
189
190 // Mark socket as passive so incoming connections can be accepted
191 if (::listen(fd: Socket, n: MaxBacklog) == -1)
192 return llvm::make_error<StringError>(Args: getLastSocketErrorCode(),
193 Args: "Listen error");
194
195 int PipeFD[2];
196#ifdef _WIN32
197 // Reserve 1 byte for the pipe and use default textmode
198 if (::_pipe(PipeFD, 1, 0) == -1)
199#else
200 if (::pipe(pipedes: PipeFD) == -1)
201#endif // _WIN32
202 return llvm::make_error<StringError>(Args: getLastSocketErrorCode(),
203 Args: "pipe failed");
204
205#ifdef _WIN32
206 return ListeningSocket{_open_osfhandle(Socket, 0), SocketPath, PipeFD};
207#else
208 return ListeningSocket{Socket, SocketPath, PipeFD};
209#endif // _WIN32
210}
211
212// If a file descriptor being monitored by ::poll is closed by another thread,
213// the result is unspecified. In the case ::poll does not unblock and return,
214// when ActiveFD is closed, you can provide another file descriptor via CancelFD
215// that when written to will cause poll to return. Typically CancelFD is the
216// read end of a unidirectional pipe.
217//
218// Timeout should be -1 to block indefinitly
219//
220// getActiveFD is a callback to handle ActiveFD's of std::atomic<int> and int
221static std::error_code
222manageTimeout(const std::chrono::milliseconds &Timeout,
223 const std::function<int()> &getActiveFD,
224 const std::optional<int> &CancelFD = std::nullopt) {
225 struct pollfd FD[2];
226 FD[0].events = POLLIN;
227#ifdef _WIN32
228 SOCKET WinServerSock = _get_osfhandle(getActiveFD());
229 FD[0].fd = WinServerSock;
230#else
231 FD[0].fd = getActiveFD();
232#endif
233 uint8_t FDCount = 1;
234 if (CancelFD.has_value()) {
235 FD[1].events = POLLIN;
236 FD[1].fd = CancelFD.value();
237 FDCount++;
238 }
239
240 // Keep track of how much time has passed in case ::poll or WSAPoll are
241 // interupted by a signal and need to be recalled
242 auto Start = std::chrono::steady_clock::now();
243 auto RemainingTimeout = Timeout;
244 int PollStatus = 0;
245 do {
246 // If Timeout is -1 then poll should block and RemainingTimeout does not
247 // need to be recalculated
248 if (PollStatus != 0 && Timeout != std::chrono::milliseconds(-1)) {
249 auto TotalElapsedTime =
250 std::chrono::duration_cast<std::chrono::milliseconds>(
251 d: std::chrono::steady_clock::now() - Start);
252
253 if (TotalElapsedTime >= Timeout)
254 return std::make_error_code(e: std::errc::operation_would_block);
255
256 RemainingTimeout = Timeout - TotalElapsedTime;
257 }
258#ifdef _WIN32
259 PollStatus = WSAPoll(FD, FDCount, RemainingTimeout.count());
260 } while (PollStatus == SOCKET_ERROR &&
261 getLastSocketErrorCode() == std::errc::interrupted);
262#else
263 PollStatus = ::poll(fds: FD, nfds: FDCount, timeout: RemainingTimeout.count());
264 } while (PollStatus == -1 &&
265 getLastSocketErrorCode() == std::errc::interrupted);
266#endif
267
268 // If ActiveFD equals -1 or CancelFD has data to be read then the operation
269 // has been canceled by another thread
270 if (getActiveFD() == -1 || (CancelFD.has_value() && FD[1].revents & POLLIN))
271 return std::make_error_code(e: std::errc::operation_canceled);
272#ifdef _WIN32
273 if (PollStatus == SOCKET_ERROR)
274#else
275 if (PollStatus == -1)
276#endif
277 return getLastSocketErrorCode();
278 if (PollStatus == 0)
279 return std::make_error_code(e: std::errc::timed_out);
280 if (FD[0].revents & POLLNVAL)
281 return std::make_error_code(e: std::errc::bad_file_descriptor);
282 return std::error_code();
283}
284
285Expected<std::unique_ptr<raw_socket_stream>>
286ListeningSocket::accept(const std::chrono::milliseconds &Timeout) {
287 auto getActiveFD = [this]() -> int { return FD; };
288 std::error_code TimeoutErr = manageTimeout(Timeout, getActiveFD, CancelFD: PipeFD[0]);
289 if (TimeoutErr)
290 return llvm::make_error<StringError>(Args&: TimeoutErr, Args: "Timeout error");
291
292 int AcceptFD;
293#ifdef _WIN32
294 SOCKET WinAcceptSock = ::accept(_get_osfhandle(FD), NULL, NULL);
295 AcceptFD = _open_osfhandle(WinAcceptSock, 0);
296#else
297 AcceptFD = ::accept(fd: FD, NULL, NULL);
298#endif
299
300 if (AcceptFD == -1)
301 return llvm::make_error<StringError>(Args: getLastSocketErrorCode(),
302 Args: "Socket accept failed");
303 return std::make_unique<raw_socket_stream>(args&: AcceptFD);
304}
305
306void ListeningSocket::shutdown() {
307 int ObservedFD = FD.load();
308
309 if (ObservedFD == -1)
310 return;
311
312 // If FD equals ObservedFD set FD to -1; If FD doesn't equal ObservedFD then
313 // another thread is responsible for shutdown so return
314 if (!FD.compare_exchange_strong(i1&: ObservedFD, i2: -1))
315 return;
316
317 ::close(fd: ObservedFD);
318 ::unlink(name: SocketPath.c_str());
319
320 // Ensure ::poll returns if shutdown is called by a separate thread
321 char Byte = 'A';
322 ssize_t written = ::write(fd: PipeFD[1], buf: &Byte, n: 1);
323
324 // Ignore any write() error
325 (void)written;
326}
327
328ListeningSocket::~ListeningSocket() {
329 shutdown();
330
331 // Close the pipe's FDs in the destructor instead of within
332 // ListeningSocket::shutdown to avoid unnecessary synchronization issues that
333 // would occur as PipeFD's values would have to be changed to -1
334 //
335 // The move constructor sets PipeFD to -1
336 if (PipeFD[0] != -1)
337 ::close(fd: PipeFD[0]);
338 if (PipeFD[1] != -1)
339 ::close(fd: PipeFD[1]);
340}
341
342//===----------------------------------------------------------------------===//
343// raw_socket_stream
344//===----------------------------------------------------------------------===//
345
346raw_socket_stream::raw_socket_stream(int SocketFD)
347 : raw_fd_stream(SocketFD, true) {}
348
349raw_socket_stream::~raw_socket_stream() = default;
350
351Expected<std::unique_ptr<raw_socket_stream>>
352raw_socket_stream::createConnectedUnix(StringRef SocketPath) {
353#ifdef _WIN32
354 WSABalancer _;
355#endif // _WIN32
356 Expected<int> FD = getSocketFD(SocketPath);
357 if (!FD)
358 return FD.takeError();
359 return std::make_unique<raw_socket_stream>(args&: *FD);
360}
361
362ssize_t raw_socket_stream::read(char *Ptr, size_t Size,
363 const std::chrono::milliseconds &Timeout) {
364 auto getActiveFD = [this]() -> int { return this->get_fd(); };
365 std::error_code Err = manageTimeout(Timeout, getActiveFD);
366 // Mimic raw_fd_stream::read error handling behavior
367 if (Err) {
368 raw_fd_stream::error_detected(EC: Err);
369 return -1;
370 }
371 return raw_fd_stream::read(Ptr, Size);
372}
373