perf(workflows): reuse one run-scoped thread pool for executor-less runs - #3052

Reviewer quick-sheet

• Diff size: the headline counter reads +1,284/−34, but +1,152 of that is one new test file (17 ownership and behavior tests). The actual engine change nets +28 lines in executor/core.py and +13 in executor/utils.py of code — 75 added and 34 removed, much of it re-indentation for the nested cleanup — plus 50 lines of in-code boundary comments.
• Blast radius: hosted and streaming paths are untouched — flat_dispatch defaults to False, they keep their existing batched dispatch, and neither http_api.py nor inference_pipeline.py appears in this diff.
• Verification: falsification 11F/6P against the unpatched base, the context-snapshot mutation fails exactly the two isolation detectors, package suite 6607 + root workflow suite 1238 green, make check_code_quality exit 0 — details below.

Summary

Executor-less workflow runs now reuse one ThreadPoolExecutor for the duration of a run instead of constructing and destroying a pool for every DAG wave.

This is the default path for standalone roboflow-workflows, SDK, and embedded usage. Hosts that already provide an executor, including the HTTP server and streaming pipeline, retain their existing behavior.

In a six-round counterbalanced benchmark, pool construction fell from one pool per wave to one pool per run:

Single-wave, host-provided-executor, and compute-heavy runs were unchanged within noise. The practical saving on the gallery workflow is about 0.09 ms per run; deep orchestration-bound workflows benefit most.

Why

When **executor=**None****, run_steps_in_parallel previously created a fresh pool inside a with block for every wave. A sequential 10-step workflow therefore paid for ten pool construction, thread startup, shutdown, and join cycles.

The standalone package extraction made this executor-less path a normal public use case. The engine is also the only component that can own this pool without requiring callers to implement a new executor lifecycle.

This does not overlap with #2448, #2940, or #2947. Those changes address server-side contention in a shared host executor during model loading. This PR only changes the path where no host executor is supplied.

Changes

• _run_workflow lazily creates one run-owned pool when the first wave executes and shuts it down in finally.
• flush_stream_pipeline_workflow uses the same ownership model.
• run_steps_in_parallel gains flat_dispatch=False:
  • run-owned pools submit the complete wave, preserving the old executor-less scheduling;
  • host-provided pools retain the existing batched dispatch;
  • if a step fails, unstarted siblings are cancelled and in-flight siblings are drained before the exception leaves the function, matching the old per-wave with behavior.
• Pool shutdown occurs before stream-pipeline closure. Nested cleanup ensures stream pipelines still close if shutdown raises.
• The package changelog records the execution-engine change.

Compatibility and ownership boundary

• Hosted and streaming paths: unchanged. Both already provide an executor, so the engine neither creates nor shuts down a pool.
• Concurrency: remains bounded by max_concurrent_steps.
• Scheduling: executor-less waves remain flat-dispatched; host pools remain batched.
• Context isolation: every submitted step still runs in its own contextvars snapshot and receives the existing framework context rebinds.
• Lifetime: engine-owned workers exist for one workflow run and cannot leak thread-local state across runs.
• Third-party thread locals: a block that leaves arbitrary threading.local state uncleared may expose it to another step in the same run. Hosted executors already have this reuse profile; block state is expected to live on block instances or managed caches.
• Failure behavior: a failed wave drains its in-flight work before propagation, later waves are not scheduled, and the owned pool is shut down exactly once.

Verification

Added 17 focused ownership and behavior tests covering:

• one owned pool per multi-wave run;
• no engine-owned pool when a host executor is supplied;
• result parity between owned and supplied executors;
• flat owned dispatch versus existing hosted batching;
• shutdown and join behavior on success and failure;
• shutdown-before-stream-close ordering;
• failure-path sibling cancellation and drain;
• flush and deferred-flush paths;
• run-scoped lifetime and consecutive-run recovery;
• ContextVar, request-thread, and cross-run thread-local isolation;
• exception abort and conditional branch termination.

Falsification against the unpatched implementation produces 11 failures / 6 passes in this file. The patched implementation produces 17 passes. Bypassing the context snapshot fails the two isolation detectors.

Repository checks run locally:

• cd workflows && pytest tests/unit_tests: 6607 passed, 113 skipped
• root workflow unit tests: 1238 passed, 30 skipped
• make check_code_quality: exit 0
• standalone wheel isolation probe: 16/16 checks passed

Risks and rollback

The behavioral change is that worker lifetime extends from one wave to one run on the executor-less path. Framework context remains task-isolated, and the pool is destroyed at run completion. The hosted path is untouched.

The test suite currently inspects CPython ThreadPoolExecutor internals (_shutdown and _threads) for several lifecycle assertions. These attributes are stable across the supported Python 3.10–3.12 interpreters; they can be replaced with recorded-shutdown fixtures if maintainers prefer avoiding internal attributes. Happy to swap the private CPython attribute assertions for public recorded-shutdown mocks if you prefer zero reliance on internal interpreter attributes.

Rollback is a direct revert of this PR, restoring one pool per wave.

Scope

Four files: the implementation in executor/core.py and executor/utils.py, one 17-test ownership suite, and the changelog entry. No server, streaming-pipeline, model-loading, or public engine lifecycle code is changed.