vf.sub
产品支持情况
- Ascend 950PR&950DT系列产品:支持
- Atlas A3系列产品:不支持
- Atlas A2系列产品:不支持
功能说明
两个reg_tensor逐元素减法。
根据mask,对源操作数src0、src1进行按元素求差操作,将结果写入目的操作数dst,计算公式如下:
dsti=src0i−src1idst_i = src0_i - src1_i
图1 vf.sub功能说明

若需要获取减法借位输出,请参考vf.subc。
函数原型
sub(src0, src1, preg, /, *, mode: Optional[MergeMode] = None) -> dst
参数说明
| 参数 | 输入/输出 | 说明 |
|---|---|---|
| src0 | 输入 | 源操作数0,reg_tensor,源操作数src0与目的操作数dst的数据类型保持一致。支持的数据类型为:DT_INT8、DT_UINT8、DT_INT16、DT_UINT16、DT_INT32、DT_UINT32、DT_FP16、DT_BF16、DT_FP32、DT_INT64、DT_UINT64。 |
| src1 | 输入 | 源操作数1,reg_tensor,数据类型与src0一致。 |
| preg | 输入 | mask_reg。 |
| mode | 输入 | 可选,对应MergeMode类型。 - pypto_pro.language.MergeMode.ZEROING(默认),preg未筛选的元素在dst中置0。 - pypto_pro.language.MergeMode.MERGING当前不支持。 |
约束说明
无。
返回值说明
返回dst目的操作数,reg_tensor,支持的数据类型和src0中的说明一致。
调用示例
基本调用示例
import os
import pypto_pro.language as pl
import torch
import torch_npu
@pl.vector_function
def example_vf(src_a, src_b, dst_tile):
preg = vf.create_mask(pattern=pl.MaskPattern.ALL, dtype=pl.DT_FP32)
reg_a = vf.load_align(src_a, 0)
reg_b = vf.load_align(src_b, 0)
reg_out = vf.sub(reg_a, reg_b, preg)
vf.store_align(dst_tile, reg_out, preg)
@pl.jit()
def example_kernel(
a: pl.Tensor[[pl.DYNAMIC, pl.DYNAMIC], pl.DT_FP32],
b: 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()
in_b_grp = pl.make_tile_group(type=tf, addrs=0x100, mutex_ids=[1])
in_b = in_b_grp.current()
t_out_grp = pl.make_tile_group(type=tf, addrs=0x200, mutex_ids=[2])
t_out = t_out_grp.current()
with pl.section_vector():
pl.load(in_a, a, [0, 0])
pl.load(in_b, b, [0, 0])
example_vf(in_a, in_b, 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)
b = 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, b, out)
torch.npu.synchronize()
torch.testing.assert_close(out, a - b, 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_a, src_b, dst_tile):
preg = vf.create_mask(pattern=pl.MaskPattern.ALL, dtype=pl.DT_INT64)
reg_a = vf.load_align(src_a, 0)
reg_b = vf.load_align(src_b, 0)
reg_out = vf.sub(reg_a, reg_b, 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],
b: 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()
in_b_grp = pl.make_tile_group(type=tf, addrs=256, mutex_ids=[1])
in_b = in_b_grp.current()
t_out_grp = pl.make_tile_group(type=tf, addrs=512, mutex_ids=[2])
t_out = t_out_grp.current()
with pl.section_vector():
pl.load(in_a, a, [0, 0])
pl.load(in_b, b, [0, 0])
example_vf_int64(in_a, in_b, 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)
b = 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, b, out)
torch.npu.synchronize()
torch.testing.assert_close(out, a - b, rtol=0, atol=0)
if __name__ == "__main__":
test_example_int64()
print("PASSED")