util: Add Windows support

Add Windows-specific code to support building and running on Windows:

- util.h: Guard ProcessId/SocketId/SocketError type aliases with WIN32
  ifdefs so they use SOCKET/uintptr_t on Windows and int on Unix.
  Add winsock2.h include on Windows.
- util.cpp: Guard Unix-specific system headers with WIN32 ifdefs. Add
  Windows-specific includes (windows.h, winsock2.h). Guard MaxFd() with
  #ifndef WIN32. Add GetCurrentThreadId() branch in ThreadName(). Add
  win32Socketpair() forward-declare. Add Windows branch in SocketPair()
  using win32Socketpair(). Add CommandLineFromArgv() helper needed to
  construct CreateProcess command lines. Add Windows branch in
  SpawnProcess() using named pipes and WSADuplicateSocket to pass socket
  to child. Add Windows branch in StartSpawned() reading socket from
  named pipe. Add Windows branch in WaitProcess() using
  WaitForSingleObject/GetExitCodeProcess.
- proxy-io.h: Add Windows branch in StreamSocketId() using
  getWin32Handle().
- proxy.cpp: Add SocketOutputStream class on Windows (analogous to
  FdOutputStream but using SOCKET/send()). Add Windows branch in
  EventLoop constructor to create m_post_writer using SocketOutputStream.

Author: ryanofsky

include/mp/proxy-io.h

@@ -214,6 +214,9 @@ std::string LongThreadName(const char* exe_name);
 inline SocketId StreamSocketId(const Stream& stream)
 {
     if (stream) KJ_IF_MAYBE(fd, stream->getFd()) return *fd;
+#ifdef WIN32
+    if (stream) KJ_IF_MAYBE(handle, stream->getWin32Handle()) return reinterpret_cast<SocketId>(*handle);
+#endif
     throw std::logic_error("Stream socket unset");
 }
 

include/mp/util.h

@@ -23,6 +23,10 @@
 #include <variant>
 #include <vector>
 
+#ifdef WIN32
+#include <winsock2.h>
+#endif
+
 namespace mp {
 
 //! Generic utility functions used by capnp code.
@@ -254,12 +258,20 @@ std::string LogEscape(const kj::StringTree& string, size_t max_size);
 
 using Stream = kj::Own<kj::AsyncIoStream>;
 
+#ifdef WIN32
+using ProcessId = uintptr_t;
+using SocketId = uintptr_t;
+constexpr SocketId SocketError{INVALID_SOCKET};
+#else
 using ProcessId = int;
 using SocketId = int;
 constexpr SocketId SocketError{-1};
+#endif
 
 //! Information about parent process passed to child process as a command-line
 //! argument. On unix this is the child socket fd number formatted as a string.
+//! On windows, this is a path to a named pipe the parent process will write
+//! WSADuplicateSocket info to.
 using ConnectInfo = std::string;
 
 //! Callback type used by SpawnProcess below.

src/mp/proxy.cpp

@@ -194,6 +194,40 @@ void Connection::removeSyncCleanup(CleanupIt it)
     m_sync_cleanup_fns.erase(it);
 }
 
+#ifdef WIN32
+//! Synchronous socket output stream. Cap'n Proto library only provides limited
+//! support for synchronous IO. It provides `FdOutputStream` which wraps unix
+//! file descriptors and calls write() internally, and `HandleOutStream` which
+//! wraps windows HANDLE values and calls WriteFile() internally. This class
+//! just provides analogous functionality wrapping SOCKET values and calls
+//! send() internally.
+class SocketOutputStream : public kj::OutputStream {
+public:
+  explicit SocketOutputStream(SOCKET socket) : m_socket(socket) {}
+
+  void write(const void* buffer, size_t size) override;
+
+private:
+  SOCKET m_socket;
+};
+
+static constexpr size_t WRITE_CLAMP_SIZE = 1u << 30;  // 1GB clamp for Windows, like FdOutputStream
+
+void SocketOutputStream::write(const void* buffer, size_t size) {
+  const char* pos = reinterpret_cast<const char*>(buffer);
+
+  while (size > 0) {
+    int n = send(m_socket, pos, static_cast<int>(kj::min(size, WRITE_CLAMP_SIZE)), 0);
+
+    KJ_WIN32(n != SOCKET_ERROR, "send() failed");
+    KJ_ASSERT(n > 0, "send() returned zero.");
+
+    pos += n;
+    size -= n;
+  }
+}
+#endif
+
 void EventLoop::addAsyncCleanup(std::function<void()> fn)
 {
     const Lock lock(m_mutex);
@@ -229,6 +263,10 @@ EventLoop::EventLoop(const char* exe_name, LogOptions log_opts, void* context)
     m_post_stream = kj::mv(pipe.ends[1]);
     KJ_IF_MAYBE(fd, m_post_stream->getFd()) {
         m_post_writer = kj::heap<kj::FdOutputStream>(*fd);
+#ifdef WIN32
+    } else KJ_IF_MAYBE(handle, m_post_stream->getWin32Handle()) {
+        m_post_writer = kj::heap<SocketOutputStream>(reinterpret_cast<SOCKET>(*handle));
+#endif
     } else {
         throw std::logic_error("Could not get file descriptor for new pipe.");
     }

src/mp/util.cpp

@@ -15,16 +15,24 @@
 #include <pthread.h>
 #include <sstream>
 #include <string>
-#include <fcntl.h>
-#include <sys/resource.h>
-#include <sys/socket.h>
-#include <sys/wait.h>
 #include <system_error>
 #include <thread> // NOLINT(misc-include-cleaner) // IWYU pragma: keep
 #include <unistd.h>
 #include <utility>
 #include <vector>
 
+#ifdef WIN32
+#include <atomic>
+#include <windows.h>
+#include <winsock2.h>
+#else
+#include <fcntl.h>
+#include <sys/resource.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+#endif
+
 #ifdef __linux__
 #include <sys/syscall.h>
 #endif
@@ -33,6 +41,11 @@
 #include <pthread_np.h>
 #endif // HAVE_PTHREAD_GETTHREADID_NP
 
+#ifdef WIN32
+// Forward-declare internal capnp function.
+namespace kj { namespace _ { int win32Socketpair(SOCKET socks[2]); } }
+#endif
+
 namespace fs = std::filesystem;
 
 namespace mp {
@@ -49,6 +62,7 @@ std::vector<char*> MakeArgv(const std::vector<std::string>& args)
     return argv;
 }
 
+#ifndef WIN32
 //! Return highest possible file descriptor.
 size_t MaxFd()
 {
@@ -59,6 +73,7 @@ size_t MaxFd()
         return 1023;
     }
 }
+#endif
 
 } // namespace
 
@@ -80,6 +95,8 @@ std::string ThreadName(const char* exe_name)
     // the former are shorter and are the same as what gdb prints "LWP ...".
 #ifdef __linux__
     buffer << syscall(SYS_gettid);
+#elif defined(WIN32)
+    buffer << GetCurrentThreadId();
 #elif defined(HAVE_PTHREAD_THREADID_NP)
     uint64_t tid = 0;
     pthread_threadid_np(nullptr, &tid);
@@ -117,10 +134,56 @@ std::string LogEscape(const kj::StringTree& string, size_t max_size)
     return result;
 }
 
+//! Generate command line that the executable being invoked will split up using
+//! the CommandLineToArgvW function, which expects arguments with spaces to be
+//! quoted, quote characters to be backslash-escaped, and backslashes to also be
+//! backslash-escaped, but only if they precede a quote character.
+std::string CommandLineFromArgv(const std::vector<std::string>& argv)
+{
+    std::string out;
+    for (const auto& arg : argv) {
+        if (!out.empty()) out += " ";
+        if (!arg.empty() && arg.find_first_of(" \t\"") == std::string::npos) {
+            // Argument has no quotes or spaces so escaping not necessary.
+            out += arg;
+        } else {
+            out += '"'; // Start with a quote
+            for (size_t i = 0; i < arg.size(); ++i) {
+                if (arg[i] == '\\') {
+                    // Count consecutive backslashes
+                    size_t backslash_count = 0;
+                    while (i < arg.size() && arg[i] == '\\') {
+                        ++backslash_count;
+                        ++i;
+                    }
+                    if (i < arg.size() && arg[i] == '"') {
+                        // Backslashes before a quote need to be doubled
+                        out.append(backslash_count * 2 + 1, '\\');
+                        out.push_back('"');
+                    } else {
+                        // Otherwise, backslashes remain as-is
+                        out.append(backslash_count, '\\');
+                        --i; // Compensate for the outer loop's increment
+                    }
+                } else if (arg[i] == '"') {
+                    // Escape double quotes with a backslash
+                    out.push_back('\\');
+                    out.push_back('"');
+                } else {
+                    out.push_back(arg[i]);
+                }
+            }
+            out += '"'; // End with a quote
+        }
+    }
+    return out;
+}
+
 std::tuple<ProcessId, SocketId> SpawnProcess(ConnectInfoToArgsFn&& connect_info_to_args)
 {
     auto fds{SocketPair()};
 
+#ifndef WIN32
     // Evaluate the callback and build the argv array before forking.
     //
     // The parent process may be multi-threaded and holding internal library
@@ -177,17 +240,66 @@ std::tuple<ProcessId, SocketId> SpawnProcess(ConnectInfoToArgsFn&& connect_info_
         _exit(127);
     }
     return {pid, fds[1]};
+#else
+    // Create windows pipe to send socket over to child process.
+    static std::atomic<int> counter{1};
+    ConnectInfo pipe_path{"\\\\.\\pipe\\mp-" + std::to_string(GetCurrentProcessId()) + "-" + std::to_string(counter.fetch_add(1))};
+    HANDLE pipe{CreateNamedPipeA(pipe_path.c_str(), PIPE_ACCESS_OUTBOUND, PIPE_TYPE_MESSAGE | PIPE_WAIT, 1, 0, 0, 0, nullptr)};
+    KJ_WIN32(pipe != INVALID_HANDLE_VALUE, "CreateNamedPipe failed");
+
+    // Start child process
+    std::string cmd{CommandLineFromArgv(connect_info_to_args(pipe_path))};
+    STARTUPINFOA si{};
+    si.cb = sizeof(si);
+    PROCESS_INFORMATION pi{};
+    KJ_WIN32(CreateProcessA(nullptr, const_cast<char*>(cmd.c_str()), nullptr, nullptr, TRUE, 0, nullptr, nullptr, &si, &pi), "CreateProcess failed");
+    CloseHandle(pi.hThread); // not needed
+
+    // Duplicate socket for the child (now that we know its PID)
+    WSAPROTOCOL_INFO info{};
+    KJ_WINSOCK(WSADuplicateSocket(fds[0], pi.dwProcessId, &info), "WSADuplicateSocket failed");
+
+    // Send socket to the child via the pipe
+    KJ_WIN32(ConnectNamedPipe(pipe, nullptr) || GetLastError() == ERROR_PIPE_CONNECTED, "ConnectNamedPipe failed");
+    DWORD wr;
+    KJ_WIN32(WriteFile(pipe, &info, sizeof(info), &wr, nullptr) && wr == sizeof(info), "WriteFile(pipe) failed");
+    CloseHandle(pipe);
+
+    return {reinterpret_cast<ProcessId>(pi.hProcess), fds[1]};
+#endif
 }
 
 SocketId StartSpawned(const ConnectInfo& connect_info)
 {
+#ifndef WIN32
     return std::stoi(connect_info);
+#else
+    HANDLE pipe = CreateFileA(connect_info.c_str(), GENERIC_READ, 0, nullptr, OPEN_EXISTING, 0, nullptr);
+    KJ_WIN32(pipe != INVALID_HANDLE_VALUE, "CreateFile(pipe) failed");
+
+    WSAPROTOCOL_INFO info{};
+    DWORD rd;
+    KJ_WIN32(ReadFile(pipe, &info, sizeof(info), &rd, nullptr) && rd == sizeof(info), "ReadFile(pipe) failed");
+    CloseHandle(pipe);
+
+    WSADATA dontcare;
+    KJ_WIN32(WSAStartup(MAKEWORD(2, 2), &dontcare) != 0, "WSAStartup() failed");
+
+    SOCKET socket{WSASocketA(FROM_PROTOCOL_INFO, FROM_PROTOCOL_INFO, FROM_PROTOCOL_INFO, &info, 0, WSA_FLAG_OVERLAPPED | WSA_FLAG_NO_HANDLE_INHERIT)};
+    KJ_WINSOCK(socket, "WSASocket(FROM_PROTOCOL_INFO) failed");
+    return socket;
+#endif
 }
 
 std::array<SocketId, 2> SocketPair()
 {
+#ifdef WIN32
+    SOCKET pair[2];
+    KJ_WINSOCK(kj::_::win32Socketpair(pair));
+#else
     int pair[2];
     KJ_SYSCALL(socketpair(AF_UNIX, SOCK_STREAM, 0, pair));
+#endif
     return {pair[0], pair[1]};
 }
 
@@ -205,11 +317,20 @@ void ExecProcess(const std::vector<std::string>& args)
 
 int WaitProcess(ProcessId pid)
 {
+#ifndef WIN32
     int status;
     if (::waitpid(pid, &status, /*options=*/0) != pid) {
         throw std::system_error(errno, std::system_category(), "waitpid");
     }
     return status;
+#else
+    HANDLE handle{reinterpret_cast<HANDLE>(pid)};
+    DWORD result{WaitForSingleObject(handle, INFINITE)};
+    KJ_WIN32(result != WAIT_OBJECT_0, "WaitForSingleObject(child) failed");
+    KJ_WIN32(GetExitCodeProcess(handle, &result), "GetExitCodeProcess failed");
+    CloseHandle(handle);
+    return result;
+#endif
 }
 
 } // namespace mp

test/mp/test/spawn_tests.cpp

@@ -9,21 +9,25 @@
 #include <chrono>
 #include <compare>
 #include <condition_variable>
-#include <csignal>
 #include <cstdlib>
 #include <mutex>
 #include <string>
-#include <sys/wait.h>
 #include <thread>
 #include <tuple>
-#include <unistd.h>
 #include <utility>
 #include <vector>
 
+#ifndef WIN32
+#include <csignal>
+#include <sys/wait.h>
+#include <unistd.h>
+#endif
+
 namespace mp {
 namespace test {
 namespace {
 
+#ifndef WIN32
 constexpr auto FAILURE_TIMEOUT = std::chrono::seconds{30};
 
 // Poll for child process exit using waitpid(..., WNOHANG) until the child exits
@@ -44,14 +48,19 @@ static bool WaitPidWithTimeout(ProcessId pid, std::chrono::milliseconds timeout,
     }
     return false;
 }
+#endif // !WIN32
 
 } // namespace
 
+#ifndef WIN32
 KJ_TEST("SpawnProcess does not run callback in child")
 {
     // This test is designed to fail deterministically if fd_to_args is invoked
     // in the post-fork child: a mutex held by another parent thread at fork
     // time appears locked forever in the child.
+    //
+    // This test is Unix-only: Windows uses CreateProcess (not fork), so the
+    // inherited-locked-mutex hazard does not apply there.
     std::mutex target_mutex;
     std::mutex control_mutex;
     std::condition_variable control_cv;
@@ -113,5 +122,6 @@ KJ_TEST("SpawnProcess does not run callback in child")
     KJ_EXPECT(exited, "Timeout waiting for child process to exit");
     KJ_EXPECT(WIFEXITED(status) && WEXITSTATUS(status) == 0);
 }
+#endif // !WIN32
 } // namespace test
 } // namespace mp