Add `makePool` method on `ThreadMap` #283

rustaceanrob commented at 2:37 PM on May 29, 2026: member

This patch introduces a pool of threads to the Connection class, and allows this pool to be populated with the thread map via makePool. When a client thread is not set in a request context, it is delegated to the pool. This is unable to handle the guarentees with server-invoked callbacks that the current API offers, but these callbacks are not yet present in the interface.

The pool is implemented as ~round-robin as it is simplest~ shortest queue, but perhaps the pool could be a queue of requests with work-stealing for threads that are available.

This was raised to me by Rust users, as they did not particularly care where work is executed on the server-side, but they have to set the thread regardless.

Tests in Rust can be seen here: https://github.com/2140-dev/bitcoin-capnp-types/pull/24

ref: https://github.com/2140-dev/bitcoin-capnp-types/blob/master/tests/util/bitcoin_core.rs#L149 ref: #281

DrahtBot commented at 2:37 PM on May 29, 2026: none

The following sections might be updated with supplementary metadata relevant to reviewers and maintainers.

Reviews

See the guideline for information on the review process.

Type	Reviewers
ACK	xyzconstant, ViniciusCestarii
Concept ACK	Eunovo
Approach ACK	ryanofsky

If your review is incorrectly listed, please copy-paste <code></code> into the comment that the bot should ignore.

Conflicts

Reviewers, this pull request conflicts with the following ones:

#288 (Create support branch for CI scripts, documentation, and examples by ryanofsky)

If you consider this pull request important, please also help to review the conflicting pull requests. Ideally, start with the one that should be merged first.

rustaceanrob renamed this:
~~Add `makePool` method on `ThreadMap`~~
WIP: Add `makePool` method on `ThreadMap`
on May 29, 2026

rustaceanrob force-pushed on May 29, 2026

rustaceanrob renamed this:
~~WIP: Add `makePool` method on `ThreadMap`~~
Add `makePool` method on `ThreadMap`
on May 29, 2026

rustaceanrob marked this as ready for review on May 29, 2026

in include/mp/proxy.capnp:51 in 158f30ddb2

  44 | @@ -45,6 +45,10 @@ interface ThreadMap $count(0) {
  45 |      # execute on. Clients create and name threads and pass the thread handle as
  46 |      # a call parameter.
  47 |      makeThread @0 (name :Text) -> (result :Thread);
  48 | +    # Pre-allocate a pool of server threads for implicit dispatch. When a
  49 | +    # request arrives with no context.thread set, the server round-robins
  50 | +    # across the pool instead of returning an error.
  51 | +    makePool @1 (count :UInt32) -> ();

ViniciusCestarii commented at 11:34 PM on May 29, 2026:

    makePool [@1](/bitcoin-core-multiprocess/contributor/1/) (name: Text, count :UInt32) -> ();

It would be usefull to have a name parameter like makeThread. Since each client may have its own thread pool, naming them would make log lines like "IPC server post request #{name} {pool/0}" attributable to a specific client

rustaceanrob commented at 9:06 AM on May 30, 2026:

Updated. How does that look?

ViniciusCestarii commented at 12:54 PM on June 1, 2026:

Nice, I just have this nit to make it more consistent with makeThread:

         const std::string thread_name = pool_name + "/pool/" + std::to_string(i);
         std::promise<ThreadContext*> thread_context;
         std::thread thread([&loop, &thread_context, thread_name]() {
-            g_thread_context.thread_name = ThreadName(loop.m_exe_name) + " (" + thread_name + ")";
+            g_thread_context.thread_name = ThreadName(loop.m_exe_name) + " (from " + thread_name + ")";
             g_thread_context.waiter = std::make_unique<Waiter>();
             Lock lock(g_thread_context.waiter->m_mutex);
             thread_context.set_value(&g_thread_context);

in src/mp/proxy.cpp:438 in 158f30ddb2 outdated

 433 | +        });
 434 | +        auto thread_server = kj::heap<ProxyServer<Thread>>(m_connection, *thread_context.get_future().get(), std::move(thread));
 435 | +        m_connection.m_thread_pool.push_back(m_connection.m_threads.add(kj::mv(thread_server)));
 436 | +    }
 437 | +    return kj::READY_NOW;
 438 | +}

ViniciusCestarii commented at 12:07 AM on May 30, 2026:

Repeated calls silently grow the pool beyond the intended size, and since naming is based on the loop index starting from 0 each time, you'd end up with duplicate names like two threads both called pool/0 making logs ambiguous.

I'm not sure what would be the correct approach but I thought about:

Return an error if m_connection.m_thread_pool is not empty
Replace the old pool by cleaning the m_connection.m_thread_pool and create a fresh one with the new count. This makes sense if the client needs to resize its dedicated pool.

rustaceanrob commented at 8:02 PM on June 1, 2026:

Added a guard statement:

    if (!m_connection.m_thread_pool.empty()) {
        throw std::runtime_error("makePool called on connection with existing pool");
    }

I think throwing is appropriate here as I would like to get user feedback first. If they anticipate pool resizing during runtime will be valuable we can add resizing APIs

rustaceanrob force-pushed on May 30, 2026

rustaceanrob force-pushed on Jun 1, 2026

ryanofsky commented at 9:43 PM on June 2, 2026: collaborator

Concept ACK. This is a good idea that should make development of rust & python clients easier, while making the behavior opt-in and not affecting c++ clients or existing clients using current behavior.

With this feature, we may also want to add a way to mark certain methods that can take long time before returning to not use the thread pool, so they do not block other work. These methods could create their own threads to run on, or require threads to be specified.

This is unable to handle the guarentees with server-invoked callbacks that the current API offers

Actually I think this is not an issue. If a server method is called with a callbackThread value it can still pass the same value back to the client when making the callback, no matter where the server thread is running.

perhaps the pool could be a queue of requests with work-stealing for threads that are available

Yes I need to look more closely at current implementation, but would seem better to have a single queue of waiting tasks instead of giving each thread its own queue. Otherwise execution order could be very unpredictable and if a single request is slow, it could create a backlog of requests assigned to the same thread. It seems like it should be possible to have a shared queue without too much complexity, so probably worth looking into.

xyzconstant commented at 5:42 PM on June 3, 2026: contributor

Concept ACK. This feature significantly reduces integration burden for clients written in other programming languages.

I just glanced at it, planning to review it deeply soon.

One thing to note: This PR will definitely need release notes.

Sjors commented at 4:38 PM on June 4, 2026: member

Would be nice to have a before and after usage example (with 2140-dev/bitcoin-capnp-types).

rustaceanrob commented at 4:46 PM on June 4, 2026: member

~Currently stale but I made an example here within the tests. No more set_thread required.~ See comment

rustaceanrob commented at 3:43 PM on June 5, 2026: member

Should I take a hack at implementing proper work delegation throughout the pool or would reviewers prefer I keep it as is for now?

xyzconstant commented at 3:53 PM on June 5, 2026: contributor

re: #283 (comment)

Should I take a hack at implementing proper work delegation throughout the pool or would reviewers prefer I keep it as is for now?

On my side, I'd love to see a more elaborate pool.

I was thinking about something like https://github.com/bitcoin/bitcoin/blob/master/src/util/threadpool.h as reference

xyzconstant commented at 8:59 PM on June 5, 2026: contributor

Just tested 0ac739b0fd0869ec153ba86d875fe95065860182 on peer-observer's ipc-extractor with a ChainNotifications callback as per bitcoin/bitcoin#29409

It's working as expected and, notably, reduces the overhead of creating a thread instance for each IPC call.

rustaceanrob force-pushed on Jun 15, 2026

rustaceanrob commented at 11:56 AM on June 15, 2026: member

Re-implemented as a "shortest queue" pool, where the thread with the lowest count of pending jobs is selected, tiebreaker by lowest index. I'd say this is more than sufficient for now (I would imagine users would not have more outstanding jobs than allocated threads in the current mining use case).

rustaceanrob force-pushed on Jun 15, 2026

in src/mp/proxy.cpp:440 in a828771f99

 435 | +            Lock lock(g_thread_context.waiter->m_mutex);
 436 | +            thread_context.set_value(&g_thread_context);
 437 | +            g_thread_context.waiter->wait(lock, [] { return !g_thread_context.waiter; });
 438 | +        });
 439 | +        auto thread_server = kj::heap<ProxyServer<Thread>>(m_connection, *thread_context.get_future().get(), std::move(thread));
 440 | +        m_connection.m_thread_pool.push_back({m_connection.m_threads.add(kj::mv(thread_server)), 0});

ViniciusCestarii commented at 2:27 PM on June 15, 2026:

On a828771f995abaf0a918cd3006cf1a54c928490e

nit: no need for explict 0, because PoolSlot.depth already defaults to 0

in include/mp/type-context.h:241 in a828771f99

 244 | +                        } else {
 245 | +                            MP_LOG(loop, Log::Error)
 246 | +                                << "IPC server error request #" << req << ", pool thread not found";
 247 | +                            throw std::runtime_error("pool thread not found");
 248 | +                        }
 249 | +                    });

ViniciusCestarii commented at 3:12 PM on June 15, 2026:

On a828771f995abaf0a918cd3006cf1a54c928490e

I believe it's better to move the depth-decrement guard up so it's created right after ++slot->depth and attached to the whole dispatch chain instead of just the inner post(). That keeps the increment and its decrement paired with no async hop between them, so the slot's depth is always released however the chain ends.

And you can use kj::defer instead of a custom struct.

++slot->depth;
auto guard = kj::defer([slot] { --slot->depth; });
return connection->m_threads.getLocalServer(slot->client)
    .then([&loop, invoke = kj::mv(invoke), req](const kj::Maybe<Thread::Server&>& pool_perhaps) mutable {
        // ... dispatch to pool_thread ...
    })
    .attach(kj::mv(guard));

ViniciusCestarii commented at 5:08 PM on June 15, 2026: contributor

Shortest-queue selection is looking good

rustaceanrob force-pushed on Jun 15, 2026

rustaceanrob commented at 6:04 PM on June 15, 2026: member

Nice feedback, updated baab0fe33e6e3a2b83acb25819547e24d89a187f

in include/mp/type-context.h:237 in baab0fe33e

 240 | +                                << "IPC server post request  #" << req << " {" << pool_thread.m_thread_context.thread_name << "}";
 241 | +                            return pool_thread.template post<typename ServerContext::CallContext>(std::move(invoke));
 242 | +                        } else {
 243 | +                            MP_LOG(loop, Log::Error)
 244 | +                                << "IPC server error request #" << req << ", pool thread not found";
 245 | +                            throw std::runtime_error("pool thread not found");

xyzconstant commented at 3:15 AM on June 16, 2026:

I'm wondering how this block could be exercised. A test for this would be great.

rustaceanrob commented at 12:54 PM on June 17, 2026:

This can only happen if a worker is removed from m_threads and not m_thread_pool, which only be possible if we added a shrink/resize method. I could add something like this, but for now I'm not sure if it's necessary. FWIW the previous else path also seems untested.

diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp
index 9be8742..cd5d5f4 100644
--- a/test/mp/test/test.cpp
+++ b/test/mp/test/test.cpp
@@ -598,5 +598,38 @@ KJ_TEST("Call async IPC method without thread or pool errors correctly")
     KJ_EXPECT(error_thrown);
 }

+KJ_TEST("Pool dispatch with torn-down slot surfaces error")
+{
+    TestSetup setup;
+    ProxyClient<messages::FooInterface>* foo = setup.client.get();
+    setup.server->m_impl->m_fn = [] {};
+
+    std::promise<void> done;
+    bool error_thrown{false};
+    foo->m_context.loop->sync([&] {
+        foo->m_context.connection->m_thread_pool.push_back(
+            {Thread::Client{KJ_EXCEPTION(FAILED, "pool slot torn down")}});
+
+        auto request{foo->m_client.callFnAsyncRequest()};
+        request.initContext();
+        foo->m_context.loop->m_task_set->add(
+            request.send().then(
+                [&](auto&&) { done.set_value(); },
+                [&](kj::Exception&& e) {
+                    error_thrown = true;
+                    KJ_EXPECT(std::string_view{e.getDescription().cStr()}.find(
+                        "pool thread not found") != std::string_view::npos);
+                    done.set_value();
+                }));
+    });
+    done.get_future().get();
+    KJ_EXPECT(error_thrown);
+}
+

xyzconstant commented at 1:58 PM on June 17, 2026:

Makes sense thanks

in include/mp/type-context.h:226 in baab0fe33e outdated

 229 | +                auto* slot = &pool[0];
 230 | +                for (size_t i = 1; i < pool.size(); ++i) {
 231 | +                    if (pool[i].depth < slot->depth) slot = &pool[i];
 232 | +                }
 233 | +                ++slot->depth;
 234 | +                auto guard = kj::defer([slot] { --slot->depth; });

xyzconstant commented at 3:16 AM on June 16, 2026:

nit: KJ_DEFER macro would be more consistent IMO.

rustaceanrob commented at 6:58 PM on June 16, 2026:

KJ_DEFER doesn't support move assignment

xyzconstant commented at 2:05 PM on June 17, 2026:

I actually meant something like this:

diff --git a/include/mp/type-context.h b/include/mp/type-context.h
index d241d95..e31ceee 100644
--- a/include/mp/type-context.h
+++ b/include/mp/type-context.h
@@ -223,9 +223,9 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                     if (pool[i].depth < slot->depth) slot = &pool[i];
                 }
                 ++slot->depth;
-                auto guard = kj::defer([slot] { --slot->depth; });
                 return connection->m_threads.getLocalServer(slot->client)
-                    .then([&loop, invoke = kj::mv(invoke), req](const kj::Maybe<Thread::Server&>& pool_perhaps) mutable {
+                    .then([&loop, invoke = kj::mv(invoke), req, slot](const kj::Maybe<Thread::Server&>& pool_perhaps) mutable {
+                        KJ_DEFER(--slot->depth);
                         KJ_IF_MAYBE (pt, pool_perhaps) {
                             auto& pool_thread = static_cast<ProxyServer<Thread>&>(*pt);
                             MP_LOG(loop, Log::Debug)
@@ -236,8 +236,7 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                                 << "IPC server error request #" << req << ", pool thread not found";
                             throw std::runtime_error("pool thread not found");
                         }
-                    })
-                    .attach(kj::mv(guard));
+                    });
             }
         }, [&loop, req](::kj::Exception&& e) -> kj::Promise<typename ServerContext::CallContext> {
             // If you see the error "(remote):0: failed: remote exception:

However, this is a nit

xyzconstant commented at 2:14 PM on June 17, 2026:

The diff above compiles and passes the tests, but so does this other:

diff --git a/include/mp/type-context.h b/include/mp/type-context.h
index d241d95..0a99610 100644
--- a/include/mp/type-context.h
+++ b/include/mp/type-context.h
@@ -223,7 +223,6 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                     if (pool[i].depth < slot->depth) slot = &pool[i];
                 }
                 ++slot->depth;
-                auto guard = kj::defer([slot] { --slot->depth; });
                 return connection->m_threads.getLocalServer(slot->client)
                     .then([&loop, invoke = kj::mv(invoke), req](const kj::Maybe<Thread::Server&>& pool_perhaps) mutable {
                         KJ_IF_MAYBE (pt, pool_perhaps) {
@@ -236,8 +235,7 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
                                 << "IPC server error request #" << req << ", pool thread not found";
                             throw std::runtime_error("pool thread not found");
                         }
-                    })
-                    .attach(kj::mv(guard));
+                    });
             }
         }, [&loop, req](::kj::Exception&& e) -> kj::Promise<typename ServerContext::CallContext> {
             // If you see the error "(remote):0: failed: remote exception:

xyzconstant commented at 5:25 PM on June 16, 2026: contributor

Thanks for the changes @rustaceanrob

I exercised this test again: #283 (comment) and downstream IPC tests, everything is working as expected.

I am still reviewing but I have left a couple of comments.

Note that doc/design.md needs to be updated too to reflect this new feature

rustaceanrob force-pushed on Jun 16, 2026

rustaceanrob commented at 7:10 PM on June 16, 2026: member

Note that doc/design.md needs to be updated too to reflect this new feature

Perhaps as a follow up? Or at least I'd like to get a few approach ACKs so we know this design won't change

rustaceanrob force-pushed on Jun 16, 2026

ryanofsky commented at 4:31 PM on June 17, 2026: collaborator

Or at least I'd like to get a few approach ACKs so we know this design won't change

Definite approach ACK for baab0fe33e6e3a2b83acb25819547e24d89a187f & sorry I haven't reviewed this more closely yet.

This is a simple and elegant way of adding a thread pool while changing existing code as little as possible.

My one concern about this is I don't think scheduling algorithm is very robust. But it's probably good enough for practical purposes, and it can be improved later. An example of a problem would be that if that if there are 10 threads, and a client makes 100 method calls, and 99 of the method calls return instantly, but one call waits for a new block to be connected and takes minutes to return. In that case, some of the 99 method calls which could complete instantaneously will be stuck queued behind the one slow call, even though there are threads sitting idle which could execute them. To fix this, it would be best to just add all incoming requests to a single queue, and have threads read from the common queue, instead of giving each thread its own individual queue. I asked claude to implement this to get an idea of what it would look like bd1f80c31b95c2041bd5feb13ae710a78698dcec (branch) but it's a more complicated approach and completely untested and I'd be happy to save something like this for a followup.

Other more minor things:

Would like to drop the name parameter from makePool, even though it could be useful for debugging. Since it only makes sense to have one thread pool per connection, the name string more of a way to identify the connection than the thread pool, and it could probably be useful in other contexts even in cases where no thread pool is needed. So I think it'd be best just to drop the name from the makePool method. And if having a name is helpful for debugging, add a separate setConnectionName method to name the connection. (We could also consider renaming ThreadMap interface to Connection to generalize it.)
I think implementation should avoid calling m_threads.getLocalServer(context_arg.getThread()) if context_arg.hasThread() is false. Should be no need to do a lookup on a null capability. This should just be a minor readability/efficiency improvement.

Otherwise this looks very good as far as I can tell and I do want to get this reviewed and merged.

rustaceanrob force-pushed on Jun 17, 2026

rustaceanrob commented at 8:38 PM on June 17, 2026: member

dd05eeab0c83d7fe5cb91804f826e7af87033415 removes name argument. I'd like to leave setConnectionName and similar refactors to a followup. Will get to the second comment soon.

rustaceanrob commented at 4:18 PM on June 18, 2026: member

@ryanofsky, r.e. second comment, is this what you mean?

diff --git a/include/mp/type-context.h b/include/mp/type-context.h
index d241d95..9751de7 100644
--- a/include/mp/type-context.h
+++ b/include/mp/type-context.h
@@ -202,7 +202,10 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
     Context::Reader context_arg = Accessor::get(params);
     auto thread_client = context_arg.getThread();
     auto* connection = server.m_context.connection;
-    auto result = connection->m_threads.getLocalServer(thread_client)
+    auto thread_promise = context_arg.hasThread()
+        ? connection->m_threads.getLocalServer(thread_client)
+        : kj::Promise<kj::Maybe<Thread::Server&>>(nullptr);
+    auto result = thread_promise
         .then([&loop, invoke = kj::mv(invoke), req, connection](const kj::Maybe<Thread::Server&>& perhaps) mutable {
             // If the client specified a thread, dispatch to it directly.
             KJ_IF_MAYBE (thread_server, perhaps) {

ryanofsky commented at 4:51 PM on June 18, 2026: collaborator

@ryanofsky, r.e. second comment, is this what you mean?

That's close but not exactly what I mean. I'd like to preserve the current "invalid thread handle" error in the case where context_arg.hasThread() is true but getLocalServer returns null. I'm actually not sure whether that case is possible to hit, but if it is possible I'd want it to be an error, not to silently fall back to using the thread pool when a thread was passed but couldn't be mapped to a local object for some reason.

So idea would be more like if (!context_arg.hasThread()) { ... your new code here ...} else { ... existing code ....} but hopefully not needing to move existing code too much.

Add `makePool` method on `ThreadMap`

This patch introduces a pool of threads to the `Connection` class, and
allows this pool to be populated with the thread map via `makePool`.
When a client thread is not set in a request context, it is delegated to
the pool.

The pool is implemented as shortest-queue, where the thread with the
shortest list of pending work handles the request. Tiebreaking is by
lowest index.

This was raised to me by Rust users, as they did not particularly care
where work is executed on the server-side, but they have to set the
thread regardless.

ref: https://github.com/2140-dev/bitcoin-capnp-types/blob/master/tests/util/bitcoin_core.rs#L149

d77e897eaf

rustaceanrob force-pushed on Jun 18, 2026

rustaceanrob commented at 8:21 PM on June 18, 2026: member

Unfortunately a larger diff but all existing code moved into the else block here d77e897eaf5e8bd8de9ea11b64233d9a6d354f25

Eunovo commented at 9:52 AM on June 22, 2026: contributor

Concept ACK

This removes the requirement for non-libmultiprocess clients to manage thread handles explicitly.

A server-wide threadpool is worth considering. The current per-connection pool allocates O(connections × pool_size) OS threads. External clients can open an unpredictable number of connections, so this scales without bound. A shared pool would cap total thread count at a fixed size across all connections.

in include/mp/proxy.capnp:51 in d77e897eaf

  44 | @@ -45,6 +45,10 @@ interface ThreadMap $count(0) {
  45 |      # execute on. Clients create and name threads and pass the thread handle as
  46 |      # a call parameter.
  47 |      makeThread @0 (name :Text) -> (result :Thread);
  48 | +    # Pre-allocate a pool of server threads for implicit dispatch. When a
  49 | +    # request arrives with no context.thread set, the server dispatches it
  50 | +    # through this pool via a shared work queue.
  51 | +    makePool @1 (count :UInt32) -> ();

xyzconstant commented at 4:45 PM on June 22, 2026:

nit: Relocate the comment below makePool definition line to maintain consistency with the file.

xyzconstant approved

xyzconstant commented at 5:23 PM on June 22, 2026: contributor

Code review ACK d77e897eaf5e8bd8de9ea11b64233d9a6d354f25

I think the changes are clean and ready to be merged.

Ran local and IPC tests on downstream, compiled a bitcoin-node instance to verify it wasn't affected, and finally tested the changes in a Rust client (see: https://github.com/xyzconstant/peer-observer/commit/c414765a0af113832c95ef2b26ba0e25fb9416fc).

This significantly improves how clients written in other languages consume libmultiprocess processes by eliminating the need to create a new thread for each IPC call, because the serving processes now automatically assign threads to requests with no threads specified.

I explored another alternative to the current approach but felt this was the most effective solution.

As a next step, based on ryanofsky's suggestion (comment), it might be helpful to rename the ThreadMap interface to something like Connection or Runtime for better clarity.

DrahtBot requested review from ryanofsky on Jun 22, 2026

xyzconstant commented at 5:31 PM on June 22, 2026: contributor

Following on from my review:

it might be helpful to rename the ThreadMap interface to something like Connection or Runtime for better clarity.

I've made some progress on top of these changes. Happy to open a follow-up PR if/once this one gets merged.

ViniciusCestarii commented at 6:32 PM on June 22, 2026: contributor

tACK d77e897eaf5e8bd8de9ea11b64233d9a6d354f25

Tested with https://github.com/ViniciusCestarii/bitcoin-ipc-demo/tree/make-pool-tester and it successfully worked passing simultaneous calls to the threads with smaller queues.