Add nonunix platform support

doc/versions.md

@@ -7,8 +7,12 @@ Library versions are tracked with simple
 Versioning policy is described in the [version.h](../include/mp/version.h)
 include.
 
-## v11
+## v12
 - Current unstable version.
+- Adds support for nonunix platforms, making API changes that are not backwards compatible.
+
+## [v11.0](https://github.com/bitcoin-core/libmultiprocess/commits/v11.0)
+- Improves debug output if EventLoop::post callback fails.
 
 ## [v10.0](https://github.com/bitcoin-core/libmultiprocess/commits/v10.0)
 - Increases spawn test timeout to avoid spurious failures.

example/calculator.cpp

@@ -6,8 +6,7 @@
 #include <init.capnp.h>
 #include <init.capnp.proxy.h> // NOLINT(misc-include-cleaner) // IWYU pragma: keep
 
-#include <charconv>
-#include <cstring>
+#include <cstring> // IWYU pragma: keep
 #include <fstream>
 #include <functional>
 #include <iostream>
@@ -16,9 +15,9 @@
 #include <kj/memory.h>
 #include <memory>
 #include <mp/proxy-io.h>
+#include <mp/util.h>
 #include <stdexcept>
 #include <string>
-#include <system_error>
 #include <utility>
 
 class CalculatorImpl : public Calculator
@@ -51,14 +50,10 @@ 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::ServeStream<InitInterface>(loop, socket, *init);
     loop.loop();
     return 0;
 }

example/example.cpp

@@ -19,20 +19,20 @@
 #include <string>
 #include <thread>
 #include <tuple>
+#include <utility>
 #include <vector>
 
 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> {
+    const auto [pid, socket] = mp::SpawnProcess([&](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 {path.string(), std::move(info)};
     });
-    return std::make_tuple(mp::ConnectStream<InitInterface>(loop, fd), pid);
+    return std::make_tuple(mp::ConnectStream<InitInterface>(loop, socket), pid);
 }
 
 static void LogPrint(mp::LogMessage log_data)

example/printer.cpp

@@ -7,18 +7,17 @@
 #include <init.capnp.h>
 #include <init.capnp.proxy.h> // NOLINT(misc-include-cleaner) // IWYU pragma: keep
 
-#include <charconv>
-#include <cstring>
+#include <cstring> // IWYU pragma: keep
 #include <fstream>
 #include <iostream>
 #include <kj/async.h>
 #include <kj/common.h>
 #include <kj/memory.h>
 #include <memory>
 #include <mp/proxy-io.h>
+#include <mp/util.h>
 #include <stdexcept>
 #include <string>
-#include <system_error>
 
 class PrinterImpl : public Printer
 {
@@ -44,14 +43,10 @@ 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::ServeStream<InitInterface>(loop, socket, *init);
     loop.loop();
     return 0;
 }

include/mp/util.h

@@ -249,25 +249,33 @@ std::string ThreadName(const char* exe_name);
 //! errors in python unit tests.
 std::string LogEscape(const kj::StringTree& string, size_t max_size);
 
+using ProcessId = int;
+using SocketId = int;
+constexpr SocketId SocketError{-1};
+
+//! 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.
+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&)>;
 
 //! 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.
-//! The fd_to_args callback is invoked in the parent process before fork().
-//! It must not rely on child pid/state, and must return the command line
-//! arguments that should be used to execute the process. Embed the remote file
-//! descriptor number in whatever format the child process expects.
-int SpawnProcess(int& pid, FdToArgsFn&& fd_to_args);
-
-//! Call execvp with vector args.
-//! Not safe to call in a post-fork child of a multi-threaded process.
-//! Currently only used by mpgen at build time.
-void ExecProcess(const std::vector<std::string>& args);
+//! pair. 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 and socket id.
+std::tuple<ProcessId, SocketId> SpawnProcess(ConnectInfoToArgsFn&& connect_info_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);
+
+//! Start a process and return its process id. Caller should call WaitProcess
+//! on the returned id.
+ProcessId 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; }

include/mp/version.h

@@ -24,7 +24,7 @@
 //! pointing at the prior merge commit. The /doc/versions.md file should also be
 //! updated, noting any significant or incompatible changes made since the
 //! previous version.
-#define MP_MAJOR_VERSION 11
+#define MP_MAJOR_VERSION 12
 
 //! Minor version number. Should be incremented in stable branches after
 //! backporting changes. The /doc/versions.md file should also be updated to

src/mp/gen.cpp

@@ -8,7 +8,6 @@
 #include <algorithm>
 #include <capnp/schema.h>
 #include <capnp/schema-parser.h>
-#include <cerrno>
 #include <cstdint>
 #include <cstdio>
 #include <cstdlib>
@@ -26,8 +25,6 @@
 #include <sstream>
 #include <stdexcept>
 #include <string>
-#include <system_error>
-#include <unistd.h>
 #include <utility>
 #include <vector>
 
@@ -180,14 +177,7 @@ static void Generate(kj::StringPtr src_prefix,
     }
     args.emplace_back("--output=" capnp_PREFIX "/bin/capnpc-c++");
     args.emplace_back(src_file);
-    const int pid = fork();
-    if (pid == -1) {
-        throw std::system_error(errno, std::system_category(), "fork");
-    }
-    if (!pid) {
-        mp::ExecProcess(args);
-    }
-    const int status = mp::WaitProcess(pid);
+    const int status = mp::WaitProcess(mp::ExecProcess(args));
     if (status) {
         throw std::runtime_error("Invoking " capnp_PREFIX "/bin/capnp failed");
     }

src/mp/proxy.cpp

@@ -204,7 +204,7 @@ EventLoop::EventLoop(const char* exe_name, LogOptions log_opts, void* context)
       m_log_opts(std::move(log_opts)),
       m_context(context)
 {
-    int fds[2];
+    SocketId fds[2];
     KJ_SYSCALL(socketpair(AF_UNIX, SOCK_STREAM, 0, fds));
     m_wait_fd = fds[0];
     m_post_fd = fds[1];

src/mp/util.cpp

@@ -10,6 +10,7 @@
 #include <filesystem>
 #include <iostream>
 #include <kj/common.h>
+#include <kj/debug.h>
 #include <kj/string-tree.h>
 #include <pthread.h>
 #include <sstream>
@@ -116,9 +117,9 @@ std::string LogEscape(const kj::StringTree& string, size_t max_size)
     return result;
 }
 
-int SpawnProcess(int& pid, FdToArgsFn&& fd_to_args)
+std::tuple<ProcessId, SocketId> SpawnProcess(ConnectInfoToArgsFn&& connect_info_to_args)
 {
-    int fds[2];
+    SocketId fds[2];
     if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds) != 0) {
         throw std::system_error(errno, std::system_category(), "socketpair");
     }
@@ -129,10 +130,10 @@ int SpawnProcess(int& pid, FdToArgsFn&& fd_to_args)
     // locks at fork time. In that case, running code that allocates memory or
     // takes locks in the child between fork() and exec() can deadlock
     // indefinitely. Precomputing arguments in the parent avoids this.
-    const std::vector<std::string> args{fd_to_args(fds[0])};
+    const std::vector<std::string> args{connect_info_to_args(std::to_string(fds[0]))};
     const std::vector<char*> argv{MakeArgv(args)};
 
-    pid = fork();
+    ProcessId pid = fork();
     if (pid == -1) {
         throw std::system_error(errno, std::system_category(), "fork");
     }
@@ -168,22 +169,31 @@ int SpawnProcess(int& pid, FdToArgsFn&& fd_to_args)
         perror("execvp failed");
         _exit(127);
     }
-    return fds[1];
+    return {pid, fds[1]};
 }
 
-void ExecProcess(const std::vector<std::string>& args)
+SocketId StartSpawned(const ConnectInfo& connect_info)
+{
+    return std::stoi(connect_info);
+}
+
+ProcessId ExecProcess(const std::vector<std::string>& args)
 {
     const std::vector<char*> argv{MakeArgv(args)};
+    ProcessId pid;
+    KJ_SYSCALL(pid = fork());
+    if (pid) return pid;
     if (execvp(argv[0], argv.data()) != 0) {
         perror("execvp failed");
         if (errno == ENOENT && !args.empty()) {
             std::cerr << "Missing executable: " << fs::weakly_canonical(args.front()) << '\n';
         }
         _exit(1);
     }
+    KJ_UNREACHABLE;
 }
 
-int WaitProcess(int pid)
+int WaitProcess(ProcessId pid)
 {
     int status;
     if (::waitpid(pid, &status, /*options=*/0) != pid) {

test/mp/test/spawn_tests.cpp

@@ -15,17 +15,21 @@
 #include <string>
 #include <sys/wait.h>
 #include <thread>
+#include <tuple>
 #include <unistd.h>
+#include <utility>
 #include <vector>
 
+namespace mp {
+namespace test {
 namespace {
 
 constexpr auto FAILURE_TIMEOUT = std::chrono::seconds{30};
 
 // Poll for child process exit using waitpid(..., WNOHANG) until the child exits
 // or timeout expires. Returns true if the child exited and status_out was set.
 // Returns false on timeout or error.
-static bool WaitPidWithTimeout(int pid, std::chrono::milliseconds timeout, int& status_out)
+static bool WaitPidWithTimeout(ProcessId pid, std::chrono::milliseconds timeout, int& status_out)
 {
     const auto deadline = std::chrono::steady_clock::now() + timeout;
     while (std::chrono::steady_clock::now() < deadline) {
@@ -86,14 +90,13 @@ KJ_TEST("SpawnProcess does not run callback in child")
         control_cv.notify_one();
     });
 
-    int pid{-1};
-    const int fd{mp::SpawnProcess(pid, [&](int child_fd) -> std::vector<std::string> {
+    const auto [pid, socket]{SpawnProcess([&](ConnectInfo connect_info) -> std::vector<std::string> {
         // If this callback runs in the post-fork child, target_mutex appears
         // locked forever (the owning thread does not exist), so this deadlocks.
         std::lock_guard<std::mutex> g(target_mutex);
-        return {"true", std::to_string(child_fd)};
+        return {"true", std::move(connect_info)};
     })};
-    ::close(fd);
+    ::close(socket);
 
     int status{0};
     // Give the child some time to exit. If it does not, terminate it and
@@ -110,3 +113,5 @@ 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);
 }
+} // namespace test
+} // namespace mp