From 1382eb13e50a4529cd5bb33682d32c42929923cd Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Fri, 17 Jul 2026 08:36:44 -0500 Subject: [PATCH 1/4] Demonstration of pinned-thread resource --- test/tbb-preview/resource_limiting_test.cpp | 84 +++++++++++++++++++++ 1 file changed, 84 insertions(+) diff --git a/test/tbb-preview/resource_limiting_test.cpp b/test/tbb-preview/resource_limiting_test.cpp index 111480526..44e56c2b8 100644 --- a/test/tbb-preview/resource_limiting_test.cpp +++ b/test/tbb-preview/resource_limiting_test.cpp @@ -2,11 +2,13 @@ #include "phlex/utilities/thread_counter.hpp" #include "catch2/catch_test_macros.hpp" +#include "fmt/std.h" #include "oneapi/tbb/flow_graph.h" #include "spdlog/sinks/basic_file_sink.h" #include "spdlog/spdlog.h" #include +#include #include #include #include @@ -44,8 +46,90 @@ namespace { spdlog::set_default_logger(logger); } + template + class managed_thread { + enum class states : std::uint8_t { off, drive, park }; + + F f_; + std::optional> args_; + std::jthread thread_; + std::binary_semaphore work_ready_{0}; + std::binary_semaphore slot_ready_{0}; + std::atomic gear_ = states::off; + + void run(std::stop_token st) + { + while (true) { + work_ready_.acquire(); + if (gear_ == states::park || st.stop_requested()) { + break; + } + + std::apply(f_, args_.value()); + slot_ready_.release(); + } + } + + public: + managed_thread(F f) : f_{std::move(f)}, thread_{[this](std::stop_token st) { run(st); }} {} + + ~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(); + } + }; + } // 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; + + managed_thread_type t1{f}; + managed_thread_type t2{f}; + flow::resource_limiter 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> geant4{ + g, + flow::unlimited, + std::tie(pinned_thread_resource), + [](unsigned int const i, auto&, managed_thread_type* geant4) { + spdlog::info("Launching from thread {} with argument {}", std::this_thread::get_id(), i); + geant4->execute(i); + }}; + + make_edge(src, geant4); + + src.activate(); + g.wait_for_all(); +} + TEST_CASE("Serialize functions based on resource", "[multithreading]") { setup_file_logger("serialize_functions_based_on_resource"); From 5a55d63e997a2277f12491030888232ae4289913 Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Thu, 23 Jul 2026 14:01:01 -0500 Subject: [PATCH 2/4] Include Opus 4.8 analysis of pinned-thread resource vs. OscarMTProducer --- .../geant4-pinned-thread-resource-2026-07.md | 176 ++++++++++++++++++ test/tbb-preview/resource_limiting_test.cpp | 8 +- 2 files changed, 180 insertions(+), 4 deletions(-) create mode 100644 docs/dev/geant4-pinned-thread-resource-2026-07.md diff --git a/docs/dev/geant4-pinned-thread-resource-2026-07.md b/docs/dev/geant4-pinned-thread-resource-2026-07.md new file mode 100644 index 000000000..ea903c55b --- /dev/null +++ b/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. diff --git a/test/tbb-preview/resource_limiting_test.cpp b/test/tbb-preview/resource_limiting_test.cpp index 44e56c2b8..4121924b2 100644 --- a/test/tbb-preview/resource_limiting_test.cpp +++ b/test/tbb-preview/resource_limiting_test.cpp @@ -115,16 +115,16 @@ TEST_CASE("std::threads as limited resources") return 0u; }}; - flow::resource_limited_node> geant4{ + flow::resource_limited_node> geant4_node{ g, flow::unlimited, std::tie(pinned_thread_resource), - [](unsigned int const i, auto&, managed_thread_type* geant4) { + [](unsigned int const i, auto&, managed_thread_type* geant4_resource) { spdlog::info("Launching from thread {} with argument {}", std::this_thread::get_id(), i); - geant4->execute(i); + geant4_resource->execute(i); }}; - make_edge(src, geant4); + make_edge(src, geant4_node); src.activate(); g.wait_for_all(); From fd0a92b612e368c83fd61ff7a4c3f1c73be6349d Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Thu, 23 Jul 2026 15:29:20 -0500 Subject: [PATCH 3/4] Move the limited-threads test to its own file --- test/tbb-preview/CMakeLists.txt | 41 ++++++- .../resource_limited_threads_test.cpp | 114 ++++++++++++++++++ test/tbb-preview/resource_limiting_test.cpp | 84 ------------- 3 files changed, 153 insertions(+), 86 deletions(-) create mode 100644 test/tbb-preview/resource_limited_threads_test.cpp diff --git a/test/tbb-preview/CMakeLists.txt b/test/tbb-preview/CMakeLists.txt index 4fd8abb10..2353517c4 100644 --- a/test/tbb-preview/CMakeLists.txt +++ b/test/tbb-preview/CMakeLists.txt @@ -35,10 +35,29 @@ 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 @@ -46,7 +65,25 @@ if(HAVE_TBB_RESOURCE_LIMITING) 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() diff --git a/test/tbb-preview/resource_limited_threads_test.cpp b/test/tbb-preview/resource_limited_threads_test.cpp new file mode 100644 index 000000000..dcbe0450f --- /dev/null +++ b/test/tbb-preview/resource_limited_threads_test.cpp @@ -0,0 +1,114 @@ +#include "catch2/catch_test_macros.hpp" +#include "fmt/std.h" +#include "oneapi/tbb/flow_graph.h" +#include "spdlog/spdlog.h" + +#include +#include +#include +#include + +using namespace oneapi::tbb; + +namespace { + + template + class managed_thread { + enum class states : std::uint8_t { off, drive, park }; + + F f_; + std::optional> args_; + std::jthread thread_; + std::binary_semaphore work_ready_{0}; + std::binary_semaphore slot_ready_{0}; + std::atomic 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; + } + + std::apply(f_, args_.value()); + slot_ready_.release(); + } + } + + public: + explicit managed_thread(F f) : + f_{std::move(f)}, thread_{[this](std::stop_token st) { + try { + run(st); + } catch (...) { + exception_ = std::current_exception(); + } + }} + { + } + 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(exception_); + } + } + }; + +} // 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; + + managed_thread_type t1{f}; + managed_thread_type t2{f}; + flow::resource_limiter 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> 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(); +} diff --git a/test/tbb-preview/resource_limiting_test.cpp b/test/tbb-preview/resource_limiting_test.cpp index 4121924b2..111480526 100644 --- a/test/tbb-preview/resource_limiting_test.cpp +++ b/test/tbb-preview/resource_limiting_test.cpp @@ -2,13 +2,11 @@ #include "phlex/utilities/thread_counter.hpp" #include "catch2/catch_test_macros.hpp" -#include "fmt/std.h" #include "oneapi/tbb/flow_graph.h" #include "spdlog/sinks/basic_file_sink.h" #include "spdlog/spdlog.h" #include -#include #include #include #include @@ -46,90 +44,8 @@ namespace { spdlog::set_default_logger(logger); } - template - class managed_thread { - enum class states : std::uint8_t { off, drive, park }; - - F f_; - std::optional> args_; - std::jthread thread_; - std::binary_semaphore work_ready_{0}; - std::binary_semaphore slot_ready_{0}; - std::atomic gear_ = states::off; - - void run(std::stop_token st) - { - while (true) { - work_ready_.acquire(); - if (gear_ == states::park || st.stop_requested()) { - break; - } - - std::apply(f_, args_.value()); - slot_ready_.release(); - } - } - - public: - managed_thread(F f) : f_{std::move(f)}, thread_{[this](std::stop_token st) { run(st); }} {} - - ~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(); - } - }; - } // 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; - - managed_thread_type t1{f}; - managed_thread_type t2{f}; - flow::resource_limiter 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> 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("Serialize functions based on resource", "[multithreading]") { setup_file_logger("serialize_functions_based_on_resource"); From 703c4717af2433afbbf51f916d2bbff5e9f02cee Mon Sep 17 00:00:00 2001 From: Kyle Knoepfel Date: Thu, 23 Jul 2026 16:23:10 -0500 Subject: [PATCH 4/4] [Checkpoint] --- .../resource_limited_threads_test.cpp | 62 ++++++++++++++++--- 1 file changed, 53 insertions(+), 9 deletions(-) diff --git a/test/tbb-preview/resource_limited_threads_test.cpp b/test/tbb-preview/resource_limited_threads_test.cpp index dcbe0450f..82ede1e4d 100644 --- a/test/tbb-preview/resource_limited_threads_test.cpp +++ b/test/tbb-preview/resource_limited_threads_test.cpp @@ -4,8 +4,10 @@ #include "spdlog/spdlog.h" #include +#include #include #include +#include #include using namespace oneapi::tbb; @@ -32,20 +34,18 @@ namespace { break; } - std::apply(f_, args_.value()); + 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) { - try { - run(st); - } catch (...) { - exception_ = std::current_exception(); - } - }} + f_{std::move(f)}, thread_{[this](std::stop_token st) { run(st); }} { } managed_thread(managed_thread const&) = delete; @@ -69,8 +69,9 @@ namespace { work_ready_.release(); slot_ready_.acquire(); args_.reset(); + if (exception_) { - std::rethrow_exception(exception_); + std::rethrow_exception(std::exchange(exception_, nullptr)); } } }; @@ -112,3 +113,46 @@ TEST_CASE("std::threads as limited resources") 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; + + managed_thread_type t1{f}; + managed_thread_type t2{f}; + flow::resource_limiter 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> 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(); +}