proxy: Name threads spawned by the event loop

include/mp/util.h

@@ -264,6 +264,9 @@ decltype(auto) TryFinally(Fn&& fn, After&& after)
     }
 }
 
+//! Set the OS-level name of the current thread
+void SetOsThreadName(const char* name);
+
 //! Format current thread name as "{exe_name}-{$pid}/{thread_name}-{$tid}".
 std::string ThreadName(const char* exe_name);
 

src/mp/proxy.cpp

@@ -358,6 +358,7 @@ void EventLoop::startAsyncThread()
         m_cv.notify_all();
     } else if (!m_async_fns->empty()) {
         m_async_thread = std::thread([this] {
+            SetOsThreadName("capnp-async");
             Lock lock(m_mutex);
             while (m_async_fns) {
                 if (!m_async_fns->empty()) {
@@ -483,7 +484,8 @@ kj::Promise<void> ProxyServer<ThreadMap>::makePool(MakePoolContext context)
     for (uint32_t i = 0; i < count; ++i) {
         const std::string thread_name = "pool/" + std::to_string(i);
         std::promise<ThreadContext*> thread_context;
-        std::thread thread([&loop, &thread_context, thread_name]() {
+        std::thread thread([&loop, &thread_context, thread_name, i]() {
+            SetOsThreadName(("capnp-pool-" + std::to_string(i)).c_str());
             CurrentThread().thread_name = ThreadName(loop.m_exe_name) + " (" + thread_name + ")";
             CurrentThread().waiter = std::make_unique<Waiter>();
             Lock lock(CurrentThread().waiter->m_mutex);
@@ -503,6 +505,7 @@ 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]() {
+        SetOsThreadName("capnp-worker");
         CurrentThread().thread_name = ThreadName(loop.m_exe_name) + " (from " + from + ")";
         CurrentThread().waiter = std::make_unique<Waiter>();
         Lock lock(CurrentThread().waiter->m_mutex);

src/mp/util.cpp

@@ -33,6 +33,10 @@
 #include <pthread_np.h>
 #endif // HAVE_PTHREAD_GETTHREADID_NP
 
+#if __has_include(<sys/prctl.h>)
+#include <sys/prctl.h>
+#endif
+
 extern "C" char **environ; // NOLINT(readability-redundant-declaration)
 
 namespace mp {
@@ -74,6 +78,22 @@ template <std::size_t N>
 
 } // namespace
 
+// Copied from https://github.com/bitcoin/bitcoin/blob/d3e40af2597/src/util/threadnames.cpp#L21-L36
+void SetOsThreadName(const char* name)
+{
+#if defined(PR_SET_NAME)
+    // Only the first 15 characters are used (16 - NUL terminator)
+    ::prctl(PR_SET_NAME, name, 0, 0, 0);
+#elif defined(HAVE_PTHREAD_GETTHREADID_NP)
+    pthread_set_name_np(pthread_self(), name);
+#elif defined(__APPLE__)
+    pthread_setname_np(name);
+#else
+    // Prevent warnings for unused parameters...
+    (void)name;
+#endif
+}
+
 std::string ThreadName(const char* exe_name)
 {
     char thread_name[16] = {0};

test/mp/test/test.cpp

@@ -25,6 +25,7 @@
 #include <kj/test.h>
 #include <map>
 #include <memory>
+#include <mp/config.h>
 #include <mp/proxy.h>
 #include <mp/proxy.capnp.h>
 #include <mp/proxy-io.h>
@@ -550,6 +551,68 @@ KJ_TEST("Call async IPC method dispatched to pool thread")
     }
 }
 
+#ifdef HAVE_PTHREAD_GETNAME_NP
+KJ_TEST("Worker thread has OS thread name")
+{
+    TestSetup setup;
+    ProxyClient<messages::FooInterface>* foo = setup.client.get();
+    foo->initThreadMap();
+
+    std::promise<std::string> thread_name;
+    setup.server->m_impl->m_fn = [&] { thread_name.set_value(ThreadName("")); };
+    foo->callFnAsync();
+
+    const std::string name{thread_name.get_future().get()};
+    KJ_EXPECT(name.find("/capnp-worker-") != std::string::npos, name);
+}
+
+KJ_TEST("Pool thread has OS thread name")
+{
+    TestSetup setup;
+    ProxyClient<messages::FooInterface>* foo = setup.client.get();
+    foo->initThreadMap();
+
+    std::promise<std::string> thread_name;
+    setup.server->m_impl->m_fn = [&] { thread_name.set_value(ThreadName("")); };
+
+    std::promise<void> pool_ready;
+    foo->m_context.loop->sync([&] {
+        auto pool_req = foo->m_context.connection->m_thread_map.makePoolRequest();
+        pool_req.setCount(1);
+        foo->m_context.loop->m_task_set->add(
+            pool_req.send().then([&](auto&&) { pool_ready.set_value(); }));
+    });
+    pool_ready.get_future().get();
+
+    std::promise<void> done;
+    foo->m_context.loop->sync([&] {
+        auto request{foo->m_client.callFnAsyncRequest()};
+        foo->m_context.loop->m_task_set->add(
+            request.send().then([&](auto&&) { done.set_value(); }));
+    });
+    // Wait for the reply before returning, so the connection is not torn down
+    // while the request is still in flight.
+    done.get_future().get();
+
+    const std::string name{thread_name.get_future().get()};
+    KJ_EXPECT(name.find("/capnp-pool-0-") != std::string::npos, name);
+}
+
+KJ_TEST("Async cleanup thread has OS thread name")
+{
+    std::promise<std::string> thread_name;
+    {
+        TestSetup setup;
+        // FooInterface has no destroy method, so the server ProxyServer runs
+        // its cleanup functions on the async thread when it is destroyed.
+        setup.server->m_context.cleanup_fns.emplace_front(
+            [&] { thread_name.set_value(ThreadName("")); });
+    }
+    const std::string name{thread_name.get_future().get()};
+    KJ_EXPECT(name.find("/capnp-async-") != std::string::npos, name);
+}
+#endif // HAVE_PTHREAD_GETNAME_NP
+
 KJ_TEST("Call async IPC method without thread or pool errors correctly")
 {
     TestSetup setup;