refactor: access ThreadContext through CurrentThread(), ci: switch Bitcoin Core to master

.github/workflows/bitcoin-core-ci.yml

@@ -18,8 +18,7 @@ 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
+  BITCOIN_CORE_REF: refs/heads/master
   LLVM_VERSION: 22
   LIBCXX_DIR: /tmp/libcxx-build/
 

include/mp/proxy-io.h

@@ -694,7 +694,7 @@ using ConnThread = ConnThreads::iterator;
 // inserted bool.
 std::tuple<ConnThread, bool> SetThread(GuardedRef<ConnThreads> threads, Connection* connection, const std::function<Thread::Client()>& make_thread);
 
-//! The thread_local ThreadContext g_thread_context struct provides information
+//! The thread_local ThreadContext struct (see CurrentThread()) provides information
 //! about individual threads and a way of communicating between them. Because
 //! it's a thread local struct, each ThreadContext instance is initialized by
 //! the thread that owns it.
@@ -934,6 +934,26 @@ extern thread_local ThreadContext g_thread_context; // NOLINT(bitcoin-nontrivial
 // cannot be thread_local" which should not be a problem on modern platforms, and
 // could lead to a small memory leak at worst on older ones.
 
+//! Return the current thread's ThreadContext.
+//!
+//! Why per-thread state is needed at all: libmultiprocess has no control over
+//! which threads the C++ application uses to call ProxyClient methods after
+//! the proxy objects are returned to it. The thread-mapping model (see
+//! "Thread Mapping" in doc/design.md) gives each application thread making
+//! IPC calls a dedicated server-side thread that executes its requests, so
+//! thread-local state and recursive mutexes work as expected across the
+//! process boundary and callbacks from the server run on the originating
+//! client thread. The client-side handles for those dedicated server threads
+//! (the ProxyClient<Thread> objects returned by ThreadMap.makeThread, stored
+//! per connection in the request_threads / callback_threads maps below) are
+//! state that must be keyed implicitly by the calling thread, and must be
+//! released when the client thread exits so the corresponding server threads
+//! are freed. A thread_local object is the C++ mechanism that provides both
+//! of these: per-thread storage plus a destructor that runs at thread exit
+//! (the C equivalent would be a pthread key destructor). This is why
+//! ThreadContext is thread_local and why its destructor is nontrivial.
+ThreadContext& CurrentThread();
+
 } // namespace mp
 
 #endif // MP_PROXY_IO_H

include/mp/proxy-types.h

@@ -691,9 +691,9 @@ void serverDestroy(Server& server)
 template <typename ProxyClient, typename GetRequest, typename... FieldObjs>
 void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, FieldObjs&&... fields)
 {
-    if (!g_thread_context.waiter) {
-        assert(g_thread_context.thread_name.empty());
-        g_thread_context.thread_name = ThreadName(proxy_client.m_context.loop->m_exe_name);
+    if (!CurrentThread().waiter) {
+        assert(CurrentThread().thread_name.empty());
+        CurrentThread().thread_name = ThreadName(proxy_client.m_context.loop->m_exe_name);
         // If next assert triggers, it means clientInvoke is being called from
         // the capnp event loop thread. This can happen when a ProxyServer
         // method implementation that runs synchronously on the event loop
@@ -702,13 +702,13 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel
         // run asynchronously off the event loop thread. This is easy to fix by
         // just adding a 'context :Proxy.Context' argument to the capnp method
         // declaration so the server method runs in a dedicated thread.
-        assert(!g_thread_context.loop_thread);
-        g_thread_context.waiter = std::make_unique<Waiter>();
+        assert(!CurrentThread().loop_thread);
+        CurrentThread().waiter = std::make_unique<Waiter>();
         MP_LOGPLAIN(*proxy_client.m_context.loop, Log::Info)
-            << "{" << g_thread_context.thread_name
+            << "{" << CurrentThread().thread_name
             << "} IPC client first request from current thread, constructing waiter";
     }
-    ThreadContext& thread_context{g_thread_context};
+    ThreadContext& thread_context{CurrentThread()};
     std::optional<ClientInvokeContext> invoke_context; // Must outlive waiter->wait() call below
     std::exception_ptr exception;
     std::string kj_exception;

include/mp/type-context.h

@@ -93,7 +93,7 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn&
         // call. In this case, the callbackThread value should point
         // to the same thread already in the map, so there is no
         // need to update the map.
-        auto& thread_context = g_thread_context;
+        auto& thread_context = CurrentThread();
         auto& request_threads = thread_context.request_threads;
         ConnThread request_thread;
         bool inserted{false};

src/mp/proxy.cpp

@@ -42,6 +42,11 @@ namespace mp {
 
 thread_local ThreadContext g_thread_context; // NOLINT(bitcoin-nontrivial-threadlocal)
 
+ThreadContext& CurrentThread()
+{
+    return g_thread_context;
+}
+
 Stream MakeStream(EventLoop&loop, SocketId socket)
 {
     Stream stream;
@@ -283,9 +288,9 @@ EventLoop::~EventLoop()
 
 void EventLoop::loop()
 {
-    assert(!g_thread_context.loop_thread);
-    g_thread_context.loop_thread = true;
-    KJ_DEFER(g_thread_context.loop_thread = false);
+    assert(!CurrentThread().loop_thread);
+    CurrentThread().loop_thread = true;
+    KJ_DEFER(CurrentThread().loop_thread = false);
 
     {
         const Lock lock(m_mutex);
@@ -479,11 +484,11 @@ kj::Promise<void> ProxyServer<ThreadMap>::makePool(MakePoolContext context)
         const std::string thread_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.waiter = std::make_unique<Waiter>();
-            Lock lock(g_thread_context.waiter->m_mutex);
-            thread_context.set_value(&g_thread_context);
-            g_thread_context.waiter->wait(lock, [] { return !g_thread_context.waiter; });
+            CurrentThread().thread_name = ThreadName(loop.m_exe_name) + " (" + thread_name + ")";
+            CurrentThread().waiter = std::make_unique<Waiter>();
+            Lock lock(CurrentThread().waiter->m_mutex);
+            thread_context.set_value(&CurrentThread());
+            CurrentThread().waiter->wait(lock, [] { return !CurrentThread().waiter; });
         });
         auto thread_server = kj::heap<ProxyServer<Thread>>(m_connection, *thread_context.get_future().get(), std::move(thread));
         m_connection.m_thread_pool.push_back({m_connection.m_threads.add(kj::mv(thread_server))});
@@ -498,14 +503,14 @@ kj::Promise<void> ProxyServer<ThreadMap>::makeThread(MakeThreadContext context)
     const std::string from = context.getParams().getName();
     std::promise<ThreadContext*> thread_context;
     std::thread thread([&loop, &thread_context, from]() {
-        g_thread_context.thread_name = ThreadName(loop.m_exe_name) + " (from " + from + ")";
-        g_thread_context.waiter = std::make_unique<Waiter>();
-        Lock lock(g_thread_context.waiter->m_mutex);
-        thread_context.set_value(&g_thread_context);
+        CurrentThread().thread_name = ThreadName(loop.m_exe_name) + " (from " + from + ")";
+        CurrentThread().waiter = std::make_unique<Waiter>();
+        Lock lock(CurrentThread().waiter->m_mutex);
+        thread_context.set_value(&CurrentThread());
         if (loop.testing_hook_makethread_created) loop.testing_hook_makethread_created();
         // Wait for shutdown signal from ProxyServer<Thread> destructor (signal
         // is just waiter getting set to null.)
-        g_thread_context.waiter->wait(lock, [] { return !g_thread_context.waiter; });
+        CurrentThread().waiter->wait(lock, [] { return !CurrentThread().waiter; });
     });
     auto thread_server = kj::heap<ProxyServer<Thread>>(m_connection, *thread_context.get_future().get(), std::move(thread));
     auto thread_client = m_connection.m_threads.add(kj::mv(thread_server));
@@ -517,7 +522,7 @@ std::atomic<int> server_reqs{0};
 
 std::string LongThreadName(const char* exe_name)
 {
-    return g_thread_context.thread_name.empty() ? ThreadName(exe_name) : g_thread_context.thread_name;
+    return CurrentThread().thread_name.empty() ? ThreadName(exe_name) : CurrentThread().thread_name;
 }
 
 kj::StringPtr KJ_STRINGIFY(Log v)

test/mp/test/test.cpp

@@ -462,7 +462,7 @@ KJ_TEST("Make simultaneous IPC calls on single remote thread")
     // that will be used for the test.
     setup.server->m_impl->m_fn = [&] {};
     foo->callFnAsync();
-    ThreadContext& tc{g_thread_context};
+    ThreadContext& tc{CurrentThread()};
     Thread::Client *callback_thread, *request_thread;
     foo->m_context.loop->sync([&] {
         Lock lock(tc.waiter->m_mutex);
@@ -516,7 +516,7 @@ KJ_TEST("Call async IPC method dispatched to pool thread")
     foo->initThreadMap();
     setup.server->m_impl->m_int_fn = [](int n) { return n * 2; };
 
-    ThreadContext& tc{g_thread_context};
+    ThreadContext& tc{CurrentThread()};
     std::atomic<size_t> running{3};
     std::promise<void> pool_ready;
     foo->m_context.loop->sync([&] {