highperfocused/bitcoin
1
0
Fork 0
mirror of https://github.com/bitcoin/bitcoin.git synced 2026-09-11 21:20:39 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge bitcoin/bitcoin#35084: ipc: Add nonunix platform support

Browse Source 
d3d74e701f ipc, refactor: Update mp::g_thread_context references (Ryan Ofsky)
2d3f72fd3f ipc, refactor: Update mp::SpawnProcess call (Ryan Ofsky)
e9f19815ca ipc, refactor: Add Stream type alias and use it (Ryan Ofsky)
3859805f05 ipc, refactor: Add SocketId type alias and use it (Ryan Ofsky)
2ee9b69c7a ipc, refactor: Add ProcessId type alias and use it (Ryan Ofsky)
3449797141 ipc: Avoid 'unistd.h' error with MSVC (Ryan Ofsky)
dbcc192dce ipc, refactor: fix include order (Ryan Ofsky)
7c86d4834e ipc, refactor: use native path separators in test (Ryan Ofsky)
00287b9a34 ipc, refactor: Change Protocol class field order (Ryan Ofsky)
33d37f3c35 ipc, refactor: Drop connect/listen/serve exe_name parameters (Ryan Ofsky)
794940469e ipc, moveonly: combine ipc_test.cpp and ipc_tests.cpp (Ryan Ofsky)

Pull request description:

  This PR makes Bitcoin Core changes needed to be compatible with https://github.com/bitcoin-core/libmultiprocess/pull/274, which changes the libmultiprocess API to stop using unix-specific types so it is compatible with windows. (Windows support is added in followups: https://github.com/bitcoin-core/libmultiprocess/pull/231 and https://github.com/bitcoin/bitcoin/pull/32387.)

  The PR uses some [compatibility shims](https://github.com/ryanofsky/bitcoin/blob/pr/ipc-wins/src/ipc/util.h) so it can be reviewed and merged without needing to merge https://github.com/bitcoin-core/libmultiprocess/pull/274 first and bump the libmultiprocess subtree. These can be deleted when the subtree is updated.

  ---

  Review note: All the changes here are refactoring, and you don't really need to know anything about IPC or Windows to review this code. It is also a mostly move-only change (131 lines added, 96 removed, 215 moved)

ACKs for top commit:
  xyzconstant:
    tACK d3d74e701f
  enirox001:
    ACK d3d74e701f
  Sjors:
    ACK d3d74e701f
  ViniciusCestarii:
    re-ACK d3d74e701f tested locally on Linux

Tree-SHA512: cd48708f9fd086ac8127dc75cfaf4bd8f8da81e07d11b2c9e65fd9061ffa33478bffc6fd6fa4b3505e86c6437752578fe6e5bd590c683c3bc9969093103a5608
This commit is contained in:
merge-script
2026-07-29 09:36:13 +02:00
parent 7dea464d6b d3d74e701f
commit a9d181f2d3
14 changed files with 365 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/ipc/CMakeLists.txt
View File

@@ -31,8 +31,8 @@ if(BUILD_TESTS)
# compiler only allows importing by relative path when the importing and
# imported files are underneath the same compilation source prefix, so the
# source prefix must be src/ipc, not src/ipc/test/
add_library(bitcoin_ipc_test STATIC EXCLUDE_FROM_ALL
test/ipc_test.cpp
add_library(bitcoin_ipc_test OBJECT EXCLUDE_FROM_ALL
test/ipc_tests.cpp
)
target_capnp_sources(bitcoin_ipc_test ${CMAKE_CURRENT_SOURCE_DIR}
test/ipc_test.capnp

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

@@ -71,36 +71,36 @@ void IpcLogFn(mp::LogMessage message)
class CapnpProtocol : public Protocol
{
public:
CapnpProtocol(const char* exe_name) : m_exe_name{exe_name} {}
~CapnpProtocol() noexcept(true)
{
m_loop_ref.reset();
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
std::unique_ptr<interfaces::Init> connect(mp::Stream stream) override
{
startLoop(exe_name);
return mp::ConnectStream<messages::Init>(*m_loop, fd);
startLoop();
return mp::ConnectStream<messages::Init>(*m_loop, std::move(stream));
}
void listen(int listen_fd, const char* exe_name, interfaces::Init& init) override
void listen(mp::SocketId listen_fd, interfaces::Init& init) override
{
startLoop(exe_name);
startLoop();
if (::listen(listen_fd, /*backlog=*/5) != 0) {
throw std::system_error(errno, std::system_category());
}
mp::ListenConnections<messages::Init>(*m_loop, listen_fd, init);
}
void serve(int fd, const char* exe_name, interfaces::Init& init, const std::function<void()>& ready_fn = {}) override
void serve(interfaces::Init& init, const std::function<mp::Stream()>& make_stream) override
{
assert(!m_loop);
mp::g_thread_context.thread_name = mp::ThreadName(exe_name);
mp::CurrentThread().thread_name = mp::ThreadName(m_exe_name);
mp::LogOptions opts = {
.log_fn = IpcLogFn,
.log_level = GetRequestedIPCLogLevel()
};
m_loop.emplace(exe_name, std::move(opts), &m_context);
if (ready_fn) ready_fn();
mp::ServeStream<messages::Init>(*m_loop, fd, init);
m_loop.emplace(m_exe_name, std::move(opts), &m_context);
mp::ServeStream<messages::Init>(*m_loop, make_stream(), init);
m_parent_connection = &m_loop->m_incoming_connections.back();
m_loop->loop();
m_loop.reset();
@@ -115,12 +115,17 @@ public:
m_loop->m_incoming_connections.remove_if([this](mp::Connection& c) { return &c != m_parent_connection; });
});
}
mp::Stream makeStream(mp::SocketId socket) override
{
startLoop();
return mp::MakeStream(*m_loop, socket);
}
void addCleanup(std::type_index type, void* iface, std::function<void()> cleanup) override
{
mp::ProxyTypeRegister::types().at(type)(iface).cleanup_fns.emplace_back(std::move(cleanup));
}
Context& context() override { return m_context; }
void startLoop(const char* exe_name)
void startLoop()
{
if (m_loop) return;
std::promise<void> promise;
@@ -130,7 +135,7 @@ public:
.log_fn = IpcLogFn,
.log_level = GetRequestedIPCLogLevel()
};
m_loop.emplace(exe_name, std::move(opts), &m_context);
m_loop.emplace(m_exe_name, std::move(opts), &m_context);
m_loop_ref.emplace(*m_loop);
promise.set_value();
m_loop->loop();
@@ -138,8 +143,8 @@ public:
});
promise.get_future().wait();
}
const char* m_exe_name;
Context m_context;
std::thread m_loop_thread;
//! EventLoop object which manages I/O events for all connections.
std::optional<mp::EventLoop> m_loop;
//! Reference to the same EventLoop. Increments the loop’s refcount on
@@ -148,9 +153,10 @@ public:
std::optional<mp::EventLoopRef> m_loop_ref;
//! Connection to parent, if this is a child process spawned by a parent process.
mp::Connection* m_parent_connection{nullptr};
std::thread m_loop_thread;
};
} // namespace
std::unique_ptr<Protocol> MakeCapnpProtocol() { return std::make_unique<CapnpProtocol>(); }
std::unique_ptr<Protocol> MakeCapnpProtocol(const char* exe_name) { return std::make_unique<CapnpProtocol>(exe_name); }
} // namespace capnp
} // namespace ipc

2
src/ipc/capnp/protocol.h
View File

@@ -10,7 +10,7 @@
namespace ipc {
class Protocol;
namespace capnp {
std::unique_ptr<Protocol> MakeCapnpProtocol();
std::unique_ptr<Protocol> MakeCapnpProtocol(const char* exe_name);
} // namespace capnp
} // namespace ipc

26
src/ipc/interfaces.cpp
View File

@@ -21,10 +21,13 @@
#include <memory>
#include <stdexcept>
#include <string>
#include <unistd.h>
#include <utility>
#include <vector>
#ifndef WIN32
#include <unistd.h>
#endif
namespace ipc {
namespace {
#ifndef WIN32
@@ -54,15 +57,14 @@ 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())
m_protocol(ipc::capnp::MakeCapnpProtocol(exe_name)), 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);
const auto [pid, socket] = m_process->spawn(new_exe_name, m_process_argv0);
LogDebug(::BCLog::IPC, "Process %s pid %i launched\n", new_exe_name, pid);
auto init = m_protocol->connect(fd, m_exe_name);
auto init = m_protocol->connect(m_protocol->makeStream(socket));
Ipc::addCleanup(*init, [this, new_exe_name, pid] {
int status = m_process->waitSpawned(pid);
LogDebug(::BCLog::IPC, "Process %s pid %i exited with status %i\n", new_exe_name, pid, status);
@@ -72,19 +74,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_exe_name, m_init);
m_protocol->serve(m_init, [&] { return m_protocol->makeStream(socket); } );
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
@@ -106,12 +108,12 @@ public:
} else {
fd = m_process->connect(gArgs.GetDataDirNet(), "bitcoin-node", address);
}
return m_protocol->connect(fd, m_exe_name);
return m_protocol->connect(m_protocol->makeStream(fd));
}
void listenAddress(std::string& address) override
{
int fd = m_process->bind(gArgs.GetDataDirNet(), m_exe_name, address);
m_protocol->listen(fd, m_exe_name, m_init);
mp::SocketId fd = m_process->bind(gArgs.GetDataDirNet(), m_exe_name, address);
m_protocol->listen(fd, m_init);
}
void disconnectIncoming() override
{

39
src/ipc/process.cpp
View File

@@ -18,11 +18,12 @@
#include <exception>
#include <iostream>
#include <stdexcept>
#include <utility>
#include <vector>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
#include <utility>
#include <vector>
using util::RemovePrefixView;
@@ -31,17 +32,17 @@ namespace {
class ProcessImpl : public Process
{
public:
int spawn(const std::string& new_exe_name, const fs::path& argv0_path, int& pid) override
std::tuple<mp::ProcessId, mp::SocketId> spawn(const std::string& new_exe_name, const fs::path& argv0_path) override
{
return mp::SpawnProcess(pid, [&](int fd) {
return mp::SpawnProcess([&](std::string connect_info) {
fs::path path = argv0_path;
path.remove_filename();
path /= fs::PathFromString(new_exe_name);
return std::vector<std::string>{fs::PathToString(path), "-ipcfd", strprintf("%i", fd)};
return std::vector<std::string>{fs::PathToString(path), "-ipcfd", std::move(connect_info)};
});
}
int waitSpawned(int pid) override { return mp::WaitProcess(pid); }
bool checkSpawned(int argc, char* argv[], int& fd) override
int waitSpawned(mp::ProcessId pid) override { return mp::WaitProcess(pid); }
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
@@ -55,17 +56,17 @@ public:
// 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.
const auto maybe_fd{ToIntegral<int32_t>(argv[2])};
if (!maybe_fd) {
throw std::runtime_error(strprintf("Invalid -ipcfd number '%s'", argv[2]));
try {
socket = mp::StartSpawned(argv[2]);
} catch (const std::exception& e) {
throw std::runtime_error(strprintf("Invalid -ipcfd number '%s' (%s)", argv[2], e.what()));
}
fd = *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,
@@ -97,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)
{
@@ -107,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) {
@@ -121,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;
@@ -137,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());
}

20
src/ipc/process.h
View File

@@ -8,6 +8,7 @@
#include <util/fs.h>
#include <memory>
#include <ipc/util.h>
#include <string>
namespace ipc {
@@ -23,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, int& pid) = 0;
//! Spawn process and return socket id for communicating with it.
virtual std::tuple<mp::ProcessId, mp::SocketId> spawn(const std::string& new_exe_name, const fs::path& argv0_path) = 0;
//! Wait for spawned process to exit and return its exit code.
virtual int waitSpawned(int pid) = 0;
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;
};

45
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>
@@ -23,8 +24,8 @@ class Protocol
public:
virtual ~Protocol() = default;
//! Return Init interface that forwards requests over given socket descriptor.
//! Socket communication is handled on a background thread.
//! Return Init interface that forwards requests over given connection
//! stream. Socket communication is handled on a background thread.
//!
//! @note It could be potentially useful in the future to add
//! std::function<void()> on_disconnect callback argument here. But there
@@ -32,31 +33,31 @@ 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, const char* exe_name) = 0;
virtual std::unique_ptr<interfaces::Init> connect(mp::Stream stream) = 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, const char* exe_name, 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
//! current thread, and this call blocks until the socket is closed.
//! Handle requests from a stream provided by the make_stream callback,
//! forwarding them to the provided Init interface. Socket communication is
//! handled on the current thread, and this call blocks until the socket is
//! closed. A callback is used to specify the stream because this method
//! initializes the event loop and it may not be possible to create the
//! stream before the event loop is initialized.
//!
//! @note: If this method is called, it needs be called before connect() or
//! listen() methods, because for ease of implementation it's inflexible and
//! always runs the event loop in the foreground thread. It can share its
//! event loop with the other methods but can't share an event loop that was
//! created by them. This isn't really a problem because serve() is only
//! called by spawned child processes that call it immediately to
//! @note: If this method is called, it needs to be called before connect()
//! or listen() methods, because for ease of implementation this method is
//! inflexible and always runs the event loop in the foreground thread. It
//! can share its event loop with the other methods but can't share an event
//! loop that was created by them. This isn't a problem because serve() is
//! only called by spawned child processes that call it immediately to
//! communicate back with parent processes.
//
//! The optional `ready_fn` callback will be called after the event loop is
//! created but before it is started. This can be useful in tests to trigger
//! 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, const char* exe_name, interfaces::Init& init, const std::function<void()>& ready_fn = {}) = 0;
virtual void serve(interfaces::Init& init, const std::function<mp::Stream()>& make_stream) = 0;
//! Make stream object from socket id.
virtual mp::Stream makeStream(mp::SocketId socket) = 0;
//! Disconnect any incoming connections that are still connected.
virtual void disconnectIncoming() = 0;

9
src/ipc/test/CMakeLists.txt
View File

@@ -2,11 +2,6 @@
# Distributed under the MIT software license, see the accompanying
# file COPYING or https://opensource.org/license/mit/.
# Do not use generator expressions in test sources because the
# SOURCES property is processed to gather test suite macros.
target_sources(test_bitcoin
PRIVATE
ipc_tests.cpp
)
target_link_libraries(test_bitcoin bitcoin_ipc_test bitcoin_ipc)
add_boost_test(${CMAKE_CURRENT_SOURCE_DIR}/ipc_tests.cpp)

9
src/ipc/test/fuzz/ipc.cpp
View File

@@ -5,6 +5,7 @@
#include <primitives/transaction.h>
#include <capnp/capability.h>
#include <capnp/rpc.h>
#include <ipc/util.h>
#include <kj/memory.h>
#include <mp/proxy-io.h>
#include <mp/proxy.h>
@@ -78,10 +79,10 @@ static void initialize_ipc()
static const auto testing_setup = MakeNoLogFileContext<>();
(void)testing_setup;
// Ensure g_thread_context is destroyed after the IPC setup, since C++
// destroys thread_local objects in reverse construction order.
mp::ThreadContext& thread_context{mp::g_thread_context};
(void)thread_context;
// Ensure the thread's ThreadContext is created before the IPC setup, so
// it is destroyed after it, since C++ destroys thread_local objects in
// reverse construction order.
mp::CurrentThread();
thread_local static IpcFuzzSetup ipc; // NOLINT(bitcoin-nontrivial-threadlocal)
g_ipc = &ipc;

193
src/ipc/test/ipc_test.cpp
View File

@@ -1,193 +0,0 @@
// Copyright (c) 2023-present 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/mining.capnp.h>
#include <ipc/capnp/protocol.h>
#include <ipc/process.h>
#include <ipc/protocol.h>
#include <ipc/test/ipc_test.capnp.h>
#include <ipc/test/ipc_test.capnp.proxy.h>
#include <ipc/test/ipc_test.h>
#include <mp/proxy-types.h>
#include <tinyformat.h>
#include <util/log.h>
#include <validation.h>
#include <future>
#include <thread>
#include <kj/common.h>
#include <kj/memory.h>
#include <kj/test.h>
#include <stdexcept>
#include <boost/test/unit_test.hpp>
static_assert(ipc::capnp::messages::MAX_MONEY == MAX_MONEY);
static_assert(ipc::capnp::messages::MAX_DOUBLE == std::numeric_limits<double>::max());
static_assert(ipc::capnp::messages::DEFAULT_BLOCK_RESERVED_WEIGHT == DEFAULT_BLOCK_RESERVED_WEIGHT);
static_assert(ipc::capnp::messages::DEFAULT_COINBASE_OUTPUT_MAX_ADDITIONAL_SIGOPS == DEFAULT_COINBASE_OUTPUT_MAX_ADDITIONAL_SIGOPS);
//! Remote init class.
class TestInit : public interfaces::Init
{
public:
std::unique_ptr<interfaces::Echo> makeEcho() override { return interfaces::MakeEcho(); }
};
//! Generate a temporary path with temp_directory_path and mkstemp
static std::string TempPath(std::string_view pattern)
{
std::string temp{fs::PathToString(fs::path{fs::temp_directory_path()} / fs::PathFromString(std::string{pattern}))};
temp.push_back('\0');
int fd{mkstemp(temp.data())};
BOOST_CHECK_GE(fd, 0);
BOOST_CHECK_EQUAL(close(fd), 0);
temp.resize(temp.size() - 1);
fs::remove(fs::PathFromString(temp));
return temp;
}
//! Unit test that tests execution of IPC calls without actually creating a
//! separate process. This test is primarily intended to verify behavior of type
//! conversion code that converts C++ objects to Cap'n Proto messages and vice
//! versa.
//!
//! The test creates a thread which creates a FooImplementation object (defined
//! in ipc_test.h) and a two-way pipe accepting IPC requests which call methods
//! on the object through FooInterface (defined in ipc_test.capnp).
void IpcPipeTest()
{
// Setup: create FooImplementation object and listen for FooInterface requests
std::promise<std::unique_ptr<mp::ProxyClient<gen::FooInterface>>> foo_promise;
std::thread thread([&]() {
mp::EventLoop loop("IpcPipeTest", [](bool raise, const std::string& log) { LogInfo("LOG%i: %s", raise, log); });
auto pipe = loop.m_io_context.provider->newTwoWayPipe();
auto connection_client = std::make_unique<mp::Connection>(loop, kj::mv(pipe.ends[0]));
auto foo_client = std::make_unique<mp::ProxyClient<gen::FooInterface>>(
connection_client->m_rpc_system->bootstrap(mp::ServerVatId().vat_id).castAs<gen::FooInterface>(),
connection_client.get(), /* destroy_connection= */ true);
(void)connection_client.release();
foo_promise.set_value(std::move(foo_client));
auto connection_server = std::make_unique<mp::Connection>(loop, kj::mv(pipe.ends[1]), [&](mp::Connection& connection) {
auto foo_server = kj::heap<mp::ProxyServer<gen::FooInterface>>(std::make_shared<FooImplementation>(), connection);
return capnp::Capability::Client(kj::mv(foo_server));
});
connection_server->onDisconnect([&] { connection_server.reset(); });
loop.loop();
});
std::unique_ptr<mp::ProxyClient<gen::FooInterface>> foo{foo_promise.get_future().get()};
// Test: make sure arguments were sent and return value is received
BOOST_CHECK_EQUAL(foo->add(1, 2), 3);
COutPoint txout1{Txid::FromUint256(uint256{100}), 200};
COutPoint txout2{foo->passOutPoint(txout1)};
BOOST_CHECK(txout1 == txout2);
UniValue uni1{UniValue::VOBJ};
uni1.pushKV("i", 1);
uni1.pushKV("s", "two");
UniValue uni2{foo->passUniValue(uni1)};
BOOST_CHECK_EQUAL(uni1.write(), uni2.write());
CMutableTransaction mtx;
mtx.version = 2;
mtx.nLockTime = 3;
mtx.vin.emplace_back(txout1);
mtx.vout.emplace_back(COIN, CScript());
CTransactionRef tx1{MakeTransactionRef(mtx)};
CTransactionRef tx2{foo->passTransaction(tx1)};
BOOST_CHECK(*Assert(tx1) == *Assert(tx2));
std::vector<CTransactionRef> txs1;
txs1.push_back(tx1);
txs1.push_back(nullptr);
std::vector<CTransactionRef> txs2(foo->passTransactions(txs1));
BOOST_CHECK_EQUAL(txs2.size(), 2);
BOOST_CHECK(*Assert(txs1[0]) == *Assert(txs2[0]));
BOOST_CHECK(!txs2[1]);
std::vector<char> vec1{'H', 'e', 'l', 'l', 'o'};
std::vector<char> vec2{foo->passVectorChar(vec1)};
BOOST_CHECK_EQUAL(std::string_view(vec1.begin(), vec1.end()), std::string_view(vec2.begin(), vec2.end()));
auto script1{CScript() << OP_11};
auto script2{foo->passScript(script1)};
BOOST_CHECK_EQUAL(HexStr(script1), HexStr(script2));
// Test cleanup: disconnect and join thread
foo.reset();
thread.join();
}
//! Test ipc::Protocol connect() and serve() methods connecting over a socketpair.
void IpcSocketPairTest()
{
int 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()};
std::promise<void> promise;
std::thread thread([&]() {
protocol->serve(fds[0], "test-serve", *init, [&] { promise.set_value(); });
});
promise.get_future().wait();
std::unique_ptr<interfaces::Init> remote_init{protocol->connect(fds[1], "test-connect")};
std::unique_ptr<interfaces::Echo> remote_echo{remote_init->makeEcho()};
BOOST_CHECK_EQUAL(remote_echo->echo("echo test"), "echo test");
remote_echo.reset();
remote_init.reset();
thread.join();
}
//! Test ipc::Process bind() and connect() methods connecting over a unix socket.
void IpcSocketTest(const fs::path& datadir)
{
std::unique_ptr<interfaces::Init> init{std::make_unique<TestInit>()};
std::unique_ptr<ipc::Protocol> protocol{ipc::capnp::MakeCapnpProtocol()};
std::unique_ptr<ipc::Process> process{ipc::MakeProcess()};
std::string invalid_bind{"invalid:"};
BOOST_CHECK_THROW(process->bind(datadir, "test_bitcoin", invalid_bind), std::invalid_argument);
BOOST_CHECK_THROW(process->connect(datadir, "test_bitcoin", invalid_bind), std::invalid_argument);
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);
BOOST_CHECK_EQUAL(address, bind_address);
protocol->listen(serve_fd, "test-serve", *init);
}};
auto connect_and_test{[&](const std::string& connect_address) {
std::string address{connect_address};
int 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, "test-connect")};
std::unique_ptr<interfaces::Echo> remote_echo{remote_init->makeEcho()};
BOOST_CHECK_EQUAL(remote_echo->echo("echo test"), "echo test");
}};
// Need to specify explicit socket addresses outside the data directory, because the data
// directory path is so long that the default socket address and any other
// addresses in the data directory would fail with errors like:
// Address 'unix' path '"/tmp/test_common_Bitcoin Core/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff/test_bitcoin.sock"' exceeded maximum socket path length
std::vector<std::string> addresses{
strprintf("unix:%s", TempPath("bitcoin_sock0_XXXXXX")),
strprintf("unix:%s", TempPath("bitcoin_sock1_XXXXXX")),
};
// Bind and listen on multiple addresses
for (const auto& address : addresses) {
bind_and_listen(address);
}
// Connect and test each address multiple times.
for (int i : {0, 1, 0, 0, 1}) {
connect_and_test(addresses[i]);
}
}

4
src/ipc/test/ipc_test.h
View File

@@ -24,8 +24,4 @@ public:
CScript passScript(CScript s) { return s; }
};
void IpcPipeTest();
void IpcSocketPairTest();
void IpcSocketTest(const fs::path& datadir);
#endif // BITCOIN_IPC_TEST_IPC_TEST_H

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

@@ -2,13 +2,202 @@
// 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/mining.capnp.h>
#include <ipc/capnp/protocol.h>
#include <ipc/process.h>
#include <ipc/protocol.h>
#include <ipc/test/ipc_test.capnp.h>
#include <ipc/test/ipc_test.capnp.proxy.h>
#include <ipc/test/ipc_test.h>
#include <mp/proxy-types.h>
#include <test/util/common.h>
#include <test/util/setup_common.h>
#include <tinyformat.h>
#include <util/log.h>
#include <validation.h>
#include <future>
#include <thread>
#include <kj/common.h>
#include <kj/memory.h>
#include <kj/test.h>
#include <stdexcept>
#include <boost/test/unit_test.hpp>
static_assert(ipc::capnp::messages::MAX_MONEY == MAX_MONEY);
static_assert(ipc::capnp::messages::MAX_DOUBLE == std::numeric_limits<double>::max());
static_assert(ipc::capnp::messages::DEFAULT_BLOCK_RESERVED_WEIGHT == DEFAULT_BLOCK_RESERVED_WEIGHT);
static_assert(ipc::capnp::messages::DEFAULT_COINBASE_OUTPUT_MAX_ADDITIONAL_SIGOPS == DEFAULT_COINBASE_OUTPUT_MAX_ADDITIONAL_SIGOPS);
//! Remote init class.
class TestInit : public interfaces::Init
{
public:
std::unique_ptr<interfaces::Echo> makeEcho() override { return interfaces::MakeEcho(); }
};
//! Generate a temporary path with temp_directory_path and mkstemp
static std::string TempPath(std::string_view pattern)
{
std::string temp{fs::PathToString(fs::path{fs::temp_directory_path()} / fs::PathFromString(std::string{pattern}))};
temp.push_back('\0');
int fd{mkstemp(temp.data())};
BOOST_CHECK_GE(fd, 0);
BOOST_CHECK_EQUAL(close(fd), 0);
temp.resize(temp.size() - 1);
fs::remove(fs::PathFromString(temp));
return temp;
}
//! Unit test that tests execution of IPC calls without actually creating a
//! separate process. This test is primarily intended to verify behavior of type
//! conversion code that converts C++ objects to Cap'n Proto messages and vice
//! versa.
//!
//! The test creates a thread which creates a FooImplementation object (defined
//! in ipc_test.h) and a two-way pipe accepting IPC requests which call methods
//! on the object through FooInterface (defined in ipc_test.capnp).
void IpcPipeTest()
{
// Setup: create FooImplementation object and listen for FooInterface requests
std::promise<std::unique_ptr<mp::ProxyClient<gen::FooInterface>>> foo_promise;
std::thread thread([&]() {
mp::EventLoop loop("IpcPipeTest", [](bool raise, const std::string& log) { LogInfo("LOG%i: %s", raise, log); });
auto pipe = loop.m_io_context.provider->newTwoWayPipe();
auto connection_client = std::make_unique<mp::Connection>(loop, kj::mv(pipe.ends[0]));
auto foo_client = std::make_unique<mp::ProxyClient<gen::FooInterface>>(
connection_client->m_rpc_system->bootstrap(mp::ServerVatId().vat_id).castAs<gen::FooInterface>(),
connection_client.get(), /* destroy_connection= */ true);
(void)connection_client.release();
foo_promise.set_value(std::move(foo_client));
auto connection_server = std::make_unique<mp::Connection>(loop, kj::mv(pipe.ends[1]), [&](mp::Connection& connection) {
auto foo_server = kj::heap<mp::ProxyServer<gen::FooInterface>>(std::make_shared<FooImplementation>(), connection);
return capnp::Capability::Client(kj::mv(foo_server));
});
connection_server->onDisconnect([&] { connection_server.reset(); });
loop.loop();
});
std::unique_ptr<mp::ProxyClient<gen::FooInterface>> foo{foo_promise.get_future().get()};
// Test: make sure arguments were sent and return value is received
BOOST_CHECK_EQUAL(foo->add(1, 2), 3);
COutPoint txout1{Txid::FromUint256(uint256{100}), 200};
COutPoint txout2{foo->passOutPoint(txout1)};
BOOST_CHECK(txout1 == txout2);
UniValue uni1{UniValue::VOBJ};
uni1.pushKV("i", 1);
uni1.pushKV("s", "two");
UniValue uni2{foo->passUniValue(uni1)};
BOOST_CHECK_EQUAL(uni1.write(), uni2.write());
CMutableTransaction mtx;
mtx.version = 2;
mtx.nLockTime = 3;
mtx.vin.emplace_back(txout1);
mtx.vout.emplace_back(COIN, CScript());
CTransactionRef tx1{MakeTransactionRef(mtx)};
CTransactionRef tx2{foo->passTransaction(tx1)};
BOOST_CHECK(*Assert(tx1) == *Assert(tx2));
std::vector<CTransactionRef> txs1;
txs1.push_back(tx1);
txs1.push_back(nullptr);
std::vector<CTransactionRef> txs2(foo->passTransactions(txs1));
BOOST_CHECK_EQUAL(txs2.size(), 2);
BOOST_CHECK(*Assert(txs1[0]) == *Assert(txs2[0]));
BOOST_CHECK(!txs2[1]);
std::vector<char> vec1{'H', 'e', 'l', 'l', 'o'};
std::vector<char> vec2{foo->passVectorChar(vec1)};
BOOST_CHECK_EQUAL(std::string_view(vec1.begin(), vec1.end()), std::string_view(vec2.begin(), vec2.end()));
auto script1{CScript() << OP_11};
auto script2{foo->passScript(script1)};
BOOST_CHECK_EQUAL(HexStr(script1), HexStr(script2));
// Test cleanup: disconnect and join thread
foo.reset();
thread.join();
}
//! Test ipc::Protocol connect() and serve() methods connecting over a socketpair.
void IpcSocketPairTest()
{
std::unique_ptr<interfaces::Init> init{std::make_unique<TestInit>()};
std::unique_ptr<ipc::Protocol> protocol{ipc::capnp::MakeCapnpProtocol("IpcSocketPairTest")};
mp::Stream client_stream;
std::promise<void> promise;
std::thread thread([&]() {
protocol->serve(*init, [&] {
auto pair{mp::SocketPair()};
client_stream = protocol->makeStream(pair[0]);
promise.set_value();
return protocol->makeStream(pair[1]);
});
});
promise.get_future().wait();
std::unique_ptr<interfaces::Init> remote_init{protocol->connect(std::move(client_stream))};
std::unique_ptr<interfaces::Echo> remote_echo{remote_init->makeEcho()};
BOOST_CHECK_EQUAL(remote_echo->echo("echo test"), "echo test");
remote_echo.reset();
remote_init.reset();
thread.join();
}
//! Test ipc::Process bind() and connect() methods connecting over a unix socket.
void IpcSocketTest(const fs::path& datadir)
{
std::unique_ptr<interfaces::Init> init{std::make_unique<TestInit>()};
std::unique_ptr<ipc::Protocol> protocol{ipc::capnp::MakeCapnpProtocol("IpcSocketTest")};
std::unique_ptr<ipc::Process> process{ipc::MakeProcess()};
std::string invalid_bind{"invalid:"};
BOOST_CHECK_THROW(process->bind(datadir, "test_bitcoin", invalid_bind), std::invalid_argument);
BOOST_CHECK_THROW(process->connect(datadir, "test_bitcoin", invalid_bind), std::invalid_argument);
auto bind_and_listen{[&](const std::string& bind_address) {
std::string address{bind_address};
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};
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(protocol->makeStream(connect_fd))};
std::unique_ptr<interfaces::Echo> remote_echo{remote_init->makeEcho()};
BOOST_CHECK_EQUAL(remote_echo->echo("echo test"), "echo test");
}};
// Need to specify explicit socket addresses outside the data directory, because the data
// directory path is so long that the default socket address and any other
// addresses in the data directory would fail with errors like:
// Address 'unix' path '"/tmp/test_common_Bitcoin Core/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff/test_bitcoin.sock"' exceeded maximum socket path length
std::vector<std::string> addresses{
strprintf("unix:%s", TempPath("bitcoin_sock0_XXXXXX")),
strprintf("unix:%s", TempPath("bitcoin_sock1_XXXXXX")),
};
// Bind and listen on multiple addresses
for (const auto& address : addresses) {
bind_and_listen(address);
}
// Connect and test each address multiple times.
for (int i : {0, 1, 0, 0, 1}) {
connect_and_test(addresses[i]);
}
}
BOOST_FIXTURE_TEST_SUITE(ipc_tests, BasicTestingSetup)
BOOST_AUTO_TEST_CASE(ipc_tests)
{
@@ -31,12 +220,13 @@ BOOST_AUTO_TEST_CASE(parse_address_test)
}
BOOST_CHECK_EQUAL(address, expect_address);
}};
check_address("unix", "unix:/var/empty/notexist/test_bitcoin.sock", "");
check_address("unix:", "unix:/var/empty/notexist/test_bitcoin.sock", "");
check_address("unix:path.sock", "unix:/var/empty/notexist/path.sock", "");
std::string prefix{fs::PathToString(datadir / "")};
check_address("unix", "unix:" + prefix + "test_bitcoin.sock", "");
check_address("unix:", "unix:" + prefix + "test_bitcoin.sock", "");
check_address("unix:path.sock", "unix:" + prefix + "path.sock", "");
check_address("unix:0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000.sock",
"unix:/var/empty/notexist/0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000.sock",
"Unix address path \"/var/empty/notexist/0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000.sock\" exceeded maximum socket path length");
"unix:" + prefix + "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000.sock",
"Unix address path \"" + prefix + "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000.sock\" exceeded maximum socket path length");
check_address("invalid", "invalid", "Unrecognized address 'invalid'");
}

64
src/ipc/util.h Normal file
View File

@@ -0,0 +1,64 @@
// Copyright (c) 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_UTIL_H
#define BITCOIN_IPC_UTIL_H
#include <tinyformat.h>
#include <util/strencodings.h>
#include <array>
#include <cstdint>
#include <functional>
#include <kj/debug.h>
#include <mp/proxy-io.h>
#include <mp/util.h>
#include <mp/version.h>
#include <sys/socket.h>
namespace mp {
// Definitions that can be deleted when libmultiprocess subtree is updated to
// v14. Having these allows Bitcoin Core changes to be decoupled from
// libmultiprocess changes so they don't have to be reviewed in a single PR.
#if MP_MAJOR_VERSION < 14
class EventLoop;
using ProcessId = int;
using SocketId = int;
constexpr SocketId SocketError{-1};
using Stream = SocketId;
inline Stream MakeStream(EventLoop&, SocketId socket)
{
return socket;
}
inline std::array<SocketId, 2> SocketPair()
{
int pair[2];
KJ_SYSCALL(socketpair(AF_UNIX, SOCK_STREAM, 0, pair));
return {pair[0], pair[1]};
}
inline std::tuple<ProcessId, SocketId> SpawnProcess(const std::function<std::vector<std::string>(std::string)>& spawn_argv)
{
ProcessId pid;
SocketId socket = SpawnProcess(pid, [&](int fd) { return spawn_argv(strprintf("%d", fd)); });
return {pid, socket};
}
inline SocketId StartSpawned(const std::string& connect_info)
{
auto socket = ToIntegral<SocketId>(connect_info);
if (!socket) throw std::invalid_argument(strprintf("Invalid socket descriptor '%s'", connect_info));
return *socket;
}
inline ThreadContext& CurrentThread()
{
return g_thread_context;
}
#endif
} // namespace mp
#endif // BITCOIN_IPC_UTIL_H

16
src/test/CMakeLists.txt
View File

@@ -171,14 +171,6 @@ target_link_libraries(test_bitcoin
$<TARGET_NAME_IF_EXISTS:USDT::headers>
)
if(ENABLE_WALLET)
add_subdirectory(${PROJECT_SOURCE_DIR}/src/wallet/test wallet)
endif()
if(ENABLE_IPC)
add_subdirectory(${PROJECT_SOURCE_DIR}/src/ipc/test ipc)
endif()
function(add_boost_test source_file)
if(NOT EXISTS ${source_file})
return()
@@ -217,6 +209,14 @@ function(add_all_test_targets)
endforeach()
endfunction()
if(ENABLE_WALLET)
add_subdirectory(${PROJECT_SOURCE_DIR}/src/wallet/test wallet)
endif()
if(ENABLE_IPC)
add_subdirectory(${PROJECT_SOURCE_DIR}/src/ipc/test ipc)
endif()
add_all_test_targets()
install_binary_component(test_bitcoin INTERNAL)