企业官网建设流程全解析

PyPTO vf.neg 向量取反算子深度解析：基于 mask 的寄存器级取反运算实战

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

vf.neg是 PyPTO 向量函数（Vector Function，VF）体系中基础算术（basic_arithmetic）模块提供的逐元素取反接口，用于在向量寄存器层面根据掩码对源数据执行dst_i = -src_i运算。本文围绕该算子的产品支持范围、函数原型、mask 驱动语义、MergeMode 行为以及 FP32/INT64 两种完整可运行示例展开讲解，并辅以源码与测试佐证，帮助读者在 Ascend 950 系列平台上快速编写、验证基于vf.neg的向量算子。

产品支持情况

vf.neg属于 SIMD-API 中寄存器计算（reg_computation）类接口，其支持范围与所在的 PyPTO 向量函数体系保持一致，具体如下：

这意味着当前版本的vf.neg仅面向 Ascend 950 系列硬件提供向量寄存器取反能力，在使用前应确认目标设备型号。

功能说明：mask 驱动的逐元素取反

vf.neg对输入寄存器src中的每个元素执行算术取反，并将结果写入目的寄存器dst，计算公式为：

$$dst_i = -src_i$$

运算的逐元素有效性由掩码寄存器preg控制：

• mask 位为 1（有效）：对应元素参与取反运算，结果写入dst对应位置；
• mask 位为 0（无效）：对应元素不参与运算，dst对应位置根据mode参数决定处理方式（默认置 0）。

这一语义与 PyPTO 向量函数的通用 mask 机制一致。mask 是 VF 计算中元素级有效性的核心控制容器，详见 vf.mask_reg：mask 寄存器总位宽固定为 256 bit，其粒度由 dtype 决定，例如 FP32 场景下每 4 bit 对应一个元素，即一个 mask 可覆盖 64 个 FP32 元素；INT64 场景下每 8 bit 对应一个元素，覆盖 32 个 INT64 元素。实际使用时，mask 的 dtype 一般与配套的数据寄存器一致，典型创建方式为vf.create_mask(pattern=pl.MaskPattern.ALL, dtype=...)，表示所有元素均参与运算。

函数原型

neg(src, preg, mode: Optional[MergeMode] = None) -> dst

该接口在 PyPTO Python 语言层的声明位于 python/pypto_pro/language/_vf_api.py，其文档字符串明确描述：对于每个 mask 位激活的通道（lane）i，计算src[i]的算术取反并写入dst[i]。它是pypto_pro.language中vf命名空间下注册的向量函数指令之一，只能在@pl.vector_function修饰的向量函数内调用。

参数说明

关于数据类型，可以从 vf.reg_tensor 的寄存器模型进一步理解：寄存器总大小固定为 256 字节，不同 dtype 决定单个寄存器的元素个数，例如 DT_FP32 为 64 个元素、DT_INT64 为 32 个元素。因此vf.neg在一次调用中处理的元素数量与 dtype 强相关，编写算子时 Tile 的 shape 应与寄存器容量匹配（如 FP32 取[1, 64]，INT64 取[1, 32]）。

MergeMode 语义详解

mode参数的类型 MergeMode 在源码中定义为一个枚举类：

class MergeMode(enum.Enum): ZEROING = ... # mask未选中位置置零（默认） MERGING = ... # mask未选中位置保留目标寄存器原值

• ZEROING（默认）：mask 未选中（位为 0）的元素，其目的寄存器对应位置被置 0。这是vf.neg的默认行为，也是当前唯一支持的模式；
• MERGING：mask 未选中位置保留目标寄存器原值。vf.neg当前不支持该模式。

在vf.neg中mode参数可省略（传None即采用默认 ZEROING 行为）。由于当前仅支持 ZEROING，实际编写代码时可省略该参数，语义上等价于"全量元素参与运算，mask 未覆盖位置输出 0"。

约束说明

vf.neg本身无额外约束。需要留意的是其依赖的数据容器约束，这些约束在使用时会一并生效：

• 寄存器在@pypto_pro.language.vector_function函数内创建和使用，函数结束后自动释放；
• 创建寄存器后必须通过vf.load_align或vf.full初始化数据，否则内容未定义；
• 数据寄存器（RegTensor）数量上限为 32，mask 寄存器上限为 16，编译器会自动复用生命周期结束的寄存器；
• FP8/FP4 类型为存储类型，不支持直接参与算术运算，因此不在vf.neg的支持类型列表中，需先通过vf.astype转换为 FP32/BF16/FP16 等计算类型。

返回值说明

返回 dst 目的操作数，类型为 reg_tensor，支持的数据类型与 src 中的说明一致（DT_INT8、DT_INT16、DT_FP16、DT_INT32、DT_FP32、DT_INT64）。该寄存器由编译器在赋值形式中自动声明，例如示例中的reg_out = vf.neg(reg_a, preg)。

调用示例

基本调用示例（DT_FP32）

以下示例展示了一个完整的取反算子：在向量函数中创建全量 mask、从 UB Tile 对齐加载数据到寄存器、执行vf.neg取反，再将对齐存储回目的 Tile，最后在 kernel 中编排 Tile 的搬运与计算，并用torch.testing.assert_close与-a的结果进行对比验证。

import os import pypto_pro.language as pl import torch import torch_npu @pl.vector_function def example_vf(src_tile, dst_tile): preg = vf.create_mask(pattern=pl.MaskPattern.ALL, dtype=pl.DT_FP32) reg_a = vf.load_align(src_tile, 0) reg_out = vf.neg(reg_a, preg) vf.store_align(dst_tile, reg_out, preg) @pl.jit() def example_kernel( a: pl.Tensor[[pl.DYNAMIC, pl.DYNAMIC], pl.DT_FP32], out: pl.Tensor[[pl.DYNAMIC, pl.DYNAMIC], pl.DT_FP32], ): tf = pl.TileType(shape=[1, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec) in_a_grp = pl.make_tile_group(type=tf, addrs=0x0, mutex_ids=[0]) in_a = in_a_grp.current() t_out_grp = pl.make_tile_group(type=tf, addrs=0x100, mutex_ids=[1]) t_out = t_out_grp.current() with pl.section_vector(): pl.load(in_a, a, [0, 0]) example_vf(in_a, t_out) pl.store(out, t_out, [0, 0]) def test_example(): device_id = int(os.environ.get("TILE_FWK_DEVICE_ID", 0)) device = f"npu:{device_id}" core_nums = 1 torch.npu.set_device(device) a = torch.randn([1, 64], device=device, dtype=torch.float32) out = torch.empty([1, 64], device=device, dtype=torch.float32) example_kernelNone, core_nums torch.npu.synchronize() torch.testing.assert_close(out, -a, rtol=1e-5, atol=1e-5) if __name__ == "__main__": test_example() print("PASSED")

示例要点说明：

• tf = pl.TileType(shape=[1, 64], dtype=pl.DT_FP32, target_memory=pl.MemorySpace.Vec)：声明 Vector 内存空间的 64 元素 FP32 Tile，与单个 FP32 寄存器容量（64 元素）对齐；
• pl.make_tile_group(type=tf, addrs=0x0, mutex_ids=[0])：以固定地址创建 Tile 组，mutex_ids用于约束共享内存互斥；
• pl.section_vector()：将向量搬移与 VF 计算放入向量指令段执行；
• 输入数据采用torch.randn，验证逻辑为out == -a（FP32 场景使用 rtol/atol 容差比较）。

INT64 数据类型示例

vf.neg同样支持 DT_INT64 数据，示例中寄存器数量随 dtype 变化：INT64 元素宽度 64 bit，单个寄存器仅容纳 32 个元素，因此 Tile shape 调整为[1, 32]，验证时使用整数精确相等比较（rtol=0, atol=0）。

import os import pypto_pro.language as pl import torch import torch_npu @pl.vector_function def example_vf_int64(src_tile, dst_tile): preg = vf.create_mask(pattern=pl.MaskPattern.ALL, dtype=pl.DT_INT64) reg_a = vf.load_align(src_tile, 0) reg_out = vf.neg(reg_a, preg) vf.store_align(dst_tile, reg_out, preg) @pl.jit() def example_kernel_int64( a: pl.Tensor[[pl.DYNAMIC, pl.DYNAMIC], pl.DT_INT64], out: pl.Tensor[[pl.DYNAMIC, pl.DYNAMIC], pl.DT_INT64], ): tf = pl.TileType(shape=[1, 32], dtype=pl.DT_INT64, target_memory=pl.MemorySpace.Vec) in_a_grp = pl.make_tile_group(type=tf, addrs=0, mutex_ids=[0]) in_a = in_a_grp.current() t_out_grp = pl.make_tile_group(type=tf, addrs=256, mutex_ids=[1]) t_out = t_out_grp.current() with pl.section_vector(): pl.load(in_a, a, [0, 0]) example_vf_int64(in_a, t_out) pl.store(out, t_out, [0, 0]) def test_example_int64(): device_id = int(os.environ.get("TILE_FWK_DEVICE_ID", 0)) device = f"npu:{device_id}" core_nums = 1 torch.npu.set_device(device) a = torch.randint(-100, 100, [1, 32], device=device, dtype=torch.int64) out = torch.empty([1, 32], device=device, dtype=torch.int64) example_kernel_int64None, core_nums torch.npu.synchronize() torch.testing.assert_close(out, -a, rtol=0, atol=0) if __name__ == "__main__": test_example_int64() print("PASSED")

该示例在仓库的向量函数测试体系中也有同类用法可供参考，例如 python/tests/st/pypto_pro/frontend/vf_api/test_vf_basic_ops.py 中通过vf.neg(reg_a, preg)构造寄存器级取反运算，验证vf.neg在 VF 前端下的行为；此外 python/tests/st/pypto_pro/frontend/element_wise/test_abs_addc_and.py 与 python/tests/st/pypto_pro/frontend/element_wise/test_vector_operations.py 展示了 Tile 级pl.neg的调用与验证方式，可作为对比参考。

运行前提

两个示例的宿主代码（test_example/test_example_int64）展示了 PyPTO VF 算子的标准验证流程：

1. 通过环境变量TILE_FWK_DEVICE_ID（默认 0）选择 NPU 设备；
2. 调用torch.npu.set_device(device)设置当前设备；
3. 以example_kernelNone, core_nums形式启动 kernel（None表示同步启动，core_nums=1指定使用 1 个 AI Core）；
4. 使用torch.npu.synchronize()等待设备侧计算完成；
5. 用torch.testing.assert_close与 PyTorch 的-a结果对比，校验通过后打印PASSED。

运行前需确保环境已安装torch、torch_npu以及本仓库的pypto_pro语言运行时，且设备为支持 Ascend 950 系列 NPU 的推理/训练环境。整体向量函数编程范式与 Tile 组、Vector section 等概念的入门介绍可参考 PyPTO 的快速上手文档 docs/zh/guide/quick_start/pro 与编程指南 docs/zh/guide/programming_guide/pro。

小结

vf.neg是 PyPTO 向量函数体系中实现元素级取反的最小算子单元，其价值体现在与 mask 机制、寄存器模型和 Tile 编排的深度协同：通过create_mask控制参与运算的元素范围，通过load_align/store_align完成 UB 与寄存器间的数据流转，再配合section_vector将整个取反流水编排进向量指令段。无论是 FP32 浮点取反还是 INT64 整数取反，读者都可以基于本文的完整示例直接修改 dtype、Tile shape 与地址参数，快速移植到自己的算子中，并借助torch.testing.assert_close完成正确性验证。

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