企业官网建设流程全解析

PyPTO 自动CV并行流水：用 preload 编译期变换把 cube/vector 串行 kernel 改写成并行流水

【免费下载链接】pyptoPyPTO（发音: pai p-t-o）：Parallel Tensor/Tile Operation编程范式。项目地址: https://gitcode.com/cann/pypto

自动CV并行流水是 pypto_pro.language.jit 提供的一项编译期变换，针对 CV 融合算子中 cube 与 vector 相互依赖、串行执行时空等对方的问题，自动把用户手写的串行流水 kernel 改写成并行流水版本，并自动插入全部所需的核间同步。读完本文，你将掌握如何通过pl.pipeline.stage划分 stage、用pl.make_tile_group声明跨核共享 Buffer（配置fwd_ids/bwd_ids）、在主循环中按 cube/vector 交替调用 stage，以及如何用PipelineConfig(preload=...)开启并逐步调优流水。

串行流水与并行流水的执行节奏对比

功能说明

CV 融合算子中，cube 和 vector 的计算相互依赖：若按串行流水执行，一个核工作时另一个核只能空等。自动CV并行流水的思路是：让上游核提前若干次迭代计算，下游核则处理上游已经算好的较早迭代的数据，两个核错开节奏、同时工作，从而把彼此的等待时间掩盖掉。图中sN·iM表示第 N 个 stage 正在处理第 M 次迭代的数据，格子宽度代表该 stage 的耗时（各 stage 耗时不同，图中为示意值）。

自动CV并行流水是pypto_pro.language.jit的一项编译期变换，包含两个能力：

• 自动流水排布：把用户手写的 CV 融合算子串行流水 kernel 代码，自动改写成并行流水版本，提升性能；
• 自动核间同步：用户开发的串行版本不需要手写任何 CV 核间同步指令，并行流水版本会插入全部所需的核间同步。

从源码看，整个变换由 python/pypto_pro/runtime/pipeline 目录下的若干模块协作完成：__init__.py对外暴露PipelineConfig、stage与transform_pipeline三个入口（见 python/pypto_pro/runtime/pipeline/init.py）；真正的 AST 改写逻辑在 _transformer.py 的transform_pipeline中。

收益上限：由较慢的那个核决定

需要注意的是，流水化的收益上限由较慢的那个核决定：图中 vector 侧单次迭代的耗时高于 cube 侧，稳态节奏就由 vector 侧决定，cube 侧会出现等待间隙。stage 划分越均衡（两个核的耗时越接近），流水填充得越满，收益越高。此外流水的建立和排空各需要若干拍，迭代次数越多，这部分开销占比越低。

这与 PipelineConfig 的实现语义一致：preload越大，掩盖的搬运/计算延迟越多，但流水填充（fill）和排空（drain）阶段的代价也越长，因此需要按 kernel 逐个调优。仓库 docs/zh/guide/figures/pro 目录下还提供了pro_pipeline_cube_bound.png（cube 侧受限的流水节奏）与pro_pipeline_with_gaps.png（带等待间隙的流水节奏）两张示意图，与本节讨论的收益边界问题直接对应，可作为深入理解流水稳态节奏的参考。

使用方法

使用自动CV并行流水需要完成四步：编写 stage 函数、声明跨核共享 Buffer、编写主循环、开启流水变换。

编写 Stage 函数

需要用户将算子划分为若干个计算流程，每个计算流程对应一个 stage 函数，通过@pypto_pro.language.pipeline.stage装饰器进行标识。该装饰器在源码层面是一个透明装饰器——它只是给函数打上pipeline_stage = True属性标记，供流水框架识别，不会改变函数行为（见 python/pypto_pro/runtime/pipeline/_stage.py 中的stage与is_pipeline_stage）。

import pypto_pro.language as pl @pl.pipeline.stage def stage1(ki, a, b_l1, a_l1_db, left_db, right_db, acc_db, mm1_vec_db): """Cube：mm1 = A_i @ B""" cur_a = a_l1_db.next() pl.load(cur_a, a, [ki * TILE, 0]) ... # 普通的 Tile/Buffer 操作，不用写任何同步

声明跨核共享Buffer

跨核共享 Buffer 使用make_tile_group接口进行声明，tile 数目由用户自主分配，通过fwd_ids和bwd_ids参数配置核间正反向同步 id，若未配置，则不会插入对应的核间同步。

其中：

• fwd_ids为正向同步（生产者写完通知消费者）：生产者 stage 之后插入 set、消费者 stage 之前插入 wait；
• bwd_ids为反向同步（消费者用完通知生产者可以覆写）：消费者 stage 之后插入 set、生产者 stage 之前插入 wait。

只配置fwd_ids时，仅保证消费者读到的是生产者写完的数据，不保证生产者下一轮覆写时消费者已经读完。每次迭代实际使用的 id 按ids[迭代序号 % len(ids)]轮转取用。这一点在 make_tile_group 的接口文档中也有明确说明：fwd_ids是标记该 group 为生产者→消费者通道的跨核事件 id（取值 0~15），每个 Tile 一个，变换按fwd_ids[i % N]每迭代取用；bwd_ids是消费者→生产者方向（Buffer 释放）的事件 id，形状与fwd_ids相同。

import pypto_pro.language as pl mm1_vec_db = pl.make_tile_group( type=pl.TileType(shape=[TILE_HALF, TILE], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec), addrs=0x0000, mutex_ids=[12, 13], fwd_ids=[0, 1], bwd_ids=[2, 3], )

编写主循环

主循环内 stage 函数按照依赖关系顺序进行书写，需要保证 stage 之间为 CV 交替。

import pypto_pro.language as pl for ki in pl.range(0, N_ITER): with pl.section_cube(): stage1(ki, a, b_l1, a_l1_db, left_db, right_db, acc_db, mm1_vec_db) with pl.section_vector(): stage2(ki, sub_id, mm1_vec_db, relu_vec_db, relu_nz_db, p_mat_db) with pl.section_cube(): stage3(ki, d_l1, p_mat_db, left_db, right_db, acc_db, out_vec_db) with pl.section_vector(): stage4(ki, sub_id, out, out_vec_db)

一个 kernel 内可以写多个独立的 for 循环，每个循环各自做流水变换，互不影响，多个循环按源码顺序先后执行，preload 可以逐循环配置（见下一节）。

import pypto_pro.language as pl for ki in pl.range(0, N_ITER): # 第一条流水 with pl.section_cube(): stage1(ki, ...) with pl.section_vector(): stage2(ki, ...) pl.system.sync_all(core_type=pl.SyncCoreType.MIX) for kj in pl.range(0, M_ITER): # 第二条流水 with pl.section_cube(): stage3(kj, ...) with pl.section_vector(): stage4(kj, ...)

开启流水变换

在pypto_pro.language.jit接口中通过PipelineConfig参数进行配置：

PipelineConfig的源码实现（python/pypto_pro/runtime/pipeline/config.py）对参数做了严格校验：preload默认值为 2；接受单个 int 或多个值组成的元组（用户写列表也会被归一化为 tuple，因为冻结 dataclass 需要可哈希字段）；每个值必须是 int 且不能是 bool、不能为负数，空序列会被拒绝。语义上，preload是同一个核上两个相邻 stage 之间的延迟步数（对应_compute_delays），ctx 环形缓冲区的深度由计算出的延迟决定（max_delay + 1），而不是直接等于 preload。

建议流程：先配置preload=0执行，确认串行版本精度正确，再逐步调大 preload 开启流水。多个流水循环时可以只把其中一条配成 0（如preload=[0, 2]），单独验证这条流水的串行精度。

import pypto_pro.language as pl @pl.jit(auto_mutex=True, pipeline=pl.pipeline.PipelineConfig(preload=0)) def pipeline_demo_kernel(...): ...

import pypto_pro.language as pl @pl.jit(auto_mutex=True, pipeline=pl.pipeline.PipelineConfig(preload=2)) def pipeline_demo_kernel(...): ...

import pypto_pro.language as pl # kernel内有两个流水循环，第一条preload=1，第二条preload=2 @pl.jit(auto_mutex=True, pipeline=pl.pipeline.PipelineConfig(preload=[1, 2])) def pipeline_demo_kernel(...): ...

查看生成的代码

框架自动生成的并行流水代码保存在编译产物目录下，文件名为pipeline_generated.py，用户可以在该代码的基础上继续修改调试。

使用约束

自动CV并行流水对 kernel 写法有明确约束，违反时会在编译期被分析器拒绝：

• 同一条流水的 stage 调用需要放在同一个 for 循环内。一个 kernel 可以有多个独立的流水循环，循环之间需要用户手动插入全核同步，两条流水循环不能嵌在同一个外层循环里。
• 流水循环的步长必须为正，不支持倒序迭代。
• preload 取值必须大于等于 0；传入列表时，列表长度必须与 kernel 内流水循环的个数相同。
• 不支持 stage 嵌套 stage。
• 跨核 Buffer 和核内 Buffer 的make_tile_group声明必须写在 kernel 函数体内，不支持在被 stage 调用的普通函数里声明。
• stage 函数不支持有返回值。
• stage 调用的普通函数不能返回 tile 或 tile group。
• 每个with pypto_pro.language.section_cube()/pypto_pro.language.section_vector()块里只放单个 stage 调用，且 stage 调用需要严格按 cube/vector 交替排列（C→V→C→V…），不允许连续两个 stage 落在同一个核上。
• 流水循环体内，第一个 stage 与最后一个 stage 之间不允许插入其他语句。
• 流水循环（含 stage 调用的那个 for 循环）体内不允许获取或操作跨核 Buffer，以及与跨核 Buffer 地址复用的核内 Buffer，请把它们放进 stage 函数体内。
• 在流水循环之外取 Tile 时，必须单独写成一条赋值语句（slot = group.next()），不能嵌在更大的表达式里；同一个变量只能取一次 Tile，需要多个 Tile 请用多个变量。
• 同一个 stage 函数在一条流水循环内只能调用一次，不允许重名 stage。
• 允许通过 if 语句判断 stage 执行场景，但分支条件必须为编译期常量。从源码看，_transformer.py 中的_prune_const_branches会在分析前把编译期常量分支折叠掉，使下游分析只看到无条件 stage；若运行时条件包裹了 stage 调用，则直接报错（动态分支 stage 不受支持）。
• fwd_ids/bwd_ids取值范围为 0~15。
• fwd_ids/bwd_ids只支持两种写法：直接写整数列表（fwd_ids=[0, 1]），或写一个在 kernel 外绑定到整数列表的变量名（IDS = [0, 1]…fwd_ids=IDS）。元素必须是编译期常量整数，不支持切片、拼接、函数调用等表达式形式。
• fwd_ids/bwd_ids的长度只能等于该 Buffer 的 Tile 数，或者等于 1。等于 1 时多个 Tile 共用同一个同步 id，交接会被串行化（性能下降但结果正确），用于同步 id 不够分配的场景。
• 允许跨核 Buffer 之间、跨核 Buffer 与核内 Buffer 之间进行地址复用，但最多允许两块 Buffer 复用，且复用双方的 Tile 数需要一致。
• 地址复用的 Buffer 之间必须声明相同的mutex_ids。mutex 锁的是地址，不同的 id 等于没有互斥。
• 一个跨核 Buffer（通过fwd_ids/bwd_ids标记）需要恰好被两个 stage 使用，且这两个 stage 分别在 cube 和 vector 上，构成一对一的生产者/消费者关系。
• 跨核 Buffer 的 Tiles 必须随迭代顺序轮转。
• stage 函数如果有结构体参数，请使用pypto_pro.language.struct()声明，不支持pypto_pro.language.struct_array()。
• 跨核 Buffer 仅支持 UB 和 L1 Buffer。
• 以_pl_开头的变量名为框架保留，kernel 内不要使用。

调用示例

以下完整示例来自文档（单个 1:2 CV 执行组，relu(A @ B) @ D两段 matmul 的 CV 融合），演示了上述全部要素的组合：4 个 stage 依次完成mm1 = A_i @ B（cube）→relu并 insert 回 L1（vector）→mm2 = relu(mm1) @ D（cube）→ 写回 GM（vector）。

import os import pypto_pro.language as pl import pytest import torch import torch_npu import pypto ST_DEVICE_ID = int(os.environ.get("TILE_FWK_DEVICE_ID", 0)) ST_DEVICE = f"npu:{ST_DEVICE_ID}" TILE = 64 TILE_HALF = TILE // 2 # dual mode 沿 M 劈开，每个 AIV 处理一半 N_ITER = 4 @pl.pipeline.stage def stage1(ki, a, b_l1, a_l1_db, left_db, right_db, acc_db, mm1_vec_db): """Cube：mm1 = A_i @ B""" cur_a = a_l1_db.next() pl.load(cur_a, a, [ki * TILE, 0]) b_slot = b_l1.current() left = left_db.next() right = right_db.next() acc = acc_db.next() pl.move(left, cur_a) pl.move(right, b_slot) pl.matmul(acc, left, right) mm1_vec = mm1_vec_db.next() # DualModeSplitM：[TILE, TILE] 的 acc 沿 M 劈开，每个 AIV 拿 [TILE_HALF, TILE] pl.move(mm1_vec, acc, acc_to_vec_mode=pl.AccToVecMode.DualModeSplitM) @pl.pipeline.stage def stage2(ki, sub_id, mm1_vec_db, relu_vec_db, relu_nz_db, p_mat_db): """Vector：对本 AIV 那半做 relu，转 NZ 后按行偏移 insert 回 L1 Buffer""" mm1_vec = mm1_vec_db.next() relu_vec = relu_vec_db.next() pl.relu(relu_vec, mm1_vec) relu_nz = relu_nz_db.next() pl.move(relu_nz, relu_vec) # ND -> NZ，insert要求源Tile为NZ p_mat = p_mat_db.next() pl.insert(p_mat, relu_nz, [sub_id * TILE_HALF, 0]) @pl.pipeline.stage def stage3(ki, d_l1, p_mat_db, left_db, right_db, acc_db, out_vec_db): """Cube：mm2 = relu(mm1) @ D""" d_slot = d_l1.current() p_mat = p_mat_db.next() left = left_db.next() right = right_db.next() acc = acc_db.next() pl.move(left, p_mat) pl.move(right, d_slot) pl.matmul(acc, left, right) out_vec = out_vec_db.next() pl.move(out_vec, acc, acc_to_vec_mode=pl.AccToVecMode.DualModeSplitM) @pl.pipeline.stage def stage4(ki, sub_id, out, out_vec_db): """Vector：每个 AIV 写回本迭代结果的一半""" out_vec = out_vec_db.next() pl.store(out, out_vec, [ki * TILE + sub_id * TILE_HALF, 0]) @pl.jit(auto_mutex=True, pipeline=pl.pipeline.PipelineConfig(preload=2)) def pipeline_demo_kernel( a: pl.Tensor[[N_ITER * TILE, TILE], pl.DT_FP32], b: pl.Tensor[[TILE, TILE], pl.DT_FP32], d: pl.Tensor[[TILE, TILE], pl.DT_FP32], out: pl.Tensor[[N_ITER * TILE, TILE], pl.DT_FP32], ): # ===== 跨核共享 Buffer：声明 fwd_ids/bwd_ids，框架据此自动插同步 ===== mm1_vec_db = pl.make_tile_group( type=pl.TileType(shape=[TILE_HALF, TILE], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec), addrs=0x0000, mutex_ids=[12, 13], fwd_ids=[0, 1], bwd_ids=[2, 3], ) p_mat_db = pl.make_tile_group( type=pl.TileType(shape=[TILE, TILE], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Mat, layout=pl.NZ), addrs=0x8000, mutex_ids=[14, 15], fwd_ids=[4, 5], bwd_ids=[6, 7], ) out_vec_db = pl.make_tile_group( type=pl.TileType(shape=[TILE_HALF, TILE], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec), addrs=0x10000, mutex_ids=[16, 17], fwd_ids=[8, 9], bwd_ids=[10, 11], ) # ===== Cube 侧局部 Buffer ===== with pl.section_cube(): a_l1_db = pl.make_tile_group( type=pl.TileType(shape=[TILE, TILE], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Mat, layout=pl.NZ), addrs=0x0000, mutex_ids=[0, 1], ) b_l1 = pl.make_tile_group( type=pl.TileType(shape=[TILE, TILE], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Mat, layout=pl.NZ), addrs=0x10000, mutex_ids=[2], ) d_l1 = pl.make_tile_group( type=pl.TileType(shape=[TILE, TILE], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Mat, layout=pl.NZ), addrs=0x14000, mutex_ids=[3], ) left_db = pl.make_tile_group( type=pl.TileType(shape=[TILE, TILE], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Left, layout=pl.NZ), addrs=0x0000, mutex_ids=[4, 5], ) right_db = pl.make_tile_group( type=pl.TileType(shape=[TILE, TILE], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Right, layout=pl.ZN), addrs=0x0000, mutex_ids=[6, 7], ) acc_db = pl.make_tile_group( type=pl.TileType( shape=[TILE, TILE], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Acc, layout=pl.NZ, fractal=1024, ), addrs=0x0000, mutex_ids=[8, 9, 10, 11], ) # B、D 在循环外一次搬入，全程复用 b_slot = b_l1.current() d_slot = d_l1.current() pl.load(b_slot, b, [0, 0]) pl.load(d_slot, d, [0, 0]) # ===== Vector 侧局部 Buffer ===== with pl.section_vector(): sub_id = pl.get_subblock_idx() relu_vec_db = pl.make_tile_group( type=pl.TileType(shape=[TILE_HALF, TILE], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec), addrs=0x8000, mutex_ids=[18, 19], ) relu_nz_db = pl.make_tile_group( type=pl.TileType( shape=[TILE_HALF, TILE], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec, layout=pl.NZ, ), addrs=0x18000, mutex_ids=[20, 21], ) # ===== 流水循环：按 cube/vector 交替调用 4 个 stage ===== for ki in pl.range(0, N_ITER): with pl.section_cube(): stage1(ki, a, b_l1, a_l1_db, left_db, right_db, acc_db, mm1_vec_db) with pl.section_vector(): stage2(ki, sub_id, mm1_vec_db, relu_vec_db, relu_nz_db, p_mat_db) with pl.section_cube(): stage3(ki, d_l1, p_mat_db, left_db, right_db, acc_db, out_vec_db) with pl.section_vector(): stage4(ki, sub_id, out, out_vec_db) @pytest.mark.soc("950") @pypto.options(pass_options={"enable_slice": False}) def test_pipeline_demo_kernel(): device = ST_DEVICE torch.npu.set_device(device) torch.manual_seed(0) a = torch.randn(N_ITER * TILE, TILE, device=device, dtype=torch.float32) b = torch.randn(TILE, TILE, device=device, dtype=torch.float32) d = torch.randn(TILE, TILE, device=device, dtype=torch.float32) out = torch.zeros(N_ITER * TILE, TILE, device=device, dtype=torch.float32) pipeline_demo_kernelNone, 1 torch.npu.synchronize() ref = torch.relu(a @ b) @ d torch.testing.assert_close(out, ref, rtol=1e-2, atol=1e-2)

[!NOTE]说明

• 本示例按单个 1:2 CV 执行组设计，因此block_dim设置为 1。扩展为多个执行组时，需要通过pypto_pro.language.get_block_idx()划分各执行组处理的 GM 数据和输出范围。block_dim的完整说明参见 Kernel核函数。

深入原理：编译期变换与同步依赖图

理解了使用方式之后，再从源码层面看自动CV并行流水的实现机制，可以帮助你更好地理解约束的成因与 preload 的调优方向。

变换入口：从 AST 到并行流水

transform_pipeline（python/pypto_pro/runtime/pipeline/_transformer.py）接收 kernel 函数的 AST 与配置，工作流程大致为：

1. 在 AST 深拷贝上工作，先折叠编译期常量分支（_prune_const_branches），保证下游分析只看到无条件的 stage 调用；
2. analyze_pipeline分析串行结构，按源码顺序返回每个流水循环的PipelineInfo；
3. 对每个流水循环，若 preload 非 0，则调用_compute_delays计算各 stage 的延迟（beat 偏移）：同一核上第一个 stage 取上游跨核 stage 的延迟 + 1，同一核上的后续 stage 取前一个同核 stage 的延迟 + preload；
4. 依据延迟计算 ctx 环形缓冲区深度（max_delay + 1），把串行循环替换为带 ctx 槽位轮转、延迟 stage 与合法性守卫（is_valid）的并行循环；循环结束后追加 drain beats，让延迟的 stage 追完剩余任务；
5. preload=0 时走_transform_serial分支：保持串行循环结构，只自动插入跨核同步。

多个流水循环共享同一个函数作用域，因此框架引入的所有变量（_pl_ctx_arr、_pl_task_id、_pl_ctx_0/_pl_ctx_negN、drain 循环变量等）都会按流水序号加后缀，并通过validate_names与用户变量做碰撞检查——这也是"以_pl_开头的变量名为框架保留"约束的由来。

同步图模型：一次建模覆盖串行与并行

所有核间同步的插入都来自 _sync_graph.py 中构建的同步依赖图，它用一个模型统一覆盖 preload=0、preload=N 与地址复用三种场景：

• node：op 级访问Access(stage, buffer, role, pipe)；
• region：物理内存区域，对地址重叠的 Buffer 做并查集合并，共址 Buffer 落入同一 region；
• lane：一个 (region, 物理槽位)，同一内存上所有竞争访问按时间排序，边只会在 lane 内部形成，因此地址复用不需要特殊处理；
• edge：RAW（读等最近的前驱写）与 WAR（写等前驱写在该槽位上的所有读者），后者天然产生地址复用的反向保护边。

每个 stage 恰好推进跨核 Buffer 一次（slot = task % slot_count），距离来自时间线的展开而非槽位算术。规划输出分三部分：prefire（循环前预释放偏斜边头部缺失的 permit）、sites（循环内每个 stage 前后的 wait/set）、drain（循环后吸收尾部多余的 set），三者必须精确配对，因为硬件要求 set 与 wait 计数严格相等。

值得注意的两个实现细节：

• 用户声明了fwd_ids却没有bwd_ids，等价于声明"该 Buffer 槽位足够多，不需要覆写保护"，该 WAR 边会被解析为不发射指令——但边仍然存在于图中，环与逆时间分析始终能看到真实依赖；
• 地址复用产生的跨 Buffer WAR 依赖是用户从未声明过 id 的，框架会从空闲事件 id 池（0~15）中自动分配一组，_pl_overlap_ids_N会被声明在循环之前；若池不够，会按槽位数量升序把部分边降级为单 id 共用（交接串行化但结果正确），仍不够则报错。

环形缓冲区与任务计数

并行版本的核心数据结构是 ctx 环形缓冲区（pl.struct_array声明）：每个 beat 通过_pl_ctx_arr[_pl_task_id % depth]选取槽位，快照本 beat 各 stage 的参数（stage 参数被推导为 ctx 字段），延迟 stage 则读取自己延迟拍之前填充的槽位，配合is_valid守卫区分有效任务与排空拍。任务计数_pl_task_id按帧递增，drain beats 在循环结束后按depth - 1次继续推进，让最后几拍的任务在各自延迟的 stage 上跑完。

调优建议与测试验证

仓库 python/tests/st/pypto_pro/frontend/auto_pipeline 目录下提供了多个针对自动流水变换的测试用例，例如test_auto_pipeline_two_loops.py（验证单 kernel 内多条流水循环）与test_pipeline_loop_body.py（验证循环体语句处理），可作为理解变换行为与回归验证的参考。结合前文，给出如下调优与排查路径：

1. 先验证串行正确性：PipelineConfig(preload=0)只插入核间同步不改写循环，是精度验证的基线；多流水循环时可逐条置 0（如preload=[0, 2]）定位问题出自哪条流水；
2. 再逐步增大 preload：从 1 开始逐级提升，观察pipeline_generated.py中 ctx 深度、延迟与同步站点的变化；收益上限受较慢核的稳态节奏约束，stage 划分越均衡收益越高；
3. 遇到同步相关报错时：优先核对跨核 Buffer 的fwd_ids/bwd_ids是否按约束声明（范围 0~15、长度等于 Tile 数或 1、写法合规）、地址复用的 Buffer 是否声明了相同的mutex_ids、跨核 Buffer 是否恰好被一个 cube stage 与一个 vector stage 一对一使用、Tiles 是否随迭代轮转。

至此，你已经掌握自动CV并行流水的完整用法：用pl.pipeline.stage划分阶段、用make_tile_group的fwd_ids/bwd_ids声明跨核通道、在jit中配置PipelineConfig(preload=...)开启变换，并能在生成的pipeline_generated.py基础上继续调试与优化。

【免费下载链接】pyptoPyPTO（发音: pai p-t-o）：Parallel Tensor/Tile Operation编程范式。项目地址: https://gitcode.com/cann/pypto