企业官网建设流程全解析

PyPTO 系统访问变量实战指南：get_block_idx / get_block_num / get_subblock_idx / get_subblock_num 多核编程完全解析

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

导读

PyPTO（Parallel Tensor/Tile Operation）的 SIMD（Vector/Tile）编程模型中，多核数据切分、跨核同步与 Cube/Vector 混合调度都依赖一组"系统访问变量"（System Variables）——get_block_idx、get_block_num、get_subblock_idx、get_subblock_num。本指南以 system_variables/index.md 文档为骨架，逐一拆解这 4 个接口的功能、返回语义、产品支持情况与典型使用场景，并结合仓库中 Python API 声明 与 CCE 后端代码生成实现 说明其底层原理。读完本文，你将掌握在纯 Vector Kernel 与混合 Kernel（AIC:AIV=1:2）中正确完成多核数据分片、条件执行与跨域协同的核心方法。

一、系统访问变量总览：四个接口的分工

在 PyPTO 中，Kernel 以"逻辑 Block"为执行单元在多个 AI Core（逻辑 AI Core）上并行运行。为了在 Kernel 内感知"我在哪、一共多少人、我这一份怎么分"，PyPTO 提供了 4 个零参数的系统级接口：

从源码看，这 4 个接口在 python/pypto_pro/language/_api.py 中以@_api_decl声明，对应 IR 层注册于 python/pypto_pro/ir/op/system_ops.py；真正落地到设备侧 CCE 代码的是 backend_cce_ops.cpp 中REGISTER_BACKEND_OP注册的一系列后端算子。其中get_subblock_num在 Python 侧注释中明确标注"Matches AscendC GetTaskRation()"，即与 AscendC 的GetTaskRation()语义对齐。

二、产品支持情况：仅限 Ascend 950 系列

上述 4 个接口的支持范围在各自文档中完全一致：

• Ascend 950PR / Ascend 950DT：支持
• Atlas A3 训练系列产品 / Atlas A3 推理系列产品：不支持
• Atlas A2 训练系列产品 / Atlas A2 推理系列产品：不支持

也就是说，系统访问变量是 950 系列设备上多核 Vector/Tile 编程的基础能力；在使用 A2/A3 平台时，多核切分需要采用其他等价手段（如 Host 侧分核传入），不能直接依赖这组接口。

三、get_block_idx()：获取当前逻辑 AI Core 的全局索引

3.1 功能与返回值

get_block_idx()获取当前执行域中逻辑 AI Core 的全局索引，用于多核控制和数据偏移计算。无参数、无约束，返回设备运行时产生的整型标量值，可用于 Kernel 内整数运算和索引。

返回值取值范围与 Kernel 的执行域（Section）密切相关：

• 仅启动 Cube（AIC）或仅启动 Vector（AIV）时：范围为[0, get_block_num())；
• 同时启动 AIC 与 AIV 时：
  • AIC 侧范围为[0, get_block_num())；
  • AIV 侧范围为[0, get_subblock_num() × get_block_num())。当前 AIC:AIV=1:2 配置下，即[0, 2 × get_block_num())。

在 Vector 段中，该接口返回的是全局 AIV 逻辑索引，编号方式为"物理核号 × get_subblock_num() + 子核号"，可直接用于数据分片和偏移计算。混合 Kernel 在 Vector 段做跨步（stride）切分时，工作单元总数为get_block_num() × get_subblock_num()。

3.2 底层实现：Vector 段如何映射为全局 AIV 索引

CCE 后端的代码生成逻辑印证了上述文档语义，见 backend_cce_ops.cpp：

static std::string MakeBlockGetBlockIdxCodegenCCE(const ir::CallPtr& op, codegen::CodegenBase& codegen_base) { CHECK(op->args_.size() == 0) << "get_block_idx requires no arguments"; auto& cg = dynamic_cast<codegen::CCECodegen&>(codegen_base); const auto target = cg.GetTarget(); if (target == ir::SectionKind::Vector) { return "(int32_t)(get_block_idx() * get_subblockdim() + get_subblockid())"; } return "(int32_t)(get_block_idx())"; }

可以看到：当编译目标为 Vector 段时，生成的设备代码是get_block_idx() * get_subblockdim() + get_subblockid()——即"物理核号 × 子核数 + 子核号"；而 Cube 段则直接返回get_block_idx()。这从实现层面保证了文档所述的两种取值范围。

3.3 调用示例一：纯 Vector Kernel 多核数据分片

以Kernel[None, NUM_CORES]形式启动 2 个逻辑 Block，每个 AIV 用get_block_idx()获取全局逻辑索引并处理 64 行逐元素加法，用 printf 打印索引值：

import os import pypto_pro.language as pl import torch NUM_CORES = 2 @pl.jit(auto_mutex=True) def multicore_add_kernel( x: pl.Tensor[[128, 128], pl.DT_FP16], y: pl.Tensor[[128, 128], pl.DT_FP16], z: pl.Tensor[[128, 128], pl.DT_FP16], ): tt = pl.TileType(shape=[64, 128], dtype=pl.DT_FP16, target_memory=pl.MemorySpace.Vec) tile_a = pl.make_tile_group(type=tt, addrs=0x0000, mutex_ids=[0]) tile_b = pl.make_tile_group(type=tt, addrs=0x4000, mutex_ids=[1]) tile_c = pl.make_tile_group(type=tt, addrs=0x8000, mutex_ids=[2]) with pl.section_vector(): vidx = pl.get_block_idx() # 当前AIV的全局逻辑索引 num_blocks = pl.get_block_num() # 实际启动的Block数 pl.printf("block_idx = %d, block_num = %d\n", vidx, num_blocks) for tile_idx in pl.range(vidx, 2, num_blocks): offset = tile_idx * 64 cur_a = tile_a.current() cur_b = tile_b.current() cur_c = tile_c.current() pl.load(cur_a, x, [offset, 0]) pl.load(cur_b, y, [offset, 0]) pl.add(cur_c, cur_a, cur_b) pl.store(z, cur_c, [offset, 0]) if __name__ == "__main__": device = f"npu:{int(os.environ.get('TILE_FWK_DEVICE_ID', 0))}" torch.npu.set_device(device) torch.manual_seed(42) x = torch.rand([128, 128], device=device, dtype=torch.float16) y = torch.rand([128, 128], device=device, dtype=torch.float16) z = torch.zeros([128, 128], device=device, dtype=torch.float16) multicore_add_kernelNone, NUM_CORES torch.npu.synchronize() torch.testing.assert_close(z, x + y, rtol=1e-2, atol=1e-2) print(f"max diff = {(z - (x + y)).abs().max().item()}")

回显（=> Vec后为核号，多核间输出顺序不固定）：

=> Vec 0 block_idx = 0, block_num = 2 => Vec 1 block_idx = 1, block_num = 2

3.4 调用示例二：混合 Kernel 中 AIC 与 AIV 的返回值差异

混合 Kernel（AIC:AIV=1:2）中，Cube 段返回物理核号[0, get_block_num())，Vector 段返回全局 AIV 逻辑编号[0, get_block_num() × get_subblock_num())：

import os import pypto_pro.language as pl import torch @pl.jit() def block_idx_mix_kernel(out: pl.Tensor[[1], pl.DT_INT32]): with pl.section_cube(): aic_idx = pl.get_block_idx() pl.printf("[cube] block_idx = %d\n", aic_idx) pl.setval(out, 0, 1) with pl.section_vector(): aiv_idx = pl.get_block_idx() pl.printf("[vector] block_idx = %d\n", aiv_idx) if __name__ == "__main__": device = f"npu:{int(os.environ.get('TILE_FWK_DEVICE_ID', 0))}" torch.npu.set_device(device) out = torch.zeros(1, device=device, dtype=torch.int32) block_idx_mix_kernelNone, 2 torch.npu.synchronize()

回显（启动 2 个逻辑 Block：AIC 侧返回 0/1，AIV 侧返回 0~3）：

=> Cube 0 [cube] block_idx = 0 => Vec 0 [vector] block_idx = 0 => Vec 1 [vector] block_idx = 1 => Cube 1 [cube] block_idx = 1 => Vec 2 [vector] block_idx = 2 => Vec 3 [vector] block_idx = 3

四、get_block_num()：读取实际启动的 Block 数并作为循环步长

4.1 功能与返回值

get_block_num()获取本次实际启动的逻辑 Block 数量，用于多核控制和数据偏移计算。返回值是设备运行时产生的整型标量，等于限核后实际启动的逻辑 Block 数。

关键语义：JIT 每次启动会通过 C++ 启动器查询实际 Stream 的有效资源限制，按 Kernel 执行域和配对比例限制 Host 请求的block_dim。因此返回值可能小于 Host 请求值——数据切分应使用本接口返回值作为循环步长，避免遗漏任务。这一点在 Python 侧 API 注释中同样强调："This can be smaller than the host's requestedkernel[stream, block_dim]. Use this runtime count as the work-distribution stride so limiting cores does not leave tiles unprocessed."（见 language/_api.py）。

分执行域的取值规则：

• 仅启动 Cube（AIC）或仅启动 Vector（AIV）时，该值等于执行域逻辑核数；
• 在 AIC:AIV 为 1:2 的混合 Kernel 中，该值表示逻辑 Block 数；AIC 逻辑核数为get_block_num()，AIV 逻辑核数为get_block_num() * get_subblock_num()。

4.2 调用示例一：按实际 Block 数循环切分

用Kernel[None, NUM_CORES]请求最多 2 个逻辑 Block，每个 AIV 以get_block_num()返回值为步长跨步处理 64 行 Tile（完整 Kernel 定义与 3.3 节相同，此处省略重复部分）：

with pl.section_vector(): vidx = pl.get_block_idx() # 当前AIV的全局逻辑索引 num_blocks = pl.get_block_num() # 实际启动的Block数 pl.printf("block_idx = %d, block_num = %d\n", vidx, num_blocks) for tile_idx in pl.range(vidx, 2, num_blocks): offset = tile_idx * 64 ...

回显：

=> Vec 0 block_idx = 0, block_num = 2 => Vec 1 block_idx = 1, block_num = 2

4.3 调用示例二：限核启动下的返回值

启动 Block 数是请求的上界，实际启动数可能更小。复用 3.3 节的 Kernel 与__main__中的 x/y，仅把启动 Block 数改为 1：

z = torch.zeros([128, 128], device=device, dtype=torch.float16) multicore_add_kernelNone, 1 torch.npu.synchronize() torch.testing.assert_close(z, x + y, rtol=1e-2, atol=1e-2) # 仍覆盖全部128行

回显：

=> Vec 0 block_idx = 0, block_num = 1

该示例同时验证了 4.1 节的核心结论：即使只启动 1 个 Block，由于循环步长来自get_block_num()（此处为 1），单核也会完整遍历所有 Tile，128 行全部被覆盖，assert_close依然通过。

五、get_subblock_idx()：区分同一 Block 内的不同 AIV

5.1 功能与返回值

get_subblock_idx()获取当前逻辑 AI Core 内 AIC 或 AIV 的 subblock 索引。返回整型标量，取值范围为[0, get_subblock_num())；在 AIC 与 AIV 比例为 1:2 的混合 Kernel 中，同一逻辑 Block 对应的两个 AIV 分别返回 0 和 1。

5.2 典型使用场景

文档给出两种典型模式：

1. insert + Cube 模式：每个子核计算部分结果，用 insert 拼入 L1 Buffer 中的 NZTile，Cube 侧读取合并后的完整数据；
2. 条件执行：根据子核号决定是否执行某段代码，例如只让 sub-core 0 执行指定操作。

[!CAUTION] 注意 纯 Vector Kernel 中的两个子核共享 MTE 搬运管道，不能由每个子核分别使用pypto_pro.language.store向 GM 的不同区域写入数据。按子核切分数据搬运时，应使用 insert + Cube 模式。

一句话总结两个索引接口的分工：get_block_idx()用于 Vector 段的全局 AIV 数据分片，get_subblock_idx()用于区分同一逻辑 Block 内的不同 AIV。

5.3 调用示例一：纯 Vector Kernel 中读取子核号

纯 Vector Kernel 中每个 Block 即一个 AIV（get_subblock_num()返回 1），get_subblock_idx()恒为 0：

import os import pypto_pro.language as pl import torch @pl.jit(auto_mutex=True) def subblock_add_kernel( x: pl.Tensor[[64, 64], pl.DT_FP32], y: 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_group(type=tt, addrs=0x0000, mutex_ids=[0]) tile_y = pl.make_tile_group(type=tt, addrs=0x4000, mutex_ids=[1]) tile_sum = pl.make_tile_group(type=tt, addrs=0x8000, mutex_ids=[2]) with pl.section_vector(): sub_idx = pl.get_subblock_idx() pl.printf("subblock_idx = %d, subblock_num = %d\n", sub_idx, pl.get_subblock_num()) cur_x = tile_x.current() cur_y = tile_y.current() cur_sum = tile_sum.current() pl.load(cur_x, x, [0, 0]) pl.load(cur_y, y, [0, 0]) pl.add(cur_sum, cur_x, cur_y) pl.store(out, cur_sum, [0, 0]) if __name__ == "__main__": device = f"npu:{int(os.environ.get('TILE_FWK_DEVICE_ID', 0))}" torch.npu.set_device(device) torch.manual_seed(42) x = torch.randn([64, 64], device=device, dtype=torch.float32) y = torch.randn([64, 64], device=device, dtype=torch.float32) out = torch.zeros([64, 64], device=device, dtype=torch.float32) subblock_add_kernel(x, y, out) torch.npu.synchronize() torch.testing.assert_close(out, x + y, rtol=1e-5, atol=1e-5) print(f"max diff = {(out - (x + y)).abs().max().item()}")

回显：

=> Vec 0 subblock_idx = 0, subblock_num = 1

5.4 调用示例二：混合 Kernel 中的子核号与条件执行

混合 Kernel 的 Vector 段中，get_block_idx()返回全局 AIV 逻辑索引用于数据分片，get_subblock_idx()用于区分同一逻辑 Block 内的两个 AIV。根据子核号可做条件执行，例如只让 sub-core 0 写结果：

import os import pypto_pro.language as pl import torch @pl.jit() def subblock_cond_kernel(out: pl.Tensor[[1], pl.DT_INT32]): with pl.section_cube(): pass with pl.section_vector(): block_idx = pl.get_block_idx() # 全局AIV逻辑索引：0~3 sub_idx = pl.get_subblock_idx() # 块内AIV编号：0或1 pl.printf("block_idx = %d, subblock_idx = %d\n", block_idx, sub_idx) # 条件执行：只让每个逻辑Block的sub-core 0写结果 if sub_idx == 0: pl.printf("sub-core 0 of core %d writes output\n", block_idx // pl.get_subblock_num()) pl.setval(out, 0, 1) if __name__ == "__main__": device = f"npu:{int(os.environ.get('TILE_FWK_DEVICE_ID', 0))}" torch.npu.set_device(device) out = torch.zeros(1, device=device, dtype=torch.int32) subblock_cond_kernelNone, 2 torch.npu.synchronize()

回显（启动 2 个逻辑 Block；=> Cube条目为 Cube 段，本例无打印）：

=> Cube 0 => Vec 0 block_idx = 0, subblock_idx = 0 sub-core 0 of core 0 writes output => Vec 1 block_idx = 1, subblock_idx = 1 => Cube 1 => Vec 2 block_idx = 2, subblock_idx = 0 sub-core 0 of core 1 writes output => Vec 3 block_idx = 3, subblock_idx = 1

注意此处block_idx // pl.get_subblock_num()正是 6.2 节所述"还原物理核号"的用法。

六、get_subblock_num()：获取 task ration 并统一 Cube/Vector 的 core_id

6.1 功能与返回值

get_subblock_num()获取当前 block 的 subblock 总数（即一个 block 关联的从核数量，又称 task ration），返回整型Expr。返回值与核类型及编译模式有关：

• AIC 核：始终返回 1（AIC 为 block，无 AIC 从核）；
• AIV 核：
  • 融合算子（mix，AIC:AIV = 1:2）：返回 2（每个 AI Core 含 2 个 AIV 从核）；
  • 纯 Vector 算子（aiv-only）：返回 1（AIV 为 block，无 subblock 划分）。

Python 侧 API 注释与文档一致："Returns 1 on AIC binaries, get_subblockdim() on AIV binaries. Matches AscendC GetTaskRation()."（见 language/_api.py）。

6.2 调用示例：用除法还原物理核号，统一两侧数据切分

在融合算子中，get_block_idx()在 AIV 核上返回的是逻辑编号（block_idx * subblock_num + subblock_idx），通过除以get_subblock_num()可还原物理 AI Core 编号，使 Cube 与 Vector 两侧用统一的core_id切分数据：

import pypto_pro.language as pl NUM_CORES = 2 @pl.jit(auto_mutex=True) def matmul_example( a: pl.Tensor[[pl.DYNAMIC, pl.DYNAMIC], pl.DT_FP16], b: pl.Tensor[[pl.DYNAMIC, pl.DYNAMIC], pl.DT_FP16], out: pl.Tensor[[pl.DYNAMIC, pl.DYNAMIC], pl.DT_FP16], ): num_cores = pl.get_block_num() # AIC/AIV两侧得到相同的物理核号，详见下方NOTE core_id = pl.get_block_idx() // pl.get_subblock_num() with pl.section_cube(): for i in pl.range(core_id, a.shape[0] // 128, num_cores): ... # Cube侧按行块i执行load/matmul/store with pl.section_vector(): for i in pl.range(core_id, a.shape[0] // 128, num_cores): ... # Vector侧用同一core_id切分，与Cube侧对齐 matmul_exampleNone, NUM_CORES

[!NOTE] 说明 该除法在 AIC 核上为block_idx // 1，在 AIV 核上为(block_idx * 2 + subblock_idx) // 2，两者均得到相同的 AI Core 编号，因此 Cube 与 Vector 可共享同一core_id做数据切分。

七、综合使用建议与注意事项

7.1 四接口组合速查

7.2 关键注意事项

1. 步长必须使用get_block_num()：实际启动数可能小于 Host 请求的block_dim，若以请求值硬编码为步长，限核时会产生数据空洞（见 4.3 节）；
2. Vector 段的索引语义与 Cube 段不同：Vector 段返回的是全局 AIV 逻辑索引（含子核维度），CCE 后端通过get_block_idx() * get_subblockdim() + get_subblockid()实现（见 backend_cce_ops.cpp）；
3. 纯 Vector Kernel 中禁止子核分别 store 到 GM：两个子核共享 MTE 搬运管道，按子核切分数据搬运时应走 insert + Cube 模式；
4. 平台限制：本组接口仅受 Ascend 950PR/950DT 支持，A2/A3 系列不支持，移植时需做平台判断；
5. 多核输出顺序不固定：使用 printf 调试时，各核打印顺序无法保证，应以内容而非顺序判断结果。

7.3 进一步阅读

• 系统访问变量索引文档
• insert（子核结果拼入 NZTile）
• printf（Kernel 内调试打印）
• Python 侧 API 声明：python/pypto_pro/language/_api.py
• IR 算子注册：python/pypto_pro/ir/op/system_ops.py
• CCE 代码生成实现：framework/src/interface/pypto_pro/backend/backend_cce_ops.cpp

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