feat(fastsac): multi-collector/multi-learner async training — structured topology, unified drain & startup hardening

[wlgys8]

概述

fastsac 支持多 collector / 多 learner 的异步训练。本 PR 描述该功能提交后的当前行为：worker 编排由结构化拓扑驱动，传输与启动经结构化握手对象完成，Learner 数据面按 collector 配对端点，单 learner 与多 learner 共享同一训练决策路径。

功能行为

结构化拓扑（async_impl/topology.py）

resolve_trainer_topology 是唯一的布局计算入口：给定 env/collector/learner 数量、设备 spec、UTD/传输配置与 actor 参数量，返回 TrainerTopology——

• learners: list[LearnerInfo] —— 每个 rank 的推理设备、NUMA 节点、CPU 列表；
• collectors: list[CollectorInfo] —— 每个 collector 的环境分片宽度、归属 rank、推理设备、NUMA 节点、CPU 列表、ring/权重传输选择；
• 派生属性与方法：env_shards、collector_owner(i)、ring_slice_for_rank(rings, rank)。

不变量（由构造保证）：collector 的 NUMA 节点恒等于其 owner 的节点（通信不跨 NUMA）；collector_cpus[i] 是预算好的绑定列表，worker 只应用、不计算；env 分片要求整除切分。

拓扑内还完成三类传输决策：transition ring 是否走 CUDA-IPC（per learner rank，all-or-nothing）、权重通道是否走 CUDA-IPC（同上 + weight_ipc_min_bytes 尺寸阈值）、learner 设备解析（spec 展开/校验，见 resolve_learner_devices）。

启动握手（transport/handshake.py）

StartupHandshake 封装启动期的两类跨进程交互：

• one-shot slot 投递：learner 构建 weight / transition-ring 的 CUDA-IPC slot 张量后，经 per-collector 队列发给各 collector（maxsize=1，恰好一条消息）；
• 就绪屏障：所有 worker 上报就绪后由 parent release()，保证 stepping 从第一帧起锁步对齐（warmup 计数、weight generation、generation merge 均从零对齐）。

barrier=False 供单 collector 手动驱动场景。

Learner 数据面（async_impl/learner.py）

• CollectorEndpoint(ring, weight_sender)：ring 与权重通道按 collector 配对；Learner 持有 endpoint 列表，num_collectors 由列表长度推导；
• GenerationAssembler：跨 collector 的代际装配——shard 按绝对代号入座，全 ring 交付的同代自动按序 pop 出来合并成 full batch；慢 collector 的缺失代安静等待，只 backpressure 它自己；
• per-owner 权重发布：每个 learner 构建并发布自己 collectors 的 weight sender（跨 GPU 时 owner 同 GPU 可走 CUDA-IPC，无需 host-shm 回退）；rank 0 另担日志 / checkpoint；
• 混合 rank 支持：同 rank 可同时持有 host ring 与 IPC ring——host 分片在装配时提升到 ingest device，merge 全程 D2D；纯 host rank（CUDA）走 DrainStaging pinned 异步 H2D 流水线（与梯度更新重叠），纯 IPC rank 直接 D2D。

训练决策（maybe_train(gstep)）

单一路径同时服务两种拓扑，gstep = collector_steps // num_collectors 为全局进度基准：

• strict：num_updates × delta；learner_bound：有进展即 num_updates；
• 单 learner：自身进度驱动；rb 为空时仅推迟更新（累计 delta 保留），不丢样本；
• 多 learner：rank 0 决策 + broadcast 复制更新数，所有 rank 执行相同次数的 backward；本 rank 数据未到时在等待循环中持续 drain（不能跳过——跳过会使 NCCL collective 失配），等待期间持续给 collector 腾出 ring 空间避免背压死锁；
• 权重发布按梯度步分块（weight_publish_interval 节奏），collector 拿到的权重不会陈旧一个量级。

面板与日志（console.py / train.py / worker.py）

• worker 进程的 logging 与 stdio 重定向到各自日志文件（runs/<run>/logs/），Rich 面板不被任何子进程输出撕裂；
• 面板底部显示 ✓ saved checkpoint <path>（checkpoint 落盘后随帧刷新）；
• boot 阶段独立面板：worker 就绪表实时刷新；启动失败时在面板退出后输出一行原因并指向 async_errors/<worker>_error.log 的完整 traceback。

使用

# 单 collector / 单 learner（1c1l）
python scripts/train.py task=g1-wbt-dance/motrix.fastsac

# 多 collector + 多 learner
python scripts/train.py task=g1-wbt-dance/motrix.fastsac \
    algo.trainer.async_options.num_collectors=8 \
    algo.trainer.async_options.num_learners=2

约束：num_envs 与 num_collectors 整除、num_collectors 与 num_learners 整除、batch_size 与 num_learners 整除；collector 推理设备缺省解析到 owner learner 的 GPU。

验证

• 单测：motrix_rl 307 passed / 4 skipped（含 drain 合并等价、代际等待、DDP 锁步决策、拓扑放置、握手行为）；motrix_env_core 全量通过；
• 实机：g1-wbt-dance 1c1l 启动训练确认（UTD ~3、权重发布 ~0.9/迭代、checkpoint 面板行稳定）。

已知事项

• 归一化统计与课程状态是否需要跨节点同步的调研与决策点见 分布式训练: 课程/自适应采样状态是否需要跨节点同步? 对收敛性的影响 #75（IsaacLab / mjlab / rsl_rl / holosoma 对比）。

[Copilot]

Copilot review overview

🔵 Needs a closer look

The change spans multi-process orchestration, DDP, NUMA binding, and CUDA-stream data movement, and includes a confirmed missing compute-stream synchronization that constitutes a data race requiring human verification.

Review effort: Balanced
Findings: 1 · 2

Open (3)

• Wait for async ingest copies before sampling · New
• Fix stale docstring reference to apply_binding · New
• Remove duplicate worker name assignment · New

What changed in this PR

This PR lands the full multi-collector / multi-learner asynchronous training path for the fastsac provider. It introduces a structured topology resolver (topology.py) that computes env sharding, device/NUMA placement, CPU binding, ring/weight transport and collector→learner ownership in one pass; consolidates cross-process startup into a StartupHandshake with keyword-argument spawns; unifies the learner drain into a single _drain_rings path with per-collector CollectorEndpoints and a cross-collector GenerationAssembler; adds DDP data-parallel learners (per-GPU processes with manual gradient/log_alpha all-reduce and cross-rank normalizer stat sync); and hardens observability (per-worker stdio redirection, boot output moved out of the Live panel, checkpoint status line via panel_queue). It fits the existing async fastsac architecture, preserving single-writer SPSC/seqlock invariants (world_size == 1 stays byte-for-byte identical).

Changes:

• Structured topology + NUMA/CPU binding, keyword-arg spawn and handshake, dead-parameter cleanup on process entrypoints.
• Unified generation-merge drain, per-collector endpoints, per-learner weight publishing, and DDP lockstep training decision.
• Design/plan docs (§8 multi-collector, §9 multi-learner) plus pipeline-equivalence tests.

File	Description
motrix_rl/​src/​motrix_rl/​fastsac/​async_impl/​learner.py	Unified drain, generation assembly, DDP-aware maybe_train; missing compute-stream sync before update.
motrix_rl/​src/​motrix_rl/​fastsac/​async_impl/​numa.py	NUMA/CPU binding primitives; docstring references renamed function.
motrix_rl/​src/​motrix_rl/​fastsac/​async_impl/​panels.py	Worker log panel rendering; contains a duplicated statement.
motrix_rl/​src/​motrix_rl/​fastsac/​async_impl/​worker.py	Keyword-arg process entrypoints, logging redirect, NCCL init, binding.
motrix_rl/​src/​motrix_rl/​fastsac/​agent.py	Manual DDP gradient/log_alpha all-reduce and compile-boundary handling.
motrix_rl/​src/​motrix_rl/​fastsac/​buffer.py	EmpiricalNormalization local accumulators + cross-rank sufficient-stats merge.
motrix_rl/​tests/​test_rl_sim_backend.py	Updated spawn to kv-args + StartupHandshake, adds tmp_path.
motrix_rl/​tests/​test_fastsac_pipeline_equivalence.py	New ring-merge / normalizer-sync equivalence tests.
wiki/​design/​fastsac-async-heterogeneous-trainer.md	Adds §8 multi-collector/NUMA and §9 multi-learner/DDP design.
wiki/​plan/​fastsac-async-multi-learner.md, wiki/​plan/​index.md	New implementation plan and index link.

---

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

[Copilot]

Copilot review overview

🔵 Needs a closer look

The change spans a large multi-process/DDP/CUDA-IPC/NUMA surface that cannot be fully validated without multi-GPU/multi-node hardware, and it contains at least one confirmed placement bug, so final human review is warranted.

Review effort: Balanced
Findings: 1

Open (1)

• Use node-local collector index when chunking CPUs · New

Resolved since last review (3)

• Wait for async ingest copies before sampling
• Remove duplicate worker name assignment
• Fix stale docstring reference to apply_binding

[Copilot]

Copilot review overview

🔵 Needs a closer look

The change spans multiprocessing, CUDA-IPC, NCCL/DDP lockstep, and NUMA binding with subtle cross-rank collective-ordering invariants that warrant final human review despite extensive tests.

Review effort: Balanced
Findings: 1

Open (1)

• Abort message hardcodes boot timeout instead of configured value · New

Resolved since last review (1)

• Use node-local collector index when chunking CPUs