Event-driven scheduler: remove 10ms poll floor and 1s dependency-chain stalls

[andre-merzky]

Summary

The run loop detected component completion only by polling future.done() once per pass, with a hardcoded asyncio.sleep(0.01) after every active pass. Two consequences:

• ~10ms latency floor per task: submission of new work waits out the tail sleep of the previous pass (measured 10.9ms round-trip for a no-op function task on LocalExecutionBackend).
• ~1s stall per dependency edge: a dependent behind a task running longer than one poll interval is only submitted when the idle event-wait hits its 1.0s timeout, since nothing woke the loop on completion (measured 1.03s for a slow(50ms) -> fast chain).

This PR makes completion wake the loop through the existing _component_change_event: every component future gets a done callback that sets the event (asyncio future callbacks run in the event loop, so this is thread-safe), and the post-activity sleep becomes a bare yield. The idle branch still waits on the same event with its existing 1s safety timeout as a backstop.

Measured after the change: no-op task round-trip 10.9ms → 2.0ms (raw ProcessPoolExecutor floor is 1.5ms); 50ms-task → dependent chain 1.03s → 0.05s.

Dependency tracking fix (first commit)

Removing the poll interval exposed a latent correctness bug, fixed separately in the first commit: dependency detection only scanned positional args for futures, while _extract_dependency_values resolves futures in args and kwargs. A future passed as a kwarg was therefore never registered as a dependency, and its consumer could be submitted while the producer was still running (InvalidStateError: Result is not set). This is reachable on current main with any kwarg producer slower than one poll interval; the faster loop made it visible even for no-op producers. The commit also deduplicates dependencies (the same future passed twice previously double-counted and deadlocked the consumer).

Tests

• tests/unit/test_scheduler_responsiveness.py: chain behind a 50ms task completes well under the 1s event-wait; sequential no-op round-trip beats the former 10ms poll floor. Both fail on current main (1.03s / 11.8ms) and pass here.
• tests/unit/test_dependencies_detection.py: kwarg-future producer is awaited before its consumer runs (fails on main with InvalidStateError); the same future passed positionally and as kwarg completes (guards the dedup).
• Full unit suite green 10×/10 (the kwarg race previously flaked ~1/6 with the fast loop); integration suite green; ruff clean on all touched files.

Context

Found while profiling ROSE's streaming learner: at batch size 1 the end-to-end window rate was pinned at ~45/s by the 22ms latency of its two chained no-op tasks — entirely scheduler sleep, not execution.

🤖 Generated with Claude Code

[Copilot]

Pull request overview

This PR updates the workflow scheduler to be event-driven (waking promptly on component completion) and fixes dependency detection so futures passed via kwargs are correctly tracked and duplicate dependencies don’t deadlock.

Changes:

• Register done-callbacks on component futures to set _component_change_event and replace the fixed 10ms post-activity sleep with an asyncio.sleep(0) yield.
• Fix dependency detection to include kwargs values and deduplicate repeated dependency futures.
• Add unit regression tests covering responsiveness (no 10ms latency floor / no 1s dependency-chain stalls) and kwarg/duplicate dependency tracking.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
src/radical/asyncflow/workflow_manager.py	Event-driven run-loop wakeups on completion; kwargs dependency detection and dependency deduplication.
tests/unit/test_scheduler_responsiveness.py	New latency/regression tests for scheduler responsiveness improvements.
tests/unit/test_dependencies_detection.py	New tests for kwarg dependency tracking and dedup behavior; updated imports.

Suppressed comments (1)

tests/unit/test_dependencies_detection.py:82

• Same cleanup issue here: if the test fails before the last line, flow.shutdown() won’t run. Use try/finally to guarantee shutdown (and avoid leaking executors across tests).

    backend = await LocalExecutionBackend()
    flow = await WorkflowEngine.create(backend=backend)

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.