Demonstration of pinned-thread resource

docs/dev/geant4-pinned-thread-resource-2026-07.md

@@ -0,0 +1,176 @@
+# Geant4 and the `pinned_thread_resource` design
+
+*Status: design analysis, July 2026. Investigates whether the pinned-thread
+resource model prototyped in `test/tbb-preview/resource_limiting_test.cpp` can
+accommodate Geant4's multi-threading constraints.*
+
+This note analyzes three inputs:
+
+- The Geant4 multi-threading design page (Book For Toolkit Developers,
+  "Parallelism in Geant4: multi-threading capabilities").
+- CMSSW's `SimG4Core/Application/plugins/OscarMTProducer.cc` (revision
+  `b8fdcd5`), which adapts an event-processing framework to Geant4's threading
+  model, together with its `edm::ThreadHandoff` helper
+  (`FWCore/Utilities/interface/ThreadHandoff.h`).
+- The `managed_thread` + `flow::resource_limiter` prototype in
+  `test/tbb-preview/resource_limiting_test.cpp` (the "std::threads as limited
+  resources" test case).
+
+The question: can the `pinned_thread_resource` approach — in which algorithm
+writers do *not* explicitly manage thread state and merely use a pinned-thread
+resource token — work given Geant4's constraints?
+
+## Short answer
+
+Yes. The `pinned_thread_resource` approach is not only possible under Geant4's
+constraints, it is a cleaner realization of the same mechanism CMSSW implements
+by hand. The `managed_thread` in the prototype is functionally the same object
+as CMSSW's `edm::ThreadHandoff`. Geant4's binding requirement — that all Geant4
+calls for a given worker happen on one specific OS thread — maps exactly onto
+what a pinned-thread resource token provides. The design goal (algorithm
+writers do not manage thread state) is achievable, with one important caveat
+about *what* gets pinned.
+
+## What Geant4 actually requires
+
+From the MT design page, the binding constraints are:
+
+1. **Event-level parallelism, master/worker model.** One `G4MTRunManager`
+   (master) creates and controls N `G4WorkerRunManager` instances. Users must
+   never instantiate a worker directly.
+2. **Thread-local storage is the thread-safety mechanism.** Split-classes
+   (`G4LogicalVolume`, `G4ParticleDefinition`, `G4VUserPhysicsList`, ...),
+   `G4ThreadLocal` statics, `G4Cache`, and thread-private singletons all key off
+   the *identity of the executing OS thread*. Geant4's `G4ThreadLocalSingleton`
+   and `G4Cache` cleanup fire at thread exit / program exit.
+3. **Worker initialization is per-thread and stateful.** A worker's TLS must be
+   initialized *on the thread that will run it* (copying constant state from the
+   fully-configured master), before any event work.
+
+The critical consequence: **a given Geant4 worker's event processing must
+always run on the same OS thread** from `initializeG4` through every `produce`
+to `endRun`. TBB tasks migrate across worker threads freely, so a TBB task
+cannot call Geant4 directly — the TLS a task sees depends on which TBB worker
+happens to run it.
+
+## What CMSSW does, and why it looks obscure
+
+`OscarMTProducer` is a `stream::EDProducer` (one instance per stream = per
+concurrent event slot). The obscurity is entirely `edm::ThreadHandoff
+m_handoff`:
+
+- `ThreadHandoff` spawns **one dedicated `pthread`** (with a configurable stack
+  size — Geant4 needs a big stack, default 10 MiB) and parks it on a condition
+  variable.
+- Every Geant4 interaction is wrapped: `m_handoff.runAndWait([...]{ ... })` in
+  the constructor, `beginRun`, `endRun`, `produce`, and the destructor.
+- `runAndWait` ships a type-erased functor to the parked thread, wakes it,
+  blocks the TBB thread until the functor completes, and rethrows any exception.
+
+So CMSSW guarantees "same OS thread for all Geant4 calls" by owning that thread
+explicitly and handing work to it. The handshaking between the master thread,
+TBB threads, and Geant4 worker threads is exactly this condition-variable
+handoff, plus the `ServiceRegistry::Operate guard{token}` dance to re-propagate
+CMSSW services across the thread boundary, plus the `StaticRandomEngineSetUnset`
+juggling of the CLHEP global engine. The algorithm writer is fully exposed to
+all of it.
+
+## The mapping to `managed_thread` / `pinned_thread_resource`
+
+Comparing `ThreadHandoff` to `managed_thread`
+(`resource_limiting_test.cpp:49-93`):
+
+| CMSSW `ThreadHandoff` | Phlex `managed_thread` |
+| --- | --- |
+| dedicated `pthread m_thread` | `std::jthread thread_` |
+| `m_threadHandoff` condition_variable + mutex | `work_ready_` / `slot_ready_` binary semaphores |
+| `runAndWait(F&&)` ships type-erased functor, blocks caller | `execute(Args...)` sets args, releases `work_ready_`, blocks on `slot_ready_` |
+| `stopThread()` / `m_stopThread` | destructor sets `gear_ = park`, releases `work_ready_` |
+| rethrows `exception_ptr` | (not yet handled — see caveats) |
+
+They are the same pattern. The difference is *who drives it*:
+
+- In CMSSW, the producer author writes every `runAndWait(...)` call by hand.
+- In Phlex, the `managed_thread` is wrapped in a `flow::resource_limiter`, and
+  the `resource_limited_node` machinery acquires a token, hands the algorithm
+  the `managed_thread*`, and the algorithm merely calls
+  `geant4_resource->execute(i)` (`resource_limiting_test.cpp:118-125`). The
+  handoff, blocking, and serialization are handled by the framework.
+
+Crucially, `resource_limiter` gives Geant4's "N workers" model for free:
+construct the limiter with N `managed_thread` tokens, and TBB admits at most N
+concurrent event tasks into the Geant4 node, each pinned to a distinct
+dedicated OS thread. That is a direct structural analog of `G4MTRunManager`
+owning N `G4WorkerRunManager`s.
+
+## Is it possible under the constraints? Yes, with these design points
+
+It works because the pinned thread is a real, persistent OS thread. Geant4's
+TLS, split-classes, and `G4Cache` only require thread *identity stability*,
+which `std::jthread` provides for its lifetime. As long as worker G4
+initialization and all subsequent event calls for that worker are routed
+through the *same* `managed_thread`, Geant4 is satisfied and cannot tell the
+difference between this and `G4THREADCREATE`.
+
+To make it fully work, the design must address:
+
+1. **Pin the whole Geant4 worker lifecycle to the token, not just `produce`.**
+   The token (`managed_thread`) must run worker `initializeG4`/`beginRun`, every
+   event, and `endRun`/teardown. If a token can be handed to different algorithm
+   invocations that correspond to *different* Geant4 workers, the TLS will not
+   match. The cleanest model: **one `managed_thread` == one Geant4 worker**, with
+   its worker state initialized lazily on first `execute` (Geant4's own
+   `G4WorkerRunManager` construction happens on that thread). The
+   `resource_limiter` holding N of them then *is* the worker pool.
+2. **Master thread / global initialization.** Geant4 requires a single
+   `G4MTRunManager` created and configured before workers spawn (CMSSW's
+   `OscarMTMasterThread` global cache, driven on its own handoff thread). Phlex
+   needs an equivalent job-level/run-level hook that constructs and configures
+   the master before the pinned worker tokens run anything. This is orthogonal
+   to the per-event resource mechanism but must exist.
+3. **Stack size.** `ThreadHandoff` takes a `stackSize` (default 10 MiB). Geant4
+   tracking is stack-hungry. `std::jthread` gives no portable control over stack
+   size, so a production `managed_thread` for Geant4 likely needs a
+   pthread-based implementation (or a platform-specific attribute) rather than
+   raw `std::jthread`.
+4. **Exception propagation.** `ThreadHandoff::runAndWait` captures and rethrows
+   `exception_ptr`. The current `managed_thread::execute` does not; a Geant4
+   exception on the pinned thread would be lost or terminate. This must be
+   added.
+5. **Thread-exit cleanup ordering.** Geant4 registers cleanup via `G4AutoDelete`
+   / `G4ThreadLocalSingleton` that fires at thread termination. The pinned
+   thread must run Geant4 worker teardown *before* the `jthread` joins, and
+   teardown must itself run on that thread. `managed_thread`'s destructor
+   currently just parks and stops; a Geant4 variant needs a "run teardown
+   functor, then stop" step.
+6. **Service/context re-propagation across the boundary.** CMSSW re-establishes
+   its `ServiceRegistry` inside each `runAndWait` (`Operate guard{token}`) and
+   manages the CLHEP global RNG engine per event. Phlex's analog: whatever
+   ambient context an algorithm needs (RNG, logging context) must be
+   re-established inside the pinned execution, or the pinned-thread body must be
+   given it. This is a framework responsibility, not the algorithm writer's —
+   which is precisely the win.
+
+## Bottom line
+
+The `pinned_thread_resource` design is compatible with Geant4 and achieves the
+stated goal: algorithm writers acquire a pinned-thread token and call
+`execute(...)`, with no visible TLS handshaking. It is the same core mechanism
+as CMSSW's `ThreadHandoff`, but inverted — instead of every author hand-writing
+`runAndWait`, the framework's `resource_limited_node` performs the handoff and
+the `resource_limiter` enforces the N-worker concurrency limit. The obscurity in
+CMSSW is not fundamental to Geant4; it is a consequence of `ThreadHandoff` being
+a manual, per-producer construct rather than a first-class framework resource.
+
+The non-negotiables to get right are:
+
+1. **One pinned thread == one Geant4 worker for its entire lifecycle** (init →
+   events → teardown all on that thread).
+2. A **master/global init hook** for `G4MTRunManager`.
+3. **Configurable stack size** (favoring a pthread-backed `managed_thread` over
+   raw `std::jthread`).
+4. **Exception propagation** from the pinned thread.
+5. **On-thread teardown** honoring Geant4's TLS cleanup.
+
+None of these conflict with the resource-limiter model; they are implementation
+requirements of the pinned token itself.

test/tbb-preview/CMakeLists.txt

@@ -35,18 +35,55 @@ set(CMAKE_REQUIRED_INCLUDES ${CMAKE_REQUIRED_INCLUDES_SAVE})
 set(CMAKE_REQUIRED_FLAGS "${CMAKE_REQUIRED_FLAGS_SAVE}")
 set(CMAKE_TRY_COMPILE_VERBOSE ${CMAKE_TRY_COMPILE_VERBOSE_SAVE})
 
+function(qualified_test)
+  cmake_parse_arguments(QT "" "NAME;TARGET;SOURCE" "LIBRARIES" ${ARGN})
+
+  if(NOT QT_NAME OR NOT QT_TARGET OR NOT QT_SOURCE)
+    message(FATAL_ERROR "qualified_test requires NAME, TARGET, and SOURCE")
+  endif()
+
+  cet_make_exec(
+    NAME ${QT_TARGET}
+    USE_CATCH2_MAIN
+    SOURCE ${QT_SOURCE}
+    LIBRARIES ${QT_LIBRARIES}
+  )
+
+  cet_test("${QT_NAME}" HANDBUILT TEST_EXEC ${QT_TARGET})
+endfunction()
+
 if(HAVE_TBB_RESOURCE_LIMITING)
-  cet_test(
+  qualified_test(
+    NAME
+    preview:resource_limiting
+    TARGET
     resource_limiting
-    USE_CATCH2_MAIN
     SOURCE
     resource_limiting_test.cpp
     LIBRARIES
     phlex::core
     TBB::tbb
     spdlog::spdlog
   )
+
+  qualified_test(
+    NAME
+    preview:resource_limited_threads
+    TARGET
+    resource_limited_threads
+    SOURCE
+    resource_limited_threads_test.cpp
+    LIBRARIES
+    phlex::core
+    TBB::tbb
+    spdlog::spdlog
+  )
+
   target_compile_definitions(resource_limiting PRIVATE TBB_PREVIEW_FLOW_GRAPH_RESOURCE_LIMITING=1)
+  target_compile_definitions(
+    resource_limited_threads
+    PRIVATE TBB_PREVIEW_FLOW_GRAPH_RESOURCE_LIMITING=1
+  )
 else()
   message(STATUS "Skipping resource_limiting test: flow::resource_limiter not available in TBB")
 endif()

test/tbb-preview/resource_limited_threads_test.cpp

@@ -0,0 +1,158 @@
+#include "catch2/catch_test_macros.hpp"
+#include "fmt/std.h"
+#include "oneapi/tbb/flow_graph.h"
+#include "spdlog/spdlog.h"
+
+#include <atomic>
+#include <mutex>
+#include <optional>
+#include <semaphore>
+#include <stdexcept>
+#include <tuple>
+
+using namespace oneapi::tbb;
+
+namespace {
+
+  template <typename F, typename... Args>
+  class managed_thread {
+    enum class states : std::uint8_t { off, drive, park };
+
+    F f_;
+    std::optional<std::tuple<Args...>> args_;
+    std::jthread thread_;
+    std::binary_semaphore work_ready_{0};
+    std::binary_semaphore slot_ready_{0};
+    std::atomic<states> gear_ = states::off;
+    std::exception_ptr exception_;
+
+    void run(std::stop_token st)
+    {
+      while (true) {
+        work_ready_.acquire();
+        if (gear_ == states::park || st.stop_requested()) {
+          break;
+        }
+
+        try {
+          std::apply(f_, args_.value());
+        } catch (...) {
+          exception_ = std::current_exception();
+        }
+        slot_ready_.release();
+      }
+    }
+
+  public:
+    explicit managed_thread(F f) :
+      f_{std::move(f)}, thread_{[this](std::stop_token st) { run(st); }}
+    {
+    }
+    managed_thread(managed_thread const&) = delete;
+    managed_thread(managed_thread&&) = delete;
+    managed_thread& operator=(managed_thread const&) = delete;
+    managed_thread& operator=(managed_thread&&) = delete;
+
+    ~managed_thread()
+    {
+      gear_ = states::park;
+      work_ready_.release();
+    }
+
+    void execute(Args const&... args)
+    {
+      if (gear_ == states::off) {
+        gear_ = states::drive;
+      }
+
+      args_ = std::make_tuple(args...);
+      work_ready_.release();
+      slot_ready_.acquire();
+      args_.reset();
+
+      if (exception_) {
+        std::rethrow_exception(std::exchange(exception_, nullptr));
+      }
+    }
+  };
+
+} // namespace
+
+TEST_CASE("std::threads as limited resources")
+{
+  auto f = [](unsigned int const i) {
+    spdlog::info("Executing on thread {} with argument {}", std::this_thread::get_id(), i);
+  };
+  using managed_thread_type = managed_thread<decltype(f), unsigned int>;
+
+  managed_thread_type t1{f};
+  managed_thread_type t2{f};
+  flow::resource_limiter<managed_thread_type*> pinned_thread_resource{&t1, &t2};
+
+  flow::graph g;
+  unsigned int i{};
+  flow::input_node src{g, [&i](flow_control& fc) {
+                         if (i < 10) {
+                           return ++i;
+                         }
+                         fc.stop();
+                         return 0u;
+                       }};
+
+  flow::resource_limited_node<unsigned int, std::tuple<>> geant4_node{
+    g,
+    flow::unlimited,
+    std::tie(pinned_thread_resource),
+    [](unsigned int const i, auto&, managed_thread_type* geant4_resource) {
+      spdlog::info("Launching from thread {} with argument {}", std::this_thread::get_id(), i);
+      geant4_resource->execute(i);
+    }};
+
+  make_edge(src, geant4_node);
+
+  src.activate();
+  g.wait_for_all();
+}
+
+TEST_CASE("Throw exception from managed thread", "[multithreading]")
+{
+  auto f = [](unsigned int const i) {
+    spdlog::info("Executing on thread {} with argument {}", std::this_thread::get_id(), i);
+    if (i == 5) {
+      throw std::runtime_error{"Exception from managed thread"};
+    }
+  };
+  using managed_thread_type = managed_thread<decltype(f), unsigned int>;
+
+  managed_thread_type t1{f};
+  managed_thread_type t2{f};
+  flow::resource_limiter<managed_thread_type*> pinned_thread_resource{&t1, &t2};
+
+  flow::graph g;
+  unsigned int i{};
+  flow::input_node src{g, [&i](flow_control& fc) {
+                         if (i < 10) {
+                           return ++i;
+                         }
+                         fc.stop();
+                         return 0u;
+                       }};
+
+  flow::resource_limited_node<unsigned int, std::tuple<>> geant4_node{
+    g,
+    flow::unlimited,
+    std::tie(pinned_thread_resource),
+    [](unsigned int const i, auto&, managed_thread_type* geant4_resource) {
+      spdlog::info("Launching from thread {} with argument {}", std::this_thread::get_id(), i);
+      try {
+        geant4_resource->execute(i);
+      } catch (...) {
+        // Ignore exceptions from the managed thread
+      }
+    }};
+
+  make_edge(src, geant4_node);
+
+  src.activate();
+  g.wait_for_all();
+}