diff --git a/.github/workflows/bitcoin-core-ci.yml b/.github/workflows/bitcoin-core-ci.yml index 83dff816..f14d5682 100644 --- a/.github/workflows/bitcoin-core-ci.yml +++ b/.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/ diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index c2a8d433..f15965bb 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -694,7 +694,7 @@ using ConnThread = ConnThreads::iterator; // inserted bool. std::tuple SetThread(GuardedRef threads, Connection* connection, const std::function& 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 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 diff --git a/include/mp/proxy-types.h b/include/mp/proxy-types.h index f1127da5..ed458209 100644 --- a/include/mp/proxy-types.h +++ b/include/mp/proxy-types.h @@ -691,9 +691,9 @@ void serverDestroy(Server& server) template 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(); + assert(!CurrentThread().loop_thread); + CurrentThread().waiter = std::make_unique(); 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 invoke_context; // Must outlive waiter->wait() call below std::exception_ptr exception; std::string kj_exception; diff --git a/include/mp/type-context.h b/include/mp/type-context.h index b63959f4..54007207 100644 --- a/include/mp/type-context.h +++ b/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}; diff --git a/src/mp/proxy.cpp b/src/mp/proxy.cpp index a0a7aba0..4c7f7666 100644 --- a/src/mp/proxy.cpp +++ b/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 ProxyServer::makePool(MakePoolContext context) const std::string thread_name = "pool/" + std::to_string(i); std::promise 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(); - 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(); + Lock lock(CurrentThread().waiter->m_mutex); + thread_context.set_value(&CurrentThread()); + CurrentThread().waiter->wait(lock, [] { return !CurrentThread().waiter; }); }); auto thread_server = kj::heap>(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 ProxyServer::makeThread(MakeThreadContext context) const std::string from = context.getParams().getName(); std::promise 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(); - 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(); + 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 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>(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 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) diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index 7a0940c8..f5f35437 100644 --- a/test/mp/test/test.cpp +++ b/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 running{3}; std::promise pool_ready; foo->m_context.loop->sync([&] {