Merge commit '707d0ded84563386f770ec17970834c65f8fa938' into pr/subtree-12

src/ipc/libmultiprocess/doc/usage.md

@@ -10,7 +10,7 @@ _libmultiprocess_ is a library and code generator that allows calling C++ class
 
 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.
 
-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.
+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.
 
 ## Example
 

src/ipc/libmultiprocess/doc/versions.md

@@ -7,8 +7,24 @@ Library versions are tracked with simple
 Versioning policy is described in the [version.h](../include/mp/version.h)
 include.
 
-## v11
+## v12
 - Current unstable version.
+- 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.
+
+## [v11.0](https://github.com/bitcoin-core/libmultiprocess/commits/v11.0)
+- 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)).
+- 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)).
+- 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)).
+- Produces more readable log output for Proxy object lifecycle events and IPC server-side failures ([#218](https://github.com/bitcoin-core/libmultiprocess/pull/218)).
+- 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)).
+- 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)).
+- 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)).
+- Fixes a race condition in a log print detected by TSan ([#286](https://github.com/bitcoin-core/libmultiprocess/pull/286)).
+- 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)).
+- 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)).
+- Used in Bitcoin Core master branch, pulled in by [#35661](https://github.com/bitcoin/bitcoin/pull/35661).
 
 ## [v10.0](https://github.com/bitcoin-core/libmultiprocess/commits/v10.0)
 - Increases spawn test timeout to avoid spurious failures.

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

@@ -13,7 +13,9 @@
 #include <capnp/rpc-twoparty.h>
 
 #include <assert.h>
+#include <algorithm>
 #include <condition_variable>
+#include <cstdlib>
 #include <functional>
 #include <kj/function.h>
 #include <map>
@@ -25,6 +27,7 @@
 
 namespace mp {
 struct ThreadContext;
+struct Listener;
 
 struct InvokeContext
 {
@@ -360,6 +363,12 @@ public:
 
     //! Hook called on the worker thread just before returning results.
     std::function<void()> testing_hook_async_request_done;
+
+    //! Hook called on the event loop thread when a client has connected.
+    std::function<void()> testing_hook_connected;
+
+    //! Hook called on the event loop thread when a client has disconnected.
+    std::function<void()> testing_hook_disconnected;
 };
 
 //! 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
 //! handles requests from the stream by calling the init object. Embed the
 //! ProxyServer in a Connection object that is stored and erased if
 //! disconnected. This should be called from the event loop thread.
-template <typename InitInterface, typename InitImpl>
-void _Serve(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream, InitImpl& init)
+template <typename InitInterface, typename InitImpl, typename OnDisconnect>
+void _Serve(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream, InitImpl& init, OnDisconnect&& on_disconnect)
 {
     loop.m_incoming_connections.emplace_front(loop, kj::mv(stream), [&](Connection& connection) {
         // 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
     });
     auto it = loop.m_incoming_connections.begin();
     MP_LOG(loop, Log::Info) << "IPC server: socket connected.";
-    it->onDisconnect([&loop, it] {
+    if (loop.testing_hook_connected) loop.testing_hook_connected();
+    it->onDisconnect([&loop, it, on_disconnect = std::forward<OnDisconnect>(on_disconnect)]() mutable {
         MP_LOG(loop, Log::Info) << "IPC server: socket disconnected.";
         loop.m_incoming_connections.erase(it);
+        on_disconnect();
+        if (loop.testing_hook_disconnected) loop.testing_hook_disconnected();
     });
 }
 
-//! Given connection receiver and an init object, handle incoming connections by
-//! calling _Serve, to create ProxyServer objects and forward requests to the
-//! init object.
-template <typename InitInterface, typename InitImpl>
-void _Listen(EventLoop& loop, kj::Own<kj::ConnectionReceiver>&& listener, InitImpl& init)
+struct Listener
 {
-    auto* ptr = listener.get();
-    loop.m_task_set->add(ptr->accept().then(
-        [&loop, &init, listener = kj::mv(listener)](kj::Own<kj::AsyncIoStream>&& stream) mutable {
-            _Serve<InitInterface>(loop, kj::mv(stream), init);
-            _Listen<InitInterface>(loop, kj::mv(listener), init);
+    explicit Listener(kj::Own<kj::ConnectionReceiver>&& receiver, std::optional<size_t> max_connections)
+        : m_receiver(kj::mv(receiver)), m_max_connections(max_connections) {}
+
+    bool atCapacity() const
+    {
+        return m_max_connections && m_active_connections >= *m_max_connections;
+    }
+
+    kj::Own<kj::ConnectionReceiver> m_receiver;
+    std::optional<size_t> m_max_connections;
+    size_t m_active_connections{0};
+};
+
+template <typename InitInterface, typename InitImpl>
+void _Listen(const std::shared_ptr<Listener>& listener, EventLoop& loop, InitImpl& init)
+{
+    if (listener->atCapacity()) return;
+
+    auto* receiver = listener->m_receiver.get();
+    loop.m_task_set->add(receiver->accept().then(
+        [&loop, &init, listener](kj::Own<kj::AsyncIoStream>&& stream) {
+            ++listener->m_active_connections;
+            _Serve<InitInterface>(loop, kj::mv(stream), init, [&loop, &init, listener] {
+                const bool resume_accept{listener->atCapacity()};
+                assert(listener->m_active_connections > 0);
+                --listener->m_active_connections;
+                if (resume_accept) _Listen<InitInterface>(listener, loop, init);
+            });
+            _Listen<InitInterface>(listener, loop, init);
         }));
 }
 
@@ -879,18 +911,22 @@ template <typename InitInterface, typename InitImpl>
 void ServeStream(EventLoop& loop, int fd, InitImpl& init)
 {
     _Serve<InitInterface>(
-        loop, loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP), init);
+        loop,
+        loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
+        init,
+        [] {});
 }
 
 //! Given listening socket file descriptor 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)
+void ListenConnections(EventLoop& loop, int fd, InitImpl& init, std::optional<size_t> max_connections = std::nullopt)
 {
     loop.sync([&]() {
-        _Listen<InitInterface>(loop,
+        auto listener{std::make_shared<Listener>(
             loop.m_io_context.lowLevelProvider->wrapListenSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
-            init);
+            max_connections)};
+        _Listen<InitInterface>(listener, loop, init);
     });
 }
 

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

@@ -24,7 +24,7 @@
 //! pointing at the prior merge commit. The /doc/versions.md file should also be
 //! updated, noting any significant or incompatible changes made since the
 //! previous version.
-#define MP_MAJOR_VERSION 11
+#define MP_MAJOR_VERSION 12
 
 //! Minor version number. Should be incremented in stable branches after
 //! backporting changes. The /doc/versions.md file should also be updated to

src/ipc/libmultiprocess/test/CMakeLists.txt

@@ -26,6 +26,7 @@ if(BUILD_TESTING AND TARGET CapnProto::kj-test)
     ${MP_PROXY_HDRS}
     mp/test/foo-types.h
     mp/test/foo.h
+    mp/test/listen_tests.cpp
     mp/test/spawn_tests.cpp
     mp/test/test.cpp
   )

src/ipc/libmultiprocess/test/mp/test/listen_tests.cpp

@@ -0,0 +1,294 @@
+// 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.
+
+#include <cassert>
+#include <filesystem>
+#include <mp/test/foo.capnp.h>
+#include <mp/test/foo.capnp.proxy.h>
+
+#include <chrono>
+#include <condition_variable>
+#include <cstdlib>
+#include <cstring>
+#include <future>
+#include <functional>
+#include <kj/async.h>
+#include <kj/common.h>
+#include <kj/debug.h>
+#include <kj/memory.h>
+#include <kj/test.h>
+#include <memory>
+#include <mp/proxy.h>
+#include <mp/proxy-io.h>
+#include <mp/util.h>
+#include <ratio> // IWYU pragma: keep
+#include <optional>
+#include <stdexcept>
+#include <string>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <thread>
+#include <unistd.h>
+
+namespace mp {
+namespace test {
+namespace {
+
+constexpr auto FAILURE_TIMEOUT = std::chrono::seconds{30};
+
+//! Owns a temporary Unix-domain listening socket used by ListenSetup. Tests call
+//! Connect() to create client socket FDs and release() to transfer the listening
+//! FD to ListenConnections().
+class UnixListener
+{
+public:
+    UnixListener()
+    {
+        std::string dir_template = (std::filesystem::temp_directory_path() / "mptest-listener-XXXXXX").string();
+        char* dir = mkdtemp(dir_template.data());
+        KJ_REQUIRE(dir != nullptr);
+        m_dir = dir;
+        m_path = m_dir + "/socket";
+
+        m_fd = socket(AF_UNIX, SOCK_STREAM, 0);
+        KJ_REQUIRE(m_fd >= 0);
+
+        sockaddr_un addr{};
+        addr.sun_family = AF_UNIX;
+        KJ_REQUIRE(m_path.size() < sizeof(addr.sun_path));
+        std::strncpy(addr.sun_path, m_path.c_str(), sizeof(addr.sun_path) - 1);
+        KJ_REQUIRE(bind(m_fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) == 0);
+        KJ_REQUIRE(listen(m_fd, SOMAXCONN) == 0);
+    }
+
+    ~UnixListener()
+    {
+        if (m_fd >= 0) close(m_fd);
+        if (!m_path.empty()) unlink(m_path.c_str());
+        if (!m_dir.empty()) rmdir(m_dir.c_str());
+    }
+
+    int release()
+    {
+        assert(m_fd >= 0);
+        int fd = m_fd;
+        m_fd = -1;
+        return fd;
+    }
+
+    int MakeConnectedSocket() const
+    {
+        int fd = socket(AF_UNIX, SOCK_STREAM, 0);
+        KJ_REQUIRE(fd >= 0);
+
+        sockaddr_un addr{};
+        addr.sun_family = AF_UNIX;
+        KJ_REQUIRE(m_path.size() < sizeof(addr.sun_path));
+        std::strncpy(addr.sun_path, m_path.c_str(), sizeof(addr.sun_path) - 1);
+        KJ_REQUIRE(connect(fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)) == 0);
+        return fd;
+    }
+
+private:
+    int m_fd{-1};
+    std::string m_dir;
+    std::string m_path;
+};
+
+//! Runs a client EventLoop on its own thread and connects one socket FD to the
+//! server. The constructed ProxyClient can be used by the test thread to make
+//! calls over that connection.
+class ClientSetup
+{
+public:
+    explicit ClientSetup(int fd)
+        : thread([this, fd] {
+              EventLoop loop("mptest-client", [](mp::LogMessage log) {
+                  KJ_LOG(INFO, log.level, log.message);
+                  if (log.level == mp::Log::Raise) throw std::runtime_error(log.message);
+              });
+              client_promise.set_value(ConnectStream<messages::FooInterface>(loop, fd));
+              loop.loop();
+          })
+    {
+        client = client_promise.get_future().get();
+    }
+
+    ~ClientSetup()
+    {
+        client.reset();
+        thread.join();
+    }
+
+    std::promise<std::unique_ptr<ProxyClient<messages::FooInterface>>> client_promise;
+    std::unique_ptr<ProxyClient<messages::FooInterface>> client;
+
+    //! Thread variable should be after other struct members so the thread does
+    //! not start until the other members are initialized.
+    std::thread thread;
+};
+
+//! Runs a server EventLoop on its own thread, starts ListenConnections() on a
+//! UnixListener socket, and records connection/disconnection counts through
+//! EventLoop test hooks
+class ListenSetup
+{
+public:
+    explicit ListenSetup(std::optional<size_t> max_connections = std::nullopt)
+        : thread([this, max_connections] {
+              EventLoop loop("mptest-server", [](mp::LogMessage log) {
+                  KJ_LOG(INFO, log.level, log.message);
+                  if (log.level == mp::Log::Raise) throw std::runtime_error(log.message);
+              });
+              loop.testing_hook_disconnected = [&] {
+                  Lock lock(counter_mutex);
+                  ++disconnected_count;
+                  counter_cv.notify_all();
+              };
+              loop.testing_hook_connected = [&] {
+                  Lock lock(counter_mutex);
+                  ++connected_count;
+                  counter_cv.notify_all();
+              };
+              m_loop_ref.emplace(loop);
+              FooImplementation foo;
+              ListenConnections<messages::FooInterface>(loop, listener.release(), foo, max_connections);
+              ready_promise.set_value();
+              loop.loop();
+          })
+    {
+        ready_promise.get_future().get();
+    }
+
+    ~ListenSetup()
+    {
+        m_loop_ref.reset();
+        thread.join();
+    }
+
+    size_t ConnectedCount()
+    {
+        Lock lock(counter_mutex);
+        return connected_count;
+    }
+
+    size_t DisconnectedCount()
+    {
+        Lock lock(counter_mutex);
+        return disconnected_count;
+    }
+
+    void WaitForConnectedCount(size_t expected_count)
+    {
+        Lock lock(counter_mutex);
+        const auto deadline = std::chrono::steady_clock::now() + FAILURE_TIMEOUT;
+        const bool matched = counter_cv.wait_until(lock.m_lock, deadline, [&]() MP_REQUIRES(counter_mutex) {
+            return connected_count >= expected_count;
+        });
+        KJ_REQUIRE(matched);
+    }
+
+    void WaitForDisconnectedCount(size_t expected_count)
+    {
+        Lock lock(counter_mutex);
+        const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
+        const bool matched = counter_cv.wait_until(lock.m_lock, deadline, [&]() MP_REQUIRES(counter_mutex) {
+            return disconnected_count >= expected_count;
+        });
+        KJ_REQUIRE(matched);
+    }
+
+    UnixListener listener;
+    std::promise<void> ready_promise;
+    std::optional<EventLoopRef> m_loop_ref;
+    Mutex counter_mutex;
+    std::condition_variable counter_cv;
+    size_t connected_count MP_GUARDED_BY(counter_mutex) {0};
+    size_t disconnected_count MP_GUARDED_BY(counter_mutex) {0};
+    //! Thread variable should be after other struct members so the thread does
+    //! not start until the other members are initialized.
+    std::thread thread;
+};
+
+KJ_TEST("ListenConnections accepts incoming connections")
+{
+    ListenSetup server;
+    KJ_EXPECT(server.ConnectedCount() == 0)
+    auto client = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
+
+    server.WaitForConnectedCount(1);
+    KJ_EXPECT(client->client->add(1, 2) == 3);
+}
+
+KJ_TEST("ListenConnections enforces a local connection limit")
+{
+    // With max-connections=1, the second socket can connect to the kernel
+    // backlog, but ListenConnections should not accept or serve it until the
+    // first accepts clients disconnects.
+
+    ListenSetup server(/*max_connections=*/1);
+
+    KJ_EXPECT(server.ConnectedCount() == 0)
+    auto client1 = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
+    server.WaitForConnectedCount(1);
+
+    KJ_EXPECT(client1->client->add(1, 2) == 3);
+
+    auto client2 = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
+    // Without this sync, ConnectedCount() == 1 might pass even if
+    // max_connections was not enforced because the event loop has not accepted
+    // client2 yet.
+    (**server.m_loop_ref).sync([] {});
+
+    KJ_EXPECT(server.ConnectedCount() == 1);
+    KJ_EXPECT(server.DisconnectedCount() == 0);
+    client1.reset();
+    server.WaitForDisconnectedCount(1);
+    server.WaitForConnectedCount(2);
+
+    KJ_EXPECT(client2->client->add(2, 3) == 5);
+
+    KJ_EXPECT(server.DisconnectedCount() == 1);
+    client2.reset();
+    server.WaitForDisconnectedCount(2);
+
+    KJ_EXPECT(server.DisconnectedCount() == 2);
+    auto client3 = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
+    server.WaitForConnectedCount(3);
+    KJ_EXPECT(client3->client->add(3, 4) == 7);
+}
+
+KJ_TEST("ListenConnections accepts multiple connections")
+{
+    // With max-connections=2, two clients should be accepted and usable at the
+    // same time, while a third waits until one active client disconnects.
+
+    ListenSetup server(/*max_connections=*/2);
+
+    KJ_EXPECT(server.ConnectedCount() == 0);
+    auto client1 = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
+    auto client2 = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
+    server.WaitForConnectedCount(2);
+
+    KJ_EXPECT(client1->client->add(1, 2) == 3);
+    KJ_EXPECT(client2->client->add(2, 3) == 5);
+
+    auto client3 = std::make_unique<ClientSetup>(server.listener.MakeConnectedSocket());
+    // Without this sync, ConnectedCount() == 2 might pass even if
+    // max_connections was not enforced because the event loop has not accepted
+    // client3 yet.
+    (**server.m_loop_ref).sync([] {});
+
+    KJ_EXPECT(server.ConnectedCount() == 2);
+    KJ_EXPECT(server.DisconnectedCount() == 0);
+    client1.reset();
+    server.WaitForDisconnectedCount(1);
+    server.WaitForConnectedCount(3);
+
+    KJ_EXPECT(client3->client->add(3, 4) == 7);
+}
+
+} // namespace
+} // namespace test
+} // namespace mp