PyPTO 同步控制(SIMD-API synchronization)完全指南:核内流水同步、互斥与跨核屏障
2026/9/20 0:13:52 网站建设 项目流程
  • 人工智能
  • 编译器
  • 模型编译
  • 高性能计算
  • 深度学习
  • CANN

【免费下载链接】pypto

PyPTO(发音: pai p-t-o):Parallel Tensor/Tile Operation编程范式。

项目地址:https://gitcode.com/cann/pypto
点击查看免费下载

导读

本文以 PyPTO 项目 SIMD-API 中的同步控制文档为主体,系统讲解 Ascend 950 系列(A5)上 Kernel 编程所需的全部同步原语:核内流水同步(sync_src/sync_dst)、单流水屏障(bar_m/bar_mte1/bar_mte2/bar_mte3/bar_fix)、全流水屏障(bar_all)、缓冲区互斥(mutex_lock/mutex_unlock)以及跨核同步(set_cross_core/wait_cross_core/sync_all)。读完本文,你将理解 AI Core 内部多条并行流水之间的数据依赖模型,掌握在 Vector/Cube 区段中正确编排同步指令、避免数据竞争与死锁的完整方法,并能在自己的 PyPTO Kernel 中直接复用文中的可运行示例。

产品支持情况:本目录全部接口仅支持Ascend 950PR/Ascend 950DT(A5 系列);Atlas A3 训练/推理系列与 Atlas A2 训练/推理系列均不支持。源码侧同步控制相关 ST 测试也通过_require_a5@pytest.mark.soc("950")限定运行设备(参见 test_pipe_barriers.py)。

一、前置知识:AI Core 的并行流水模型

理解同步控制的前提是认识 AI 处理器内部的并行流水结构。PyPTO 用pypto_pro.language.PipeType枚举标记操作运行在哪条硬件流水上,其完整定义与职责如下(详见 PipeType.md):

枚举值职责
MTE1搬运流水 1:L1 Buffer → L0A/L0B/BiasTable/L0A_MX/L0B_MX Buffer
MTE2搬运流水 2:GM → L1 Buffer/UB 的 load 搬入
MTE3搬运流水 3:UB → GM 的 store 搬出、UB → L1 Buffer 的 move
M矩阵计算流水:Cube/MAD 等 matmul 计算
V向量计算流水:element-wise、reduce、cast、quant、dequant 等
S标量流水:getval、setval 等标量操作
FIXFixpipe 流水:L0C Buffer 结果读出及随路量化/反量化
ALL本 AI Core 的全部流水(V、M、MTE1、MTE2、MTE3、FIX 等)

数据搬运操作(load/store/move)的流水由源/目的内存空间自动决定:load(GM→UB/L1)走 MTE2,store(UB→GM)走 MTE3,store(L0C→GM)走 FIX,move(L1→L0A/L0B)走 MTE1,move(UB→UB)走 V 等。由于不同流水并行执行,上游流水尚未完成的数据读写,下游流水可能已经开始执行,这正是同步控制要解决的问题。

二、核内流水同步:sync_src 与 sync_dst

2.1 功能与原理

sync_srcsync_dst用于同一 AI Core 内不同流水之间的点对点同步,二者必须配对使用:

  • sync_src(set_pipe=..., wait_pipe=..., event_id=...):当set_pipe中位于本接口之前的所有数据读写操作完成后,将对应标志位置 1(Set Flag)。该接口不会阻塞set_pipe中位于其后的操作。
  • sync_dst(set_pipe=..., wait_pipe=..., event_id=...):当wait_pipe执行到本接口时,若对应标志位为 0,则wait_pipe中位于本接口之后的操作被阻塞;若标志位为 1,则清零并继续执行(Wait Flag)。

这一"置位不阻塞发送方、等待阻塞接收方"的设计让搬运流水(MTE2/MTE3)与计算流水(V/M/FIX)之间可以流水化并行,只在数据依赖点精确停靠。底层实现中,两个接口在 system_ops.py 分别生成system.sync_srcsystem.sync_dstIR 调用,校验set_pipe/wait_pipe为具体流水并检查event_id范围。

2.2 参数说明

参数输入/输出说明
set_pipe输入源流水(发送同步事件的流水)。必须为具体流水,不支持PipeType.ALL,且必须与wait_pipe不同
wait_pipe输入目的流水(等待同步事件的流水)。必须为与set_pipe不同的具体流水,不支持PipeType.ALL
event_id输入事件 ID。支持 Python int、整型常量表达式或整型运行时 Scalar,不支持 bool,取值范围[0, 7]

2.3 支持的流水组合

Vector 段支持的组合:

set_pipe支持的 wait_pipe
MTE2V、MTE3、S
MTE3V、MTE2、S
VMTE2、MTE3、S
SV、MTE2、MTE3

Cube 段支持的组合:

set_pipe支持的 wait_pipe
MTE1MTE2、M、MTE3、FIX
MTE2MTE1、M、MTE3、S、FIX
MTE3MTE1、MTE2、S
MMTE1、MTE2、FIX、S
SMTE2、MTE3
FIXM、MTE2、S、MTE3、MTE1

2.4 典型同步模式

场景set_pipewait_pipe说明
load 后进行向量计算MTE2VGM 数据搬入 UB/L1 后再计算
向量计算后 storeVMTE3计算完成后再搬出到 GM
store 后再次 loadMTE3MTE2搬出到 GM 后再向相同 Buffer 搬入新数据
L1→L0A 搬运后矩阵计算MTE1M确保矩阵操作数就位
matmul 后 FIX 读出结果MFIX确保矩阵计算完成
标量读写MTE2/MTE3Sgetval/setval 走标量流水

2.5 调用示例

以下示例先经 MTE2 把两个 FP32 输入从 GM 搬入 UB,再用MTE2→V同步确保搬运完成,执行pl.add;计算完成后用V→MTE3同步确保计算完成,最后pl.store写回 GM。该示例与 ST 测试 test_sync_src_dst 中的sync_src_dst_kernel完全一致,测试在 Ascend 950 上断言out == a + b

import pypto_pro.language as pl @pl.jit() def sync_src_dst_kernel( a: pl.Tensor[[64, 64], pl.DT_FP32], b: pl.Tensor[[64, 64], pl.DT_FP32], out: pl.Tensor[[64, 64], pl.DT_FP32], ): tt = pl.TileType(shape=[64, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec) tile_a = pl.make_tile(tt, addr=0x0000) tile_b = pl.make_tile(tt, addr=0x4000) tile_out = pl.make_tile(tt, addr=0x8000) with pl.section_vector(): pl.load(tile_a, a, [0, 0]) pl.load(tile_b, b, [0, 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_out, tile_a, tile_b) pl.system.sync_src(set_pipe=pl.PipeType.V, wait_pipe=pl.PipeType.MTE3, event_id=1) pl.system.sync_dst(set_pipe=pl.PipeType.V, wait_pipe=pl.PipeType.MTE3, event_id=1) pl.store(out, tile_out, [0, 0])

2.6 约束要点

  • 两个接口的set_pipewait_pipeevent_id必须完全一致,且sync_srcsync_dst
  • 同一 set_pipe/wait_pipe 组合下,同一event_id只能在前一次sync_dst完成等待后复用;连续调用sync_src、漏写配对接口或两个接口所在控制流路径不一致,都可能造成数据竞争或死锁
  • auto_mutex=True同时使用时,应确保显式同步与自动 mutex 分别处理明确的依赖,避免对同一依赖重复同步。

三、单流水屏障:bar_m / bar_mte1 / bar_mte2 / bar_mte3 / bar_fix

当只需要等待某一条流水时,应使用对应的单流水屏障接口,避免bar_all带来的全局停顿开销。这些接口均无参数、无返回值,仅等待当前 AI Core 内对应流水中此前下发的操作完成,不执行跨核同步。IR 层统一通过_create_barrier_op生成system.bar_*调用(参见 system_ops.py)。

接口等待的流水可调用区段特别约束
bar_mM(矩阵计算)仅 Cube 区段等待 M 流水矩阵操作完成
bar_mte1MTE1仅 Cube 区段等待 MTE1 搬运完成
bar_mte2MTE2Cube 或 Vector 区段连续 MTE2 搬入写入的片上地址重叠时,两次搬入之间必须调用,否则产生错误数据
bar_mte3MTE3Cube 或 Vector 区段连续 MTE3 搬出写入的GM 地址重叠时,两次搬出之间必须调用,否则产生错误数据
bar_fixFIX仅 Cube 区段等待 FIX 流水操作完成

bar_mte2bar_mte3的"地址重叠"约束是最常见的正确性陷阱:当循环复用同一块片上 Buffer 反复搬入/搬出数据时,若不在轮次之间插入对应屏障,前一次 DMA 尚未完成、下一次 DMA 已改写同一地址,将读到错误数据。例如循环内复用 UB 的场景:

with pl.section_vector(): for i in pl.range(0, 128, 64): pl.load(tile_x, x, [i, 0]) # MTE2 搬入 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_out, tile_x, tile_x) # V 计算 pl.system.sync_src(set_pipe=pl.PipeType.V, wait_pipe=pl.PipeType.MTE3, event_id=1) pl.system.sync_dst(set_pipe=pl.PipeType.V, wait_pipe=pl.PipeType.MTE3, event_id=1) pl.store(out, tile_out, [i, 0]) # MTE3 搬出

Cube 区段中的bar_m可用于强制两次连续 matmul 串行化(如累加复用 L0C 的场景,见 bar_m.md 示例)。

四、全流水屏障:bar_all

pypto_pro.language.system.bar_all() -> None

bar_all等待当前 AI Core 内 V、M、MTE1、MTE2、MTE3 和 FIX 等全部流水此前下发的操作完成,可在 Cube 区段或 Vector 区段调用。需要特别注意:

  • 仅是核内全流水屏障,不是多核之间的全局屏障;多核同步请使用sync_all
  • 它会让当前 AI Core 的全部流水停顿,可能影响性能;只需等待一条流水时应优先使用对应单流水屏障接口。

典型用途是"循环体顶部的全流水同步":确保上一轮迭代的 store 全部完成后再开始本轮 load。文档示例与 ST 测试 test_bar_all 中的bar_all_kernel一致:

import pypto_pro.language as pl @pl.jit() def bar_all_kernel( x: pl.Tensor[[128, 64], pl.DT_FP16], out: pl.Tensor[[128, 64], pl.DT_FP16], ): tt = pl.TileType(shape=[64, 64], dtype=pl.DT_FP16, target_memory=pl.MemorySpace.Vec) tile_x = pl.make_tile(tt, addr=0x0000) tile_out = pl.make_tile(tt, addr=0x2000) with pl.section_vector(): for i in pl.range(0, 128, 64): pl.system.bar_all() # 等待上一轮全部流水完成 pl.load(tile_x, x, [i, 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_out, tile_x, tile_x) pl.system.sync_src(set_pipe=pl.PipeType.V, wait_pipe=pl.PipeType.MTE3, event_id=1) pl.system.sync_dst(set_pipe=pl.PipeType.V, wait_pipe=pl.PipeType.MTE3, event_id=1) pl.store(out, tile_out, [i, 0])

五、缓冲区互斥:mutex_lock 与 mutex_unlock

5.1 功能与原理

mutex_lock在指定 pipe 上获取mutex_id对应的缓冲区互斥资源(buffer-id token),资源尚未释放时当前 pipe 等待,直到能够获取;mutex_unlock释放该资源,使等待的其他 pipe 继续执行。二者用于防止多条 pipe 同时访问同一片上缓冲区,其 IR 实现位于 system_ops.py。

pypto_pro.language.system.mutex_lock( *, pipe: PipeType, mutex_id: Union[int, Scalar], ) -> None pypto_pro.language.system.mutex_unlock( *, pipe: PipeType, mutex_id: Union[int, Scalar], ) -> None

5.2 参数说明

参数输入/输出说明
pipe输入PipeType枚举值,必须是 MTE1/MTE2/MTE3/V/M/S/FIX 中的一条具体 pipe;不允许PipeType.ALL
mutex_id输入Python 整数、结果为整数的常量表达式,或整型的运行时 Scalar 表达式。静态 ID 取值范围 [0, 31],不接受 bool;动态 ID 运行时取值需在 [0, 31] 内

5.3 约束要点

  • mutex_lock必须与同一 pipe、同一 mutex_idmutex_unlock成对使用,先 lock 后 unlock。
  • 同一 pipe 上,前一次 lock 未被对应 unlock 释放时,不得再次获取同一 mutex_id,否则第二次获取会持续等待导致死锁;手动与自动生成的 mutex 操作也不得在同一 pipe 上重复获取尚未释放的同一 ID。
  • 同一 mutex_id 的 lock/unlock不得嵌套使用(无论各组操作的 pipe 是否相同);使用自动 mutex 时也须避免与显式 mutex 操作形成同一 ID 的嵌套。
  • 同一 pipe 上连续使用相同 mutex_id 的多组 lock/unlock,不能保证该 pipe 中各组操作依次完成。需要保证同一流水内前序操作完成后再执行后序操作时,应优先使用对应单流水屏障接口;当前流水没有对应单流水屏障接口时,可使用bar_all
  • lock/unlock 必须位于对称的控制流路径中,确保每次获取的互斥资源均被释放。
  • auto_mutex=True仅对带 mutex 元数据的 Tile 自动生成互斥操作;显式调用的mutex_lock仍会保留,自动同步与手动同步可在同一 Kernel 中混用。
  • 常规单缓冲、双缓冲与 N 缓冲场景,推荐使用 make_tile_group 配合auto_mutex=True;只有需要精确控制加锁 pipe 与插入位置时才使用手动 mutex。

5.4 调用示例

以下 Kernel 在auto_mutex=False下计算out = x + x:输入 UB 用 mutex ID 0 约束 MTE2 与 V 的访问顺序,输出 UB 用 mutex ID 1 约束 V 与 MTE3 的访问顺序。每次 lock 之后均在同一 pipe 上调用对应 unlock:

import pypto_pro.language as pl @pl.jit(auto_mutex=False) def mutex_kernel( x: pl.Tensor[[64, 64], pl.DT_FP32], out: pl.Tensor[[64, 64], pl.DT_FP32], ): tt = pl.TileType(shape=[64, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec) tile_x = pl.make_tile(tt, addr=0x0000) tile_out = pl.make_tile(tt, addr=0x4000) with pl.section_vector(): pl.system.mutex_lock(pipe=pl.PipeType.MTE2, mutex_id=0) pl.load(tile_x, x, [0, 0]) pl.system.mutex_unlock(pipe=pl.PipeType.MTE2, mutex_id=0) pl.system.mutex_lock(pipe=pl.PipeType.V, mutex_id=0) pl.system.mutex_lock(pipe=pl.PipeType.V, mutex_id=1) pl.add(tile_out, tile_x, tile_x) pl.system.mutex_unlock(pipe=pl.PipeType.V, mutex_id=1) pl.system.mutex_unlock(pipe=pl.PipeType.V, mutex_id=0) pl.system.mutex_lock(pipe=pl.PipeType.MTE3, mutex_id=1) pl.store(out, tile_out, [0, 0]) pl.system.mutex_unlock(pipe=pl.PipeType.MTE3, mutex_id=1)

六、跨核信号同步:set_cross_core 与 wait_cross_core

6.1 功能与原理

set_cross_corewait_cross_core是一对核间同步接口,基于计数信号量实现:每个event_id对应一个初始值为 0 的计数器。set_cross_core在指定 pipe 的前序指令完成后使对应计数器加 1;配对的wait_cross_core执行时,若计数器为 0 则阻塞指定流水中的后续指令,若大于 0 则减 1并继续。

接口支持三类同步场景:同类 AIC 或 AIV 的全核同步、同一 AI Core 内 AIV 之间的同步、同一 AI Core 内 AIC 与 AIV 之间的同步。参与同步的核与信号配对方式由sync_mode决定。

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

IR 层实现会对pipe做具体流水校验(_validate_concrete_pipe),并按event_id是否为运行时Expr分发到静态或动态(_dyn)指令(参见 system_ops.py)。

6.2 参数说明

pipe(发送侧 set_cross_core):表示发送信号所在的硬件流水,该流水前序指令完成后 SET 才生效。INTER_BLOCK/INTER_SUBBLOCK/INTRA_BLOCK时取 M、V、MTE1、MTE2、MTE3、FIX(不支持 S 和 ALL);UNICAST_BLOCK时还可取 S,但仍不支持 ALL。可与配对wait_cross_core的 pipe 不同。

pipe(等待侧 wait_cross_core):表示等待期间被阻塞的流水。只阻塞该流水中尚未下发的后续指令,已下发的指令仍可继续执行。四种模式均支持 M、V、MTE1、MTE2、MTE3、FIX 和 S,不支持 ALL;无需与配对的 set 侧 pipe 相同。

event_id:核间同步事件 ID,支持 Python 整型常量或运行时整数 Scalar 表达式:

  • Python 整型常量当前只能取0~15;动态表达式须由调用方保证运行时取值合法:INTER_BLOCK/INTER_SUBBLOCK/INTRA_BLOCK取 0~15;UNICAST_BLOCK在 AIV 侧取 0~15、在 AIC 侧取 0~31。
  • UNICAST_BLOCK配对规则:AIV0 发送的 0~15 与 AIC 等待的 0~15 配对,AIV1 发送的 0~15 与 AIC 等待的 16~31 配对;AIC 发送的 0~15 与 AIV0 等待的 0~15 配对,AIC 发送的 16~31 与 AIV1 等待的 0~15 配对。
  • 每个事件 ID 的计数器取值范围为0~15;同一事件的信号未被 WAIT 消费时,连续发送超过 15 次 SET 会触发异常并中断执行
  • 复用事件 ID 或将其用于不同同步模式前,必须完成前一同步过程中的所有 SET 和 WAIT;与sync_all同时使用时须避开 HARD 模式占用的事件 ID;使用自动流水编排时还应避免与其分配的事件 ID 冲突。
  • 同一核连续发送多个 SET 时,不保证不同事件 ID 之间的生效顺序;存在先后依赖时应先完成前一组 SET/WAIT。

sync_mode:核间同步模式,须与配对的 wait/set 使用相同模式,取值见 CrossCoreSyncMode.md:

模式说明
INTER_BLOCK0多个 AI Core 之间的同类核全核同步。AIC 场景同步本次 Kernel 启动的所有 AIC,AIV 场景同步本次 Kernel 启动的所有 AIV;AIC 与 AIV 不会在该模式下互相同步
INTER_SUBBLOCK1同一 AI Core 内的 AIV0 与 AIV1 同步,不同 AI Core 之间互不影响
INTRA_BLOCK2同一 AI Core 内的 AIC 与全部 AIV 同步。AIV→AIC 方向须由 AIV0 和 AIV1 分别发送信号、AIC 等待两路信号;AIC→AIV 方向由 AIC 发送信号、AIV0 和 AIV1 分别等待。默认同步模式
UNICAST_BLOCK3同一 AI Core 内的 AIC 与单个 AIV 同步。AIC 侧事件 ID 0~15 对应 AIV0,16~31 对应 AIV1;AIV 侧事件 ID 取 0~15

6.3 约束要点

  • 必须存在与当前调用匹配的配对接口,且所有参与同步的核均能到达同步点,否则可能死锁。
  • 使用INTER_BLOCK时,若多流/多算子并发执行且并发算子申请的核数总和超过物理核数,当至少两个并发算子使用核间同步时,部分核可能因未被调度而无法到达同步点造成死锁;须保证每个同步算子所需的核能够同时执行。

6.4 调用示例

各模式的典型调用片段(详见 set_cross_core.md):

with pl.section_vector(): # INTER_BLOCK:本AIV上的前置操作完成后,全核同类AIV同步 pl.system.set_cross_core( pipe=pl.PipeType.MTE3, event_id=0, sync_mode=pl.CrossCoreSyncMode.INTER_BLOCK, ) with pl.section_vector(): # INTER_SUBBLOCK:AIV0和AIV1各自完成前置操作后互相同步 pl.system.set_cross_core( pipe=pl.PipeType.V, event_id=1, sync_mode=pl.CrossCoreSyncMode.INTER_SUBBLOCK, ) with pl.section_cube(): # INTRA_BLOCK:AIC等待AIV0和AIV1的两路信号 pl.system.wait_cross_core( pipe=pl.PipeType.MTE1, event_id=2, sync_mode=pl.CrossCoreSyncMode.INTRA_BLOCK, ) with pl.section_cube(): # UNICAST_BLOCK:仅等待AIV0的信号 pl.system.wait_cross_core( pipe=pl.PipeType.S, event_id=15, sync_mode=pl.CrossCoreSyncMode.UNICAST_BLOCK, )

完整的跨核 Kernel 示例实现"Vector 侧(AIV0/AIV1 各算一半)把x+y结果以 NZ 布局写入v1_mat,经INTRA_BLOCK通知 Cube 侧 AIC,Cube 侧等待信号后与rhs做 matmul 并写回 out"的流水线协作,其完整代码见 set_cross_core.md 与 wait_cross_core.md。核心结构为:Vector 段内先做MTE2→VV→MTE3核内同步,搬出完成后set_cross_core(INTRA_BLOCK);Cube 段内wait_cross_core阻塞 MTE1,待 AIV 信号到达后再执行move(v1_left, v1_mat)matmul

七、全局核屏障:sync_all

7.1 功能与原理

sync_all在多个 AIV 核、多个 AIC 核,或 AIV 与 AIC 核之间建立核间屏障:参与同步的核到达sync_all后等待,直到本轮所有参与核均到达,再继续执行。当前仅支持HARD 模式(使用 FFTS 硬件同步,不需要 workspace),源码中sync_all在 system_ops.py 实现,HARD 模式传入空 workspace 元组,生成system.sync_all指令;SOFT 模式(GM 共享状态同步)已在枚举中预留但当前暂不支持

pypto_pro.language.system.sync_all( workspaces: Optional[List] = None, *, core_type: SyncCoreType = pypto_pro.language.SyncCoreType.MIX, mode: SyncAllMode = pypto_pro.language.SyncAllMode.HARD, ) -> None

7.2 参数说明

参数输入/输出说明
workspaces输入可选。当前仅支持 HARD 模式,不使用 workspace,保持默认None或传空列表
core_type输入SyncCoreType 枚举值,指定参与屏障的核类型,默认MIX。该参数不指定参与核数量
mode输入SyncAllMode 枚举值,指定同步实现模式,默认HARD,当前仅支持HARD

SyncCoreType取值:AIV_ONLY(仅同步参与执行的 AIV 核)、AIC_ONLY(仅同步参与执行的 AIC 核)、MIX(同步参与执行的 AIC 核和 AIV 核)。

7.3 约束要点

  • 所有参与同步的核必须以相同顺序执行相同次数的sync_all。循环次数或分支条件不一致导致部分核少执行或多执行 sync_all 时,可能死锁。MIX 模式下 AIC 侧与 AIV 侧的调用必须一一对应。
  • 纯 Vector Kernel 使用AIV_ONLY,纯 Cube Kernel 使用AIC_ONLY。MIX 模式要求 Cube 侧 AIC 与 Vector 侧 AIV 都执行对应的 sync_all;只在一侧调用会使另一侧无法到达屏障,导致Kernel 超时
  • 多流/多算子并发执行且核数总和超过物理核数时,若至少两个并发算子使用 sync_all,部分核可能因未被调度而无法到达屏障造成死锁;须保证每个同步算子所需的核能够同时执行。
  • sync_all 只建立屏障;屏障前后需要跨核读写 GM 数据时,还需满足相应的数据可见性要求。
  • set_cross_core/wait_cross_core并用时,两侧的 MIX 屏障必须位于该 SET/WAIT 对的同一侧;禁止 Cube 侧先执行 sync_all 再 SET、Vector 侧先 WAIT 再执行 sync_all,否则会形成环形等待
  • HARD 模式会占用核间同步事件 IDAIV_ONLY在 AIV 侧占用 14;AIC_ONLY在 AIC 侧占用 11;MIX在 AIC 侧占用 11~13、在 AIV 侧占用 12~13,MIX 1:2 场景的 AIC 还会占用 28 和 29。与set_cross_core/wait_cross_core同时使用时,不得将这些事件 ID 用于尚未完成的手工核间同步。

7.4 调用示例

纯 Vector Kernel 使用 HARDAIV_ONLY屏障的完整示例(各 AIV 按核号处理互不重叠的行,TileGroupauto_mutex负责核内 pipe 依赖,sync_all 位于循环外使所有参与 AIV 在本阶段结束后再越过屏障):

import pypto_pro.language as pl @pl.jit(auto_mutex=True) def sync_all_kernel( x: pl.Tensor[[2048, 64], pl.DT_FP32], out: pl.Tensor[[2048, 64], pl.DT_FP32], ): tt = pl.TileType(shape=[1, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec) input_tiles = pl.make_tile_group( type=tt, addrs=[0x0000, 0x0100], mutex_ids=[0, 1]) output_tiles = pl.make_tile_group( type=tt, addrs=[0x0200, 0x0300], mutex_ids=[2, 3]) with pl.section_vector(): for row in pl.range(pl.get_block_idx(), x.shape[0], pl.get_block_num()): tile_x = input_tiles.next() tile_out = output_tiles.next() pl.load(tile_x, x, [row, 0]) pl.add(tile_out, tile_x, tile_x) pl.store(out, tile_out, [row, 0]) pl.system.sync_all( mode=pl.SyncAllMode.HARD, core_type=pl.SyncCoreType.AIV_ONLY, )

HARD 模式下三种core_type的放置方式(分别对应纯 Vector、纯 Cube、Cube+Vector 共存 Kernel,详见 sync_all.md):

# 纯Vector Kernel:所有参与AIV都执行 with pl.section_vector(): # ... Vector阶段计算 pl.system.sync_all(mode=pl.SyncAllMode.HARD, core_type=pl.SyncCoreType.AIV_ONLY) # 纯Cube Kernel:所有参与AIC都执行 with pl.section_cube(): # ... Cube阶段计算 pl.system.sync_all(mode=pl.SyncAllMode.HARD, core_type=pl.SyncCoreType.AIC_ONLY) # Cube、Vector共存的Kernel:AIC和AIV必须到达同一个MIX屏障 with pl.section_cube(): # ... Cube阶段计算 pl.system.sync_all(mode=pl.SyncAllMode.HARD, core_type=pl.SyncCoreType.MIX) with pl.section_vector(): # ... Vector阶段计算 pl.system.sync_all(mode=pl.SyncAllMode.HARD, core_type=pl.SyncCoreType.MIX)

八、同步原语选型:从核内到跨核

综合全部分组,可按下述原则快速选型:

  1. 同一流水的串行化:连续 MTE2/MTE3 操作访问重叠地址,或需要同流水前序操作完成后才执行后序操作时,优先使用对应单流水屏障bar_mte2/bar_mte3(流水无对应屏障时用bar_all)。
  2. 同一 AI Core 内不同流水之间的点对点依赖(如 load→计算→store):使用sync_src/sync_dst配对,按 PipeType.md 的典型模式选择 set_pipe/wait_pipe,event_id取 [0, 7]。
  3. 同一片上缓冲区的跨 pipe 互斥访问:常规多缓冲场景用make_tile_group+auto_mutex=True;需要精确控制加锁 pipe 与插入位置时手动mutex_lock/mutex_unlockmutex_id取 [0, 31]。
  4. 核内全部流水统一停靠:需要等待当前 AI Core 全部流水完成时使用bar_all(注意其性能开销且非跨核屏障)。
  5. 跨核协同:需要同类核全核同步、同核 AIV 间同步或 AIC/AIV 间同步时,使用set_cross_core/wait_cross_core并按CrossCoreSyncMode选择配对模式;需要全局多核屏障时使用sync_all(HARD 模式),并严格避开其占用的事件 ID(AIV 14、AIC 11、MIX 11~13/12~13 等)。

所有接口的完整函数原型、参数表与可运行示例均可在 synchronization 目录 下按名查阅,IR 层实现集中在 system_ops.py,ST 验证用例见 test_pipe_barriers.py 与 test_manual_mutex.py。在 Ascend 950PR/Ascend 950DT 上编写 Kernel 时,遵循"配对成双、路径对称、ID 不冲突、核全到达"四条铁律,即可写出无数据竞争、无死锁的高效同步代码。

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

【免费下载链接】pypto

PyPTO(发音: pai p-t-o):Parallel Tensor/Tile Operation编程范式。

项目地址:https://gitcode.com/cann/pypto
点击查看免费下载

相关推荐

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询