Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
f2c4952
doc: Bump version 12 > 13
ryanofsky Jul 10, 2026
c67439a
util, refactor: Add ProcessId type alias and use it
ryanofsky Apr 30, 2025
eb2e797
util, refactor: Add SocketId type alias and use it
ryanofsky Apr 30, 2025
3589a2a
util, refactor: Add SpawnConnectInfo type alias and use it
ryanofsky Apr 30, 2025
dbf2e50
util, refactor: Do not fork() and exec() separately
ryanofsky Apr 17, 2026
4df5bc0
util, refactor: Add SocketPair() and use it in SpawnProcess
ryanofsky Apr 30, 2025
a7c2938
proxy, refactor: Replace EventLoop wakeup fd integers with KJ stream …
ryanofsky Apr 30, 2025
c5fd319
cmake: Bump minimum required Cap'n Proto version to 0.9
ryanofsky Apr 16, 2026
d9ceffc
proxy, refactor: Change ConnectStream and ServeStream to accept strea…
ryanofsky Apr 30, 2025
31cd357
proxy: Call shutdownWrite() in Connection destructor
ryanofsky Apr 30, 2025
46a508d
proxy: Fix shutdownWrite() exception handling on macOS with dynamic l…
ryanofsky Apr 17, 2026
f26c217
util, refactor: Fix PtrOrValue constructor for move-only types on MSVC
ryanofsky Apr 17, 2026
a8e6f82
proxy, refactor: Fix C4305 truncation warning in Accessor on MSVC
ryanofsky Apr 22, 2026
02ea993
types: Replace SFINAE with requires clauses to avoid MSVC C2039 error
ryanofsky Jun 22, 2026
4781bbc
ci: Check out bitcoin/bitcoin PR #35084 instead of master
ryanofsky Apr 16, 2026
d4d043c
ipc: Wrap mpgen main() in try-catch to print errors
ryanofsky Apr 20, 2026
491f1e8
doc: Remove trailing whitespace
ryanofsky Apr 17, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions .github/workflows/bitcoin-core-ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@ concurrency:

env:
BITCOIN_REPO: bitcoin/bitcoin
# Temporary: use PR #35084 until it merges; revert to refs/heads/master after
BITCOIN_CORE_REF: refs/pull/35084/merge
LLVM_VERSION: 22
LIBCXX_DIR: /tmp/libcxx-build/

Expand Down Expand Up @@ -78,6 +80,7 @@ jobs:
uses: actions/checkout@v4
with:
repository: ${{ env.BITCOIN_REPO }}
ref: ${{ env.BITCOIN_CORE_REF }}
fetch-depth: 1

- name: Checkout libmultiprocess
Expand Down Expand Up @@ -194,6 +197,7 @@ jobs:
uses: actions/checkout@v4
with:
repository: ${{ env.BITCOIN_REPO }}
ref: ${{ env.BITCOIN_CORE_REF }}
fetch-depth: 1

- name: Checkout libmultiprocess
Expand Down
2 changes: 1 addition & 1 deletion CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ endif()
include("cmake/compat_find.cmake")

find_package(Threads REQUIRED)
find_package(CapnProto 0.7 QUIET NO_MODULE)
find_package(CapnProto 0.9 QUIET NO_MODULE)
if(NOT CapnProto_FOUND)
message(FATAL_ERROR
"Cap'n Proto is required but was not found.\n"
Expand Down
2 changes: 1 addition & 1 deletion ci/configs/olddeps.bash
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
CI_DESC="CI job using old Cap'n Proto and cmake versions"
CI_DIR=build-olddeps
export CXXFLAGS="-Werror -Wall -Wextra -Wpedantic -Wno-unused-parameter -Wno-error=array-bounds"
NIX_ARGS=(--argstr capnprotoVersion "0.7.1" --argstr cmakeVersion "3.12.4")
NIX_ARGS=(--argstr capnprotoVersion "0.9.2" --argstr cmakeVersion "3.12.4")
BUILD_ARGS=(-k)
4 changes: 2 additions & 2 deletions doc/design.md
Original file line number Diff line number Diff line change
Expand Up @@ -120,7 +120,7 @@ sequenceDiagram
participant PMT as ProxyMethodTraits
participant Impl as Actual C++ Method

serverInvoke->>SF1: SF1::invoke
serverInvoke->>SF1: SF1::invoke
SF1->>SF2: SF2::invoke
SF2->>SR: SR::invoke
SR->>SC: SC::invoke
Expand Down Expand Up @@ -167,7 +167,7 @@ Thread mapping enables each client thread to have a dedicated server thread proc
Thread mapping is initialized by defining an interface method with a `ThreadMap` parameter and/or response. The example below adds `ThreadMap` to the `construct` method because libmultiprocess calls the `construct` method automatically.

```capnp
interface InitInterface $Proxy.wrap("Init") {
interface InitInterface $Proxy.wrap("Init") {
construct @0 (threadMap: Proxy.ThreadMap) -> (threadMap :Proxy.ThreadMap);
}
```
Expand Down
5 changes: 4 additions & 1 deletion doc/versions.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,11 @@ Library versions are tracked with simple
Versioning policy is described in the [version.h](../include/mp/version.h)
include.

## v12
## v13
- Current unstable version.
- Adds support for nonunix platforms, making API changes that are not backwards compatible.

## [v12.0](https://github.com/bitcoin-core/libmultiprocess/commits/v12.0)
- Adds an optional per-listener `max_connections` parameter to `ListenConnections()`
so servers can stop accepting new connections when a local connection cap is reached,
and resume accepting after existing connections disconnect.
Expand Down
13 changes: 4 additions & 9 deletions example/calculator.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand All @@ -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
Expand Down Expand Up @@ -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])};

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In c7a939b "util, refactor: Add SpawnConnectInfo type alias and use it"

Didn't we lose our error handling here? std::stoi throws exceptions on bad input, so shouldn't we add a try-catch to prevent a crash?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

re: #274 (comment)

Didn't we lose our error handling here? std::stoi throws exceptions on bad input, so shouldn't we add a try-catch to prevent a crash?

Yes previous error handling provided more context. Changed to catch and throw a more descriptive exception. It can be up to callers whether to crash in this case (which probably indicates a bug not a runtime error) and crashing in the examples seems ok

mp::EventLoop loop("mpcalculator", LogPrint);
std::unique_ptr<Init> init = std::make_unique<InitImpl>();
mp::ServeStream<InitInterface>(loop, fd, *init);
mp::ServeStream<InitInterface>(loop, mp::MakeStream(loop.m_io_context, socket), *init);
loop.loop();
return 0;
}
8 changes: 4 additions & 4 deletions example/example.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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::SpawnConnectInfo 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, mp::MakeStream(loop.m_io_context, socket)), pid);
}

static void LogPrint(mp::LogMessage log_data)
Expand Down
13 changes: 4 additions & 9 deletions example/printer.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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
{
Expand All @@ -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, mp::MakeStream(loop.m_io_context, socket), *init);
loop.loop();
return 0;
}
44 changes: 23 additions & 21 deletions include/mp/proxy-io.h
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
#include <cstdlib>
#include <functional>
#include <kj/function.h>
#include <kj/io.h>
#include <map>
#include <memory>
#include <optional>
Expand Down Expand Up @@ -213,6 +214,12 @@ class Logger

std::string LongThreadName(const char* exe_name);

//! Wrap a socket file descriptor as an async stream, taking ownership of the fd.
inline Stream MakeStream(kj::AsyncIoContext& io_context, SocketId socket)
{
return io_context.lowLevelProvider->wrapSocketFd(socket, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP);
}

//! Event loop implementation.
//!
//! Cap'n Proto threading model is very simple: all I/O operations are
Expand Down Expand Up @@ -311,11 +318,12 @@ class EventLoop
//! Callback functions to run on async thread.
std::optional<CleanupList> m_async_fns MP_GUARDED_BY(m_mutex);

//! Pipe read handle used to wake up the event loop thread.
int m_wait_fd = -1;
//! Socket pair used to post and wait for wakeups to the event loop thread.
kj::Own<kj::AsyncIoStream> m_wait_stream;
kj::Own<kj::AsyncIoStream> m_post_stream;

//! Pipe write handle used to wake up the event loop thread.
int m_post_fd = -1;
//! Synchronous writer used to write to m_post_stream.
kj::Own<kj::OutputStream> m_post_writer;

//! Number of EventLoopRef instances referencing this event loop. This is a
//! sum of the number of client and server objects (Connection, ProxyClient,
Expand Down Expand Up @@ -451,7 +459,7 @@ class Connection
//! Capability::Client handles owned by ProxyClient objects), then schedules
//! asynchronous cleanup functions to run in a worker thread (to run
//! destructors of m_impl instances owned by ProxyServer objects).
~Connection();
~Connection() noexcept(false);

//! Register synchronous cleanup function to run on event loop thread (with
//! access to capnp thread local variables) when disconnect() is called.
Expand Down Expand Up @@ -824,17 +832,15 @@ kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
return ret;
}

//! Given stream file descriptor, make a new ProxyClient object to send requests
//! over the stream. Also create a new Connection object embedded in the
//! client that is freed when the client is closed.
//! Given a stream, make a new ProxyClient object to send requests over it.
//! Also create a new Connection object embedded in the client that is freed
//! when the client is closed.
template <typename InitInterface>
std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, int fd)
std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, Stream stream)
{
typename InitInterface::Client init_client(nullptr);
std::unique_ptr<Connection> connection;
loop.sync([&] {
auto stream =
loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP);
connection = std::make_unique<Connection>(loop, kj::mv(stream));
init_client = connection->m_rpc_system->bootstrap(ServerVatId().vat_id).castAs<InitInterface>();
Connection* connection_ptr = connection.get();
Expand Down Expand Up @@ -905,22 +911,18 @@ void _Listen(const std::shared_ptr<Listener>& listener, EventLoop& loop, InitImp
}));
}

//! Given stream file descriptor and an init object, handle requests on the
//! stream by calling methods on the Init object.
//! Given a stream and an init object, handle requests on the stream by calling
//! methods on the Init object.
template <typename InitInterface, typename InitImpl>
void ServeStream(EventLoop& loop, int fd, InitImpl& init)
void ServeStream(EventLoop& loop, Stream stream, InitImpl& init)
{
_Serve<InitInterface>(
loop,
loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
init,
[] {});
_Serve<InitInterface>(loop, kj::mv(stream), init, [] {});
}

//! Given listening socket file descriptor and an init object, handle incoming
//! Given listening socket identifier and an init object, handle incoming
//! connections and requests by calling methods on the Init object.
template <typename InitInterface, typename InitImpl>
void ListenConnections(EventLoop& loop, int fd, InitImpl& init, std::optional<size_t> max_connections = std::nullopt)
void ListenConnections(EventLoop& loop, SocketId fd, InitImpl& init, std::optional<size_t> max_connections = std::nullopt)
{
loop.sync([&]() {
auto listener{std::make_shared<Listener>(
Expand Down
10 changes: 5 additions & 5 deletions include/mp/proxy.h
Original file line number Diff line number Diff line change
Expand Up @@ -315,21 +315,21 @@ template <typename Field, int flags>
struct Accessor : public Field
{
//! Field is present from the Cap'n Proto Params struct (client -> server).
static const bool in = flags & FIELD_IN;
static constexpr bool in = (flags & FIELD_IN) != 0;
//! Field is present from the Cap'n Proto Results struct (server -> client).
static const bool out = flags & FIELD_OUT;
static constexpr bool out = (flags & FIELD_OUT) != 0;
//! Field has a companion has{Name} boolean field in the Cap'n Proto struct.
//! This is used to represent optional primitive values (e.g. C++
//! std::optional<int>) because Cap'n Proto doesn't allow primitive fields to
//! be unset.
static const bool optional = flags & FIELD_OPTIONAL;
static constexpr bool optional = (flags & FIELD_OPTIONAL) != 0;
//! Results field has a companion want{Name} boolean field in the Params
//! struct. Used for optional output parameters (e.g. C++ int*) and set to
//! true if the caller passed a non-null pointer and wants the result.
static const bool requested = flags & FIELD_REQUESTED;
static constexpr bool requested = (flags & FIELD_REQUESTED) != 0;
//! Field is a Cap'n Proto pointer type (struct, list, text, data,
//! interface) as opposed to a primitive type (bool, int, float, enum).
static const bool boxed = flags & FIELD_BOXED;
static constexpr bool boxed = (flags & FIELD_BOXED) != 0;
};

//! Wrapper around std::function for passing std::function objects between client and servers.
Expand Down
4 changes: 2 additions & 2 deletions include/mp/type-context.h
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,11 @@

namespace mp {
template <typename Output>
requires FieldTypeIs<Output, Context::Builder>
void CustomBuildField(TypeList<>,
Priority<1>,
ClientInvokeContext& invoke_context,
Output&& output,
typename std::enable_if<std::is_same<decltype(output.get()), Context::Builder>::value>::type* enable = nullptr)
Output&& output)
{
auto& connection = invoke_context.connection;
auto& thread_context = invoke_context.thread_context;
Expand Down
20 changes: 10 additions & 10 deletions include/mp/type-interface.h
Original file line number Diff line number Diff line change
Expand Up @@ -21,12 +21,12 @@ kj::Own<typename Interface::Server> CustomMakeProxyServer(InvokeContext& context
}

template <typename Impl, typename Value, typename Output>
requires InterfaceField<Output>
void CustomBuildField(TypeList<std::unique_ptr<Impl>>,
Priority<1>,
InvokeContext& invoke_context,
Value&& value,
Output&& output,
typename Decay<decltype(output.get())>::Calls* enable = nullptr)
Output&& output)
{
if (value) {
using Interface = typename decltype(output.get())::Calls;
Expand All @@ -35,12 +35,12 @@ void CustomBuildField(TypeList<std::unique_ptr<Impl>>,
}

template <typename Impl, typename Value, typename Output>
requires InterfaceField<Output>
void CustomBuildField(TypeList<std::shared_ptr<Impl>>,
Priority<2>,
InvokeContext& invoke_context,
Value&& value,
Output&& output,
typename Decay<decltype(output.get())>::Calls* enable = nullptr)
Output&& output)
{
if (value) {
using Interface = typename decltype(output.get())::Calls;
Expand All @@ -49,12 +49,12 @@ void CustomBuildField(TypeList<std::shared_ptr<Impl>>,
}

template <typename Impl, typename Output>
requires InterfaceField<Output>
void CustomBuildField(TypeList<Impl&>,
Priority<1>,
InvokeContext& invoke_context,
Impl& value,
Output&& output,
typename decltype(output.get())::Calls* enable = nullptr)
Output&& output)
{
// Disable deleter so proxy server object doesn't attempt to delete the
// wrapped implementation when the proxy client is destroyed or
Expand All @@ -77,12 +77,12 @@ std::unique_ptr<Impl> CustomMakeProxyClient(InvokeContext& context, typename Int
}

template <typename LocalType, typename Input, typename ReadDest>
requires InterfaceField<Input>
decltype(auto) CustomReadField(TypeList<std::unique_ptr<LocalType>>,
Priority<1>,
InvokeContext& invoke_context,
Input&& input,
ReadDest&& read_dest,
typename Decay<decltype(input.get())>::Calls* enable = nullptr)
ReadDest&& read_dest)
{
using Interface = typename Decay<decltype(input.get())>::Calls;
if (CustomHasField(TypeList<LocalType>(), invoke_context, input)) {
Expand All @@ -93,12 +93,12 @@ decltype(auto) CustomReadField(TypeList<std::unique_ptr<LocalType>>,
}

template <typename LocalType, typename Input, typename ReadDest>
requires InterfaceField<Input>
decltype(auto) CustomReadField(TypeList<std::shared_ptr<LocalType>>,
Priority<1>,
InvokeContext& invoke_context,
Input&& input,
ReadDest&& read_dest,
typename Decay<decltype(input.get())>::Calls* enable = nullptr)
ReadDest&& read_dest)
{
using Interface = typename Decay<decltype(input.get())>::Calls;
if (CustomHasField(TypeList<LocalType>(), invoke_context, input)) {
Expand Down
Loading
Loading