算子系统¶

类型 (Type) 安全的算子定义，支持自动类型推导，按模块化分类组织（TensorOp、TileOp、SyncOp、CrossCoreOp）。

算子分类¶

分类	类型	用途	文件位置
TensorOp	TensorType	支持广播的 N 维张量 (Tensor) 操作	`src/ir/op/tensor_ops/`
TileOp	TileType	硬件优化的 Tile 操作	`src/ir/op/tile_ops/`
SyncOp	UnknownType（屏障）；ScalarType（task / 启动形状查询）	流水线屏障、同步、TaskId 与 SPMD 启动形状查询	`src/ir/op/sync_ops/`
CrossCoreOp	UnknownType/TileType	AIC↔AIV 跨核通信	`src/ir/op/sync_ops/cross_core.cpp`

主要特性：流式 API、自动类型推导、kwargs 元数据、NumPy 风格广播、类型提升、动态维度（kDynamicDim）

类型系统¶

// Dynamic dimensions (pypto/core/common.h)
constexpr int64_t kDynamicDim = -1;
auto dynamic_dim = make_int(kDynamicDim);

类型	维度	用途	内存
TensorType	N 维	通用张量、函数参数/返回值	DDR（可选 MemRef）
TileType	N 维	统一缓冲区中的硬件优化 Tile	统一缓冲区（可选 MemRef）
ScalarType	0 维	标量值	寄存器
UnknownType	无	无返回值（同步操作）	无

REGISTER_OP 流式 API¶

方法	用途	示例
`set_op_category(str)`	算子分类	`.set_op_category("TensorOp")`
`set_description(str)`	人类可读描述	`.set_description("Element-wise add")`
`add_argument(name, desc)`	位置 Expr 参数	`.add_argument("lhs", "Left tensor")`
`no_argument()`	无参数（同步操作）	`.no_argument()`
`set_attr<T>(name)`	Kwarg 模式（T: bool, int, DataType 等）	`.set_attr<bool>("a_trans")`
`f_deduce_type(fn)`	类型推导函数	`.f_deduce_type(DeduceAddType)`

类型推导签名：

std::function<TypePtr(const std::vector<ExprPtr>& args,
                      const std::vector<std::pair<std::string, std::any>>& kwargs)>

C++ 注册示例¶

简单逐元素算子¶

// src/ir/op/tensor_ops/elementwise.cpp
REGISTER_OP("tensor.add")
    .set_op_category("TensorOp")
    .add_argument("lhs", "Left tensor")
    .add_argument("rhs", "Right tensor")
    .f_deduce_type([](const std::vector<ExprPtr>& args,
                      const std::vector<std::pair<std::string, std::any>>& kwargs) {
      CHECK(args.size() == 2);
      auto t1 = std::dynamic_pointer_cast<const TensorType>(args[0]->GetType());
      auto t2 = std::dynamic_pointer_cast<const TensorType>(args[1]->GetType());
      auto dtype = PromoteDataTypes(t1->dtype_, t2->dtype_);
      auto shape = BroadcastShapes(t1->shape_, t2->shape_);
      return std::make_shared<TensorType>(shape.shape, *dtype);
    });

带 Kwargs 的算子¶

// src/ir/op/tensor_ops/matmul.cpp
TypePtr DeduceMatMul(const std::vector<ExprPtr>& args,
                     const std::vector<std::pair<std::string, std::any>>& kwargs) {
  auto lhs = std::dynamic_pointer_cast<const TensorType>(args[0]->GetType());
  auto rhs = std::dynamic_pointer_cast<const TensorType>(args[1]->GetType());

  auto get = [&](const std::string& k, bool d) {
    for (const auto& [name, val] : kwargs)
      if (name == k) return std::any_cast<bool>(val);
    return d;
  };

  DataType dtype = [&]() {
    for (const auto& [k, v] : kwargs)
      if (k == "out_dtype") return static_cast<DataType>(std::any_cast<int>(v));
    return *PromoteDataTypes(lhs->dtype_, rhs->dtype_);
  }();

  bool a_t = get("a_trans", false), b_t = get("b_trans", false);
  ExprPtr m = a_t ? lhs->shape_[1] : lhs->shape_[0];
  ExprPtr n = b_t ? rhs->shape_[0] : rhs->shape_[1];
  return std::make_shared<TensorType>(std::vector<ExprPtr>{m, n}, dtype);
}

REGISTER_OP("tensor.matmul")
    .set_op_category("TensorOp")
    .add_argument("lhs", "Left matrix")
    .add_argument("rhs", "Right matrix")
    .set_attr<DataType>("out_dtype")
    .set_attr<bool>("a_trans")
    .set_attr<bool>("b_trans")
    .f_deduce_type(DeduceMatMul);

在 tile 层，tile.batch_matmul 为 TileType 操作数提供批量语义。它接受 rank >= 2 的 tile，广播前导批量维度，并保持与 tile.matmul 相同的纯操作数接口风格。如果批量操作数需要转置语义，可以通过两种等价方式表达：在输入上显式使用 tile.transpose(...)，或在自然 tile.load 上叠加零拷贝 tile.transpose_view(...)。在后续降级到 2D tile.matmul 时，这两种写法都会被统一识别为操作数转置语义。

tile.batch_matmul_acc(acc, lhs, rhs) 是批量路径上的累加版本：acc = acc + lhs @ rhs，遵循与 tile.batch_matmul 一致的 rank>=2 + batch 广播规则。acc 的 batch 形状必须与 lhs/rhs 广播后的 batch 形状完全一致；matmul 的 (M, N) 必须与 acc 的末两维一致；K 维必须与 lhs/rhs 内层匹配。累加器的内部 dtype 默认为浮点 → FP32、整型 → INT32（与 tile.matmul_acc 对齐）。在 conversion 阶段，ConvertTensorToTileOps 会把 tensor.matmul / tensor.matmul_acc 在任一操作数 rank > 2 时分派到该批量路径；后续由 FlattenTileNdTo2D 将其展开为逐 batch 的 2D 操作。

Python 用法¶

from pypto.pypto_core import DataType, ir
from pypto.ir import op

span = ir.Span.unknown()
dim4, dim8 = ir.ConstInt(4, DataType.INT32, span), ir.ConstInt(8, DataType.INT32, span)

# Create tensors
tensor_a = ir.Var("a", ir.TensorType([dim4, dim8], DataType.FP32), span)
tensor_b = ir.Var("b", ir.TensorType([dim8], DataType.FP32), span)

# Simple operators
result = op.tensor.add(tensor_a, tensor_b)  # Broadcasting: [4,8] + [8] → [4,8]

# Operators with kwargs
dim64, dim128 = ir.ConstInt(64, DataType.INT32, span), ir.ConstInt(128, DataType.INT32, span)
a = ir.Var("a", ir.TensorType([dim64, dim128], DataType.FP16), span)
b = ir.Var("b", ir.TensorType([dim128, dim64], DataType.FP16), span)
matmul = op.tensor.matmul(a, b, out_dtype=DataType.FP32, a_trans=True)

# Query registry
assert ir.is_op_registered("tensor.add")
op_instance = ir.get_op("tensor.add")

Kwargs（关键字参数）¶

Call 表达式 (Expression) 将 Expr 参数与元数据参数通过 kwargs 分离。

Kwargs vs Args vs 属性 (Property)¶

-	Args	Kwargs	Op 属性
类型	`ExprPtr`	`std::any`	类型擦除
作用域	每次调用	每次调用	全局
用途	张量、维度、偏移	`out_dtype`、标志、模式	设备、分类
访问方式	`call.args_`	`call.kwargs_`	`op.get_attr()`

C++ - 读取 Kwargs¶

TypePtr DeduceCastType(const std::vector<ExprPtr>& args,
                       const std::vector<std::pair<std::string, std::any>>& kwargs) {
  auto input = std::dynamic_pointer_cast<const TensorType>(args[0]->GetType());

  // `kwargs` is a vector of pairs, not a map — scan it to look a key up.
  auto find_kwarg = [&kwargs](const std::string& key) {
    return std::find_if(kwargs.begin(), kwargs.end(),
                        [&key](const auto& kv) { return kv.first == key; });
  };

  // Required kwargs — `cast` declares both `target_type` and `mode`, and codegen
  // reads `mode` unconditionally, so a missing one must fail here rather than
  // silently default to round_mode NONE.
  auto it = find_kwarg("target_type");
  CHECK(it != kwargs.end()) << "tensor.cast requires 'target_type'";
  DataType target = static_cast<DataType>(std::any_cast<int>(it->second));

  CHECK(find_kwarg("mode") != kwargs.end()) << "tensor.cast requires 'mode'";

  return std::make_shared<TensorType>(input->shape_, target);
}

真正可选的 kwarg（codegen 读取时带回退值，例如 tile.log 的 high_precision）应使用 Call::GetKwarg<T>(key, default_value) 读取，而不是 CHECK——参见 include/pypto/ir/expr.h。

Python - 使用 Kwargs¶

result = op.tensor.matmul(a, b, out_dtype=DataType.FP32, a_trans=True)
print(result.kwargs)  # {'out_dtype': 51, 'a_trans': True}

广播与类型提升¶

NumPy 风格广播¶

维度从右向左对齐：

[4, 8] + [4, 8] → [4, 8]  # Exact match
[4, 8] + [8]    → [4, 8]  # Missing left dimension = 1
[4, 1] + [8]    → [4, 8]  # Size 1 broadcasts
[1, 8] + [4, 8] → [4, 8]  # Size 1 broadcasts
[4, 8] + [5]    → Error   # 8 ≠ 5

类型提升¶

标准数值规则：浮点 > 整数，大尺寸 > 小尺寸，有符号 > 无符号（相同大小时）。

INT32 + INT32 → INT32
INT32 + FP32  → FP32   (float precedence)
INT32 + INT64 → INT64  (larger size)
UINT32 + INT32 → INT32 (signed precedence)

TensorOp：N 维张量操作¶

用途：支持完整广播的通用 N 维张量类型：TensorType（任意维度）位置：src/ir/op/tensor_ops/ Python API：from pypto.ir.op import tensor

操作： tensor.add/sub/mul/div（逐元素，支持完整 N 维广播），tensor.maximum/minimum（逐元素 max/min；rhs 可为 tensor 或 scalar — ConvertTensorToTileOps 根据 rhs 类型分发到 tile.maximum/minimum 或 tile.maximums/minimums），tensor.set_validshape（内部 API，更新 valid_shape 元数据，不搬移数据 — 仅供编译器生成代码使用），tensor.sort32 / tensor.mrgsort_format1 / tensor.mrgsort_format2（排序；分别对应 tile.sort32 / tile.mrgsort 的 tensor 层接口，由 ConvertTensorToTileOps 转换为 tile 操作），tensor.gather（按维索引；MVP 仅支持 2D 输入 + dim=-1，由 ConvertTensorToTileOps 按后端分策略下降 —— A5（Ascend950）将末维 gather 展开为对扁平元素偏移 flat[i, j] = i * src_cols + index[i, j] 的单次整块 tile.gather，并在此之前把带 stride 的 tile 源（如 tile.slice 视图）物化为连续 tile，使扁平索引能正确寻址；A2A3（Ascend910B）保留 legacy 的按行 tile.gather 循环，此时每个单行切片内的列索引即等于扁平索引），tensor.gather_mask（掩码模式选择；对应 tile.gather_mask，支持可选同位宽 output_dtype；见掩码模式），tensor.scatter（按列散布；tensor.gather 的按列逆操作，MVP 仅支持 2D 输入 + dim=-1 —— out[b, index[b, k]] = src[b, k]，index 与 src 同形状 —— 由 ConvertTensorToTileOps 下降到 tile.scatter），tensor.scatter_mask（按掩码模式散布；对应 tile.scatter_mask，将紧凑 input 按掩码扩展到 dst 的对应列 —— 见掩码模式），tensor.ci / tensor.arange（生成连续整数序列，下层降到 tile.ci；同时通过 pl.arange 暴露在顶层 namespace）

tensor.view 是只修改元数据的零拷贝 shape/layout 重新解释操作。它注册为 TensorOp，并在 ConvertTensorToTileOps 中作为 passthrough 处理；PTO in-core codegen 会将其降级为基于原始 base pointer 的 pto.make_tensor_view。目标 rank 至少为 1（DN 至少为 2）；编排层仅支持 ND shape 重新解释，且不能同时改变 layout。对部分有效的源张量进行 shape 重新解释时，仅支持把 packed ND 的 leading dimensions 折叠为 2D，或把连续前缀线性折叠为 [1, product(shape)]；两种形式都必须显式提供目标 valid_shape，并会保留源张量类型及其底层元数据。

pl.reinterpret_view(data, dtype, *, shape=None) 会根据输入分派到等价的 pl.tensor 或 pl.tile 算子，并保持返回类型种类不变。它是覆盖完全相同字节的零拷贝视图，因此 dtype 必须不同，且仅支持有/无符号 8/16/32/64 位整数、FP16、BF16 与 FP32。省略 shape 时，ND/row-major 缩放最后一轴，DN/col-major 按源/目标字节宽度比例缩放倒数第二轴。显式 shape 必须字节数相等；除非能证明它与自动推导 shape 等价，否则必须完全静态。部分有效的 valid_shape 只能使用与自动推导结果等价的 shape。零值/null padding 元数据会保留，依赖 dtype 的 max/min padding 则会清除。初始可执行路径支持 packed ND in-core tensor 及 packed、flat（none_box）row/col-major tile；DN tensor 可做类型推导但 Tensor-to-Tile 下降会拒绝，编排层 tensor 暂不支持。

示例：

from pypto.ir.op import tensor

ib = IRBuilder()
with ib.function("tensor_example") as f:
    input_a = f.param("input_a", ir.TensorType([128, 64, 32], DataType.FP32))
    input_b = f.param("input_b", ir.TensorType([128, 64, 32], DataType.FP32))
    f.return_type(ir.TensorType([128, 64, 32], DataType.FP32))
    result = ib.let("result", tensor.add(input_a, input_b))
    ib.return_stmt(result)

TileOp：硬件优化 Tile 操作¶

用途：带有显式内存管理的硬件优化 Tile 操作类型：TileType（统一缓冲区中的 Tile）位置：src/ir/op/tile_ops/ Python API：from pypto.ir.op import tile

设计：使用 TileType（而非单独的 BlockType）以保持一致性。命名空间 tile.* + TileType 清楚地表示硬件优化的 Tile 操作。

操作列表¶

分类	操作	描述
内存	`tile.get_block_idx`	获取 block 索引（返回 UINT64 标量）
-	`tile.load`	TensorType → TileType（DDR 到统一缓冲区）
-	`tile.store`	TileType → TensorType（统一缓冲区到 DDR）
逐元素	`tile.add/sub/mul/div`	Tile-Tile 操作
-	`tile.adds/subs/muls/divs`	Tile-Scalar 操作。常量标量操作数会采用 tile 的元素 dtype（裸整数字面量否则会被解析为 `index`，而任何 `pto.ts` 算子都不接受它）——但整数 tile 上的浮点字面量仍保持 FP32，以保留类型提升语义。显式的 `pl.const(v, dtype)` 属于用户的有意标注，与任何非常量表达式一样保持不变；非常量的 `index` 标量（循环变量、`pl.dim`）会被拒绝——需用 `pl.cast` 转换。`tensor.s` 同理。
一元	`tile.sqrt`	逐元素平方根
变换	`tile.slice`	提取子 tile，静态 shape，可选动态 valid_shape
-	`tile.extract`	从 `src` 在 `(index_row, index_col)` 处提取子 tile —— ISA TEXTRACT Variant 1（Mat→Left/Right，Acc→Mat）
-	`tile.reshape`	重塑 tile 维度（元素总数须一致）。会把源的 `valid_shape` 带到结果上，且绝不扩大 —— 见reshape 与有效区域（valid region）
-	`tile.reinterpret_view`	以不同 dtype 对完全相同的字节做零拷贝视图；可选 shape 默认按 layout 推导（仅支持紧密、非分形 tile）
-	`tile.transpose`	交换 tile 的两个轴
-	`tile.set_validshape`	更新 valid_shape 元数据，不搬移数据
-	`tile.ci`	生成连续整数序列（升序 start+k 或降序 start-k）；dtype ∈ {INT16, INT32}；最内维 != 1
规约	`tile.row_` / `tile.col_`	方向特定的规约（`row_sum`/`row_max`/`row_min`/`row_prod` 折叠最后一轴；`col_*` 折叠第 0 轴）。不存在以 axis 参数化的规约算子 —— ISA 只提供方向特定的指令（`pto.trowsum`、`pto.tcolsum` 等）
散布	`tile.scatter`	按行索引把 `src` 散布到 `dst`（`pto.tscatter` 索引形式；DPS：`dst` 为 in/out，结果别名为 `dst`）。`src` / `dst` dtype ∈ {I8, I16, I32, FP16, FP32, BF16}；`indexes` dtype ∈ {I16, I32}；元素宽度匹配规则：4 字节 dst ↔ INT32，2 字节 dst ↔ INT16，1 字节 dst ↔ INT16。
-	`tile.scatter_mask`	按掩码模式把 `src` 行写入 `dst` 中由掩码选中的列（DPS：`dst` 为 in/out）。这是 PyPTO codegen 层形式，下降为 `pto.tscatter` 掩码发射 —— 并非独立的 pto-isa 指令（与 `tile.gather_mask` 不同）。掩码语义见掩码模式。

tile.reshape 保持 dtype、元素总数以及源的有效区域（见下）；tile.reinterpret_view(data, dtype, *, shape=None) 改变 dtype，但要求前后总字节数完全相同。省略 shape 时，它会根据源/目标 dtype 字节宽度和 tile layout 缩放物理连续轴。在 PTOAS 内存规划下，无论 shape 是否变化，都会下降为保持别名关系的 PTO treshape 原语。

reshape 与有效区域（valid region）¶

reshape 是零拷贝视图，无法凭空产生数据：tensor.reshape 与 tile.reshape 共用同一条规则，把源的 valid_shape 映射到目标 shape，且绝不扩大。有效区域只能表示为以原点为锚的矩形框，因此并非所有源区域都能在重新切分后保留：

源区域	结果
完全有效	`new_shape` —— 会被规范化掉，不产生 view，已有程序不受影响
可证明为空	全零矩形框
仅增删完全有效的单位轴	保留的轴按 1:1 映射，可精确保留任意矩形
连续的扁平前缀	`new_shape` 中覆盖同一批元素的矩形（若存在）
其他情况	拒绝 —— `valid_shape` 无法描述 reshape 后的区域

因此 [8, 16] valid [5, 16]（80 个元素的扁平前缀）可映射为 [16, 8] valid [10, 8] 或 [128] valid [80]，而 [4, 32] 会被拒绝 —— 80 个元素不是整数行（每行 32）。[1, 8, 16] valid [1, 8, 5] 根本不是扁平前缀，但映射到 [8, 16] valid [8, 5] 是精确的，因为丢弃完全有效的单位轴不改变行列关系。 tensor.reshape 可选的第三个 valid_shape 操作数只能收窄推导出的区域，不能声称拥有该区域之外的数据。

数据流： TensorType (DDR) → tile.load → TileType (Unified Buffer) → tile.{ops} → TileType → tile.store → TensorType (DDR)

掩码模式¶

*.gather_mask / *.scatter_mask 使用编译期 MaskPattern（pl.tile.MaskPattern，整数取值 1–7，与硬件 VREDUCEv2 的 pattern mode 一致）按行标记列的一个子集（模式名从右往左读，最右位对应列 0）。同一标记集合驱动两个算子做相反方向的操作。gather_mask 选择并紧凑：从宽输入中读取被标记的列，紧凑写入较窄输出的前若干列（out_cols = cols / stride）；这是真实的 pto-isa 指令（pto.tgather 掩码形式），A2/A3 与 A5 均支持。scatter_mask 放置并扩展：把紧凑输入写入更宽 dst 的被标记列（dst_cols = cols * stride），未标记列保留 dst 原值（DPS）；这是 PyPTO codegen 层形式，并非独立的 pto-isa 指令 —— 不存在 pto.tscatter 掩码指令（与 gather 不同）—— PyPTO 为 A2/A3 / CPU-sim 类下降路径发射它。例如对 [a0 a1 a2 a3 a4 a5 a6 a7]：gather P0101 → [a0 a2 a4 a6]；对 [s0 s1 s2 s3] 做 scatter P0101 → [s0 · s1 · s2 · s3 ·]（· 表示保留的 dst）。

模式	整数	标记列 `c` 的条件	被标记的列	步长
`P0101`	1	`c % 2 == 0`	0, 2, 4, …	2
`P1010`	2	`c % 2 == 1`	1, 3, 5, …	2
`P0001`	3	`c % 4 == 0`	0, 4, 8, …	4
`P0010`	4	`c % 4 == 1`	1, 5, 9, …	4
`P0100`	5	`c % 4 == 2`	2, 6, 10, …	4
`P1000`	6	`c % 4 == 3`	3, 7, 11, …	4
`P1111`	7	全选	全部	1

最后一维须能被步长整除。gather_mask 另接受可选的同位宽 output_dtype（按位重解释，而非数值转换）。参考：gather 的选择语义见 pto-isa 的 MaskSelect（include/pto/cpu/TGather.hpp）；pypto 类型推导见 src/ir/op/tile_ops/gather.cpp（gather）/ src/ir/op/tile_ops/scatter.cpp（scatter）。

使用示例¶

from pypto.ir.op import tile

ib = IRBuilder()
with ib.function("tile_computation") as f:
    input_a = f.param("input_a", ir.TensorType([128, 128], DataType.FP32))
    input_b = f.param("input_b", ir.TensorType([128, 128], DataType.FP32))
    output = f.param("output", ir.TensorType([128, 1], DataType.FP32))
    f.return_type(ir.TensorType([128, 1], DataType.FP32))

    # Load, compute, reduce, store
    tile_a = ib.let("tile_a", tile.load(input_a, [0, 0], [32, 128]))
    tile_b = ib.let("tile_b", tile.load(input_b, [0, 0], [32, 128]))
    tile_mul = ib.let("tile_mul", tile.mul(tile_a, tile_b))
    tile_sqrt = ib.let("tile_sqrt", tile.sqrt(tile_mul))
    # row_sum 折叠最后一轴 -> [32, 1]。scratch tile 必须与输入 dtype 和 rank 相同，
    # 且每一维都不小于输入的对应维度。
    tmp_tile = ib.let("tmp_tile", tile.create([32, 128], DataType.FP32))
    tile_sum = ib.let("tile_sum", tile.row_sum(tile_sqrt, tmp_tile))
    result = ib.let("result", tile.store(tile_sum, [0, 0], output))
    ib.return_stmt(result)

SyncOp：同步操作¶

用途：硬件同步与屏障，以及共用 system. 命名空间的 TaskId 与 SPMD 启动形状查询类型：屏障类为 UnknownType（无返回值，在 EvalStmt 中使用）；查询类为 ScalarType，会绑定一个值（task_invalid、task_is_valid、available_cluster_count、available_aiv_count）位置：src/ir/op/sync_ops/ —— sync.cpp（屏障）、task.cpp（TaskId）、launch.cpp（启动形状查询） Python API：from pypto.ir.op import system

操作	描述	Kwargs
`system.bar_all`	全局屏障	无
`system.bar_v`	向量屏障	无
`system.bar_m`	矩阵屏障	无
`system.fence`	全局内存屏障（下降为 `pto.fence.barrier_all #pto.fence_scope<gm>`）	无
`system.cacheinvalid`	使 tensor 某个子区域对应的 cache line 失效。参数：`tensor`、`shapes`（N 维）、`offsets`（N 维）。`shapes` 全为 1（标量写）时下降为 `pto.addptr` + `pto.cmo.cacheinvalid %write_ptr single_cache_line`；区域更大（TStore 写）时下降为 `pto.partition_view` + `pto.cmo.cacheinvalid %payload_view single_cache_line : !pto.partition_tensor_view<...>`	无
`system.syncall`	跨核全员屏障（`pto::SYNCALL`）。`mode="hard"`（FFTS，无 operand）或 `mode="soft"`（GM 轮询，带 operand）	`core_type`（`"aiv_only"` \| `"aic_only"` \| `"mix"`）、`mode`（`"hard"` \| `"soft"`）
`system.sync_src`	设置同步标志	`set_pipe`, `wait_pipe`, `event_id`
`system.sync_dst`	等待同步标志	`set_pipe`, `wait_pipe`, `event_id`
`system.task_invalid`	`PTO2TaskId::invalid()` 哨兵——TaskId carry 的 "暂无 producer" 种子	无
`system.task_is_valid`	测试某个 `TASK_ID` 值是否为有效（非哨兵）handle	无；唯一位置参数是 TaskId Var
`system.available_cluster_count`	本次运行的 MIX cluster（= AIC）数，由设备读回。结果为 `Scalar[INT32]`	无
`system.available_aiv_count`	本次运行的独立 AIV 核数，由设备读回。结果为 `Scalar[INT32]`	无

system.syncall 有两种 mode。hard 形态（mode="hard"，默认）下沉为 FFTS 屏障，等待所选 core_type 的全部物理核到达；kernel 必须以满占用方式启动（每个物理核一个 block）且带 sync_start=True（使所有 block 同时驻留——非 sync_start 启动可能分波次派发 block 而使屏障死锁），否则屏障死锁（AICore 错误 507018）。soft 形态（mode="soft"）轮询一段共享 GM workspace，因此可在部分占用下工作。gm_workspace 是共享、清零的 GM INT32 tensor，含 used_cores * 8 个 slot（请作为 kernel 参数传入，使所有 block 共享同一缓冲）；暂存 tile 由编译器合成；used_cores 是参与核数。soft 形态对每种 core_type 都支持，operand 随参与核集合而不同：

aiv_only：[gm_workspace, ub_scratch, used_cores] —— 一个 UB（Vec）暂存 tile。
aic_only：[gm_workspace, l1_scratch, used_cores] —— 一个扁平 L1（Mat，slayout=none_box）暂存 tile。
mix：[gm_workspace, ub_scratch, l1_scratch, used_cores] —— UB 与扁平 L1 各一个。该屏障汇合 AIC + AIV 核，故 used_cores 是总参与数（AIC block 数 + AIV subblock 数）。该 op 会被复制到 cube 与 vector 两条流上，每条流各用自己的 tile（另一个在该流上是死代码），与 pto-isa 的 soft-mix 下沉一致。

扁平 L1 暂存 tile 通过 pl.tile.create(..., target_memory=pl.Mem.Mat, flat_layout=True) 创建，保持连续的 slayout=none_box 布局（普通的 boxed NZ Mat tile 会错位 8 个 int32 计数槽）。

统一的 mode= 关键字 API（mode="hard" / mode="soft"）是 DSL 层接口（pl.system.syncall）。pypto.ir.op.system 下的 Python IR 辅助函数则是拆开的：syncall(core_type=...) 构造 hard 形态，syncall_soft(core_type, args) 构造 soft 形态。

system.available_cluster_count / system.available_aiv_count 是 SPMD 启动形状查询：把它作为 pl.spmd(...) 的 core_num 传入，启动宽度即按本次运行落到的设备自适应。Orchestration codegen 分别下沉为 rt_available_cluster_count() / rt_available_aiv_count()。混合（AIC+AIV）或纯 cube kernel 用 cluster 数（每个 core-group 一个 block），纯 vector kernel 用 AIV 数。这是唯一能跨设备保持满占用的启动宽度，而 hard system.syncall 正需要满占用；HardSyncallOccupancy verifier 对这类宽度不再做数量比较，并会拒绝用错核类型的查询。请把调用内联传入（pl.spmd(pl.system.available_cluster_count())），不要先绑定到变量名——变量名会以「定义在调用方的变量」形式落到外提出的 Spmd 包装函数上，IR printer 无法重新解析。源码：src/ir/op/sync_ops/launch.cpp。

system.task_invalid 返回类型为 ScalarType(DataType::TASK_ID)。当 Python 字面量 None 出现在 TaskId 位置（deps=[None] 条目或 TaskId 循环 iter_arg 种子）时，它就是 None 在 with pl.manual_scope(): 区域内的下沉目标。不存在 system.task_id_of op —— producer task id 由 pl.submit(...) parser construct 返回的二元组第二个元素获得，而非来自 builtin。源码：src/ir/op/sync_ops/task.cpp。

CrossCoreOp：AIC↔AIV 跨核通信¶

用途：AIC (Cube) 和 AIV (Vector) 内核之间的跨核同步、数据传输和管道管理类型：UnknownType（sync/push/init/buffer/free 操作）或 TileType 透传（pop 操作）位置：src/ir/op/tile_ops/cross_core.cpp（tpush/tpop）和 src/ir/op/sync_ops/cross_core.cpp（sync/tfree/管道初始化/缓冲区） Python API：import pypto.language as pl（提升的操作）或 from pypto.ir.op import tile, system

显式事件同步¶

操作	参数	描述	Kwargs
`system.sync_set`	0 或 1（`event_id_dyn`）	从一种核类型发出 `pto.sync.set`	`pipe`、静态 `event_id`、可选 `ffts_mode`、可选 `core_type`
`system.sync_wait`	0 或 1（`event_id_dyn`）	在对端核类型发出 `pto.sync.wait`	`pipe`、静态 `event_id`、可选 `core_type`
`system.set_ffts`	1（`workspace`）	声明 A3 显式跨核事件所需的 FFTS 设置	—

在显式指定类型的 AIC/AIV kernel 中使用 pl.system.sync_set(event_id, pipe=..., ffts_mode=...) 和 pl.system.sync_wait(event_id, pipe=...)。在混合 InCore kernel 中，传入 core_type="aiv" 或 core_type="aic"，以便 kernel 展开时将各事件操作保留在目标核通道上。在 A3 上，每个参与同步的 AIC/AIV 函数都必须在首次显式事件操作前调用 pl.system.set_ffts(workspace)；workspace 必须是至少包含 256 个元素的一维 INT64 张量，并作为 PTOAS 的设置操作数。PyPTO 的常驻运行时会持续安装硬件 FFTS 控制地址，因此生成的运行时封装不会用该操作数覆盖此地址。A5 不需要该设置。event_id 可以是用户可用范围 0–13 内的整数，也可以是动态 pl.Scalar[pl.INDEX]；ID 14 和 15 为保留值。sync_set 的可选 ffts_mode 必须为 0、1 或 2。手写跨核协议的作者负责正确配对事件 ID 和 pipe。PyPTO 的常规核内自动依赖插入仍保持启用，并使用独立的 set_flag/wait_flag 机制，因此不会占用这些显式跨核事件 ID。

数据传输操作¶

操作	参数	描述	Kwargs
`tile.tpush_to_aiv`	1 (tile)	从 Cube 推送 tile 到 Vector	`split`，可选 `id`
`tile.tpush_to_aic`	1 (tile)	从 Vector 推送 tile 到 Cube	`split`，可选 `id`
`tile.tpop_from_aic`	0	从 Cube 管道弹出 tile（→ TileType）	`split`，可选 `id`
`tile.tpop_from_aiv`	0	从 Vector 管道弹出 tile（→ TileType）	`split`，可选 `id`
`system.tfree_to_aic`	1 (tile)	向 Cube 生产者释放槽位	可选 `id`
`system.tfree_to_aiv`	1 (tile)	向 Vector 生产者释放槽位	可选 `id`

管道初始化操作¶

操作	参数	描述	Kwargs
`system.aic_initialize_pipe`	2	在 Cube 侧初始化跨核管道（位置参数：`c2v_consumer_buf`、`v2c_consumer_buf`，i32 SSA）	`dir_mask`, `slot_size`，可选 `slot_num`，可选 `local_slot_num`，可选 `id`
`system.aiv_initialize_pipe`	2	在 Vector 侧初始化跨核管道（位置参数：`c2v_consumer_buf`、`v2c_consumer_buf`，i32 SSA）	`dir_mask`, `slot_size`，可选 `slot_num`，可选 `local_slot_num`，可选 `id`

slot_num（设置时必须 > 0）显式指定 GM 环形缓冲区的槽数量；省略时由 PTOAS 取默认值（单向 8，双向每方向 4）。
local_slot_num（仅 a2/a3，必须 > 0 且 <= slot_num）显式指定本地槽数量。
预留/导入缓冲区大小需由用户自行设置，且与架构相关：a3 为 slot_size * local_slot_num；a5 为 slot_size * slot_num。

缓冲区管理操作¶

操作	参数	描述	Kwargs
`system.reserve_buffer`	0	预留跨核通信命名缓冲区（消费者侧）	`name`, `size`, `base`*
`system.import_peer_buffer`	0	从同组对等函数导入缓冲区（生产者侧）	`name`, `peer_func`

* base 默认为 AUTO (-1)，由编译器自动分配地址。

DSL 示例（跨核 V2C 单向）¶

dir_mask=2 仅启用 V2C，因此 C2V 侧缓冲区实参需为未使用方向的占位（0、pl.const(0, pl.INT32)）；启用侧将 reserve_buffer / import_peer_buffer 的句柄作为第一个位置实参传入。

import pypto.language as pl

@pl.program
class CrossCoreExample:
    @pl.function(type=pl.FunctionType.InCore)
    def vector_producer(self, a: pl.Tensor[[16, 16], pl.FP16]):
        peer = pl.import_peer_buffer(name="v2c_buf", peer_func="cube_consumer")
        pl.aiv_initialize_pipe(pl.const(0, pl.INT32), peer, dir_mask=2, slot_size=512)

        tile_a: pl.Tile[[16, 16], pl.FP16] = pl.load(a, [0, 0], [16, 16])
        pl.tpush_to_aic(tile_a, split=0)

    @pl.function(type=pl.FunctionType.InCore)
    def cube_consumer(self, out: pl.Tensor[[16, 16], pl.FP32]) -> pl.Tensor[[16, 16], pl.FP32]:
        buf = pl.reserve_buffer(name="v2c_buf", size=4096, base=0x1000)
        pl.aic_initialize_pipe(pl.const(0, pl.INT32), buf, dir_mask=2, slot_size=512)

        received: pl.Tile[[16, 16], pl.FP16] = pl.tpop_from_aiv(split=0)
        pl.tfree_to_aiv(received)
        result: pl.Tensor[[16, 16], pl.FP32] = pl.store(received, [0, 0], out)
        return result

参阅 TPUSH/TPOP ISA 参考和缓冲区管理了解硬件细节。

文件组织¶

目录/文件	内容
`src/ir/op/type_inference.cpp`	共享的类型推断工具
`tensor_ops/elementwise.cpp`	TensorOp: add, sub, mul, div
`tile_ops/memory.cpp`	TileOp: load, store, read, get_block_idx
`tile_ops/elementwise.cpp`	TileOp: add, mul, div, adds, muls 等
`tile_ops/reduction.cpp`	TileOp: sum（含 axis, keepdim）
`tile_ops/unary.cpp`	TileOp: sqrt
`sync_ops/sync.cpp`	SyncOp: sync_src, sync_dst, barriers
`sync_ops/task.cpp`	SyncOp：TaskId 哨兵与判定
`sync_ops/launch.cpp`	SyncOp：SPMD 启动形状查询
`sync_ops/cross_core.cpp`	CrossCoreOp: tpush, tpop, pipe init, buffers

优势：

模块化：自包含的算子分类
构建性能：修改一个分类不会重新构建其他分类
可维护性：易于定位和修改算子
可扩展性：直接添加新算子

添加新操作¶

选择分类文件：src/ir/op/tensor_ops/elementwise.cpp、matmul.cpp、reduction.cpp，或 src/ir/op/tile_ops/memory.cpp、unary.cpp
实现类型推导：

TypePtr DeduceType(const std::vector<ExprPtr>& args,
                   const std::vector<std::pair<std::string, std::any>>& kwargs) {
  CHECK(args.size() == 2) << "op requires 2 arguments";
  // Validate types, read kwargs, compute output type
  return result_type;
}

注册：

REGISTER_OP("tensor.matmul")
    .set_op_category("TensorOp")
    .add_argument("lhs", "Left tensor")
    .add_argument("rhs", "Right tensor")
    .set_attr<DataType>("out_dtype")
    .f_deduce_type(DeduceType);

Python 封装 (python/pypto/ir/op/tensor_ops.py)：

def matmul(lhs: Expr, rhs: Expr, out_dtype=None, a_trans=False) -> Call:
    kwargs = {}
    if out_dtype: kwargs["out_dtype"] = out_dtype.code() if isinstance(out_dtype, DataType) else out_dtype
    if a_trans: kwargs["a_trans"] = a_trans
    return _ir_core.create_op_call("tensor.matmul", [lhs, rhs], kwargs, Span.unknown())

添加测试，位于 tests/ut/ir/，如需要则更新 CMakeLists.txt

参考¶

核心定义位于 include/pypto/core/common.h 和 include/pypto/ir/；注册表与类型推断实现在 src/ir/，算子实现按类别位于 src/ir/op/{tensor_ops,tile_ops,sync_ops}/。