Schedulable Pipeline Engine可调度流水线引擎

2.1 Three layers2.1 三层结构

2.2 Module dependency2.2 模块依赖

# Engine core — pure stdlib + torch only
commons.pipeline.engine
├── task.py       # Task, DataSlot
├── context.py    # SlotStore, BatchRing, TaskContext (thread-local)
├── streams.py    # StreamPool with anchor_device + use(stream) ctx
├── schedule.py   # Schedule, Stage
├── deps.py       # infer_cross_stream_waits
├── pipeline.py   # SchedulablePipeline driver + _seed_first_batch
├── executor.py   # Sequential + Threaded + _NcclOrderedLock
├── _presets.py   # T1/T2 forward/backward/optimizer task factories
└── autosched/
    ├── validator.py        # 8-rule SPEC §4.2 schedule validity
    ├── cost_model.py       # CostProfiler, CostModel (JSON roundtrip)
    └── list_scheduler.py   # Critical-path list scheduler

# HSTU adapter — depends on torchrec, megatron, dynamicemb
commons.pipeline.hstu_pipeline
├── pipeline.py   # HSTUPipeline + HSTU_DEFAULT_THREAD_MAP
├── tasks.py      # 14 task factories + PipelineState
└── factory.py    # HSTUPipelineFactory

2.3 Hard invariants2.3 硬性不变量

• Engine framework-free引擎层框架无关: commons.pipeline.engine.* never imports torchrec, megatron, fbgemm_gpu, or commons.distributed.*. Enforced by test_engine_import_hygiene.py. commons.pipeline.engine.* 不引入 torchrec、megatron、fbgemm_gpu 或 commons.distributed.*。由 test_engine_import_hygiene.py 强制约束。
• Legacy untouched遗留代码字节级不动: train_pipeline.py, train_pipeline_factory.py, utils.py stay byte-identical. Enforced by test_engine_legacy_untouched.py. train_pipeline.py、train_pipeline_factory.py、utils.py 保持字节级一致。由 test_engine_legacy_untouched.py 强制约束。
• v1-only contexts in HSTUHSTU 仅允许 v1 context: create_torchrec_ctx asserts ctx.version == 1. Enforced by test_no_v0_context_references_in_hstu_pipeline (regex scan). create_torchrec_ctx 断言 ctx.version == 1。由 test_no_v0_context_references_in_hstu_pipeline（正则扫描）强制约束。
• Out-of-slot shared state needs an engine-visible happens-beforeslot 之外的共享状态需要引擎可见的 happens-before: Rule. Engine only inserts sync for edges it can see: same-slot reads/writes, matching-la depends_on, or same_progress_sync. Anything outside slots — module attributes, PipelineState fields, globals — is invisible. If two tasks mutate the same out-of-slot object without one of those declared edges, author must colocate them on one thread via thread_map (same thread = serial, no race). Convention only; no runtime check.
  
  Example. Two tasks both write module.forward._context: start_input_dist (la=1) and forward (la=0). Within one progress, ordering is fine — forward reads torchrec_ctx that start_input_dist wrote (slot edge). But across progresses forward(P_n) and start_input_dist(P_n) actually run concurrently in the same progress on different lookaheads, touching different batches' contexts but the same Python attribute → race. Fix used: drop the shared mutation entirely (PostProc cleanup); the colocation guard that previously enforced "put both on one thread" was deleted along with it. 规则。引擎只对它能看见的边加 sync：同 slot 的 reads/writes、la 相等的 depends_on、或 same_progress_sync。slot 之外的东西—— module 属性、PipelineState 字段、全局变量——引擎看不见。如果两个 task 改 slot 之外的同一个对象、又没声明上述任意一种边，必须用 thread_map 放到同一个线程（单线程内自然串行）。约定式约束，运行时不检查。
  
  例子。两个 task 都写 module.forward._context：start_input_dist(la=1) 与 forward(la=0)。同一次 progress 内顺序没问题——forward 读 start_input_dist 写过的 torchrec_ctx slot（slot 边）。但 P_n 里的 forward 与 start_input_dist 同 progress 不同 la 并发跑，触碰不同批次的 context 但是同一个 Python 属性 → 竞态。修法：直接砍掉这个共享 mutation（PostProc 清理），原本"把两者放同线程"的运行时校验也一并删除。

3.1 Task & DataSlot3.1 Task 与 DataSlot

A Task is a unit of work with declarative metadata for the engine to schedule: Task 是一个带声明式元数据的工作单元，由引擎进行调度：

class Task:
    name: str
    stream: str                                 # named CUDA stream context
    priority: int                              # NVTX priority hint
    lookahead: int                             # SPEC_p4 v2: 0 = current batch, k = k iters ahead
    reads: Tuple[DataSlot, ...]                # slots consumed (str shorthand auto-wraps)
    writes: Tuple[DataSlot, ...]               # slots produced
    depends_on: Tuple[str, ...]                # SAME-batch logical edge: this task on
                                                 #   batch K waits for X to have processed K
    cross_iter_depends_on: Tuple[Tuple[str, int], ...]
                                                 # DIFFERENT-batch logical edge: ("X",) is
                                                 #   shorthand for (X, -1); ("Y", -N) for N≠1.
                                                 #   Positive/zero offsets rejected.
    same_progress_sync: Tuple[str, ...]        # SAME-progress GPU/stream coherency wait,
                                                 #   regardless of which batches X/self process
                                                 #   (mirrors legacy default_stream.wait_stream)
    nccl: bool                                 # participate in NCCL ordered lock
    nvtx_tag: Optional[str]                      # auto-emitted via nvtx.annotate

    def run(self, ctx: TaskContext) -> None: ...

SPEC_p4 v3 user API note: the three dependency fields encode three distinct intents and are not interchangeable. depends_on is for same-batch logical waits (e.g. backward.depends_on=("zero_grad",)); cross_iter_depends_on is for "wait for X from N batches ago" (rare in HSTU); same_progress_sync is for "wait for X's GPU work in this progress to finish" (e.g. backward.same_progress_sync=("prefetch_embeddings",) in the prefetch variant — a stream coherency wait, not a logical batch dependency). Names mentioned in two fields raise at Task.__init__. Subclasses can set the three as class attributes; normalization runs unconditionally so bare-name shorthand works in both kwargs and class-attr forms. SPEC_p4 v3 用户 API 说明：三个依赖字段表达三种不同的意图，不可互换。depends_on 用于同 batch 的逻辑等待（如 backward.depends_on=("zero_grad",)）；cross_iter_depends_on 用于"等 X 处理 N 个 batch 之前的工作"（HSTU 暂无用例）；same_progress_sync 用于"等 X 在本次 progress 内的 GPU 工作做完"（如 prefetch 变体的 backward.same_progress_sync=("prefetch_embeddings",) —— 是 stream 协同等待，不是逻辑 batch 依赖）。同一个名字出现在两个字段会在 Task.__init__ 抛错。子类可以以类属性形式设置三者；规范化无条件运行，bare-name 简写在 kwargs 和 class-attr 两种写法下都生效。

SPEC_p4 v2 user API note (commits 2060ee62, 9a2c7a12, faa7a30f): the legacy batch_offset=N kwarg is removed (Phase C hard break). Pass lookahead=N; reads / writes accept bare strings. Pseudo-slot names (e.g. module_input_post_prefetch) are no longer required to model HSTU's prefetch→forward mutation chain. SPEC_p4 v2 用户 API 说明（提交 2060ee62、9a2c7a12、faa7a30f）：legacy 的 batch_offset=N 形参已移除（Phase C hard break）。改用 lookahead=N；reads / writes 直接传字符串。HSTU 的 prefetch→forward mutation chain 不再需要 pseudo-slot 名（如 module_input_post_prefetch）。

DataSlot is an opaque named handle: (name: str, batch_offset: int). Two slots are equal iff both fields match. Cross-iter data flow happens via BatchRing.advance() shifting the slot store rather than copying values. DataSlot 是一个不透明的命名句柄：(name: str, batch_offset: int)。两个 slot 相等当且仅当两个字段都相等。跨迭代的数据流是通过 BatchRing.advance() 平移 slot store 实现的，并不真的拷贝数据。

3.2 Schedule & Stage3.2 Schedule 与 Stage

A Schedule is a frozen DAG declaration: Schedule 是一个冻结的 DAG 声明：

schedule = Schedule(
    stages=(
        Stage(tasks=(h2d_task, fwd_task, bwd_task, opt_task)),
    ),
    stream_slots=("default", "memcpy"),
)

Stages run in the order they appear in Schedule.stages. The tasks within a single-stage schedule are reordered at SchedulablePipeline.__init__ by topological_sort() (autosched/deps.py): edges come from exact-slot reads/writes match, depends_on with matching lookahead, and same_progress_sync; cross_iter_depends_on and cross-lookahead depends_on contribute no in-progress edge (cross-iter is satisfied by ring rotation). Author-declaration order is the tie-break for incomparable tasks. The §4.8 mask + executor parallelism choice still gate runtime triggering. in_flight_batches is derived from max(task.lookahead) + 1. Stage 按 Schedule.stages 中出现的顺序运行。单 stage 内的任务由 SchedulablePipeline.__init__ 调用 topological_sort()（autosched/deps.py）重新排序：边来自 reads/writes 精确 slot 匹配、lookahead 相等的 depends_on、以及 same_progress_sync；cross_iter_depends_on 与跨 lookahead 的 depends_on 不入边（跨迭代由 ring 平移满足）。声明顺序作为不可比任务之间的 tie-break。运行期触发仍由 §4.8 掩码与执行器并行策略约束。in_flight_batches 由 max(task.lookahead) + 1 推导。

The validator (autosched/validator.py) enforces 8 rules at SchedulablePipeline.__init__: validator（autosched/validator.py）在 SchedulablePipeline.__init__ 时强制执行 8 条规则：

3.3 BatchRing3.3 BatchRing 批次环

The ring holds N in-flight batches' slot stores. at(k) is the slot store at offset k from current. advance() drops offset 0, shifts each store down by one, and appends a fresh empty store at the highest offset. 环里持有 N 个在飞批次的 slot store。at(k) 返回距当前偏移 k 的 slot store。advance() 丢弃 offset 0，把所有 store 向下平移一格，并在最大偏移处追加一个空的新 store。

3.4 §4.8 mask formula3.4 §4.8 触发掩码公式

Determines whether a task at lookahead = k fires at iteration i (max_lookahead = max(task.lookahead) across the schedule): 决定 lookahead = k 的任务是否在迭代 i 触发（max_lookahead 取 schedule 中所有 task 的 lookahead 最大值）：

fire iff   (max_lookahead - k) ≤ i < M_known + (max_lookahead - k)

where M_known = pulled while pulling is live, otherwise the final M. 其中 M_known 在还在拉数据时等于已拉数 pulled，拉完之后等于最终的 M。

• The prefill phase (i < max_lookahead) runs only deep-lookahead tasks; compute @ lookahead=0 doesn't fire yet. 预热阶段（i < max_lookahead）只跑 lookahead 较深的任务；lookahead=0 上的 compute 还不触发。
• The steady phase (max_lookahead ≤ i < M) runs all qualifying tasks each iter. 稳态阶段（max_lookahead ≤ i < M）每轮都跑所有满足条件的任务。
• The drain phase (i ≥ M) runs only shallow-lookahead tasks until the ring is empty. 排空阶段（i ≥ M）只跑 lookahead 较浅的任务，直到环清空。

Engine raises StopIteration when no future iteration can satisfy any task's mask. 当未来任何迭代都不能让任意任务的掩码成立时，引擎抛出 StopIteration。

3.5 SchedulablePipeline driver3.5 SchedulablePipeline 驱动器

Public API surface (8 symbols total at commons.pipeline.engine): 公开 API（commons.pipeline.engine 共 8 个符号）：

3.6 Event-sync graph & slot_offset3.6 事件同步图与 slot_offset

deps.infer_cross_stream_event_deps(schedule) compiles the four user-facing dependency declarations (reads/writes, depends_on, cross_iter_depends_on, same_progress_sync) into a flat per-consumer list of (producer, producer_stream, slot_offset) triples. At runtime the executor performs ring.at(slot_offset).get_event(producer) + consumer.wait_event(event). Two views below clarify (a) which declarations become topological-sort edges vs. only event-emit edges, and (b) how each edge's slot_offset is derived from ring rotation. deps.infer_cross_stream_event_deps(schedule) 把用户层四种依赖声明（reads/writes、depends_on、cross_iter_depends_on、same_progress_sync）编译成每个 consumer 的扁平三元组列表 (producer, producer_stream, slot_offset)。运行期 executor 执行 ring.at(slot_offset).get_event(producer) + consumer.wait_event(event)。下面两张图分别说明：(a) 哪些声明会成为 topological_sort 的边、哪些只 emit 事件；(b) 每种边的 slot_offset 如何由 ring 旋转推出。

Common structure: producer always records its event at ring.at(producer.batch_offset) (executor.py:165) in its own progress call. The consumer's slot_offset answer is "where has that slot rotated to by the time consumer runs" — and the four formulas above are just the four answers when ring rotation count is 0 (same_progress_sync), X.la − consumer.la (depends_on cross-la), N + X.la − consumer.la (cross_iter), or directly given by read_slot.batch_offset (reads/writes). 共同结构：producer 永远在本次 progress 调用中往 ring.at(producer.batch_offset)（executor.py:165）录 event。consumer 的 slot_offset 答案就是"那个 slot 在 consumer 跑的时候已经转到哪儿"——上面四个公式分别是 ring 旋转次数 = 0（same_progress_sync）、X.la − consumer.la（depends_on 跨 la）、N + X.la − consumer.la（cross_iter）、由 read_slot.batch_offset 直接给出（reads/writes）这四种场景。

3.7 same_progress_sync — when to use it3.7 same_progress_sync 的使用场景

Reads as: "this task must wait for X's work in this same progress() call to finish, regardless of which batches X and self are processing." Engine guarantees: 语义："本 task 等 X 这同一次 progress() 内的工作做完，不管 X 和自己处理的是哪个 batch。"引擎承诺：

• Adds a topo-DAG edge X → self (X is guaranteed to fire before self within the progress).加一条 topo-DAG 边 X → self（保证 X 在 progress 内先于 self 触发）。
• For threaded execution, adds a CPU-side threading.Event wait (cross-thread case).线程执行时跨线程加一条 CPU 端 threading.Event 等待。
• For cross-stream, emits wait_event at slot[producer.la]; same-stream relies on CUDA stream FIFO.跨流时在 slot[producer.la] emit wait_event；同流靠 CUDA stream FIFO 保序。

Three known use cases: 三种已知用法：

Direction convention. The field lives on the consumer side; X.same_progress_sync=() always lists producers the consumer is waiting on. producer.la and consumer.la may differ — the field is la-agnostic (in contrast to depends_on which requires producer.la ≥ consumer.la to avoid future-read). 方向约定。字段写在 consumer 一侧；X.same_progress_sync=() 列的都是 X 等待的 producer。producer.la 与 consumer.la 可以不相等——这个字段对 lookahead 不敏感（区别于 depends_on 必须 producer.la ≥ consumer.la 以避免 future-read）。

Distinguishing from depends_on and cross_iter_depends_on: 与 depends_on 和 cross_iter_depends_on 的区别：

• depends_on=("X",) — same-batch logical wait. Both must process the same batch K. Constrained by producer.la ≥ consumer.la.depends_on=("X",) —— 同 batch 逻辑等待。两者必须处理同一个 batch K。约束 producer.la ≥ consumer.la。
• cross_iter_depends_on=((X, -N),) — different-batch wait. "Consumer on batch K waits for X on batch K−N." Δ ≥ 1 is the cross-progress, ring-rotated case; Δ=0 (i.e. producer.la + N == consumer.la) is auto-promoted to the same_progress_sync mechanical contract — same topo edge, CPU edge, and current-progress event lookup at slot[producer.la] (per SPEC_cross_iter_delta0_autoconvert.md).cross_iter_depends_on=((X, -N),) —— 跨 batch 等待。"consumer 在 batch K 时等 X 在 batch K−N 的工作"。Δ ≥ 1 走 ring-rotated 跨 progress 路径；Δ=0（即 producer.la + N == consumer.la）会被自动 promote 为 same_progress_sync 的机制契约——相同的 topo 边、CPU 边、以及在 slot[producer.la] 上 lookup 当前 progress 的 event（详见 SPEC_cross_iter_delta0_autoconvert.md）。
• same_progress_sync=("X",) — same-progress wait, batch-agnostic. The two ARE allowed to process different batches; what matters is X completes its current-progress GPU work before consumer reads.same_progress_sync=("X",) —— 同 progress 等待，不关心 batch。两者允许处理不同 batch；重点是 X 在本次 progress 的 GPU 工作完成后 consumer 才读。

4.1 thread_map strategies4.1 thread_map 策略

Threads and CUDA streams are decoupled: a task's stream selects the CUDA stream context inside StreamPool.use(); thread_map selects the CPU worker that submits the task. The same stream may be submitted from different threads (different iterations or — with caution — same iteration). 线程与 CUDA 流是 解耦的：任务的 stream 字段决定 StreamPool.use() 内的 CUDA 流上下文；thread_map 决定提交任务的 CPU 线程。同一个流可以来自不同线程（来自不同迭代，或谨慎地来自同一迭代）。

4.2 CPU dependency computation4.2 CPU 依赖计算

_compute_cpu_deps builds a dict {task_name → [predecessor_completion_events]} by walking the topo-sorted active task list and adding four edge types: _compute_cpu_deps 沿拓扑序遍历活跃任务，构造 {task_name → [predecessor_completion_events]} 字典，添加四类边：

1. Slot-based基于 slot: any writer of a slot the task reads (excluding self). 该任务所读 slot 的任意 writer（自己除外）。
2. Same-batch logical (depends_on)同 batch 逻辑（depends_on）: every task.depends_on name with matching lookahead that is in active. Cross-lookahead names go through ring rotation as a cross-stream wait, not a CPU edge. task.depends_on 中且 lookahead 相等、又在活跃集合中的任务名。跨 lookahead 的名字走 ring 平移变成跨流 wait，不进 CPU 边。
3. Same-progress sync (same_progress_sync)同 progress 协同（same_progress_sync）: every task.same_progress_sync name that is in active — same-progress GPU coherency wait, regardless of lookahead. task.same_progress_sync 中且在活跃集合中的任务名 —— 同一次 progress 内的 GPU 协同等待，与 lookahead 无关。
4. Same-stream FIFO同流 FIFO: the most recent task on the same stream in topological order — preserves CUDA stream FIFO semantics when thread_map splits same-stream tasks across threads. 拓扑序中同一个流上的最近一个任务 —— 当 thread_map 把同流任务拆到不同线程时，这条边保住了 CUDA 流的 FIFO 语义。

The cross-thread filter then keeps only edges where thread_id_of(producer) ≠ thread_id_of(consumer); same-thread predecessors are already serialized by their thread loop. 随后通过跨线程过滤，仅保留 thread_id_of(producer) ≠ thread_id_of(consumer) 的边；同线程的前驱已经被线程内的循环串行化。

4.3 NCCL ordered lock4.3 NCCL 顺序锁

Tasks tagged nccl=True acquire a ticket-based lock that guarantees topological-order execution across all ranks. Tickets are assigned by the topo-sorted task tuple (deterministic across ranks because the schedule and tie-break are deterministic). Without this, a multi-threaded executor on different ranks may submit collectives in different orders → mismatched NCCL calls → deadlock. 标了 nccl=True 的任务要获取一把基于票据的锁，保证所有 rank 都按拓扑序执行集合通信。票据由拓扑排序后的任务元组分配（schedule 与 tie-break 都是确定性的，所以各 rank 一致）。如果没有它，多线程执行器在不同 rank 上可能以不同顺序提交集合操作 → NCCL 调用错配 → 死锁。

4.4 Cancellation & abort4.4 取消与中止

Each execute_stage call sets up: 每次 execute_stage 调用会准备：

• A threading.Event per active task — signaled when the task finishes (or its thread errors out).每个活跃任务一个 threading.Event —— 任务完成（或所在线程出错）时置位。
• A shared cancelled Event — set by the first failing thread.一个共享的 cancelled Event —— 由第一个失败的线程置位。
• An errors_lock + errors list to collect exceptions across threads.errors_lock + errors 列表用于跨线程收集异常。

The except handler in _run_thread_chain does three things in order: _run_thread_chain 的 except 分支按顺序做三件事：

1. cancelled.set() — stops further task starts on all chains.阻止所有线程链上后续任务的启动。
2. self._nccl_lock.abort() — wakes any worker blocked inside acquire(ticket > current).唤醒任何阻塞在 acquire(ticket > current) 的 worker。
3. Append exception to errors; the finally block sets every remaining task's completion event so peer threads waiting on cpu_deps Events don't deadlock.把异常追加进 errors；finally 块把剩余任务的完成事件全部置位，避免其他线程在 cpu_deps 上死锁。

After the stage barrier (future.result() on every chain), execute_stage raises errors[0]. stage 屏障（对每条链 future.result()）之后，execute_stage 抛出 errors[0]。

5.1 14 tasks5.1 14 个任务

HSTUPipeline._build_schedule assembles a single-stage schedule from these task factories. Author declaration order is just a tie-break for incomparable tasks — actual execution order is decided by topological_sort over reads/writes + depends_on + same_progress_sync. HSTUPipeline._build_schedule 用这些任务工厂拼出一个单 stage 的 schedule。作者的声明顺序只在不可比的任务之间作为 tie-break 用；真正的执行顺序由 topological_sort 基于 reads/writes + depends_on + same_progress_sync 决定。

np = non-prefetch variant, pf = prefetch variant. Both keep max_lookahead = 2 at prefetch_depth=1 (matches legacy 3-batch in-flight). For prefetch_depth=K, max_lookahead = K+1; in_flight_batches = max_lookahead + 1. np = 无 prefetch 变体，pf = prefetch 变体。两者在 prefetch_depth=1 时都是 max_lookahead = 2（与遗留的 3 批在飞匹配）。prefetch_depth=K 时 max_lookahead = K+1；in_flight_batches = max_lookahead + 1。

5.2 HSTU_DEFAULT_THREAD_MAP5.2 HSTU_DEFAULT_THREAD_MAP

Two thread groups: 两个线程组：

HSTU_DEFAULT_THREAD_MAP = {
    # io thread — pure data movement
    "h2d": "io",
    "start_shuffle": "io",
    "finish_shuffle": "io",
    # compute thread — everything that touches shared module state
    "start_input_dist": "compute",
    "wait_input_dist": "compute",
    "prefetch_embeddings": "compute",
    "zero_grad": "compute",
    "global_tokens_allreduce": "compute",
    "nccl_safety_barrier": "compute",
    "forward": "compute",
    "backward": "compute",
    "finalize_model_grads": "compute",
    "optimizer_step": "compute",
    "watchdog_step": "compute",
}

5.3 Bootstrap flow5.3 引导流程

The first call to HSTUPipeline.progress() runs a one-time bootstrap because torchrec's _rewrite_model needs a real batch for FX tracing. Sequence diagram in §6.1. 首次调用 HSTUPipeline.progress() 会执行一次性引导，因为 torchrec 的 _rewrite_model 需要一个真实 batch 做 FX 追踪。时序图见 §6.1。

Key invariants: 关键不变量：

• StreamPool built first先建 StreamPool — H2D + bootstrap collectives run on the engine's actual streams, not throwaway streams.H2D + 引导阶段的集合通信都跑在引擎真正用的流上，而非临时流。
• Real per-batch context真实的 per-batch context — bootstrap context is created by state.create_torchrec_ctx() with version=1 (the v0 single-shared-context branch is forbidden).引导 context 由 state.create_torchrec_ctx() 创建，version=1（v0 单共享 context 分支被禁用）。
• Peek batch is seeded into the ringpeek batch 被植入 ring — _pipe._seed_first_batch(...) populates ring.at(max_offset) with {batch_cpu, batch_gpu, torchrec_ctx, shuffled_batch (identity only)}; idempotent guards in h2d/start_input_dist/finish_shuffle skip the work that's already been done._pipe._seed_first_batch(...) 把 {batch_cpu, batch_gpu, torchrec_ctx, shuffled_batch（仅 identity）} 写入 ring.at(max_offset)；h2d/start_input_dist/finish_shuffle 中的幂等保护会跳过已完成的工作。
• No 1-batch loss不丢一个 batch — the peek batch flows through forward/backward/optimizer like any other batch.peek batch 与普通 batch 一样走完 forward/backward/optimizer。

5.4 native vs prefetch variant5.4 native 与 prefetch 变体

Same task list, plus or minus prefetch_embeddings. The non-prefetch variant uses PipelinedForward which reads input_dist_tensors_requests directly. The prefetch variant uses PrefetchPipelinedForward which reads from module_input_post_prefetch, populated in the previous iter by prefetch_embeddings. 两者任务列表相同，区别是有没有 prefetch_embeddings。无 prefetch 变体用 PipelinedForward，直接读 input_dist_tensors_requests；prefetch 变体用 PrefetchPipelinedForward，读上一轮 prefetch_embeddings 写入的 module_input_post_prefetch。

Critical ordering (within one progress)关键顺序（同一次 progress 内）: prefetch_embeddings must run after forward in every progress, not before. Reason: dynamicemb's _prefetch_outstanding_keys counter is incremented by prefetch() and decremented at end of forward. The ordering is enforced by backward.same_progress_sync=("prefetch_embeddings",): backward (and therefore this progress's whole compute chain after forward) waits for prefetch's GPU work in the same progress to finish — a stream coherency wait, not a batch dependency. Within one progress: prefetch_embeddings 必须在每次 progress 中跑在 forward 之后，不能反过来。原因：dynamicemb 的 _prefetch_outstanding_keys 计数器在 prefetch() 时加，在 forward 结束时减。这一顺序由 backward.same_progress_sync=("prefetch_embeddings",) 强制：backward（以及由此 progress 中 forward 之后的整条计算链）等本次 progress 内 prefetch 的 GPU 工作完成 —— 是 stream 协同等待，不是 batch 依赖。同一次 progress 内：

• forward → prefetch (correct)forward → prefetch（正确）: forward consumes prev-iter's prefetched data first (counter decremented), then this-iter's prefetch adds new keys. Steady-state outstanding ≈ 1 batch.forward 先消费上一轮 prefetch 的数据（计数减），然后本轮 prefetch 新增 key。稳态在飞 ≈ 1 个 batch。
• prefetch → forward (wrong)prefetch → forward（错误）: prefetch adds new keys before forward consumes prev's. Steady-state outstanding ≈ 2 batches → cache overflow.prefetch 在 forward 消费前一轮之前就加新 key。稳态在飞 ≈ 2 个 batch → cache 溢出。

Why same_progress_sync and not depends_on? prefetch_embeddings has lookahead=1 (it pre-loads next batch K+1) while backward has lookahead=0 (consumes batch K). They process different batches, so a same-batch depends_on would be incorrect; what we actually need is "drain prefetch's GPU work in this progress before the next progress starts" — exactly the same-progress GPU coherency intent. 为什么用 same_progress_sync 而不是 depends_on？prefetch_embeddings 的 lookahead=1（提前为 K+1 batch 加载），而 backward 的 lookahead=0（消费 batch K）。两者处理的是不同 batch，写 depends_on（同 batch 等待）语义就错了；真正要表达的是"在本次 progress 结束前，把 prefetch 的 GPU 工作排空" —— 正好是同 progress 的 GPU 协同意图。

5.5 Auto-scheduler v2: joint search5.5 自动调度器 v2：联合搜索

The v2 design searches the coupled space thread_map × stream_map × fire_ordering for the HSTU schedule. It does not replace the current lookahead-only scheduler in autosched/fire_order.py; instead, every candidate schedule is first materialized with its proposed thread/stream/fire-order choices, then passed through auto_assign_lookaheads(..., max_in_flight=5). The returned lookahead map becomes part of the scored candidate. v2 设计在 HSTU schedule 上联合搜索 thread_map × stream_map × fire_ordering。它不会替换当前 autosched/fire_order.py 中的 lookahead-only 调度器；每个候选 schedule 会先按候选的线程、流、触发顺序物化，再交给 auto_assign_lookaheads(..., max_in_flight=5)。返回的 lookahead map 会成为该候选的一部分并参与打分。

The objective is modeled steady-state latency: simulate several warmup/steady/drain progress calls with the task costs from hstu_prefetch_caching_1node.json, then score the interval between successive optimizer_step completions. The overlap matrix is the resource model: same stream serializes, same NCCL communicator serializes even on different streams, PCIe-bound tasks contend, and all other pairs may overlap if the DAG permits. 目标函数是建模后的稳态延迟：用 hstu_prefetch_caching_1node.json 中的任务 cost 模拟若干 warmup/steady/drain progress，再对连续 optimizer_step 完成之间的间隔打分。overlap matrix 是资源模型：同 stream 串行，同 NCCL communicator 即使在不同 stream 上也串行，PCIe-bound 任务竞争带宽，其余任务只要 DAG 允许就可重叠。

# Pseudocode sketch only; full design is in tasks/SPEC_autosched_joint_search.md.
beam = [current_hstu_candidate]
for round in search_rounds:
    expanded = mutate_thread_map(beam) + mutate_stream_map(beam) + mutate_fire_ordering(beam)
    for candidate in expanded:
        schedule = rebuild_schedule(candidate.thread_map, candidate.stream_map, candidate.fire_ordering)
        candidate.lookahead_map = auto_assign_lookaheads(schedule, cost_model, max_in_flight=5)
        candidate.score = modeled_optimizer_step_interval(schedule, candidate.lookahead_map)
    beam = top_k(valid_candidates, by="score")

Full pseudocode (6 procedures)完整伪代码（6 个过程）

Python-like, but not runnable. The pseudocode below mirrors tasks/SPEC_autosched_joint_search.md; use <details> to expand each procedure individually. Python 风格但非可运行代码；与 tasks/SPEC_autosched_joint_search.md 一致。点击下方 <details> 展开每个过程。

procedure JOINT_HSTU_SEARCH(
    base_schedule,
    cost_model,
    hstu_thread_presets,
    beam_width,
    max_threads = 4,
    max_in_flight = 5,
):
    require max_threads <= 4
    require max_in_flight <= 5

    frozen = {
        "forward",
        "backward",
        "finalize_model_grads",
        "optimizer_step",
    }

    legal_thread_presets = []
    for preset_name, preset in hstu_thread_presets:
        if preset_name == "per_task":
            continue                       # exceeds max_threads <= 4
        legal_thread_presets.append((preset_name, preset))

    canonical = Candidate(
        thread_map    = HSTU_DEFAULT_THREAD_MAP,
        stream_map    = CURRENT_HSTU_STREAMS(base_schedule),
        fire_ordering = DECLARATION_ORDER(base_schedule),
        lookahead_map = CURRENT_LOOKAHEADS(base_schedule),
    )

    canonical = NORMALIZE_AND_SCORE(
        canonical, base_schedule, cost_model,
        frozen, max_threads, max_in_flight,
    )

    beam = TOP_BY_SCORE([canonical], beam_width)
    best = canonical

    for round in 1..SEARCH_ROUNDS:
        expanded = []
        for candidate in beam:
            expanded += MUTATE_THREAD_MAP(candidate, legal_thread_presets)
            expanded += MUTATE_STREAM_MAP(candidate)
            expanded += MUTATE_FIRE_ORDERING(candidate)

        scored = []
        for candidate in UNIQUE(expanded):
            normalized = NORMALIZE_AND_SCORE(
                candidate, base_schedule, cost_model,
                frozen, max_threads, max_in_flight,
            )
            if normalized is valid:
                scored.append(normalized)

        beam = TOP_BY_SCORE(DOMINANCE_PRUNE(scored), beam_width)

        if beam is empty:                    break
        if SCORE(beam[0]) < SCORE(best):    best = beam[0]
        if NO_SCORE_IMPROVEMENT_FOR_PATIENCE_ROUNDS():  break

    return best

procedure NORMALIZE_AND_SCORE(
    candidate, base_schedule, cost_model,
    frozen, max_threads, max_in_flight,
):
    if VIOLATES_FROZEN_STREAM_OR_LOOKAHEAD(candidate, frozen):
        return invalid

    resolved_thread_map = RESOLVE_THREAD_MAP(candidate.thread_map,
                                            candidate.stream_map)
    if COUNT_THREADS(resolved_thread_map) > max_threads:
        return invalid

    provisional_schedule = REBUILD_SCHEDULE(
        base_schedule,
        stream_map    = candidate.stream_map,
        fire_ordering = candidate.fire_ordering,
        lookahead_map = CURRENT_LOOKAHEADS(base_schedule),
    )

    try:
        derived_lookahead = auto_assign_lookaheads(
            provisional_schedule, cost_model,
            max_in_flight = max_in_flight,
        )
    except SchedulerRejectsCandidate:
        return invalid

    if ANY(derived_lookahead[t] != 0 for t in frozen):
        return invalid

    scored_schedule = REBUILD_SCHEDULE(
        base_schedule,
        stream_map    = candidate.stream_map,
        fire_ordering = candidate.fire_ordering,
        lookahead_map = derived_lookahead,
    )

    overlap_matrix = compute_overlap_matrix(scored_schedule.tasks)
    latency = RESOURCE_LIST_SCHEDULE_LATENCY(
        scored_schedule, cost_model,
        resolved_thread_map, overlap_matrix, max_in_flight,
    )

    candidate.lookahead_map = derived_lookahead
    candidate.score         = latency
    return candidate

procedure MUTATE_THREAD_MAP(candidate, legal_thread_presets):
    for preset_name, preset in legal_thread_presets:
        yield candidate with thread_map = preset

procedure MUTATE_STREAM_MAP(candidate):
    movable_groups = [
        {"h2d", "start_shuffle", "finish_shuffle"},
        {"start_input_dist", "wait_input_dist"},
        {"prefetch_embeddings"},
    ]

    for group in movable_groups:
        if group is absent from schedule:
            continue

        for stream_choice in LEGAL_STREAM_CHOICES(group):
            next_map = COPY(candidate.stream_map)
            for task in group:
                if task in BIT_EXACT_FROZEN_TASKS:
                    continue                  # never move forward/bwd/opt/finalize
                next_map[task] = stream_choice

            yield candidate with stream_map = next_map

procedure MUTATE_FIRE_ORDERING(candidate):
    policies = [
        CRITICAL_DEFAULT_CHAIN_FIRST,
        LONGEST_OFF_DEFAULT_FIRST,
        NCCL_QUEUE_EARLY_WITH_TICKET_ORDER,
        PCIE_PREP_EARLY,
        MIN_SLACK_FIRST,
    ]

    for policy in policies:
        priority = BUILD_PRIORITY_VECTOR(candidate, policy)
        yield candidate with fire_ordering = priority

    # local search refinement: any topo-incomparable adjacent pair
    # in the current order is allowed to swap
    for adjacent_pair in INCOMPARABLE_ADJACENT_PAIRS(candidate.fire_ordering):
        yield candidate with that pair swapped

procedure DOMINANCE_PRUNE(candidates):
    grouped = GROUP_BY(
        candidates,
        keys = (thread_map, stream_map, lookahead_map),
    )
    for group in grouped:
        keep the lowest-score fire_ordering
        discard candidates with same or worse score and no lower thread count

Full SPEC (objective definition, constraint encoding, complexity bound, worked example): tasks/SPEC_autosched_joint_search.md (685 lines). 完整 SPEC（目标函数、约束编码、复杂度边界、worked example）：tasks/SPEC_autosched_joint_search.md（685 行）。

→ Interactive visualization (Diagram 5: search-space landscape). → 交互式可视化（图 5：搜索空间）。

→ Thread-map preset comparison (6 hand-coded presets, stream × thread × lookahead per HSTU task). → 6 个 thread_map preset 对照（每个 HSTU task 的 stream × thread × lookahead 布局）。

6.1 First progress() call (bootstrap)6.1 首次 progress() 调用（引导）

6.2 Steady-state iteration6.2 稳态迭代

6.3 NCCL ticket lock6.3 NCCL 票据锁

6.4 Multi-batch overlap (timing diagram)6.4 多批次重叠（时序图）

7.1 Engine classes7.1 引擎类

7.2 Executor classes7.2 执行器类

7.3 HSTU classes7.3 HSTU 类

8.1 T1: vanilla preset (≤ 8-line diff)8.1 T1：基础预设（≤ 8 行 diff）

from commons.pipeline.engine import SchedulablePipeline

# Before — manual loop:
# for batch in loader:
#     opt.zero_grad()
#     loss = model(batch).sum()
#     loss.backward()
#     opt.step()

# After — preset:
pipe = SchedulablePipeline.basic(model, optimizer, loss_fn=lambda out: out.sum())
for batch in loader:
    pipe.step(batch)

8.2 T2: with prefetch + memcpy stream8.2 T2：带 prefetch + memcpy 流

pipe = SchedulablePipeline.basic(
    model, optimizer,
    loss_fn=lambda out: out.sum(),
    prefetch=True,
    memcpy_stream=True,
)
# Drive with cpu-resident batches; engine moves them to GPU on memcpy stream
for cpu_batch in cpu_loader:
    pipe.step(cpu_batch)

8.3 HSTU usage8.3 HSTU 使用

from commons.pipeline.hstu_pipeline import HSTUPipelineFactory

pipe = HSTUPipelineFactory.create(
    "hstu_prefetch_sparse_dist",
    model=model_train,
    optimizer=dense_optimizer,
    device=torch.device("cuda"),
    batch_shuffler=batch_shuffler,
)

# Create the iterator ONCE outside the loop. SPEC_p4 v2's
# iterator-identity reset means a fresh `iter(dataloader)` per
# step would restart the engine each time and discard in-flight
# batches.
dataloader_iter = iter(dataloader)
for _ in range(max_train_iters):
    try:
        loss, tokens, output = pipe.progress(dataloader_iter)
    except StopIteration:
        break
pipe.shutdown()

Matches legacy's return signature; use case is drop-in replacement of JaggedMegatronTrainPipelineSparseDist / JaggedMegatronPrefetchTrainPipelineSparseDist. 返回签名与遗留一致；可作为 JaggedMegatronTrainPipelineSparseDist / JaggedMegatronPrefetchTrainPipelineSparseDist 的直接替换。

8.4 Custom thread_map8.4 自定义 thread_map

# by_stream (default for non-HSTU SchedulablePipeline.basic)
pipe = SchedulablePipeline.basic(model, opt, threaded=True, thread_map="by_stream")

# per_task — every task its own thread
pipe = SchedulablePipeline.basic(model, opt, threaded=True, thread_map="per_task")

# Explicit dict
pipe = SchedulablePipeline.basic(
    model, opt,
    threaded=True,
    thread_map={
        "h2d": "io",
        "forward": "compute",
        "backward": "compute",
        "optimizer_step": "compute",
    },
)

# Callable
pipe = SchedulablePipeline.basic(
    model, opt,
    threaded=True,
    thread_map=lambda task: "io" if task.stream == "memcpy" else "compute",
)

9.1 Custom Task subclass9.1 自定义 Task 子类

from commons.pipeline.engine import Task, DataSlot

class CountTokensTask(Task):
    name = "count_tokens"
    stream = "default"
    lookahead = 0
    reads = (DataSlot("batch_gpu", 0),)
    writes = (DataSlot("token_count", 0),)

    def run(self, ctx):
        batch = ctx.slots["batch_gpu"]
        ctx.slots.set("token_count", batch.numel())

9.2 Hand-built Schedule9.2 手工搭建 Schedule

from commons.pipeline.engine import (
    Schedule, Stage, StreamPool, Task, DataSlot, SchedulablePipeline,
)

# Two tasks: producer on memcpy stream, consumer on default
producer = Task.from_fn(
    "producer",
    lambda ctx: ctx.slots.set("x", torch.randn(8, device="cuda")),
    stream="memcpy",
    writes=(DataSlot("x", 0),),
)
consumer = Task.from_fn(
    "consumer",
    lambda ctx: ctx.slots.set("step_result", ctx.slots["x"].sum()),
    stream="default",
    reads=(DataSlot("x", 0),),
    writes=(DataSlot("step_result", 0),),
)

schedule = Schedule(
    stages=(Stage(tasks=(producer, consumer)),),
    stream_slots=("default", "memcpy"),
)
pool = StreamPool({
    "default": torch.cuda.default_stream(),
    "memcpy": torch.cuda.Stream(priority=-1),
})
pipe = SchedulablePipeline(schedule, pool, executor="threaded")
result = pipe.step(None)  # engine fills batch_cpu but our tasks don't use it

9.3 Plug in a custom Executor9.3 接入自定义 Executor

# An executor must implement: execute_stage(...) and shutdown().
class VerboseExecutor:
    def __init__(self, inner):
        self._inner = inner

    def execute_stage(self, stage, ctx, iter_count, should_run, waits, pool):
        for task in stage.tasks:
            if should_run(task):
                print(f"[iter {iter_count}] running {task.name}")
        return self._inner.execute_stage(stage, ctx, iter_count, should_run, waits, pool)

    def shutdown(self):
        self._inner.shutdown()

# Use:
from commons.pipeline.engine import SequentialExecutor
pipe = SchedulablePipeline(schedule, pool, executor=VerboseExecutor(SequentialExecutor()))

A.1 DefinitionA.1 定义

Δ is primitively defined as the progress lag between two task submissions — equivalently, the number of ring.advance() calls (one per progress()) elapsed between when a producer records an event and when a consumer reads it back: Δ 的本质定义是两次任务提交之间的 progress 跨度——等价地，producer 录 event 到 consumer 读 event 之间 ring.advance()（每次 progress() 调一次）调用的次数：

$$\Delta \;\stackrel{\text{def}}{=}\; q_c - q_p$$

where $q_c$ is the progress() index in which consumer reads the event and $q_p$ is the progress() index in which producer recorded it. This definition is universal — it applies to any cross-progress wait, independent of which sync field the user declared. cross-iter requires $q_p < q_c$, i.e. $\Delta \ge 1$. 其中 $q_c$ 是 consumer 读 event 时所在的 progress() 编号，$q_p$ 是 producer 录 event 时所在的 progress() 编号。这个定义是普适的——任何跨 progress 等待都适用，与用户用了哪种 sync 字段无关。cross-iter 要求 $q_p < q_c$，即 $\Delta \ge 1$。

Closed-form value in the cross_iter case. When the wait is declared via consumer.cross_iter_depends_on=(("X", -N),) with $N > 0$, the lookahead mechanics determine $q_c$ and $q_p$ uniquely; working through them (see §A.2) gives: cross_iter 场景下的闭式取值。当依赖通过 consumer.cross_iter_depends_on=(("X", -N),) ($N > 0$) 声明时，lookahead 机制把 $q_c$ 和 $q_p$ 唯一确定下来；按 §A.2 的推导，可以算出：

$$\Delta \;=\; X.\mathrm{la} + N - \mathrm{consumer.la} \quad\text{(cross\_iter only)}$$

This is a derived equality, not a definition — it holds in the cross_iter context because of how lookahead pins down which batch each task processes. Other sync types would yield different closed forms (e.g. same_progress_sync always has $\Delta = 0$ since both tasks are in the same progress). 这是个推导出来的等式，不是定义——在 cross_iter 上下文里成立，是因为 lookahead 把每个 task 处理的 batch 钉死了。其他 sync 类型会有不同的闭式（例如 same_progress_sync 永远 $\Delta = 0$，因为两个 task 在同一次 progress 内）。

Useful decomposition: $\Delta = N + (X.\mathrm{la} - \mathrm{consumer.la})$. When producer and consumer have equal lookahead, $\Delta = N$ exactly (the user's literal "wait for N batches ago"); when producer is deeper in the ring ($X.\mathrm{la} > \mathrm{consumer.la}$), $\Delta > N$ (extra rotations to bring the slot down to consumer's view). 有用的分解：$\Delta = N + (X.\mathrm{la} - \mathrm{consumer.la})$。producer 与 consumer lookahead 相等时 $\Delta = N$（即用户字面理解的"N 个 batch 前"）；producer 在 ring 更深处 ($X.\mathrm{la} > \mathrm{consumer.la}$) 时 $\Delta > N$（slot 需要多平移几步才到 consumer 视角）。

A.2 Derivation of Δ in the cross_iter caseA.2 cross_iter 场景下 Δ 的推导

Goal: take the primitive Δ definition $q_c - q_p$ from §A.1 and reduce it to a closed-form expression in $X.\mathrm{la}$, $N$, and $\mathrm{consumer.la}$ when the wait is declared as cross_iter_depends_on=(("X", -N),). 目标：把 §A.1 给出的本质定义 $\Delta = q_c - q_p$，在 cross_iter_depends_on=(("X", -N),) 这种声明下化简成只含 $X.\mathrm{la}$、$N$、$\mathrm{consumer.la}$ 的闭式。

Convention: in P_n (the n-th call to progress(), 0-indexed), a task with lookahead k processes batch B_(n+k) and records its event at slot[k]. Equivalently, slot[k] in P_n currently holds B_(n+k). 约定：在 P_n（第 n 次调用 progress()，0-indexed）里，lookahead 为 k 的 task 处理 batch B_(n+k)，并在 slot[k] 记录 event。等价地，P_n 里 slot[k] 当前装着 B_(n+k)。

Consumer at P_n = q_c processes B_(n + consumer.la). The cross_iter declaration says it waits for X's work on B_(n + consumer.la − N). The progress in which X processes B_(n + consumer.la − N) is P_(n + consumer.la − N − X.la): consumer 在 P_n = q_c 处理 B_(n + consumer.la)。cross_iter 要等 X 在 B_(n + consumer.la − N) 上的工作。而处理 batch B_(n + consumer.la − N) 的 progress 是在 P_(n + consumer.la − N − X.la)：

$$ \begin{aligned} q_p &= n + \mathrm{consumer.la} - N - X.\mathrm{la} \\ \Delta &= q_c - q_p \\ &= n - (n + \mathrm{consumer.la} - N - X.\mathrm{la}) \\ &= X.\mathrm{la} + N - \mathrm{consumer.la} \end{aligned} $$

Where consumer actually looks up the event. Producer recorded its event at $\mathrm{slot}[X.\mathrm{la}]$ in its own progress $P_{q_p}$. Each ring.advance() uniformly shifts every slot by $-1$. After $\Delta$ advances, the slot that was at offset $X.\mathrm{la}$ has rotated to offset $X.\mathrm{la} - \Delta$ in consumer's progress $P_{q_c}$. The engine's consumer-side ring index is exactly this: consumer 实际去查的 slot。producer 在自己的 progress $P_{q_p}$ 里把 event 录在 $\mathrm{slot}[X.\mathrm{la}]$；每次 ring.advance() 把所有 slot 整体 $-1$ 平移，经过 $\Delta$ 次 advance 后那个 slot 在 consumer 的 progress $P_{q_c}$ 里已经落到偏移 $X.\mathrm{la} - \Delta$。这个偏移就是引擎在 consumer 端实际查询的 ring index：

$$\mathrm{slot\_offset} \;\stackrel{\text{def}}{=}\; X.\mathrm{la} - \Delta$$

$X.\mathrm{la}$ does appear in $\Delta$, but it cancels when we substitute: $\Delta$ 里其实带着 $X.\mathrm{la}$，代入展开后两个 $X.\mathrm{la}$ 相消：

$$\mathrm{slot\_offset} \;=\; X.\mathrm{la} - (X.\mathrm{la} + N - \mathrm{consumer.la}) \;=\; \mathrm{consumer.la} - N$$

That is why the consumer-side lookup index depends only on $\mathrm{consumer.la}$ and $N$, not on the producer's lookahead — ring rotation is uniform, so producer.la is fully absorbed into $\Delta$. The slot_offset formula in deps.py writes this directly as consumer.batch_offset + neg_offset (where neg_offset = -N). 这就是为什么 consumer 端实际查询的 slot 编号只跟 $\mathrm{consumer.la}$ 和 $N$ 有关，与 producer 的 lookahead 无关——ring 旋转是整体均匀平移，producer.la 的信息全部被 $\Delta$ 吸收。deps.py 里直接写成 consumer.batch_offset + neg_offset（其中 neg_offset = -N）。

A.3 Validity constraintsA.3 合法性约束

Combined cross-stream interval: cross-stream 合并区间：

$$N \;\le\; \mathrm{consumer.la} \;\le\; X.\mathrm{la} + N - 1 \quad\Longleftrightarrow\quad 1 \;\le\; \Delta \;\le\; X.\mathrm{la}$$

Same-stream relaxes to just Δ ≥ 1. 同 stream 放宽为只需 Δ ≥ 1。

Validator code: deps.py lines 437-501 (Δ=0 / Δ<0), and deps.py line 515 (slot_offset<0). 校验代码：deps.py 437-501 行（Δ=0 / Δ<0），deps.py 515 行（slot_offset<0）。

A.4 Plain-language summaryA.4 通俗解释

Read cross_iter_depends_on=(("X", -N),) as: "the consumer task, when running on batch $K$, needs X's GPU work for batch $K - N$ to be already finished and recorded into the ring." $\Delta$ tells the engine how stale X's record will be by the time consumer reads it — measured in progress() calls. 把 cross_iter_depends_on=(("X", -N),) 读作："consumer 在 batch $K$ 上跑时，需要 X 在 batch $K - N$ 上的 GPU 工作已经做完并录进 ring。"$\Delta$ 告诉引擎：等 consumer 真正读这条 event 时，它在 ring 里待机了多少个 progress() 步。

• $\Delta = 1$: producer ran in the immediately preceding progress. Most common case.$\Delta = 1$：producer 在紧邻的上一次 progress 跑完。最常见。
• $\Delta = 2, 3, \ldots$: producer ran further back. Engine still finds the event in the ring as long as $\Delta \le X.\mathrm{la}$ (the slot hasn't rotated out yet).$\Delta = 2, 3, \ldots$：producer 在更早的 progress 跑完。只要 $\Delta \le X.\mathrm{la}$（slot 还没被旋转出去），引擎仍能在 ring 里找到 event。
• $\Delta = 0$: producer and consumer are both in the same progress. Engine auto-promotes the declaration to same_progress_sync semantics — same topo edge, same CPU edge, same slot[producer.la] event lookup. Equivalent to writing same_progress_sync=(X,) directly; the cross_iter syntax just lets users keep a batch-flow mental model.$\Delta = 0$：producer 和 consumer 在同一次 progress 里。引擎自动提升这条声明为 same_progress_sync 语义——相同的 topo 边、CPU 边、slot[producer.la] event 查询。与直接写 same_progress_sync=(X,) 等价；cross_iter 语法只是让 batch-flow 思维的用户少改一行。
• $\Delta < 0$: consumer would be asking for something producer hasn't computed yet. Logically impossible.$\Delta < 0$：consumer 在等 producer 还没算出来的东西。逻辑上不可能。

Quick mnemonic: if you wrote cross_iter_depends_on=(("X", -1),), you usually mean "wait for last progress's X". That works exactly when $\Delta = 1$, which forces $X.\mathrm{la} = \mathrm{consumer.la}$. Any la mismatch causes $\Delta$ to drift and may collide with one of the boundary checks above. 速记：写 cross_iter_depends_on=(("X", -1),) 通常是想表达"等上一次 progress 的 X"。这恰好对应 $\Delta = 1$，强制 $X.\mathrm{la} = \mathrm{consumer.la}$。任何 la 不一致都会让 $\Delta$ 偏离，可能撞上上面的边界检查。

A.5 Worked examplesA.5 实例