feat(collectives): ring allreduce TPUT push model on HOST builtin + InCore composite (requires PTOAS >= 0.55)

docs/en/dev/passes/12-lower_composite_ops.md

@@ -188,7 +188,7 @@ Running `LowerCompositeOps` twice produces identical IR after the first run: the
 
 ## `pld.tensor.*` distributed collectives
 
-The pass also lowers the `pld.tensor.*` family of window-bound distributed collectives. Each collective is a single composite `Call` that expands into a notify / wait + data-movement recipe, plus a self-clearing epilogue. The data-movement primitive differs by op: `allgather` uses `pld.tile.put` (TPUT-based, auto-chunks through a VEC staging tile), `broadcast` relocates window data with `pld.tile.get` (GM→GM copy), while `allreduce` and `reduce_scatter` pull peer chunks into a UB tile with `pld.tile.remote_load`. Allreduce selects `tile.add`, `tile.maximum`, `tile.minimum`, or `tile.mul`; reduce-scatter currently accumulates with `tile.add`. All seven rules share the same **self-clearing credit-barrier protocol** (`LoweringBuilder::EmitBarrier` + `EmitEpilogueReset`) — see [Barrier-signal protocol](#barrier-signal-protocol) below — so a `signal` buffer is reusable across back-to-back calls, and even inside `for` / `while` / `if`.
+The pass also lowers the `pld.tensor.*` family of window-bound distributed collectives. Each collective is a single composite `Call` that expands into a notify / wait + data-movement recipe, plus a self-clearing epilogue. The data-movement primitive differs by op: `allgather` and ring `allreduce` use `pld.tile.put` (TPUT-based push, auto-chunking through a VEC staging tile), `broadcast` relocates window data with `pld.tile.get` (GM→GM copy), while mesh `allreduce` and `reduce_scatter` pull peer chunks into a UB tile with `pld.tile.remote_load`. Allreduce selects `tile.add`, `tile.maximum`, `tile.minimum`, or `tile.mul`; reduce-scatter currently accumulates with `tile.add`. All seven rules share the same **self-clearing credit-barrier protocol** (`LoweringBuilder::EmitBarrier` + `EmitEpilogueReset`) — see [Barrier-signal protocol](#barrier-signal-protocol) below — so a `signal` buffer is reusable across back-to-back calls, and even inside `for` / `while` / `if`.
 
 ### Barrier-signal protocol
 
@@ -206,7 +206,7 @@ The allreduce rule starts with a cross-rank ready barrier on shared `signal` cel
 
 For a fully-valid packed target, mesh lowering creates a logical `[1, product(all dimensions)]` view and traverses it with physical tiles of at most 16 KiB. A statically known extent smaller than the budget shrinks the chunk to the smallest 32-byte-aligned physical width that covers it, so small allreduces do not reserve a full 16-KiB tile while remaining legal PTO tiles. The tail carries `valid_shape=[1, min(chunk, remaining)]` through both `tile.load` and `pld.tile.remote_load`, so the allocation stays static while the read/store extent is exact. If an ND target carries a partial `TensorView.valid_shape`, the pass preserves and reduces the representable `[rows, cols]` rectangle through the established single-rectangle path. Constant valid rectangles use their compact shape; symbolic valid extents fall back to the source's physical rectangle when that statically bounded rectangle fits within one 16-KiB chunk. Oversized partial rectangles, strided targets, DN partial views, and partial boxes that cannot be represented by the leading-dimension collapse are rejected explicitly.
 
-Ring lowering uses one packed 2D view for its reduce-scatter and allgather phases. A fully valid target becomes `[1, SIZE]`; a contiguous partial prefix keeps physical shape `[1, product(target.shape)]` and carries logical `TensorView.valid_shape=[1, product(target.valid_shape)]`. FP32 retains balanced `floor(i * SIZE / NR)` segment boundaries. FP16 rounds each interior boundary up to 16 elements and caps it at `SIZE`; consequently every non-empty segment and every UB subchunk starts at a 32-byte-aligned address. A ragged FP16 remote load may read the aligned physical tail reserved by the communication domain, then `tile.set_validshape` restores the logical extent before reduction and store. This supports non-divisible inputs and `SIZE < NR` without inserting holes into the public tensor layout. Every subchunk of every round barriers on a call-local ready + read-complete generation pair; the epilogue then subtracts `2 * chunk_count` (uniform across rounds, since every round's subchunk loop shares the same bound) from every row of the `[2*(NR-1), NR]` signal.
+Ring lowering uses one packed 2D view for its reduce-scatter and allgather phases, and moves data with **TPUT pushes** (`pld.tile.put`, non-atomic) instead of remote loads. A fully valid target becomes `[1, SIZE]`; a contiguous partial prefix keeps physical shape `[1, product(target.shape)]` and carries logical `TensorView.valid_shape=[1, product(target.valid_shape)]`. FP32 retains balanced `floor(i * SIZE / NR)` segment boundaries. FP16 rounds each interior boundary up to 16 elements and caps it at `SIZE`; consequently every non-empty segment and every UB subchunk starts at a 32-byte-aligned address. Per subchunk, each rank first reads its OWN value of the receive slot into a register tile (the slot is stable — own value only — until the left neighbour's push lands), barriers on the ready generation (2k+1), then TPUTs its send subchunk into the RIGHT neighbour's slot of the same index; a push-done barrier (2k+2) precedes the local read + reduce + store of the receive slot. The single shared VEC staging tile is narrowed with `tile.set_validshape` to each transfer's exact `valid_cols`, and the push transfer carries that dynamic extent — PTOAS >= v0.55 accepts dynamic partition-view shapes for `tput` (hw-native-sys/PTOAS#1069), preserving ragged and FP16 tails without padding the window. The non-atomic push + local reduce keeps every `ReduceOp` (Sum/Max/Min/Prod) working — only a remote-atomic `TPUT<AtomicAdd>` would be Sum-only. This supports non-divisible inputs and `SIZE < NR` without inserting holes into the public tensor layout. Every subchunk of every round barriers on a call-local ready + push-done generation pair; the epilogue then subtracts `2 * chunk_count` (uniform across rounds, since every round's subchunk loop shares the same bound) from every row of the `[2*(NR-1), NR]` signal.
 
 Any symbolic target or partial-valid extent that survives lowering must be runtime-bound by a kernel scalar, loop variable, or physical tensor-shape parameter; a type-metadata-only symbol is rejected during PTO codegen. A fully dynamic physical target dimension is bound from that tensor parameter.
 

docs/en/dev/passes/40-lower_host_tensor_collectives.md

@@ -90,6 +90,17 @@ Ring allreduce currently supports only `ReduceOp.Sum` with `dtype=FP32`.
 with `mode="ring"`. Ring allreduce also supports at most 16 participating
 devices (`world_size <= 16`).
 
+The `builtin.tensor.allreduce_ring` kernel is **push-based**: data movement uses
+`pto::comm::TPUT` (remote write) — the reduce-scatter phase accumulates into the
+right neighbour's slot via `TPUT<AtomicAdd>`, and the allgather phase forwards
+each finalized chunk with a non-atomic `TPUT`, mirroring the in-tree `allgather`
+/ `all_to_all` host builtins.  Ordering is `pipe_barrier(PIPE_ALL)` around each
+transfer plus `dsb(DSB_DDR)` before every `TNOTIFY` (not
+`pto.fence.barrier_all`, which does not drain the MTE DMA pipe).  Cross-rank
+synchronisation uses the O(1) `NeighborBarrier` (notify/wait the two ring
+neighbours only) — safe on NPU because the TPUT write pipeline orders the data
+ahead of the signal, which the old pull model (TLOAD/TSTORE) did not.
+
 ## Pass properties
 
 | Field | Value |

docs/zh/dev/passes/12-lower_composite_ops.md

@@ -188,7 +188,7 @@ sin 与 cos 共用同一组多项式系数：cos 路径只在区间归约阶段
 
 ## `pld.tensor.*` 分布式集合通信算子
 
-本 Pass 同时降级 `pld.tensor.*` 系列的窗口绑定 (window-bound) 分布式集合通信算子。每个集合通信算子都是一个组合 `Call`，展开为 notify / wait + 数据搬运序列，外加自清理尾声。数据搬运原语因算子而异：`allgather` 使用 `pld.tile.put`（基于 TPUT 的推送，经 VEC staging tile 自动分块），`broadcast` 用 `pld.tile.get` 搬运窗口数据（GM→GM 拷贝），`allreduce` 与 `reduce_scatter` 用 `pld.tile.remote_load` 把 peer chunk 拉进 UB tile。allreduce 根据规约类型选择 `tile.add`、`tile.maximum`、`tile.minimum` 或 `tile.mul`；reduce-scatter 当前仍用 `tile.add`。七条规则共享同一套**自清理信用屏障协议**（`LoweringBuilder::EmitBarrier` + `EmitEpilogueReset`）—— 参见下方[屏障-信号协议](#屏障-信号协议) —— 因此 `signal` buffer 可以在连续调用之间复用，甚至在 `for` / `while` / `if` 内部也可以。
+本 Pass 同时降级 `pld.tensor.*` 系列的窗口绑定 (window-bound) 分布式集合通信算子。每个集合通信算子都是一个组合 `Call`，展开为 notify / wait + 数据搬运序列，外加自清理尾声。数据搬运原语因算子而异：`allgather` 与 ring `allreduce` 使用 `pld.tile.put`（基于 TPUT 的推送，经 VEC staging tile 自动分块），`broadcast` 用 `pld.tile.get` 搬运窗口数据（GM→GM 拷贝），mesh `allreduce` 与 `reduce_scatter` 用 `pld.tile.remote_load` 把 peer chunk 拉进 UB tile。allreduce 根据规约类型选择 `tile.add`、`tile.maximum`、`tile.minimum` 或 `tile.mul`；reduce-scatter 当前仍用 `tile.add`。七条规则共享同一套**自清理信用屏障协议**（`LoweringBuilder::EmitBarrier` + `EmitEpilogueReset`）—— 参见下方[屏障-信号协议](#屏障-信号协议) —— 因此 `signal` buffer 可以在连续调用之间复用，甚至在 `for` / `while` / `if` 内部也可以。
 
 ### 屏障-信号协议
 
@@ -227,16 +227,24 @@ allreduce 仍预留完整 16-KiB tile，又满足 PTO tile 的对齐要求。尾
 过大的 partial 矩形、strided 目标、DN partial view 和无法按 leading-dimension collapse
 表示的 partial 区域会被明确拒绝。
 
-ring 降级在 reduce-scatter 和 allgather 阶段使用同一个 packed 2D 视图。
+ring 降级在 reduce-scatter 和 allgather 阶段使用同一个 packed 2D 视图，并用
+**TPUT 推送**（`pld.tile.put`，非原子）代替远程加载来搬运数据。
 完全有效的目标会变为 `[1, SIZE]`；连续 partial prefix 保留物理 shape
 `[1, product(target.shape)]`，并携带逻辑
 `TensorView.valid_shape=[1, product(target.valid_shape)]`。FP32 保留均衡的
 `floor(i * SIZE / NR)` segment 边界；FP16 把每个内部
 边界向上对齐到 16 个元素并限制在 `SIZE` 内，因此每个非空 segment 和 UB
-subchunk 都从 32 字节对齐地址开始。FP16 的 ragged remote load 可以读取通信域
-预留的对齐物理尾部，然后通过 `tile.set_validshape` 在归约和写回前恢复逻辑范围。
+subchunk 都从 32 字节对齐地址开始。每个 subchunk 中，各 rank 先把接收 slot 的
+**自身值**读入寄存器 tile（该 slot 在左邻居的推送落地前保持稳定 —— 只有自身值），
+随后在 ready generation (2k+1) 上做屏障，再把发送 subchunk 通过 TPUT 推送到
+**右邻居**的同一下标 slot；push-done 屏障 (2k+2) 之后才本地读取、归约并写回接收 slot。
+共享的 VEC staging tile 通过 `tile.set_validshape` 收窄到每次传输的精确
+`valid_cols`，推送传输携带该动态范围 —— PTOAS >= v0.55 接受 `tput` 的动态
+partition-view 形状（hw-native-sys/PTOAS#1069），因此无需填充窗口即可保留
+ragged 与 FP16 尾部。非原子推送 + 本地归约保留了所有 `ReduceOp`
+（Sum/Max/Min/Prod）——只有远端原子 `TPUT<AtomicAdd>` 才只支持 Sum。
 该方案无需在公开 tensor 布局中插入空洞，也能支持非整除输入和 `SIZE < NR`。
-每一轮的每个 subchunk 都使用本调用局部的 ready + read-complete generation 对
+每一轮的每个 subchunk 都使用本调用局部的 ready + push-done generation 对
 做屏障；尾声随后把 `2 * chunk_count`（跨轮统一，因为每轮的 subchunk 循环
 共享相同边界）从 signal 的每一行中减去。
 

docs/zh/dev/passes/40-lower_host_tensor_collectives.md

@@ -78,6 +78,15 @@ Ring allreduce 目前仅支持 `ReduceOp.Sum` 和 `dtype=FP32`。
 `mode="ring"` 下尚未支持。Ring allreduce 最多支持 16 个参与设备
 （`world_size <= 16`）。
 
+`builtin.tensor.allreduce_ring` 内核采用**推送（push）模型**：数据搬运使用
+`pto::comm::TPUT`（远端写）——reduce-scatter 阶段通过 `TPUT<AtomicAdd>` 将部分和
+累加到右邻居的 slot，allgather 阶段用非原子 `TPUT` 转发每个已归约的 chunk，
+与树内 `allgather` / `all_to_all` host builtin 保持一致。顺序保证为每次传输前后
+`pipe_barrier(PIPE_ALL)`，并在每次 `TNOTIFY` 前加 `dsb(DSB_DDR)`（而非
+`pto.fence.barrier_all`，后者不会排空 MTE DMA 流水线）。跨 rank 同步使用 O(1) 的
+`NeighborBarrier`（只通知/等待左右两个 ring 邻居）——在 NPU 上安全是因为 TPUT
+写流水线保证数据先于信号可见，而旧的拉取（pull）模型（TLOAD/TSTORE）不具备该保证。
+
 ## Pass 属性
 
 | 字段 | 取值 |