PyPTO 核间同步之 wait_cross_core：接口语义、同步模式与实战示例

• 人工智能
• 编译器
• 模型编译
• 高性能计算
• 深度学习
• CANN

pypto_pro.language.system.wait_cross_core是 CANN PyPTO（Parallel Tensor/Tile Operation 编程范式）中用于核间（跨 AI Core / AIV）同步消费的底层控制接口，与 set_cross_core 配对使用，实现不同核、不同流水之间的信号握手。本文以该接口为主线，完整讲解其函数原型、参数语义、四种CrossCoreSyncMode同步模式、事件 ID 配对规则与死锁约束，并结合仓库源码与测试用例说明其底层实现原理，帮助读者在自研 Kernel 中正确编排多核并行计算。

一、接口定位：核间同步的"等待-消费"半边

在多核（多 AI Core）并行编程中，一个 Kernel 的计算可能被拆分到多个核（AIC / AIV）上执行，核之间往往存在数据依赖：例如 AIV 计算出的中间结果需要交给同一 AI Core 的 AIC 做矩阵乘。PyPTO 通过set_cross_core/wait_cross_core这一对接口实现核间同步：

• set_cross_core：发送同步信号（SET）。每个事件 ID 对应一个初始值为 0 的计数器，指定流水中的前序指令完成后执行 SET，使对应计数器加 1；
• wait_cross_core：等待并消费同步信号（WAIT）。执行时若计数器为 0，则阻塞指定流水中的后续指令；若计数器大于 0，则计数器减 1 并放行后续指令。

从计数语义可以清晰看出，wait_cross_core是"带消费"的同步：每一条 WAIT 恰好抵消一条 SET，二者必须成对出现，这与sync_all（全核栅栏）等一次性屏障不同，因而天然支持多轮、多事件的细粒度流水同步。其同步机制细节以 set_cross_core 为准。

在 Python 侧，该接口由 python/pypto_pro/ir/op/system_ops.py 中的wait_cross_core构造 IR 算子调用实现，pipe、event_id、sync_mode均为仅限关键字参数（keyword-only）。

二、产品支持情况

即该接口目前仅在 Ascend 950 系列产品上可用，使用前需确认目标硬件平台。

三、函数原型

pypto_pro.language.system.wait_cross_core( *, pipe: PipeType, event_id: Union[int, Scalar], sync_mode: CrossCoreSyncMode = pypto_pro.language.CrossCoreSyncMode.INTRA_BLOCK, ) -> None

三个参数均为仅限关键字参数（调用时必须写成pipe=...形式）。函数无返回值。

四、参数说明

4.1 pipe 的语义细节

pipe决定的是"哪条硬件流水被卡住"，因此它既可以是产生数据的流水（如 MTE3 搬出完成后再等待），也可以是消费数据的流水（如 MTE1 等待搬入前提）。由于 WAIT 只阻塞指定流水尚未下发的后续指令，已下发的指令不受影响，这为流水级并行留出了空间——例如可以在 MTE3 等待期间让 V 流水继续执行不依赖同步数据的指令。

同时需要注意：WAIT 的 pipe 与 SET 的 pipe 不必相同。典型场景是"MTE3 上发送信号、V 流水上等待"，允许跨流水组合出更灵活的同步拓扑。

4.2 event_id 的静态与动态形式

源码 system_ops.py 展示了两种形式的底层分流：

if isinstance(event_id, Expr): return _ir_core.create_op_call( "system.wait_cross_core_dyn", [event_id], {"pipe": pipe, "sync_mode": sync_mode}, actual_span ) _check_id_range(event_id, MAX_EVENT_ID, "event_id") kwargs = {"pipe": pipe, "event_id": event_id, "sync_mode": sync_mode} return _ir_core.create_op_call("system.wait_cross_core", [], kwargs, actual_span)

• 传入 Python 整型常量时，会经过_check_id_range校验（MAX_EVENT_ID = 15，见 system_ops.py），越界直接报错；
• 传入运行时 Scalar 表达式（Expr）时，生成动态事件 ID 算子system.wait_cross_core_dyn，取值合法性（0～15，UNICAST_BLOCK 的 AIC 侧为 0～31）由调用方保证。

4.3 对 ALL 的拒绝

_validate_concrete_pipe明确拒绝PipeType.ALL：

if pipe == PipeType.ALL: raise InvalidArgument(f"{name} must identify one concrete pipe, got PipeType.ALL", span=span)

见 system_ops.py。也就是说，wait_cross_core必须指定一条具体的硬件流水，这与sync_all、bar_all等全局屏障语义有本质区别。

五、同步模式详解：CrossCoreSyncMode 四种取值

sync_mode的类型为 pypto_pro.language.CrossCoreSyncMode，枚举定义：

PYPTO_DECLARE_ENUM(CrossCoreSyncMode, INTER_BLOCK, INTER_SUBBLOCK, INTRA_BLOCK, UNICAST_BLOCK )

四种模式参与同步的核与信号配对方式如下：

5.1 UNICAST_BLOCK 的事件 ID 配对细节

UNICAST_BLOCK 是四者中唯一存在"事件 ID 映射偏移"的模式，也是最容易用错的点：

• AIV0 发送 0～15 ↔ AIC 等待 0～15；
• AIV1 发送 0～15 ↔ AIC 等待 16～31；
• AIC 发送 0～15 ↔ AIV0 等待 0～15；
• AIC 发送 16～31 ↔ AIV1 等待 0～15。

即"谁在等待"决定了 AIC 侧事件 ID 是否要加 16 的偏移，而 AIV 侧永远只使用 0～15。编写 UNICAST_BLOCK 同步时，务必根据收发双方核对 ID，避免错配导致死锁或误放行。

六、约束说明

• 必须存在匹配的 SET：调用wait_cross_core前必须存在与之配对的 set_cross_core 调用，并保证所有参与同步的核均能到达同步点，否则可能发生死锁（某个核一直等待永远不来的信号）。
• INTER_BLOCK 的并发死锁风险：使用 INTER_BLOCK 时还需满足 set_cross_core 的约束：当多流或多个算子并发执行，且并发算子申请的核数总和超过物理核数时，如果至少两个并发算子使用核间同步，部分核可能因未被调度而无法到达同步点，造成死锁。必须保证每个同步算子所需的核能够同时执行。
• 同一事件 ID 的复用前提：对于 INTER_BLOCK、INTER_SUBBLOCK 和 INTRA_BLOCK，同一核复用事件 ID 或将同一事件 ID 用于不同同步模式前，必须完成该事件 ID 在前一同步过程中的所有 SET 和 WAIT。
• 计数器溢出：每个事件 ID 对应的计数器取值范围为 0～15；同一事件的信号未被 WAIT 消费时，连续发送超过 15 次 SET 会触发异常并中断执行。
• 与 sync_all 及自动流水编排的冲突：与 sync_all 同时使用时，须避开 HARD 模式占用的事件 ID；使用自动流水编排时，还应避免与其分配的事件 ID 冲突。
• SET 顺序不确定性：同一核连续发送多个 SET 时，不保证不同事件 ID 之间的生效顺序；存在先后依赖时，应先完成前一组 SET/WAIT。

七、调用示例

7.1 INTER_BLOCK：全核屏障式等待

with pl.section_vector(): pl.system.wait_cross_core( pipe=pl.PipeType.MTE3, event_id=0, sync_mode=pl.CrossCoreSyncMode.INTER_BLOCK, ) # 所有AIV均到达同步点后执行的操作。

对应配对的 SET 侧（所有 AIV 各自执行完前置操作后发送信号）：

with pl.section_vector(): # 本AIV上的前置操作。 pl.system.set_cross_core( pipe=pl.PipeType.MTE3, event_id=0, sync_mode=pl.CrossCoreSyncMode.INTER_BLOCK, )

7.2 INTER_SUBBLOCK：同核内 AIV0/AIV1 握手

with pl.section_vector(): pl.system.wait_cross_core( pipe=pl.PipeType.V, event_id=1, sync_mode=pl.CrossCoreSyncMode.INTER_SUBBLOCK, ) # 同一AI Core内的AIV0和AIV1均到达后继续。

配对 SET 侧由 AIV0 和 AIV1 各自执行前置操作后发送：

with pl.section_vector(): # AIV0和AIV1各自执行前置操作。 pl.system.set_cross_core( pipe=pl.PipeType.V, event_id=1, sync_mode=pl.CrossCoreSyncMode.INTER_SUBBLOCK, )

7.3 INTRA_BLOCK：AIC 与全部 AIV 同步（默认模式）

with pl.section_cube(): # 等待AIV0和AIV1的信号。 pl.system.wait_cross_core( pipe=pl.PipeType.MTE1, event_id=2, sync_mode=pl.CrossCoreSyncMode.INTRA_BLOCK, )

AIC 侧需要等待两路信号（AIV0 与 AIV1 分别发送），因此配对的 SET 侧是：

with pl.section_vector(): # AIV0和AIV1完成前置操作后分别发送信号。 pl.system.set_cross_core( pipe=pl.PipeType.MTE3, event_id=2, sync_mode=pl.CrossCoreSyncMode.INTRA_BLOCK, )

7.4 UNICAST_BLOCK：AIC 与单个 AIV 同步

with pl.section_cube(): # 仅等待AIV0的信号。 pl.system.wait_cross_core( pipe=pl.PipeType.S, event_id=15, sync_mode=pl.CrossCoreSyncMode.UNICAST_BLOCK, )

配对 SET 侧需要显式限定只有 AIV0 发送信号：

with pl.section_vector(): if pl.get_subblock_idx() == 0: # 仅AIV0发送信号。 pl.system.set_cross_core( pipe=pl.PipeType.S, event_id=15, sync_mode=pl.CrossCoreSyncMode.UNICAST_BLOCK, )

7.5 完整 Kernel 示例：AIV 计算 + AIC 矩阵乘的 INTRA_BLOCK 同步

下面的完整示例体现了核间同步的真实用法：AIV 子核分别加载x、y并按子核索引切片做加法，将结果写入共享的v1_mat（Mat 空间）；AIC 侧加载rhs后在 MTE1 上等待 AIV 的信号，再读取v1_mat做矩阵乘。注意其中pl.system.set_cross_core在section_vector中发送，pl.system.wait_cross_core在section_cube中等待，二者同为INTRA_BLOCK、event_id=2：

import pypto_pro.language as pl @pl.jit() def cross_core_kernel( x: pl.Tensor[[64, 64], pl.DT_FP32], y: pl.Tensor[[64, 64], pl.DT_FP32], rhs: pl.Tensor[[64, 64], pl.DT_FP32], out: pl.Tensor[[64, 64], pl.DT_FP32], ): v1_mat = pl.make_tile( pl.TileType(shape=[64, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Mat, layout=pl.NZ), addr=0x10000) with pl.section_vector(): sub_index = pl.get_subblock_idx() off = sub_index * 32 tile_x = pl.make_tile( pl.TileType(shape=[32, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec), addr=0x0000) tile_y = pl.make_tile( pl.TileType(shape=[32, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec), addr=0x2000) tile_sum = pl.make_tile( pl.TileType(shape=[32, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec), addr=0x4000) tile_nz = pl.make_tile( pl.TileType(shape=[32, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec, layout=pl.NZ), addr=0x6000) pl.load(tile_x, x, [off, 0]) pl.load(tile_y, y, [off, 0]) pl.system.sync_src(set_pipe=pl.PipeType.MTE2, wait_pipe=pl.PipeType.V, event_id=0) pl.system.sync_dst(set_pipe=pl.PipeType.MTE2, wait_pipe=pl.PipeType.V, event_id=0) pl.add(tile_sum, tile_x, tile_y) pl.move(tile_nz, tile_sum) pl.system.sync_src(set_pipe=pl.PipeType.V, wait_pipe=pl.PipeType.MTE3, event_id=2) pl.system.sync_dst(set_pipe=pl.PipeType.V, wait_pipe=pl.PipeType.MTE3, event_id=2) pl.insert(v1_mat, tile_nz, [off, 0]) pl.system.set_cross_core( pipe=pl.PipeType.MTE3, event_id=2, sync_mode=pl.CrossCoreSyncMode.INTRA_BLOCK, ) with pl.section_cube(): rhs_mat = pl.make_tile( pl.TileType(shape=[64, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Mat, layout=pl.NZ), addr=0x0000) v1_left = pl.make_tile( pl.TileType(shape=[64, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Left, layout=pl.NZ), addr=0x0000) rhs_right = pl.make_tile( pl.TileType(shape=[64, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Right, layout=pl.ZN), addr=0x0000) c_l0c = pl.make_tile( pl.TileType(shape=[64, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Acc, layout=pl.NZ, fractal=1024), addr=0x0000) pl.load(rhs_mat, rhs, [0, 0]) pl.system.sync_src(set_pipe=pl.PipeType.MTE2, wait_pipe=pl.PipeType.MTE1, event_id=0) pl.system.sync_dst(set_pipe=pl.PipeType.MTE2, wait_pipe=pl.PipeType.MTE1, event_id=0) pl.move(rhs_right, rhs_mat) pl.system.wait_cross_core( pipe=pl.PipeType.MTE1, event_id=2, sync_mode=pl.CrossCoreSyncMode.INTRA_BLOCK, ) pl.move(v1_left, v1_mat) pl.system.sync_src(set_pipe=pl.PipeType.MTE1, wait_pipe=pl.PipeType.M, event_id=0) pl.system.sync_dst(set_pipe=pl.PipeType.MTE1, wait_pipe=pl.PipeType.M, event_id=0) pl.matmul(c_l0c, v1_left, rhs_right) pl.system.sync_src(set_pipe=pl.PipeType.M, wait_pipe=pl.PipeType.FIX, event_id=0) pl.system.sync_dst(set_pipe=pl.PipeType.M, wait_pipe=pl.PipeType.FIX, event_id=0) pl.store(out, c_l0c, [0, 0])

示例中的要点：

1. 流水内同步先行：在核间 SET 之前，先用sync_src/sync_dst保证insert到 Mat 空间的写已完成（V→MTE3 事件 ID 2）；
2. SET 与 WAIT 分属不同 section：AIV 侧在section_vector中发送，AIC 侧在section_cube中等待，符合 INTRA_BLOCK 的 AIV→AIC 方向语义；
3. WAIT 之后才消费共享数据：wait_cross_core(pipe=MTE1, event_id=2)阻塞 MTE1，确保pl.move(v1_left, v1_mat)发生在 AIV 的insert完成之后；
4. 事件 ID 可复用不同同步域：事件 ID 0 被用于流水内sync_src/sync_dst，事件 ID 2 被用于核间同步，二者同步域不同、互不冲突。

八、源码级实现解析

8.1 IR 层的算子构造

wait_cross_core在 python/pypto_pro/ir/op/system_ops.py 中实现，核心逻辑是：

1. 校验pipe必须是具体的PipeType（拒绝ALL）；
2. 对静态event_id做0～15的范围校验（_check_id_range，配合MAX_EVENT_ID = 15）；
3. 动态event_id（Expr）走system.wait_cross_core_dyn算子；
4. 静态event_id生成system.wait_cross_core算子调用。

set_cross_core的实现与之对称（system_ops.py），二者共享同一套 pipe 校验与 ID 范围校验逻辑，这也解释了文档中"配对使用、参数须一致"的要求为何会在编译期被强制。

8.2 与自动流水编排的关系

在 PyPTO 的自动流水编排（pipeline scheduling）框架中，存在专门的跨核同步扫描逻辑 python/pypto_pro/runtime/pipeline/_cross_core_scanner.py，其文档注释表明：扫描结果会驱动在流水阶段边界自动插入 wait/set_cross_core。也就是说，wait_cross_core既是开发者可以手动调用的底层原语，也是自动流水编排内部用于生成跨核依赖同步的构件之一。这解释了文档中"使用自动流水编排时，应避免与其分配的事件 ID 冲突"这一约束的由来。

8.3 枚举的 Python 导出

CrossCoreSyncMode由 python/pypto_pro/language/init.py 导出（源码中同时出现在"CrossCoreSyncMode"的__all__列表与 import 列表中），并在 C++ 绑定层 python/src/bindings/ir/ir.cpp 中注册，因此可以在 Kernel 中以pl.CrossCoreSyncMode.INTRA_BLOCK形式直接使用。

九、测试用例验证

仓库测试 python/tests/st/pypto_pro/frontend/system/test_manual_sync_dynamic.py 中给出了核间同步与核内手动同步共存的验证场景：

with pl.section_vector(): pl.system.set_cross_core( pipe=pl.PipeType.V, event_id=14, sync_mode=pl.CrossCoreSyncMode.INTER_SUBBLOCK ) pl.system.wait_cross_core( pipe=pl.PipeType.V, event_id=14, sync_mode=pl.CrossCoreSyncMode.INTER_SUBBLOCK )

该用例的注释说明：跨子核同步可以与核内所有手动同步形式共存，测试仍以端到端精度结果校验核内流水行为。此外 python/tests/st/pypto_pro/frontend/system/test_cube_vector_sync.py、python/tests/ut/pypto_pro/codegen/test_manual_sync_dynamic_shape.py 等测试也覆盖了核间/核内同步的多种组合，可作为编写自身用例时的参考范式。

十、使用要点小结

1. 成对使用：每个wait_cross_core必须匹配一个同event_id、同sync_mode的set_cross_core，并保证所有参与核都能到达同步点；
2. pipe 各管各的：SET 与 WAIT 的 pipe 可以不同，WAIT 的 pipe 只决定哪条流水被阻塞；
3. 模式决定核集合：INTER_BLOCK（跨核同类核）、INTER_SUBBLOCK（同核 AIV0/AIV1）、INTRA_BLOCK（同核 AIC↔全部 AIV，默认）、UNICAST_BLOCK（同核 AIC↔单个 AIV，注意 AIC 侧 16～31 的 ID 偏移）；
4. 事件 ID 是有限资源：静态 ID 0～15，计数器上限 15，复用前须确认前一轮 SET/WAIT 全部完成，同时避开sync_allHARD 模式与自动流水编排占用的 ID；
5. 警惕死锁：INTER_BLOCK 模式下并发算子核数总和超过物理核数时可能因部分核未被调度而死锁，需保证同步算子所需核能同时执行；
6. 平台限制：当前仅 Ascend 950PR / Ascend 950DT 支持。

核间同步是编写高效多核 Kernel 的关键一环，建议读者结合 set_cross_core 与 CrossCoreSyncMode 两份文档通读，并以上述完整 Kernel 示例为起点，在目标硬件上实测验证同步语义。

• 人工智能
• 编译器
• 模型编译
• 高性能计算
• 深度学习
• CANN