highperfocused/bitcoin
1
0
Fork 0
mirror of https://github.com/bitcoin/bitcoin.git synced 2026-09-12 05:32:22 +02:00
Code Issues Packages Projects Releases Wiki Activity

ipc, refactor: Add SocketId type alias and use it

Browse Source 
Use SocketId type instead of int to represent socket ids to be
compatible with an upcoming version of libmultiprocess which adds
windows support.
This commit is contained in:
Ryan Ofsky
2025-04-30 08:39:29 -04:00
parent 2ee9b69c7a
commit 3859805f05
7 changed files with 42 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/ipc/capnp/protocol.cpp
View File

@@ -78,12 +78,12 @@ public:
if (m_loop_thread.joinable()) m_loop_thread.join();
assert(!m_loop);
};
std::unique_ptr<interfaces::Init> connect(int fd) override
std::unique_ptr<interfaces::Init> connect(mp::SocketId socket) override
{
startLoop();
return mp::ConnectStream<messages::Init>(*m_loop, fd);
return mp::ConnectStream<messages::Init>(*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<messages::Init>(*m_loop, listen_fd, init);
}
void serve(int fd, interfaces::Init& init, const std::function<void()>& ready_fn = {}) override
void serve(mp::SocketId socket, interfaces::Init& init, const std::function<void()>& 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<messages::Init>(*m_loop, fd, init);
mp::ServeStream<messages::Init>(*m_loop, socket, init);
m_parent_connection = &m_loop->m_incoming_connections.back();
m_loop->loop();
m_loop.reset();

12
src/ipc/interfaces.cpp
View File

@@ -63,7 +63,7 @@ public:
std::unique_ptr<interfaces::Init> 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<interfaces::Init> 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

22
src/ipc/process.cpp
View File

@@ -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());
}

17
src/ipc/process.h
View File

@@ -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;
};

11
src/ipc/protocol.h
View File

@@ -6,6 +6,7 @@
#define BITCOIN_IPC_PROTOCOL_H
#include <interfaces/init.h>
#include <ipc/util.h>
#include <functional>
#include <memory>
@@ -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<interfaces::Init> connect(int fd) = 0;
virtual std::unique_ptr<interfaces::Init> 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<void()>& ready_fn = {}) = 0;
virtual void serve(mp::SocketId fd, interfaces::Init& init, const std::function<void()>& ready_fn = {}) = 0;
//! Disconnect any incoming connections that are still connected.
virtual void disconnectIncoming() = 0;

8
src/ipc/test/ipc_tests.cpp
View File

@@ -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<interfaces::Init> init{std::make_unique<TestInit>()};
std::unique_ptr<ipc::Protocol> 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<interfaces::Init> remote_init{protocol->connect(connect_fd)};
std::unique_ptr<interfaces::Echo> remote_echo{remote_init->makeEcho()};

2
src/ipc/util.h
View File

@@ -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