From 3859805f05e6fb642140c89e6d7378203e04df2c Mon Sep 17 00:00:00 2001 From: Ryan Ofsky Date: Wed, 30 Apr 2025 08:39:29 -0400 Subject: [PATCH] ipc, refactor: Add SocketId type alias and use it Use SocketId type instead of int to represent socket ids to be compatible with an upcoming version of libmultiprocess which adds windows support. --- src/ipc/capnp/protocol.cpp | 10 +++++----- src/ipc/interfaces.cpp | 12 ++++++------ src/ipc/process.cpp | 22 +++++++++++----------- src/ipc/process.h | 17 ++++++++--------- src/ipc/protocol.h | 11 ++++++----- src/ipc/test/ipc_tests.cpp | 8 ++++---- src/ipc/util.h | 2 ++ 7 files changed, 42 insertions(+), 40 deletions(-) diff --git a/src/ipc/capnp/protocol.cpp b/src/ipc/capnp/protocol.cpp index ca183603e73..0790986d90b 100644 --- a/src/ipc/capnp/protocol.cpp +++ b/src/ipc/capnp/protocol.cpp @@ -78,12 +78,12 @@ public: if (m_loop_thread.joinable()) m_loop_thread.join(); assert(!m_loop); }; - std::unique_ptr connect(int fd) override + std::unique_ptr connect(mp::SocketId socket) override { startLoop(); - return mp::ConnectStream(*m_loop, fd); + return mp::ConnectStream(*m_loop, socket); } - void listen(int listen_fd, interfaces::Init& init) override + void listen(mp::SocketId listen_fd, interfaces::Init& init) override { startLoop(); if (::listen(listen_fd, /*backlog=*/5) != 0) { @@ -91,7 +91,7 @@ public: } mp::ListenConnections(*m_loop, listen_fd, init); } - void serve(int fd, interfaces::Init& init, const std::function& ready_fn = {}) override + void serve(mp::SocketId socket, interfaces::Init& init, const std::function& ready_fn = {}) override { assert(!m_loop); mp::g_thread_context.thread_name = mp::ThreadName(m_exe_name); @@ -101,7 +101,7 @@ public: }; m_loop.emplace(m_exe_name, std::move(opts), &m_context); if (ready_fn) ready_fn(); - mp::ServeStream(*m_loop, fd, init); + mp::ServeStream(*m_loop, socket, init); m_parent_connection = &m_loop->m_incoming_connections.back(); m_loop->loop(); m_loop.reset(); diff --git a/src/ipc/interfaces.cpp b/src/ipc/interfaces.cpp index 73a616bfb95..66b5e8ec4f6 100644 --- a/src/ipc/interfaces.cpp +++ b/src/ipc/interfaces.cpp @@ -63,7 +63,7 @@ public: std::unique_ptr spawnProcess(const char* new_exe_name) override { mp::ProcessId pid; - int fd = m_process->spawn(new_exe_name, m_process_argv0, pid); + mp::SocketId fd = m_process->spawn(new_exe_name, m_process_argv0, pid); LogDebug(::BCLog::IPC, "Process %s pid %i launched\n", new_exe_name, pid); auto init = m_protocol->connect(fd); Ipc::addCleanup(*init, [this, new_exe_name, pid] { @@ -75,19 +75,19 @@ public: 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)) { + mp::SocketId socket{mp::SocketError}; + if (!m_process->checkSpawned(argc, argv, socket)) { return false; } IgnoreCtrlC(strprintf("[%s] SIGINT received — waiting for parent to shut down.\n", m_exe_name)); - m_protocol->serve(fd, m_init); + m_protocol->serve(socket, m_init); exit_status = EXIT_SUCCESS; return true; } std::unique_ptr connectAddress(std::string& address) override { if (address.empty() || address == "0") return nullptr; - int fd; + mp::SocketId fd; if (address == "auto") { // Treat "auto" the same as "unix" except don't treat it an as error // if the connection is not accepted. Just return null so the caller @@ -113,7 +113,7 @@ public: } void listenAddress(std::string& address) override { - int fd = m_process->bind(gArgs.GetDataDirNet(), m_exe_name, address); + mp::SocketId fd = m_process->bind(gArgs.GetDataDirNet(), m_exe_name, address); m_protocol->listen(fd, m_init); } void disconnectIncoming() override diff --git a/src/ipc/process.cpp b/src/ipc/process.cpp index ec658a9af26..d7f040780d5 100644 --- a/src/ipc/process.cpp +++ b/src/ipc/process.cpp @@ -32,7 +32,7 @@ namespace { class ProcessImpl : public Process { public: - int spawn(const std::string& new_exe_name, const fs::path& argv0_path, mp::ProcessId& pid) override + mp::SocketId spawn(const std::string& new_exe_name, const fs::path& argv0_path, mp::ProcessId& pid) override { return mp::SpawnProcess(pid, [&](int fd) { fs::path path = argv0_path; @@ -42,7 +42,7 @@ public: }); } int waitSpawned(mp::ProcessId pid) override { return mp::WaitProcess(pid); } - bool checkSpawned(int argc, char* argv[], int& fd) override + bool checkSpawned(int argc, char* argv[], mp::SocketId& socket) 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 @@ -60,13 +60,13 @@ public: if (!maybe_fd) { throw std::runtime_error(strprintf("Invalid -ipcfd number '%s'", argv[2])); } - fd = *maybe_fd; + socket = *maybe_fd; return true; } - int connect(const fs::path& data_dir, + mp::SocketId connect(const fs::path& data_dir, const std::string& dest_exe_name, std::string& address) override; - int bind(const fs::path& data_dir, const std::string& exe_name, std::string& address) override; + mp::SocketId bind(const fs::path& data_dir, const std::string& exe_name, std::string& address) override; }; static bool ParseAddress(std::string& address, @@ -98,7 +98,7 @@ static bool ParseAddress(std::string& address, return false; } -int ProcessImpl::connect(const fs::path& data_dir, +mp::SocketId ProcessImpl::connect(const fs::path& data_dir, const std::string& dest_exe_name, std::string& address) { @@ -108,8 +108,8 @@ int ProcessImpl::connect(const fs::path& data_dir, throw std::invalid_argument(error); } - int fd; - if ((fd = ::socket(addr.sun_family, SOCK_STREAM, 0)) == -1) { + mp::SocketId fd; + if ((fd = ::socket(addr.sun_family, SOCK_STREAM, 0)) == mp::SocketError) { throw std::system_error(errno, std::system_category()); } if (::connect(fd, (struct sockaddr*)&addr, sizeof(addr)) == 0) { @@ -122,7 +122,7 @@ int ProcessImpl::connect(const fs::path& data_dir, throw std::system_error(connect_error, std::system_category()); } -int ProcessImpl::bind(const fs::path& data_dir, const std::string& exe_name, std::string& address) +mp::SocketId ProcessImpl::bind(const fs::path& data_dir, const std::string& exe_name, std::string& address) { struct sockaddr_un addr; std::string error; @@ -138,8 +138,8 @@ int ProcessImpl::bind(const fs::path& data_dir, const std::string& exe_name, std } } - int fd; - if ((fd = ::socket(addr.sun_family, SOCK_STREAM, 0)) == -1) { + mp::SocketId fd; + if ((fd = ::socket(addr.sun_family, SOCK_STREAM, 0)) == mp::SocketError) { throw std::system_error(errno, std::system_category()); } diff --git a/src/ipc/process.h b/src/ipc/process.h index 13a004d7ca3..54ca204cd55 100644 --- a/src/ipc/process.h +++ b/src/ipc/process.h @@ -24,25 +24,24 @@ 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, mp::ProcessId& pid) = 0; + //! Spawn process and return socket id for communicating with it. + virtual mp::SocketId spawn(const std::string& new_exe_name, const fs::path& argv0_path, mp::ProcessId& pid) = 0; //! Wait for spawned process to exit and return its exit code. virtual int waitSpawned(mp::ProcessId 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 + //! process. If so, return true and a socket id for communicating //! with the parent process. - virtual bool checkSpawned(int argc, char* argv[], int& fd) = 0; + virtual bool checkSpawned(int argc, char* argv[], mp::SocketId& socket) = 0; - //! Canonicalize and connect to address, returning socket descriptor. - virtual int connect(const fs::path& data_dir, + //! Canonicalize and connect to address, returning socket id. + virtual mp::SocketId connect(const fs::path& data_dir, const std::string& dest_exe_name, std::string& address) = 0; - //! Create listening socket, bind and canonicalize address, and return socket descriptor. - virtual int bind(const fs::path& data_dir, + //! Create listening socket, bind and canonicalize address, and return socket id. + virtual mp::SocketId bind(const fs::path& data_dir, const std::string& exe_name, std::string& address) = 0; }; diff --git a/src/ipc/protocol.h b/src/ipc/protocol.h index d14c7f20013..4c3c1bdc8d9 100644 --- a/src/ipc/protocol.h +++ b/src/ipc/protocol.h @@ -6,6 +6,7 @@ #define BITCOIN_IPC_PROTOCOL_H #include +#include #include #include @@ -32,12 +33,12 @@ public: //! up its own state (calling ProxyServer destructors, etc) on disconnect, //! and any client calls will just throw ipc::Exception errors after a //! disconnect. - virtual std::unique_ptr connect(int fd) = 0; + virtual std::unique_ptr connect(mp::SocketId fd) = 0; - //! Listen for connections on provided socket descriptor, accept them, and - //! handle requests on accepted connections. This method doesn't block, and + //! Listen for connections on provided socket id, accept them, and handle + //! requests on accepted connections. This method doesn't block, and //! performs I/O on a background thread. - virtual void listen(int listen_fd, interfaces::Init& init) = 0; + virtual void listen(mp::SocketId listen_fd, interfaces::Init& init) = 0; //! Handle requests on provided socket descriptor, forwarding them to the //! provided Init interface. Socket communication is handled on the @@ -56,7 +57,7 @@ public: //! client connections from another thread as soon as the event loop is //! available, but should not be necessary in normal code which starts //! clients and servers independently. - virtual void serve(int fd, interfaces::Init& init, const std::function& ready_fn = {}) = 0; + virtual void serve(mp::SocketId fd, interfaces::Init& init, const std::function& ready_fn = {}) = 0; //! Disconnect any incoming connections that are still connected. virtual void disconnectIncoming() = 0; diff --git a/src/ipc/test/ipc_tests.cpp b/src/ipc/test/ipc_tests.cpp index 08f2aa06e4d..f09c2d161d3 100644 --- a/src/ipc/test/ipc_tests.cpp +++ b/src/ipc/test/ipc_tests.cpp @@ -129,7 +129,7 @@ void IpcPipeTest() //! Test ipc::Protocol connect() and serve() methods connecting over a socketpair. void IpcSocketPairTest() { - int fds[2]; + mp::SocketId fds[2]; BOOST_CHECK_EQUAL(socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0); std::unique_ptr init{std::make_unique()}; std::unique_ptr protocol{ipc::capnp::MakeCapnpProtocol("IpcSocketPairTest")}; @@ -159,15 +159,15 @@ void IpcSocketTest(const fs::path& datadir) auto bind_and_listen{[&](const std::string& bind_address) { std::string address{bind_address}; - int serve_fd = process->bind(datadir, "test_bitcoin", address); - BOOST_CHECK_GE(serve_fd, 0); + mp::SocketId serve_fd = process->bind(datadir, "test_bitcoin", address); + BOOST_CHECK_NE(serve_fd, mp::SocketError); BOOST_CHECK_EQUAL(address, bind_address); protocol->listen(serve_fd, *init); }}; auto connect_and_test{[&](const std::string& connect_address) { std::string address{connect_address}; - int connect_fd{process->connect(datadir, "test_bitcoin", address)}; + mp::SocketId connect_fd{process->connect(datadir, "test_bitcoin", address)}; BOOST_CHECK_EQUAL(address, connect_address); std::unique_ptr remote_init{protocol->connect(connect_fd)}; std::unique_ptr remote_echo{remote_init->makeEcho()}; diff --git a/src/ipc/util.h b/src/ipc/util.h index 58912a8575d..5e591d7b21c 100644 --- a/src/ipc/util.h +++ b/src/ipc/util.h @@ -15,6 +15,8 @@ namespace mp { // libmultiprocess changes so they don't have to be reviewed in a single PR. #if MP_MAJOR_VERSION < 14 using ProcessId = int; +using SocketId = int; +constexpr SocketId SocketError{-1}; #endif } // namespace mp