Fix some windows build issues

Author: cpsughrue

llvm/include/llvm/Support/raw_socket_stream.h

@@ -59,15 +59,16 @@ class WSABalancer {
 ///
 class ListeningSocket {
 
-  /// If ListeningSocket::shutdown is used by a signal handler to clean up
-  /// ListeningSocket resources FD may be closed while ::poll is waiting for
-  /// FD to become ready to perform I/O. When FD is closed ::poll will
-  /// continue to block so use the self-pipe trick to get ::poll to return
-  int PipeFD[2];
-  std::mutex PipeMutex;
   std::atomic<int> FD;
-  std::string SocketPath; // Never modified
-  ListeningSocket(int SocketFD, StringRef SocketPath);
+  std::string SocketPath; // Never modified after construction
+
+  /// If a seperate thread calls ListeningSocket::shutdown, the ListeningSocket
+  /// file descriptor (FD) could be closed while ::poll is waiting for it to be
+  /// ready to performa I/O operations. ::poll with continue to block even after
+  /// FD is closed so use a self-pipe mechanism to get ::poll to return
+  int PipeFD[2]; // Never modified after construction
+
+  ListeningSocket(int SocketFD, StringRef SocketPath, int PipeFD[2]);
 
 #ifdef _WIN32
   WSABalancer _;
@@ -79,14 +80,20 @@ class ListeningSocket {
   ListeningSocket(const ListeningSocket &LS) = delete;
   ListeningSocket &operator=(const ListeningSocket &) = delete;
 
-  /// Closes the socket's FD and unlinks the socket file from the file system.
-  /// The method is thread and signal safe
+  /// Closes the FD, unlinks the socket file, and writes to PipeFD.
+  ///
+  /// After the construction of the ListeningSocket, shutdown is signal safe if
+  /// it is called during the lifetime of the object. shutdown can be called
+  /// concurrently with ListeningSocket::accept as writing to PipeFD will cause
+  /// a blocking call to ::poll to return.
+  ///
+  /// Once shutdown is called there is no way to reinitialize ListeningSocket.
   void shutdown();
 
   /// Accepts an incoming connection on the listening socket. This method can
   /// optionally either block until a connection is available or timeout after a
   /// specified amount of time has passed. By default the method will block
-  /// until the socket has recieved a connection
+  /// until the socket has recieved a connection.
   ///
   /// \param Timeout An optional timeout duration in milliseconds
   ///

llvm/lib/Support/raw_socket_stream.cpp

@@ -71,36 +71,40 @@ static sockaddr_un setSocketAddr(StringRef SocketPath) {
 
 static Expected<int> getSocketFD(StringRef SocketPath) {
 #ifdef _WIN32
-  SOCKET MaybeSocket = socket(AF_UNIX, SOCK_STREAM, 0);
-  if (MaybeSocket == INVALID_SOCKET) {
+  SOCKET Socket = socket(AF_UNIX, SOCK_STREAM, 0);
+  if (Socket == INVALID_SOCKET) {
 #else
-  int MaybeSocket = socket(AF_UNIX, SOCK_STREAM, 0);
-  if (MaybeSocket == -1) {
+  int Socket = socket(AF_UNIX, SOCK_STREAM, 0);
+  if (Socket == -1) {
 #endif // _WIN32
     return llvm::make_error<StringError>(getLastSocketErrorCode(),
                                          "Create socket failed");
   }
 
   struct sockaddr_un Addr = setSocketAddr(SocketPath);
-  if (::connect(MaybeSocket, (struct sockaddr *)&Addr, sizeof(Addr)) == -1)
+  if (::connect(Socket, (struct sockaddr *)&Addr, sizeof(Addr)) == -1)
     return llvm::make_error<StringError>(getLastSocketErrorCode(),
                                          "Connect socket failed");
 
 #ifdef _WIN32
-  return _open_osfhandle(MaybeWinsocket, 0);
+  return _open_osfhandle(Socket, 0);
 #else
-  return MaybeSocket;
+  return Socket;
 #endif // _WIN32
 }
 
-ListeningSocket::ListeningSocket(int SocketFD, StringRef SocketPath)
-    : FD(SocketFD), SocketPath(SocketPath) {}
+ListeningSocket::ListeningSocket(int SocketFD, StringRef SocketPath,
+                                 int PipeFD[2])
+    : FD(SocketFD), SocketPath(SocketPath), PipeFD{PipeFD[0], PipeFD[1]} {}
 
 ListeningSocket::ListeningSocket(ListeningSocket &&LS)
-    : FD(LS.FD.load()), SocketPath(LS.SocketPath) {
+    : FD(LS.FD.load()),
+      SocketPath(LS.SocketPath), PipeFD{LS.PipeFD[0], LS.PipeFD[1]} {
 
-  LS.SocketPath.clear();
   LS.FD = -1;
+  LS.SocketPath.clear();
+  LS.PipeFD[0] = -1;
+  LS.PipeFD[1] = -1;
 }
 
 Expected<ListeningSocket>
@@ -135,36 +139,38 @@ ListeningSocket::createListeningUnixSocket(StringRef SocketPath,
 
 #ifdef _WIN32
   WSABalancer _;
-  SOCKET MaybeSocket = socket(AF_UNIX, SOCK_STREAM, 0);
-  if (MaybeSocket == INVALID_SOCKET) {
+  SOCKET Socket = socket(AF_UNIX, SOCK_STREAM, 0);
+  if (Socket == INVALID_SOCKET) {
 #else
-  int MaybeSocket = socket(AF_UNIX, SOCK_STREAM, 0);
-  if (MaybeSocket == -1) {
+  int Socket = socket(AF_UNIX, SOCK_STREAM, 0);
+  if (Socket == -1)
 #endif
     return llvm::make_error<StringError>(getLastSocketErrorCode(),
                                          "socket create failed");
-  }
 
   struct sockaddr_un Addr = setSocketAddr(SocketPath);
-  if (::bind(MaybeSocket, (struct sockaddr *)&Addr, sizeof(Addr)) == -1) {
+  if (::bind(Socket, (struct sockaddr *)&Addr, sizeof(Addr)) == -1) {
     // Grab error code from call to ::bind before calling ::close
     std::error_code EC = getLastSocketErrorCode();
-    ::close(MaybeSocket);
+    ::close(Socket);
     return llvm::make_error<StringError>(EC, "Bind error");
   }
 
   // Mark socket as passive so incoming connections can be accepted
-  if (::listen(MaybeSocket, MaxBacklog) == -1)
+  if (::listen(Socket, MaxBacklog) == -1)
     return llvm::make_error<StringError>(getLastSocketErrorCode(),
                                          "Listen error");
 
-  int Socket;
+  int PipeFD[2];
+  if (::pipe(PipeFD) == -1)
+    return llvm::make_error<StringError>(getLastSocketErrorCode(),
+                                         "pipe failed");
+
 #ifdef _WIN32
-  Socket = _open_osfhandle(MaybeWinsocket, 0);
+  return ListeningSocket{_open_osfhandle(Socket, 0), SocketPath, PipeFD};
 #else
-  Socket = MaybeSocket;
+  return ListeningSocket{Socket, SocketPath, PipeFD};
 #endif // _WIN32
-  return ListeningSocket{Socket, SocketPath};
 }
 
 Expected<std::unique_ptr<raw_socket_stream>>
@@ -180,12 +186,7 @@ ListeningSocket::accept(std::optional<std::chrono::milliseconds> Timeout) {
 #endif
 
   FDs[1].events = POLLIN;
-  PipeMutex.lock();
-  if (::pipe(PipeFD) == -1)
-    return llvm::make_error<StringError>(getLastSocketErrorCode(),
-                                         "pipe failed");
   FDs[1].fd = PipeFD[0];
-  PipeMutex.unlock();
 
   int TimeoutCount = Timeout.value_or(std::chrono::milliseconds(-1)).count();
   int PollStatus = ::poll(FDs, 2, TimeoutCount);
@@ -223,14 +224,22 @@ void ListeningSocket::shutdown() {
   ::close(FD);
   ::unlink(SocketPath.c_str());
 
+  // Ensure ::poll returns if shutdown is called by a seperate thread
   char Byte = 'A';
-  PipeMutex.lock();
-  write(PipeFD[1], &Byte, 1);
-  PipeMutex.unlock();
+  ::write(PipeFD[1], &Byte, 1);
+
   FD = -1;
 }
 
-ListeningSocket::~ListeningSocket() { shutdown(); }
+ListeningSocket::~ListeningSocket() {
+  shutdown();
+
+  // Close the pipe's FDs in the destructor instead of within
+  // ListeningSocket::shutdown to avoid unnecessary synchronization issues that
+  // would occur as PipeFD's values would have to be changed to -1
+  ::close(PipeFD[0]);
+  ::close(PipeFD[1]);
+}
 
 //===----------------------------------------------------------------------===//
 //  raw_socket_stream