multiprocess: Add Ipc interface implementation

10 files changed, 371 insertions, 0 deletions

diff --git a/src/ipc/capnp/.gitignore b/src/ipc/capnp/.gitignore
new file mode 100644
index 0000000000..036df1430c
--- /dev/null
+++ b/src/ipc/capnp/.gitignore
@@ -0,0 +1,2 @@
+# capnp generated files
+*.capnp.*
diff --git a/src/ipc/capnp/init-types.h b/src/ipc/capnp/init-types.h
new file mode 100644
index 0000000000..c84b94802a
--- /dev/null
+++ b/src/ipc/capnp/init-types.h
@@ -0,0 +1,7 @@
+// Copyright (c) 2021 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#ifndef BITCOIN_IPC_CAPNP_INIT_TYPES_H
+#define BITCOIN_IPC_CAPNP_INIT_TYPES_H
+#endif // BITCOIN_IPC_CAPNP_INIT_TYPES_H
diff --git a/src/ipc/capnp/init.capnp b/src/ipc/capnp/init.capnp
new file mode 100644
index 0000000000..be6eecb4b9
--- /dev/null
+++ b/src/ipc/capnp/init.capnp
@@ -0,0 +1,16 @@
+# Copyright (c) 2021 The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+@0xf2c5cfa319406aa6;
+
+using Cxx = import "/capnp/c++.capnp";
+$Cxx.namespace("ipc::capnp::messages");
+
+using Proxy = import "/mp/proxy.capnp";
+$Proxy.include("interfaces/init.h");
+$Proxy.includeTypes("ipc/capnp/init-types.h");
+
+interface Init $Proxy.wrap("interfaces::Init") {
+    construct @0 (threadMap: Proxy.ThreadMap) -> (threadMap :Proxy.ThreadMap);
+}
diff --git a/src/ipc/capnp/protocol.cpp b/src/ipc/capnp/protocol.cpp
new file mode 100644
index 0000000000..74c66c899a
--- /dev/null
+++ b/src/ipc/capnp/protocol.cpp
@@ -0,0 +1,90 @@
+// Copyright (c) 2021 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#include <interfaces/init.h>
+#include <ipc/capnp/init.capnp.h>
+#include <ipc/capnp/init.capnp.proxy.h>
+#include <ipc/capnp/protocol.h>
+#include <ipc/exception.h>
+#include <ipc/protocol.h>
+#include <kj/async.h>
+#include <logging.h>
+#include <mp/proxy-io.h>
+#include <mp/proxy-types.h>
+#include <mp/util.h>
+#include <util/threadnames.h>
+
+#include <assert.h>
+#include <errno.h>
+#include <future>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <thread>
+
+namespace ipc {
+namespace capnp {
+namespace {
+void IpcLogFn(bool raise, std::string message)
+{
+    LogPrint(BCLog::IPC, "%s\n", message);
+    if (raise) throw Exception(message);
+}
+
+class CapnpProtocol : public Protocol
+{
+public:
+    ~CapnpProtocol() noexcept(true)
+    {
+        if (m_loop) {
+            std::unique_lock<std::mutex> lock(m_loop->m_mutex);
+            m_loop->removeClient(lock);
+        }
+        if (m_loop_thread.joinable()) m_loop_thread.join();
+        assert(!m_loop);
+    };
+    std::unique_ptr<interfaces::Init> connect(int fd, const char* exe_name) override
+    {
+        startLoop(exe_name);
+        return mp::ConnectStream<messages::Init>(*m_loop, fd);
+    }
+    void serve(int fd, const char* exe_name, interfaces::Init& init) override
+    {
+        assert(!m_loop);
+        mp::g_thread_context.thread_name = mp::ThreadName(exe_name);
+        m_loop.emplace(exe_name, &IpcLogFn, nullptr);
+        mp::ServeStream<messages::Init>(*m_loop, fd, init);
+        m_loop->loop();
+        m_loop.reset();
+    }
+    void addCleanup(std::type_index type, void* iface, std::function<void()> cleanup) override
+    {
+        mp::ProxyTypeRegister::types().at(type)(iface).cleanup.emplace_back(std::move(cleanup));
+    }
+    void startLoop(const char* exe_name)
+    {
+        if (m_loop) return;
+        std::promise<void> promise;
+        m_loop_thread = std::thread([&] {
+            util::ThreadRename("capnp-loop");
+            m_loop.emplace(exe_name, &IpcLogFn, nullptr);
+            {
+                std::unique_lock<std::mutex> lock(m_loop->m_mutex);
+                m_loop->addClient(lock);
+            }
+            promise.set_value();
+            m_loop->loop();
+            m_loop.reset();
+        });
+        promise.get_future().wait();
+    }
+    std::thread m_loop_thread;
+    std::optional<mp::EventLoop> m_loop;
+};
+} // namespace
+
+std::unique_ptr<Protocol> MakeCapnpProtocol() { return std::make_unique<CapnpProtocol>(); }
+} // namespace capnp
+} // namespace ipc
diff --git a/src/ipc/capnp/protocol.h b/src/ipc/capnp/protocol.h
new file mode 100644
index 0000000000..eb057949d2
--- /dev/null
+++ b/src/ipc/capnp/protocol.h
@@ -0,0 +1,17 @@
+// Copyright (c) 2021 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#ifndef BITCOIN_IPC_CAPNP_PROTOCOL_H
+#define BITCOIN_IPC_CAPNP_PROTOCOL_H
+
+#include <memory>
+
+namespace ipc {
+class Protocol;
+namespace capnp {
+std::unique_ptr<Protocol> MakeCapnpProtocol();
+} // namespace capnp
+} // namespace ipc
+
+#endif // BITCOIN_IPC_CAPNP_PROTOCOL_H
diff --git a/src/ipc/exception.h b/src/ipc/exception.h
new file mode 100644
index 0000000000..53dee8124a
--- /dev/null
+++ b/src/ipc/exception.h
@@ -0,0 +1,20 @@
+// Copyright (c) 2021 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#ifndef BITCOIN_IPC_EXCEPTION_H
+#define BITCOIN_IPC_EXCEPTION_H
+
+#include <stdexcept>
+
+namespace ipc {
+//! Exception class thrown when a call to remote method fails due to an IPC
+//! error, like a socket getting disconnected.
+class Exception : public std::runtime_error
+{
+public:
+    using std::runtime_error::runtime_error;
+};
+} // namespace ipc
+
+#endif // BITCOIN_IPC_EXCEPTION_H
diff --git a/src/ipc/interfaces.cpp b/src/ipc/interfaces.cpp
new file mode 100644
index 0000000000..ad4b78ed81
--- /dev/null
+++ b/src/ipc/interfaces.cpp
@@ -0,0 +1,77 @@
+// Copyright (c) 2021 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#include <fs.h>
+#include <interfaces/init.h>
+#include <interfaces/ipc.h>
+#include <ipc/capnp/protocol.h>
+#include <ipc/process.h>
+#include <ipc/protocol.h>
+#include <logging.h>
+#include <tinyformat.h>
+#include <util/system.h>
+
+#include <functional>
+#include <memory>
+#include <stdexcept>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <string>
+#include <unistd.h>
+#include <utility>
+#include <vector>
+
+namespace ipc {
+namespace {
+class IpcImpl : public interfaces::Ipc
+{
+public:
+    IpcImpl(const char* exe_name, const char* process_argv0, interfaces::Init& init)
+        : m_exe_name(exe_name), m_process_argv0(process_argv0), m_init(init),
+          m_protocol(ipc::capnp::MakeCapnpProtocol()), m_process(ipc::MakeProcess())
+    {
+    }
+    std::unique_ptr<interfaces::Init> spawnProcess(const char* new_exe_name) override
+    {
+        int pid;
+        int fd = m_process->spawn(new_exe_name, m_process_argv0, pid);
+        LogPrint(::BCLog::IPC, "Process %s pid %i launched\n", new_exe_name, pid);
+        auto init = m_protocol->connect(fd, m_exe_name);
+        Ipc::addCleanup(*init, [this, new_exe_name, pid] {
+            int status = m_process->waitSpawned(pid);
+            LogPrint(::BCLog::IPC, "Process %s pid %i exited with status %i\n", new_exe_name, pid, status);
+        });
+        return init;
+    }
+    bool startSpawnedProcess(int argc, char* argv[], int& exit_status) override
+    {
+        exit_status = EXIT_FAILURE;
+        int32_t fd = -1;
+        if (!m_process->checkSpawned(argc, argv, fd)) {
+            return false;
+        }
+        m_protocol->serve(fd, m_exe_name, m_init);
+        exit_status = EXIT_SUCCESS;
+        return true;
+    }
+    void addCleanup(std::type_index type, void* iface, std::function<void()> cleanup) override
+    {
+        m_protocol->addCleanup(type, iface, std::move(cleanup));
+    }
+    const char* m_exe_name;
+    const char* m_process_argv0;
+    interfaces::Init& m_init;
+    std::unique_ptr<Protocol> m_protocol;
+    std::unique_ptr<Process> m_process;
+};
+} // namespace
+} // namespace ipc
+
+namespace interfaces {
+std::unique_ptr<Ipc> MakeIpc(const char* exe_name, const char* process_argv0, Init& init)
+{
+    return std::make_unique<ipc::IpcImpl>(exe_name, process_argv0, init);
+}
+} // namespace interfaces
diff --git a/src/ipc/process.cpp b/src/ipc/process.cpp
new file mode 100644
index 0000000000..43ed1f1bae
--- /dev/null
+++ b/src/ipc/process.cpp
@@ -0,0 +1,61 @@
+// Copyright (c) 2021 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#include <fs.h>
+#include <ipc/process.h>
+#include <ipc/protocol.h>
+#include <mp/util.h>
+#include <tinyformat.h>
+#include <util/strencodings.h>
+
+#include <cstdint>
+#include <exception>
+#include <iostream>
+#include <stdexcept>
+#include <stdlib.h>
+#include <string.h>
+#include <system_error>
+#include <unistd.h>
+#include <utility>
+#include <vector>
+
+namespace ipc {
+namespace {
+class ProcessImpl : public Process
+{
+public:
+    int spawn(const std::string& new_exe_name, const fs::path& argv0_path, int& pid) override
+    {
+        return mp::SpawnProcess(pid, [&](int fd) {
+            fs::path path = argv0_path;
+            path.remove_filename();
+            path.append(new_exe_name);
+            return std::vector<std::string>{path.string(), "-ipcfd", strprintf("%i", fd)};
+        });
+    }
+    int waitSpawned(int pid) override { return mp::WaitProcess(pid); }
+    bool checkSpawned(int argc, char* argv[], int& fd) override
+    {
+        // If this process was not started with a single -ipcfd argument, it is
+        // not a process spawned by the spawn() call above, so return false and
+        // do not try to serve requests.
+        if (argc != 3 || strcmp(argv[1], "-ipcfd") != 0) {
+            return false;
+        }
+        // If a single -ipcfd argument was provided, return true and get the
+        // file descriptor so Protocol::serve() can be called to handle
+        // requests from the parent process. The -ipcfd argument is not valid
+        // in combination with other arguments because the parent process
+        // should be able to control the child process through the IPC protocol
+        // without passing information out of band.
+        if (!ParseInt32(argv[2], &fd)) {
+            throw std::runtime_error(strprintf("Invalid -ipcfd number '%s'", argv[2]));
+        }
+        return true;
+    }
+};
+} // namespace
+
+std::unique_ptr<Process> MakeProcess() { return std::make_unique<ProcessImpl>(); }
+} // namespace ipc
diff --git a/src/ipc/process.h b/src/ipc/process.h
new file mode 100644
index 0000000000..4bb2930d9c
--- /dev/null
+++ b/src/ipc/process.h
@@ -0,0 +1,42 @@
+// Copyright (c) 2021 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#ifndef BITCOIN_IPC_PROCESS_H
+#define BITCOIN_IPC_PROCESS_H
+
+#include <memory>
+#include <string>
+
+namespace ipc {
+class Protocol;
+
+//! IPC process interface for spawning bitcoin processes and serving requests
+//! in processes that have been spawned.
+//!
+//! There will be different implementations of this interface depending on the
+//! platform (e.g. unix, windows).
+class Process
+{
+public:
+    virtual ~Process() = default;
+
+    //! Spawn process and return socket file descriptor for communicating with
+    //! it.
+    virtual int spawn(const std::string& new_exe_name, const fs::path& argv0_path, int& pid) = 0;
+
+    //! Wait for spawned process to exit and return its exit code.
+    virtual int waitSpawned(int pid) = 0;
+
+    //! Parse command line and determine if current process is a spawned child
+    //! process. If so, return true and a file descriptor for communicating
+    //! with the parent process.
+    virtual bool checkSpawned(int argc, char* argv[], int& fd) = 0;
+};
+
+//! Constructor for Process interface. Implementation will vary depending on
+//! the platform (unix or windows).
+std::unique_ptr<Process> MakeProcess();
+} // namespace ipc
+
+#endif // BITCOIN_IPC_PROCESS_H
diff --git a/src/ipc/protocol.h b/src/ipc/protocol.h
new file mode 100644
index 0000000000..af955b0007
--- /dev/null
+++ b/src/ipc/protocol.h
@@ -0,0 +1,39 @@
+// Copyright (c) 2021 The Bitcoin Core developers
+// Distributed under the MIT software license, see the accompanying
+// file COPYING or http://www.opensource.org/licenses/mit-license.php.
+
+#ifndef BITCOIN_IPC_PROTOCOL_H
+#define BITCOIN_IPC_PROTOCOL_H
+
+#include <interfaces/init.h>
+
+#include <functional>
+#include <memory>
+#include <typeindex>
+
+namespace ipc {
+//! IPC protocol interface for calling IPC methods over sockets.
+//!
+//! There may be different implementations of this interface for different IPC
+//! protocols (e.g. Cap'n Proto, gRPC, JSON-RPC, or custom protocols).
+class Protocol
+{
+public:
+    virtual ~Protocol() = default;
+
+    //! Return Init interface that forwards requests over given socket descriptor.
+    //! Socket communication is handled on a background thread.
+    virtual std::unique_ptr<interfaces::Init> connect(int fd, const char* exe_name) = 0;
+
+    //! Handle requests on provided socket descriptor, forwarding them to the
+    //! provided Init interface. Socket communication is handled on the
+    //! current thread, and this call blocks until the socket is closed.
+    virtual void serve(int fd, const char* exe_name, interfaces::Init& init) = 0;
+
+    //! Add cleanup callback to interface that will run when the interface is
+    //! deleted.
+    virtual void addCleanup(std::type_index type, void* iface, std::function<void()> cleanup) = 0;
+};
+} // namespace ipc
+
+#endif // BITCOIN_IPC_PROTOCOL_H