Add windows support

Add support for running on windows. These changes make the libmultiprocess API
more generic, using stream types instead of file descriptors. All features are
supported, including spawning processes with socket connections to the parent
process. These changes were originally made in
bitcoin/bitcoin#32387

Author: ryanofsky

example/calculator.cpp

@@ -51,14 +51,11 @@ int main(int argc, char** argv)
         std::cout << "Usage: mpcalculator <fd>\n";
         return 1;
     }
-    int fd;
-    if (std::from_chars(argv[1], argv[1] + strlen(argv[1]), fd).ec != std::errc{}) {
-        std::cerr << argv[1] << " is not a number or is larger than an int\n";
-        return 1;
-    }
+    mp::SocketId socket{mp::StartSpawned(argv[1])};
     mp::EventLoop loop("mpcalculator", LogPrint);
     std::unique_ptr<Init> init = std::make_unique<InitImpl>();
-    mp::ServeStream<InitInterface>(loop, fd, *init);
+    mp::Stream stream{loop.m_io_context.lowLevelProvider->wrapSocketFd(socket, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP)};
+    mp::ServeStream<InitInterface>(loop, kj::mv(stream), *init);
     loop.loop();
     return 0;
 }

example/example.cpp

@@ -25,14 +25,14 @@ namespace fs = std::filesystem;
 
 static auto Spawn(mp::EventLoop& loop, const std::string& process_argv0, const std::string& new_exe_name)
 {
-    int pid;
-    const int fd = mp::SpawnProcess(pid, [&](int fd) -> std::vector<std::string> {
+    auto pair{mp::SocketPair()};
+    mp::ProcessId pid{mp::SpawnProcess(pair[0], [&](mp::ConnectInfo info) -> std::vector<std::string> {
         fs::path path = process_argv0;
         path.remove_filename();
         path.append(new_exe_name);
-        return {path.string(), std::to_string(fd)};
-    });
-    return std::make_tuple(mp::ConnectStream<InitInterface>(loop, fd), pid);
+        return {path.string(), std::move(info)};
+    })};
+    return std::make_tuple(mp::ConnectStream<InitInterface>(loop, loop.m_io_context.lowLevelProvider->wrapSocketFd(pair[1])), pid);
 }
 
 static void LogPrint(mp::LogMessage log_data)

example/printer.cpp

@@ -44,14 +44,11 @@ int main(int argc, char** argv)
         std::cout << "Usage: mpprinter <fd>\n";
         return 1;
     }
-    int fd;
-    if (std::from_chars(argv[1], argv[1] + strlen(argv[1]), fd).ec != std::errc{}) {
-        std::cerr << argv[1] << " is not a number or is larger than an int\n";
-        return 1;
-    }
+    mp::SocketId socket{mp::StartSpawned(argv[1])};
     mp::EventLoop loop("mpprinter", LogPrint);
     std::unique_ptr<Init> init = std::make_unique<InitImpl>();
-    mp::ServeStream<InitInterface>(loop, fd, *init);
+    mp::Stream stream{loop.m_io_context.lowLevelProvider->wrapSocketFd(socket, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP)};
+    mp::ServeStream<InitInterface>(loop, std::move(stream), *init);
     loop.loop();
     return 0;
 }

include/mp/proxy-io.h

@@ -185,6 +185,17 @@ class Logger
 
 std::string LongThreadName(const char* exe_name);
 
+using Stream = kj::Own<kj::AsyncIoStream>;
+
+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");
+}
+
 //! Event loop implementation.
 //!
 //! Cap'n Proto threading model is very simple: all I/O operations are
@@ -283,11 +294,12 @@ class EventLoop
     //! Callback functions to run on async thread.
     std::optional<CleanupList> m_async_fns MP_GUARDED_BY(m_mutex);
 
-    //! Pipe read handle used to wake up the event loop thread.
-    int m_wait_fd = -1;
+    //! Socket pair used to post and wait for wakeups to the event loop thread.
+    kj::Own<kj::AsyncIoStream> m_wait_stream;
+    kj::Own<kj::AsyncIoStream> m_post_stream;
 
-    //! Pipe write handle used to wake up the event loop thread.
-    int m_post_fd = -1;
+    //! Synchronous writer used to write to m_post_stream.
+    kj::Own<kj::OutputStream> m_post_writer;
 
     //! Number of clients holding references to ProxyServerBase objects that
     //! reference this event loop.
@@ -679,13 +691,11 @@ struct ThreadContext
 //! over the stream. Also create a new Connection object embedded in the
 //! client that is freed when the client is closed.
 template <typename InitInterface>
-std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, int fd)
+std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, kj::Own<kj::AsyncIoStream> stream)
 {
     typename InitInterface::Client init_client(nullptr);
     std::unique_ptr<Connection> connection;
     loop.sync([&] {
-        auto stream =
-            loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP);
         connection = std::make_unique<Connection>(loop, kj::mv(stream));
         init_client = connection->m_rpc_system->bootstrap(ServerVatId().vat_id).castAs<InitInterface>();
         Connection* connection_ptr = connection.get();
@@ -735,10 +745,9 @@ void _Listen(EventLoop& loop, kj::Own<kj::ConnectionReceiver>&& listener, InitIm
 //! Given stream file descriptor and an init object, handle requests on the
 //! stream by calling methods on the Init object.
 template <typename InitInterface, typename InitImpl>
-void ServeStream(EventLoop& loop, int fd, InitImpl& init)
+void ServeStream(EventLoop& loop, kj::Own<kj::AsyncIoStream> stream, InitImpl& init)
 {
-    _Serve<InitInterface>(
-        loop, loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP), init);
+    _Serve<InitInterface>(loop, kj::mv(stream), init);
 }
 
 //! Given listening socket file descriptor and an init object, handle incoming

include/mp/util.h

@@ -19,6 +19,10 @@
 #include <variant>
 #include <vector>
 
+#ifdef WIN32
+#include <winsock2.h>
+#endif
+
 namespace mp {
 
 //! Generic utility functions used by capnp code.
@@ -216,22 +220,44 @@ std::string ThreadName(const char* exe_name);
 //! errors in python unit tests.
 std::string LogEscape(const kj::StringTree& string, size_t max_size);
 
+#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.  On unix this is
+//! just the inherited int file descriptor formatted as a string. On windows,
+//! this is a path to a named path pipe the parent process will write
+//! WSADuplicateSocket info to.
+using ConnectInfo = std::string;
+
 //! Callback type used by SpawnProcess below.
-using FdToArgsFn = std::function<std::vector<std::string>(int fd)>;
+using ConnectInfoToArgsFn = std::function<std::vector<std::string>(const ConnectInfo&)>;
+
+//! Create a socket pair that can be used to communicate within a process or
+//! between parent and child processes.
+std::array<SocketId, 2> SocketPair();
+
+//! Spawn a new process that communicates with the current process over provided
+//! socket argument. Calls connect_info_to_args callback with a connection
+//! string that needs to be passed to the child process, and executes the
+//! argv command line it returns. Returns child process id.
+ProcessId SpawnProcess(SocketId socket, ConnectInfoToArgsFn&& connect_info_to_args);
 
-//! Spawn a new process that communicates with the current process over a socket
-//! pair. Returns pid through an output argument, and file descriptor for the
-//! local side of the socket. Invokes fd_to_args callback with the remote file
-//! descriptor number which returns the command line arguments that should be
-//! used to execute the process, and which should have the remote file
-//! descriptor embedded in whatever format the child process expects.
-int SpawnProcess(int& pid, FdToArgsFn&& fd_to_args);
+//! Initialize spawned child process using the ConnectInfo string passed to it,
+//! returning a socket id for communicating with the parent process.
+SocketId StartSpawned(const ConnectInfo& connect_info);
 
 //! Call execvp with vector args.
 void ExecProcess(const std::vector<std::string>& args);
 
 //! Wait for a process to exit and return its exit code.
-int WaitProcess(int pid);
+int WaitProcess(ProcessId pid);
 
 inline char* CharCast(char* c) { return c; }
 inline char* CharCast(unsigned char* c) { return (char*)c; }

src/mp/proxy.cpp

@@ -30,12 +30,15 @@
 #include <optional>
 #include <stdexcept>
 #include <string>
-#include <sys/socket.h>
 #include <thread>
 #include <tuple>
-#include <unistd.h>
 #include <utility>
 
+#ifndef WIN32
+#include <sys/socket.h>
+#include <unistd.h>
+#endif
+
 namespace mp {
 
 thread_local ThreadContext g_thread_context;
@@ -66,10 +69,9 @@ void EventLoopRef::reset(bool relock) MP_NO_TSA
         loop->m_num_clients -= 1;
         if (loop->done()) {
             loop->m_cv.notify_all();
-            int post_fd{loop->m_post_fd};
             loop_lock->unlock();
             char buffer = 0;
-            KJ_SYSCALL(write(post_fd, &buffer, 1)); // NOLINT(bugprone-suspicious-semicolon)
+            loop->m_post_writer->write(&buffer, 1);
             // By default, do not try to relock `loop_lock` after writing,
             // because the event loop could wake up and destroy itself and the
             // mutex might no longer exist.
@@ -96,6 +98,20 @@ Connection::~Connection()
     // after the calls finish.
     m_rpc_system.reset();
 
+    // shutdownWrite is needed on Windows so pending data in the m_stream socket
+    // will be sent instead of discarded when m_stream is destroyed. On unix,
+    // this doesn't seem to be needed because data is sent more reliably.
+    //
+    // Sending pending data is important if the connection is a socketpair
+    // because when one side of the socketpair is closed, the other side doesn't
+    // seem to receive any onDisconnect event. So it is important for the other
+    // side to instead receive Cap'n Proto "release" messages (see `struct
+    // Release` in capnp/rpc.capnp) from local Client objects being being
+    // destroyed so the remote side can free resources and shut down cleanly.
+    // Without this call, Server objects corresponding to the Client objects on
+    // the other side of the connection are not freed by Cap'n Proto.
+    m_stream->shutdownWrite();
+
     // ProxyClient cleanup handlers are in sync list, and ProxyServer cleanup
     // handlers are in the async list.
     //
@@ -192,17 +208,59 @@ void EventLoop::addAsyncCleanup(std::function<void()> fn)
     startAsyncThread();
 }
 
+#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 analagous 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
+
 EventLoop::EventLoop(const char* exe_name, LogOptions log_opts, void* context)
     : m_exe_name(exe_name),
       m_io_context(kj::setupAsyncIo()),
       m_task_set(new kj::TaskSet(m_error_handler)),
       m_log_opts(std::move(log_opts)),
       m_context(context)
 {
-    int fds[2];
-    KJ_SYSCALL(socketpair(AF_UNIX, SOCK_STREAM, 0, fds));
-    m_wait_fd = fds[0];
-    m_post_fd = fds[1];
+    auto pipe = m_io_context.provider->newTwoWayPipe();
+    m_wait_stream = kj::mv(pipe.ends[0]);
+    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.");
+    }
 }
 
 EventLoop::~EventLoop()
@@ -211,8 +269,8 @@ EventLoop::~EventLoop()
     const Lock lock(m_mutex);
     KJ_ASSERT(m_post_fn == nullptr);
     KJ_ASSERT(!m_async_fns);
-    KJ_ASSERT(m_wait_fd == -1);
-    KJ_ASSERT(m_post_fd == -1);
+    KJ_ASSERT(!m_wait_stream);
+    KJ_ASSERT(!m_post_stream);
     KJ_ASSERT(m_num_clients == 0);
 
     // Spin event loop. wait for any promises triggered by RPC shutdown.
@@ -232,9 +290,7 @@ void EventLoop::loop()
         m_async_fns.emplace();
     }
 
-    kj::Own<kj::AsyncIoStream> wait_stream{
-        m_io_context.lowLevelProvider->wrapSocketFd(m_wait_fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP)};
-    int post_fd{m_post_fd};
+    kj::Own<kj::AsyncIoStream>& wait_stream{m_wait_stream};
     char buffer = 0;
     for (;;) {
         const size_t read_bytes = wait_stream->read(&buffer, 0, 1).wait(m_io_context.waitScope);
@@ -246,7 +302,7 @@ void EventLoop::loop()
             m_cv.notify_all();
         } else if (done()) {
             // Intentionally do not break if m_post_fn was set, even if done()
-            // would return true, to ensure that the EventLoopRef write(post_fd)
+            // would return true, to ensure that the EventLoopRef write(post_stream)
             // call always succeeds and the loop does not exit between the time
             // that the done condition is set and the write call is made.
             break;
@@ -256,10 +312,9 @@ void EventLoop::loop()
     m_task_set.reset();
     MP_LOG(*this, Log::Info) << "EventLoop::loop bye.";
     wait_stream = nullptr;
-    KJ_SYSCALL(::close(post_fd));
     const Lock lock(m_mutex);
-    m_wait_fd = -1;
-    m_post_fd = -1;
+    m_wait_stream = nullptr;
+    m_post_stream = nullptr;
     m_async_fns.reset();
     m_cv.notify_all();
 }
@@ -274,10 +329,9 @@ void EventLoop::post(kj::Function<void()> fn)
     EventLoopRef ref(*this, &lock);
     m_cv.wait(lock.m_lock, [this]() MP_REQUIRES(m_mutex) { return m_post_fn == nullptr; });
     m_post_fn = &fn;
-    int post_fd{m_post_fd};
     Unlock(lock, [&] {
         char buffer = 0;
-        KJ_SYSCALL(write(post_fd, &buffer, 1));
+        m_post_writer->write(&buffer, 1);
     });
     m_cv.wait(lock.m_lock, [this, &fn]() MP_REQUIRES(m_mutex) { return m_post_fn != &fn; });
 }