Skip to content

Commit daee541

Browse files
committed
Merge bitcoin/bitcoin#35684: Update libmultiprocess subtree to add max_connections option
707d0de Squashed 'src/ipc/libmultiprocess/' changes from 16bf05dea02..28e056576a3 (Ryan Ofsky) Pull request description: The changes can be verified by running `test/lint/git-subtree-check.sh src/ipc/libmultiprocess` as described in [developer notes](https://github.com/bitcoin/bitcoin/blob/master/doc/developer-notes.md#subtrees) and [lint instructions](https://github.com/bitcoin/bitcoin/tree/master/test/lint#git-subtree-checksh). Change since last subtree update (#35661): - Adds an optional `max_connections` parameter to `ListenConnections` ([#269](bitcoin-core/libmultiprocess#269)) This is needed for bitcoin/bitcoin#35037 which lets the maximum number of IPC of incoming connections be configured in bitcoin core. ACKs for top commit: sedited: ACK a9d1b652f324126ef7e80d9ab0b9e4f60019dade Tree-SHA512: 1c3ec5c4eb98717c7414a32a3faf63e551b402f7318146745d840fb0e80cbd9e5006892476eb0869eb659cc551ac26d7a3fb2e43509883fb276ac0639a2a7c79
2 parents d911313 + 21ded7f commit daee541

6 files changed

Lines changed: 366 additions & 19 deletions

File tree

‎src/ipc/libmultiprocess/doc/usage.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ _libmultiprocess_ is a library and code generator that allows calling C++ class
1010

1111
The `*.capnp` data definition files are consumed by the _libmultiprocess_ code generator and each `X.capnp` file generates `X.capnp.c++`, `X.capnp.h`, `X.capnp.proxy-client.c++`, `X.capnp.proxy-server.c++`, `X.capnp.proxy-types.c++`, `X.capnp.proxy-types.h`, and `X.capnp.proxy.h` output files. The generated files include `mp::ProxyClient<Interface>` and `mp::ProxyServer<Interface>` class specializations for all the interfaces in the `.capnp` files. These allow methods on C++ objects in one process to be called from other processes over IPC sockets.
1212

13-
The `ProxyServer` objects help translate IPC requests from a socket to method calls on a local object. The `ProxyServer` objects are just used internally by the `mp::ServeStream(loop, socket, wrapped_object)` and `mp::ListenConnections(loop, socket, wrapped_object)` functions, and aren't exposed externally. The `ProxyClient` classes are exposed, and returned from the `mp::ConnectStream(loop, socket)` function and meant to be used directly. The classes implement methods described in `.capnp` definitions, and whenever any method is called, a request with the method arguments is sent over the associated IPC connection, and the corresponding `wrapped_object` method on the other end of the connection is called, with the `ProxyClient` method blocking until it returns and forwarding back any return value to the `ProxyClient` method caller.
13+
The `ProxyServer` objects help translate IPC requests from a socket to method calls on a local object. The `ProxyServer` objects are just used internally by the `mp::ServeStream(loop, socket, wrapped_object)` and `mp::ListenConnections(loop, socket, wrapped_object[, max_connections])` functions, and aren't exposed externally. The `ProxyClient` classes are exposed, and returned from the `mp::ConnectStream(loop, socket)` function and meant to be used directly. The classes implement methods described in `.capnp` definitions, and whenever any method is called, a request with the method arguments is sent over the associated IPC connection, and the corresponding `wrapped_object` method on the other end of the connection is called, with the `ProxyClient` method blocking until it returns and forwarding back any return value to the `ProxyClient` method caller.
1414

1515
## Example
1616

‎src/ipc/libmultiprocess/doc/versions.md‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,24 @@ Library versions are tracked with simple
77
Versioning policy is described in the [version.h](../include/mp/version.h)
88
include.
99

10-
## v11
10+
## v12
1111
- Current unstable version.
12+
- Adds an optional per-listener `max_connections` parameter to `ListenConnections()`
13+
so servers can stop accepting new connections when a local connection cap is reached,
14+
and resume accepting after existing connections disconnect.
15+
16+
## [v11.0](https://github.com/bitcoin-core/libmultiprocess/commits/v11.0)
17+
- Adds `makePool` method on `ThreadMap` to support thread pool routing, allowing requests without a specific client thread to be dispatched to a pool using a shortest-queue strategy ([#283](https://github.com/bitcoin-core/libmultiprocess/pull/283)).
18+
- Adds `std::unordered_set` support, a `BuildList` helper, and a `ReadList` helper to reduce duplication in list build and read handlers ([#277](https://github.com/bitcoin-core/libmultiprocess/pull/277), [#285](https://github.com/bitcoin-core/libmultiprocess/pull/285)).
19+
- Adds support for translating C++ `std::optional<T>` struct fields to pairs of `T` + `hasT :Bool` Cap'n Proto struct fields, allowing unset optional primitive fields to be represented ([#243](https://github.com/bitcoin-core/libmultiprocess/pull/243)).
20+
- Produces more readable log output for Proxy object lifecycle events and IPC server-side failures ([#218](https://github.com/bitcoin-core/libmultiprocess/pull/218)).
21+
- Handles exceptions thrown by `destroy` methods by logging instead of aborting ([#273](https://github.com/bitcoin-core/libmultiprocess/pull/273)). This can prevent server crashes when non-libmultiprocess clients disconnect without destroying objects, in the case where a server object owns client objects and the server destructor tries to call the disconnected client to free them ([#219](https://github.com/bitcoin-core/libmultiprocess/issues/219)).
22+
- Handles unexpected exceptions thrown by callbacks (that should never happen) by logging errors instead of deadlocking ([#260](https://github.com/bitcoin-core/libmultiprocess/pull/260)).
23+
- Fixes a rare mptest hang on musl builds caused by a lost wakeup bug in `Waiter` ([#295](https://github.com/bitcoin-core/libmultiprocess/pull/295)).
24+
- Fixes a race condition in a log print detected by TSan ([#286](https://github.com/bitcoin-core/libmultiprocess/pull/286)).
25+
- Build improvements: makes `target_capnp_sources` work correctly when libmultiprocess is used as a CMake subproject ([#289](https://github.com/bitcoin-core/libmultiprocess/pull/289)), adds `mp_headers` target for better lint tool support ([#291](https://github.com/bitcoin-core/libmultiprocess/pull/291)), and fixes compatibility with recent Nix and CMake 4.0 ([#238](https://github.com/bitcoin-core/libmultiprocess/pull/238)).
26+
- Test, CI, documentation, and minor code improvements: design document corrections ([#278](https://github.com/bitcoin-core/libmultiprocess/pull/278)), field constant comments ([#279](https://github.com/bitcoin-core/libmultiprocess/pull/279)), clang-tidy fix ([#292](https://github.com/bitcoin-core/libmultiprocess/pull/292)), new smoke test for double-precision float values ([#294](https://github.com/bitcoin-core/libmultiprocess/pull/294)), new test for recursive async IPC calls ([#301](https://github.com/bitcoin-core/libmultiprocess/pull/301)), removal of libevent from Core CI builds ([#299](https://github.com/bitcoin-core/libmultiprocess/pull/299)), and rename of `EventLoop::m_num_clients` to `m_num_refs` ([#302](https://github.com/bitcoin-core/libmultiprocess/pull/302)).
27+
- Used in Bitcoin Core master branch, pulled in by [#35661](https://github.com/bitcoin/bitcoin/pull/35661).
1228

1329
## [v10.0](https://github.com/bitcoin-core/libmultiprocess/commits/v10.0)
1430
- Increases spawn test timeout to avoid spurious failures.

‎src/ipc/libmultiprocess/include/mp/proxy-io.h‎

Lines changed: 52 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,9 @@
1313
#include <capnp/rpc-twoparty.h>
1414

1515
#include <assert.h>
16+
#include <algorithm>
1617
#include <condition_variable>
18+
#include <cstdlib>
1719
#include <functional>
1820
#include <kj/function.h>
1921
#include <map>
@@ -25,6 +27,7 @@
2527

2628
namespace mp {
2729
struct ThreadContext;
30+
struct Listener;
2831

2932
struct InvokeContext
3033
{
@@ -360,6 +363,12 @@ class EventLoop
360363

361364
//! Hook called on the worker thread just before returning results.
362365
std::function<void()> testing_hook_async_request_done;
366+
367+
//! Hook called on the event loop thread when a client has connected.
368+
std::function<void()> testing_hook_connected;
369+
370+
//! Hook called on the event loop thread when a client has disconnected.
371+
std::function<void()> testing_hook_disconnected;
363372
};
364373

365374
//! Single element task queue used to handle recursive capnp calls. (If the
@@ -842,8 +851,8 @@ std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, int f
842851
//! handles requests from the stream by calling the init object. Embed the
843852
//! ProxyServer in a Connection object that is stored and erased if
844853
//! disconnected. This should be called from the event loop thread.
845-
template <typename InitInterface, typename InitImpl>
846-
void _Serve(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream, InitImpl& init)
854+
template <typename InitInterface, typename InitImpl, typename OnDisconnect>
855+
void _Serve(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream, InitImpl& init, OnDisconnect&& on_disconnect)
847856
{
848857
loop.m_incoming_connections.emplace_front(loop, kj::mv(stream), [&](Connection& connection) {
849858
// Disable deleter so proxy server object doesn't attempt to delete the
@@ -853,23 +862,46 @@ void _Serve(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream, InitImpl& init
853862
});
854863
auto it = loop.m_incoming_connections.begin();
855864
MP_LOG(loop, Log::Info) << "IPC server: socket connected.";
856-
it->onDisconnect([&loop, it] {
865+
if (loop.testing_hook_connected) loop.testing_hook_connected();
866+
it->onDisconnect([&loop, it, on_disconnect = std::forward<OnDisconnect>(on_disconnect)]() mutable {
857867
MP_LOG(loop, Log::Info) << "IPC server: socket disconnected.";
858868
loop.m_incoming_connections.erase(it);
869+
on_disconnect();
870+
if (loop.testing_hook_disconnected) loop.testing_hook_disconnected();
859871
});
860872
}
861873

862-
//! Given connection receiver and an init object, handle incoming connections by
863-
//! calling _Serve, to create ProxyServer objects and forward requests to the
864-
//! init object.
874+
struct Listener
875+
{
876+
explicit Listener(kj::Own<kj::ConnectionReceiver>&& receiver, std::optional<size_t> max_connections)
877+
: m_receiver(kj::mv(receiver)), m_max_connections(max_connections) {}
878+
879+
bool atCapacity() const
880+
{
881+
return m_max_connections && m_active_connections >= *m_max_connections;
882+
}
883+
884+
kj::Own<kj::ConnectionReceiver> m_receiver;
885+
std::optional<size_t> m_max_connections;
886+
size_t m_active_connections{0};
887+
};
888+
865889
template <typename InitInterface, typename InitImpl>
866-
void _Listen(EventLoop& loop, kj::Own<kj::ConnectionReceiver>&& listener, InitImpl& init)
890+
void _Listen(const std::shared_ptr<Listener>& listener, EventLoop& loop, InitImpl& init)
867891
{
868-
auto* ptr = listener.get();
869-
loop.m_task_set->add(ptr->accept().then(
870-
[&loop, &init, listener = kj::mv(listener)](kj::Own<kj::AsyncIoStream>&& stream) mutable {
871-
_Serve<InitInterface>(loop, kj::mv(stream), init);
872-
_Listen<InitInterface>(loop, kj::mv(listener), init);
892+
if (listener->atCapacity()) return;
893+
894+
auto* receiver = listener->m_receiver.get();
895+
loop.m_task_set->add(receiver->accept().then(
896+
[&loop, &init, listener](kj::Own<kj::AsyncIoStream>&& stream) {
897+
++listener->m_active_connections;
898+
_Serve<InitInterface>(loop, kj::mv(stream), init, [&loop, &init, listener] {
899+
const bool resume_accept{listener->atCapacity()};
900+
assert(listener->m_active_connections > 0);
901+
--listener->m_active_connections;
902+
if (resume_accept) _Listen<InitInterface>(listener, loop, init);
903+
});
904+
_Listen<InitInterface>(listener, loop, init);
873905
}));
874906
}
875907

@@ -879,18 +911,22 @@ template <typename InitInterface, typename InitImpl>
879911
void ServeStream(EventLoop& loop, int fd, InitImpl& init)
880912
{
881913
_Serve<InitInterface>(
882-
loop, loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP), init);
914+
loop,
915+
loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
916+
init,
917+
[] {});
883918
}
884919

885920
//! Given listening socket file descriptor and an init object, handle incoming
886921
//! connections and requests by calling methods on the Init object.
887922
template <typename InitInterface, typename InitImpl>
888-
void ListenConnections(EventLoop& loop, int fd, InitImpl& init)
923+
void ListenConnections(EventLoop& loop, int fd, InitImpl& init, std::optional<size_t> max_connections = std::nullopt)
889924
{
890925
loop.sync([&]() {
891-
_Listen<InitInterface>(loop,
926+
auto listener{std::make_shared<Listener>(
892927
loop.m_io_context.lowLevelProvider->wrapListenSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
893-
init);
928+
max_connections)};
929+
_Listen<InitInterface>(listener, loop, init);
894930
});
895931
}
896932

‎src/ipc/libmultiprocess/include/mp/version.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@
2424
//! pointing at the prior merge commit. The /doc/versions.md file should also be
2525
//! updated, noting any significant or incompatible changes made since the
2626
//! previous version.
27-
#define MP_MAJOR_VERSION 11
27+
#define MP_MAJOR_VERSION 12
2828

2929
//! Minor version number. Should be incremented in stable branches after
3030
//! backporting changes. The /doc/versions.md file should also be updated to

‎src/ipc/libmultiprocess/test/CMakeLists.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ if(BUILD_TESTING AND TARGET CapnProto::kj-test)
2626
${MP_PROXY_HDRS}
2727
mp/test/foo-types.h
2828
mp/test/foo.h
29+
mp/test/listen_tests.cpp
2930
mp/test/spawn_tests.cpp
3031
mp/test/test.cpp
3132
)

0 commit comments

Comments
 (0)