vf.abs

产品支持情况

• Ascend 950PR&950DT系列产品：支持

• Atlas A3系列产品：不支持

• Atlas A2系列产品：不支持

功能说明

• 对实数类型（DT_INT8、DT_INT16、DT_INT32、DT_FP16、DT_FP32）

  对src中的有效元素逐个取绝对值，并将结果写入dst对应位置，计算公式如下：

  $$ds{t}_i=|sr{c}_i|$$

• 对复数类型（DT_FP16双寄存器模式、DT_FP32双寄存器模式）

  复数类型通过双寄存器模式实现：DT_FP16双寄存器模式对应由两个DT_FP16组成的复数（实部和虚部各16位，共32位），DT_FP32双寄存器模式对应由两个DT_FP32组成的复数（实部和虚部各32位，共64位）。reg[0]存储实部，reg[1]存储虚部。

  对src中有效元素逐个取模，并将结果写入dst对应位置，计算公式如下：

  $$ds{t}_i=|sr{c}_i|=({\alpha }^2+{\beta }^2{)}^{1/2}$$

  其中$sr{c}_i=\alpha +\beta i$，α为复数的实部，β为复数的虚部。

函数原型

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

参数说明

参数名	输入/输出	说明
src	输入	源操作数，reg_tensor。源操作数src与目的操作数dst的数据类型保持一致。支持的数据类型为：DT_INT8、DT_INT16、DT_INT32、DT_FP16、DT_FP32、DT_INT64。其中DT_FP16和DT_FP32支持双寄存器模式，用于复数取模运算。
preg	输入	mask_reg。
mode	输入	可选，对应MergeMode类型。 - pypto_pro.language.MergeMode.ZEROING（默认），preg未筛选的元素在dst中置0。 - pypto_pro.language.MergeMode.MERGING当前不支持。

约束说明

无。

返回值说明

返回dst目的操作数，reg_tensor，支持的数据类型和src中的说明一致。整型数据的计算结果如果超出数据类型的表示范围会采取非饱和截断，比如DT_INT8类型，src为-128，其绝对值128会被截断成-128。

调用示例

基本调用示例

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.abs(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_kernel[None, core_nums](a, out)
    torch.npu.synchronize()
    torch.testing.assert_close(out, torch.abs(a), rtol=1e-5, atol=1e-5)

if __name__ == "__main__":
    test_example()
    print("PASSED")

INT64数据类型示例

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.abs(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_int64[None, core_nums](a, out)
    torch.npu.synchronize()
    torch.testing.assert_close(out, torch.abs(a), rtol=0, atol=0)

if __name__ == "__main__":
    test_example_int64()
    print("PASSED")