[DataFlow runtime · online] O1.2 — named builder + interleaved async loop

[maocheng23]

Stage O1.2 of the online roadmap (#618): the named single-process runtime builder and the interleaved async loop that replaces drain-then-fit. Stacked on O1.1 (#624) — review/merge that first; the diff here is against the O1.1 branch.

Problem

The colocated online path was drain-then-fit: generate the whole prompt pool, then train. (Root cause: the in-process SampleRefQueue.get is non-blocking, so the loader ended the instant the queue was momentarily empty.)

Change

• run_disagg_online_interleaved: the producer streams refs on a daemon thread while trainer.fit consumes on the main thread (StreamingRefQueue blocks until closed-and-drained, so the trainer tracks the producer). Symmetric, hang-free shutdown:
  • trainer finishes first (e.g. max_steps) → a cooperative should_stop winds the producer down instead of blocking on the in-flight watermark after the consumer stops draining;
  • producer finishes first → closes the channel, the loader drains the tail;
  • producer raises → channel closed in finally (consumer can't hang) + the exception re-raised on the main thread.
• build_disagg_online_eagle3_runtime: the named single-process builder from the roadmap; composes producer + consumer over one shared metadata store, one consume-once feature store, and a producer/consumer StreamingRefChannel pair, returning (trainer, loader, run). Uses the in-process generate_eagle3_data stub — no live SGLang server and no Ray (those are O1.3 / O2).
• drive_producer gains a cooperative should_stop predicate + a finally-close so it never strands the consumer.

Tests

tests/test_runtime/test_disagg_online_interleave.py (CPU): the three shutdown paths (full-drain-then-terminate, trainer-stops-first cooperative wind-down, producer-exception propagation + consumer unblock). A GPU end-to-end test for the named builder's interleaved run() is added to test_disagg_online_launch.py.

Next

O1.3 — live SGLang-server hidden-state capture (the roadmap's gating risk; needs a GPU + the capture spike first).

🤖 Generated with Claude Code

[gemini-code-assist]

Warning

You have reached your daily quota limit. Please wait up to 24 hours and I will start processing your requests again!

[gemini-code-assist]

Warning

You have reached your daily quota limit. Please wait up to 24 hours and I will start processing your requests again!

[maocheng23]

Code review

No high-confidence issues found. Checked for bugs in the interleaved producer/trainer shutdown paths, backpressure stop handling, and named online runtime wiring.

[maocheng23]

Addressed review feedback (self-review pass).

• run_disagg_online_interleaved: hardened shutdown. After thread.join(timeout) it now checks is_alive() and raises instead of silently returning with a leaked, still-running daemon producer; a trainer exception no longer masks a producer exception (they are chained, producer surfaced as root cause); a producer failure raised after the join no longer goes unreported.

Validated: full tests/test_runtime = 200 OK (2 skipped, 1 xfail), zero failures, on a 2-node H200 pod. Lint clean (black 24.10.0 / isort 5.13.2 / autoflake).