跳转至

PyPTO 语言指南

pypto.languagepl)模块的完整参考。

类型系统

数据类型(DataType)

常量 位数 说明
pl.BOOL 1 布尔值
pl.INT4 / pl.UINT4 4 有符号 / 无符号 4 位整数
pl.INT8 / pl.UINT8 8 有符号 / 无符号 8 位整数
pl.INT16 / pl.UINT16 16 有符号 / 无符号 16 位整数
pl.INT32 / pl.UINT32 32 有符号 / 无符号 32 位整数
pl.INT64 / pl.UINT64 64 有符号 / 无符号 64 位整数
pl.FP16 16 IEEE 半精度浮点
pl.BF16 16 Brain Float 16
pl.FP32 32 IEEE 单精度浮点
pl.FP4 4 4 位浮点(打包 MXFP4 E2M1×2;PTOAS !pto.f4E2M1x2
pl.FP8E4M3FN 8 8 位浮点(e4m3fn);MXFP8 数据
pl.FP8E5M2 8 8 位浮点(e5m2);MXFP8 数据
pl.FP8E8M0 8 MX 块缩放指数(E8M0);PTOAS !pto.f8E8M0 / pto-isa float8_e8m0_t
pl.HF4 / pl.HF8 4/8 昇腾浮点格式
pl.INDEX 64 索引计算类型 —— 循环变量、维度

容器类型

pl.Tensor[[shape], dtype] —— DDR 内存数组(片外全局内存)。

x: pl.Tensor[[64, 128], pl.FP32]        # 二维,64×128,float32
y: pl.Tensor[[256], pl.FP16]            # 一维,256 个元素,float16
z: pl.Tensor[[64, 128], pl.FP16, pl.NZ] # 带 NZ 布局

pl.Tile[[shape], dtype] —— 片上内存缓冲区(默认统一缓冲区)。

t: pl.Tile[[64, 64], pl.FP32]           # 二维 tile,64×64

pl.Scalar[dtype] —— 单个标量值。

s: pl.Scalar[pl.FP32]                   # float32 标量
idx: pl.Scalar[pl.INDEX]                # 索引标量

张量布局(TensorLayout)

pl.Tensor[...] annotation 写 runtime 行优先 shape,不写 layout 标记。layout 是 IR 内部概念,由派生/消费视图的 op 推导,不需要在 annotation 上表达。

# ✅ 推荐 —— 写源 tensor shape,不写 layout 标记:
b: pl.Tensor[[N, K], pl.FP32]
# ⚠️ 已弃用(RFC #1300 补充 1):
b: pl.Tensor[[K, N], pl.FP32, pl.DN]   # → 解析期触发 DeprecationWarning

为什么弃用 pl.Tensor[..., pl.DN] layout-only 简写迫使用户脑子里同时持有两套坐标系(IR 逻辑后视图 shape 与 runtime 行优先 shape)—— 恰恰是 RFC #1300 想要消除的歧义。改用:去掉 layout 标记,写 runtime shape —— matmul B^T 场景给 pl.matmulb_trans=True(或 a_trans=True),或自然 load 后用 pl.tile.transpose_view(...)(参见下文「数据搬运」);DN-producing op 之后的 slice 自动继承父 layout。

如需 NZ(硬件 tile layout),写 pl.Tile[..., pl.NZ] —— NZ 是 tile-only,不允许作为 TensorType annotation。pl.NZ 常量保留用于 tile annotation 和 IR 内部使用。

若需要在 IR 层面写 DN tensor(如测试 fixture 或 round-trip 打印的 IR),用 pl.TensorView(stride=[...], layout=pl.TensorLayout.DN) —— 强制写显式 stride,避免隐式坐标翻转的隐患。

动态形状(Dynamic Shapes)

使用 pl.dynamic() 声明运行时确定的维度:

M = pl.dynamic("M")
N = pl.dynamic("N")

@pl.function
def dynamic_kernel(
    a: pl.Tensor[[M, N], pl.FP32],
) -> pl.Tensor[[M, N], pl.FP32]:
    ...

参数方向(Parameter Directions)

默认情况下,参数为只读输入。使用包装器声明输出参数:

方向 语法 说明
输入(默认) a: pl.Tensor[...] 只读
输出 a: pl.Out[pl.Tensor[...]] 只写输出
输入/输出 a: pl.InOut[pl.Tensor[...]] 读写
@pl.function
def kernel(
    input_a: pl.Tensor[[64], pl.FP32],                    # In
    output_b: pl.Out[pl.Tensor[[64], pl.FP32]],            # Out
    accum_c: pl.InOut[pl.Tensor[[64], pl.FP32]],           # InOut
) -> pl.Tensor[[64], pl.FP32]:
    ...

操作

分发模型(Dispatch Model)

PyPTO 操作分为三个层级:

命名空间 层级 说明
pl.* 统一 根据输入类型(Tensor 或 Tile)自动分发
pl.tensor.* Tensor DDR 级别的 Tensor 操作
pl.tile.* Tile 片上 Tile 操作

推荐: 尽量使用 pl.*(统一接口)。分发器会选择正确的实现。

# 统一接口 —— Tensor 和 Tile 都适用
result = pl.add(a, b)       # 分发到 tensor.add 或 tile.add
result = pl.mul(a, scalar)   # 分发到 tensor.muls 或 tile.muls

# 显式 tile 级别(需要 tile 特定操作时)
tile = pl.tile.load(tensor, [0, 0], [64, 64])
tile = pl.tile.adds(tile, 1.0)

Python 运算符

标准 Python 运算符映射到 IR 操作:

Python IR 操作 示例
a + b add c = a + b
a - b sub c = a - b
a * b mul c = a * b
a / b div c = a / b
a == b eq(比较) if a == 0:
a != b ne(比较) if a != 0:
a < b lt(比较) if a < n:
a > b gt(比较) if a > 0:

统一操作(Unified Operations)

常用 pl.* 操作 —— 完整列表参见操作参考

c = pl.add(a, b)            # 算术(还有 sub、mul、div)
c = pl.add(a, 1.0)          # 标量右操作数自动检测
c = pl.cast(a, pl.FP16)     # 类型转换
c = pl.reshape(a, [16, 8])  # 形状操作(还有 transpose、slice)
c = pl.matmul(a, b)         # 线性代数
c = pl.row_sum(a)            # 归约(还有 row_max)

并非每个 pl.cast 都是一条指令。某个 (src, dst) 组合是映射到一条硬件 pto.tcvt, 还是被展开成多跳链,取决于目标架构——例如 INT32 -> FP16 在 Ascend910B 上是一条指令, 在 Ascend950 上则下降为 INT32 -> FP32 -> FP16。展开链每跳一条 tcvt;当中转类型比源 类型更窄时,结果可能与直接舍入的转换相差目标类型的 1 个 ULP。各架构的原生组合、展开组 合及其跳数见 LegalizeTileCast

Tensor 和 Tile 类型支持 Python 下标语法作为 slice/read 的语法糖:

row = A[0:16, :]       # 等价于 pl.slice(A, [16, N], [0, 0])
elem = A[i, j]         # 等价于 pl.tensor.read(A, [i, j]) / pl.tile.read(A, [i, j])
block = A[0:16, 0:32]  # 等价于 pl.slice(A, [16, 32], [0, 0])

对称的写入形式 dst[<slices...>] = srcpl.assemble 的语法糖:

out[i:i+16, j:j+32] = src   # 等价于 out = pl.assemble(out, src, [i, j])

该语法糖仅在 SSA 转换前可用——它会重新绑定 dst,与严格 SSA 不兼容。在 @pl.function(strict_ssa=True) 或任何 SSA 后的上下文中,请显式调用 pl.assemble(...)

需要 tile 特定操作(内存搬运、广播、位运算等)时使用 pl.tile.*

变量赋值与 SSA

PyPTO 的 IR 同时支持 SSA(静态单赋值,Static Single Assignment)和非 SSA 两种形式。在 SSA 形式中,每个变量只被赋值一次;在非 SSA 形式中,可以对同一个变量名多次赋值。

编写风格

非 SSA(默认) —— 像普通 Python 一样自由重新赋值:

@pl.function
def example(x: pl.Tensor[[64], pl.FP32]) -> pl.Tensor[[64], pl.FP32]:
    result: pl.Tensor[[64], pl.FP32] = pl.mul(x, 2.0)
    result: pl.Tensor[[64], pl.FP32] = pl.add(result, 1.0)  # 重新赋值,没问题
    return result

SSA 风格 —— 每个变量只赋值一次,使用不同的名称:

@pl.function
def example(x: pl.Tensor[[64], pl.FP32]) -> pl.Tensor[[64], pl.FP32]:
    result_0: pl.Tensor[[64], pl.FP32] = pl.mul(x, 2.0)
    result_1: pl.Tensor[[64], pl.FP32] = pl.add(result_0, 1.0)
    return result_1

两种方式都能生成有效的 IR。选择你更习惯的风格即可。

自动 SSA 转换

大多数优化 pass 需要 SSA 形式。编译流水线会在早期自动运行 ConvertToSSA,因此你无需担心 —— 直接编写非 SSA 代码,编译器会自动处理转换。

严格 SSA 模式

传入 strict_ssa=True 可在解析阶段强制要求 SSA。如果重新赋值变量,解析器将报错:

@pl.function(strict_ssa=True)
def example(x: pl.Tensor[[64], pl.FP32]) -> pl.Tensor[[64], pl.FP32]:
    result: pl.Tensor[[64], pl.FP32] = pl.mul(x, 2.0)
    result: pl.Tensor[[64], pl.FP32] = pl.add(result, 1.0)  # 报错:SSAViolationError
    return result

这对于捕获无意的变量覆盖很有用,但完全是可选的。

为什么需要 yield_

在 SSA 形式中,控制流(循环、if/else)不能简单地重新赋值变量 —— 每次赋值必须是唯一的。pl.yield_() 是将值从控制流作用域中传出的机制:

  • 循环pl.yield_() 将更新后的累加器传递给下一次迭代
  • If/else:两个分支中的 pl.yield_() 创建一个合并点(phi 节点),产生一个结果变量

这就是为什么带累加器的循环需要 init_values + yield_,以及为什么产生值的 if/else 分支必须都使用 yield_

控制流

For 循环 —— pl.range()

简单循环:

for i in pl.range(10):
    # i = 0, 1, 2, ..., 9
    ...

for i in pl.range(2, 10):
    # i = 2, 3, ..., 9
    ...

for i in pl.range(0, 100, 4):
    # i = 0, 4, 8, ..., 96
    ...

带累加器的循环(init_values):

累加器在迭代之间传递值。每次迭代接收前一次的值,必须 yield_ 新值:

@pl.function
def sum_16_elements(data: pl.Tensor[[16], pl.FP32]) -> pl.Tensor[[1], pl.FP32]:
    init_sum: pl.Tensor[[1], pl.FP32] = pl.create_tensor([1], dtype=pl.FP32)

    for i, (running_sum,) in pl.range(16, init_values=(init_sum,)):
        chunk: pl.Tensor[[1], pl.FP32] = pl.slice(data, [1], [i])
        new_sum: pl.Tensor[[1], pl.FP32] = pl.add(running_sum, chunk)
        sum_out: pl.Tensor[[1], pl.FP32] = pl.yield_(new_sum)

    # 循环结束后 sum_out 保存最终累加值
    return sum_out

多个累加器:

@pl.function
def find_max_and_sum(
    data: pl.Tensor[[4, 64], pl.FP32],
) -> pl.Tensor[[1, 64], pl.FP32]:
    init_max: pl.Tensor[[1, 64], pl.FP32] = pl.create_tensor([1, 64], dtype=pl.FP32)
    init_sum: pl.Tensor[[1, 64], pl.FP32] = pl.create_tensor([1, 64], dtype=pl.FP32)

    for i, (acc_max, acc_sum) in pl.range(4, init_values=(init_max, init_sum)):
        row: pl.Tensor[[1, 64], pl.FP32] = pl.slice(data, [1, 64], [i, 0])
        new_max: pl.Tensor[[1, 64], pl.FP32] = pl.maximum(acc_max, row)
        new_sum: pl.Tensor[[1, 64], pl.FP32] = pl.add(acc_sum, row)
        out_max, out_sum = pl.yield_(new_max, new_sum)

    return out_sum

并行循环 —— pl.parallel()

语法与 pl.range() 相同,但迭代可以并行执行:

for i in pl.parallel(0, num_blocks):
    # 迭代相互独立,可以并行运行
    ...

While 循环 —— pl.while_()

始终需要 init_values。条件通过 pl.cond() 作为循环体的第一条语句设置:

for (x,) in pl.while_(init_values=(0,)):
    pl.cond(x < 10)          # 当 x < 10 时继续
    new_x = x + 1
    x_out = pl.yield_(new_x)

If/Else 与 pl.yield_()

产生值的分支必须 yield_ 这些值。这会创建 SSA phi 节点 —— 两个分支必须 yield 相同数量和类型的值:

@pl.function
def conditional_update(
    a: pl.Tensor[[64], pl.FP32],
    delta: pl.Tensor[[64], pl.FP32],
) -> pl.Tensor[[64], pl.FP32]:
    init: pl.Tensor[[64], pl.FP32] = pl.create_tensor([64], dtype=pl.FP32)

    for i, (prev,) in pl.range(4, init_values=(init,)):
        if i == 0:
            result: pl.Tensor[[64], pl.FP32] = pl.yield_(a)
        else:
            updated: pl.Tensor[[64], pl.FP32] = pl.add(prev, delta)
            result: pl.Tensor[[64], pl.FP32] = pl.yield_(updated)
        # result 保存执行的那个分支的值
        out: pl.Tensor[[64], pl.FP32] = pl.yield_(result)

    return out

规则: 如果一个分支 yield,另一个也必须 yield。两个分支 yield 相同数量的值。

程序与函数

@pl.function

将 Python 函数解析为 IR:

@pl.function
def my_func(x: pl.Tensor[[64], pl.FP32]) -> pl.Tensor[[64], pl.FP32]:
    ...

指定函数类型:

@pl.function(type=pl.FunctionType.InCore)
def compute_kernel(...):
    ...

@pl.function(type=pl.FunctionType.Orchestration)
def task_graph(...):
    ...
函数类型 说明 典型用途
Opaque 未指定上下文(默认) 独立函数
InCore AICore 计算内核 Load/compute/store 模式
Orchestration 主机端协调器 创建张量、调度 InCore 任务

@pl.program

将多个函数组成可编译的程序:

@pl.program
class MyProgram:
    @pl.function(type=pl.FunctionType.InCore)
    def kernel(self, ...):
        ...

    @pl.function(type=pl.FunctionType.Orchestration)
    def main(self, ...):
        result = self.kernel(...)   # 跨函数调用
        return result

规则:

  • 每个方法必须有 self 作为第一个参数(从 IR 中去除)
  • 跨函数调用使用 self.method_name(...)
  • 装饰后的类成为 ir.Program,不是 Python 类

@pl.jit 家族

@pl.jit 装饰器让你直接以普通 Python 函数的形式编写内核,首次调用时 被 specialize 成 @pl.program 源码(无需类边界)。五个变体分别对应 一种 IR 函数类型,让一个程序可以跨越 host、chip 与 core 三个层级:

装饰器 对应 IR 用途
@pl.jit FunctionType.Orchestration 芯片级入口——分发 InCore 工作的顶层内核
@pl.jit.host level=HOST, role=Orchestrator HOST 级入口——分布式(L3+)程序中分配窗口缓冲、按 rank 分发芯片级 orchestrator
@pl.jit.incore FunctionType.InCore 独立的 InCore 子函数(接受 level= 指定具体层级)
@pl.jit.inline FunctionType.Inline 在每个调用点由 InlineFunctions pass 展开的辅助函数
@pl.jit.opaque FunctionType.Opaque 独立 IR 函数,可包裹编排循环和 pl.at 作用域

子函数依赖(.incore / .inline / .opaque)会从入口函数体自动发现; 用户只需按名字调用。@pl.jit.host 入口还会额外发现 @pl.jit 芯片级 编排依赖,因此完整的分布式程序可以不依赖任何 @pl.program 类来编写:

import pypto.language as pl
import pypto.language.distributed as pld

@pl.jit.inline
def reduce_step(local, peer, out): ...

@pl.jit
def chip_orch(
    inp: pl.Tensor, out: pl.Out[pl.Tensor],
    data: pl.InOut[pld.DistributedTensor], peer: pl.Scalar[pl.INT32],
):
    return reduce_step(inp, peer, out)   # 自动发现的子函数

@pl.jit.host
def host_orch(
    inputs: pl.Tensor[[2, 1, 256], pl.FP32],
    outputs: pl.Out[pl.Tensor[[2, 1, 256], pl.FP32]],
):
    data_buf = pld.alloc_window_buffer(256 * pl.FP32.get_byte())
    for r in pl.range(pld.world_size()):
        data = pld.window(data_buf, [1, 256], dtype=pl.FP32)
        chip_orch(inputs[r], outputs[r], data, (r + 1) % pld.world_size(),
                  device=r)            # device= 按 rank 分发
    return outputs

@pl.jit.host 拒绝 level=(HOST 是隐式的),specialize 成 @pl.function(level=pl.Level.HOST, role=pl.Role.Orchestrator)。 普通的 @pl.jit 入口不会自动发现其他 @pl.jit 入口——只有 .host 会跨越芯片边界,以防两个无关的顶层内核被悄悄折叠成同一个程序。

默认情况下编译器会为你插入 AUTO 运行时 scope(PTO2_SCOPE)。 若要用 with pl.scope() 手动放置,传入 auto_scope=False

@pl.jit(auto_scope=False)              # Orchestration 入口
def orchestrator(a: pl.Tensor, b: pl.Tensor, out: pl.Out[pl.Tensor]):
    with pl.scope():
        out = tile_add(a, b, out)
    return out

@pl.jit.host(auto_scope=False)         # HOST orchestrator
def host_orch(...): ...

@pl.jit.inline(auto_scope=False)       # inline 子函数
def layer(...):
    with pl.scope():                   # 内联后落入调用方
        ...

auto_scope=False 在 Orchestration 入口(@pl.jit)、HOST orchestrator(@pl.jit.host)和 inline 子函数(@pl.jit.inline)上 被接受——inline 函数体会被拼接进调用方,手放的 scope 因此落入调用方 (入口通常也应设 auto_scope=False;入口 True + inline False 合法, 只是手放 scope 会嵌套在编译器 AUTO scope 之内)。.incore / .opaque 仍会拒绝它——它们外提为独立 kernel。它会 specialize 成 @pl.function(..., auto_scope=False)——具体的 scope 放置语义见 MaterializeRuntimeScopes pass

@pl.inline

定义一个在每个调用点展开其函数体的函数(程序中不会有单独的函数):

@pl.inline
def normalize(x: pl.Tensor[[64], pl.FP32]) -> pl.Tensor[[64], pl.FP32]:
    result: pl.Tensor[[64], pl.FP32] = pl.mul(x, 2.0)
    return result

@pl.program
class MyProgram:
    @pl.function
    def main(self, x: pl.Tensor[[64], pl.FP32]) -> pl.Tensor[[64], pl.FP32]:
        y: pl.Tensor[[64], pl.FP32] = normalize(x)  # 函数体在此处内联
        return y

外部函数调用

独立的 @pl.function 可以在 @pl.program 内被调用。它会作为单独的函数添加到程序中:

@pl.function
def softmax(x: pl.Tensor[[64], pl.FP32]) -> pl.Tensor[[64], pl.FP32]:
    ...

@pl.program
class Model:
    @pl.function
    def main(self, x: pl.Tensor[[64], pl.FP32]) -> pl.Tensor[[64], pl.FP32]:
        y: pl.Tensor[[64], pl.FP32] = softmax(x)  # 调用外部函数
        return y

InCore 作用域

将代码区域标记为 InCore 执行,无需创建单独的函数:

with pl.at(level=pl.Level.CORE_GROUP):
    y: pl.Tensor[[64], pl.FP32] = pl.add(x, x)

如需为 ExpandMixedKernel Pass 指定跨核 split 模式,使用 pl.split(...)

# InCore + split 提示:
with pl.at(level=pl.Level.CORE_GROUP,
           optimizations=[pl.split(pl.SplitMode.UP_DOWN)]):
    y: pl.Tensor[[64], pl.FP32] = pl.add(x, x)

如需指定自动跨核 pipe 的槽位数(环深),使用 pl.cross_core_slot(...)。它只决定数据 通道的大小,而不划分计算,因此与 split 模式相互独立,二者可自由组合:

# 把环收缩到 4 槽,以腾出 buffer 空间给更大的 tile:
with pl.at(level=pl.Level.CORE_GROUP,
           optimizations=[pl.split(pl.SplitMode.UP_DOWN),
                          pl.cross_core_slot(slot_num=4)]):
    y: pl.Tensor[[64], pl.FP32] = pl.add(x, x)

省略该条目时沿用默认值(单向生效时 8 槽,双向时每方向 4 槽)。

内存与数据搬运

内存层次结构

DDR(片外,全局内存)
 ├── Vec(统一缓冲区,片上)         ← pl.load() / pl.store()
 │    └── 计算(向量运算)
 ├── Mat(L1 缓冲区)               ← pl.load(..., target_memory=pl.Mem.Mat)
 │    ├── Left(L0A)               ← pl.move(..., target_memory=pl.Mem.Left)
 │    └── Right(L0B)              ← pl.move(..., target_memory=pl.Mem.Right)
 │         └── Acc(L0C)           ← pl.matmul() 结果
 │              └── DDR             ← pl.store()

内存空间(MemorySpace,简称 Mem)

空间 枚举 说明
DDR pl.Mem.DDR 片外全局内存(Tensor 参数)
Vec pl.Mem.Vec 统一向量缓冲区(pl.load 默认目标)
Mat pl.Mem.Mat L1 矩阵缓冲区
Left pl.Mem.Left L0A —— 矩阵乘法左操作数
Right pl.Mem.Right L0B —— 矩阵乘法右操作数
Acc pl.Mem.Acc L0C —— 矩阵乘法累加器
Bias pl.Mem.Bias 偏置缓冲区(AIC 核心)

兼容性说明: pl.Mempl.MemorySpace 的简写别名,两者等价,均可使用。

数据搬运操作

tile = pl.load(tensor, [0, 0], [64, 64])                              # DDR → Vec
tile_l1 = pl.load(tensor, [0, 0], [32, 32], target_memory=pl.Mem.Mat)  # DDR → Mat
tile_l0a = pl.move(tile_l1, target_memory=pl.Mem.Left)                # Mat → Left
out = pl.store(tile, [0, 0], output)                                  # Tile → DDR

模式:矩阵乘法(DDR → Mat → Left/Right → Acc → DDR)

a_l1 = pl.load(a, [0, 0], [32, 32], target_memory=pl.Mem.Mat)
b_l1 = pl.load(b, [0, 0], [32, 32], target_memory=pl.Mem.Mat)
a_l0a = pl.move(a_l1, target_memory=pl.Mem.Left)
b_l0b = pl.move(b_l1, target_memory=pl.Mem.Right)
c_acc = pl.matmul(a_l0a, b_l0b)                     # 结果 → Acc
out = pl.store(c_acc, [0, 0], output)      # Acc → DDR

编译

ir.compile()

from pypto import ir
from pypto.backend import BackendType

output_dir = ir.compile(
    program,
    output_dir=None,                           # 为 None 时自动生成
    strategy=ir.OptimizationStrategy.Default,  # 或 DebugTileOptimization
    dump_passes=True,                          # 将 IR 快照写入 output_dir/passes_dump/
    backend_type=BackendType.Ascend910B,
)
参数 选项 说明
program ir.Program 必填,待编译的程序对象(来自 @pl.program 等)
strategy OptimizationStrategy.DefaultDebugTileOptimization Default = 完整 tensor 导向流水线。DebugTileOptimization = 仅用于调试的 PTO tile 流水线,不包含 tensor-only pass
backend_type BackendType.Ascend910BBackendType.Ascend950 Pass 与代码生成的目标硬件(从 pypto.backend 导入 BackendType
dump_passes True/False True 时在每个 pass 后将 IR 快照写入 <output_dir>/passes_dump/(默认 True
skip_ptoas True/False 跳过 ptoas;只生成原始 .pto(MLIR),不生成已编译的 C++ 包装代码(默认 False
output_dir 路径或 None None 时使用 <base>/<program_name>_<timestamp>,其中 <base> 取自 PYPTO_PROG_BUILD_DIR 环境变量,未设置时为 build_output;目录按需创建
verification_level Noneir.VerificationLevel.NONEBASIC None 表示使用默认(BASIC,或由环境变量 PYPTO_VERIFY_LEVEL 覆盖);否则显式指定校验级别

优化流水线

Default 策略按顺序运行以下 pass:

  1. UnrollLoops —— 展开循环迭代
  2. CtrlFlowTransform —— 将控制流改写为结构化 IR
  3. ConvertToSSA —— 转换为静态单赋值形式
  4. FlattenCallExpr —— 展平嵌套函数调用
  5. OutlineHierarchyScopes —— 提取 hierarchy 作用域
  6. OutlineIncoreScopes —— 将 InCore 作用域提取为独立函数
  7. OutlineClusterScopes —— 提取 cluster 作用域
  8. ConvertTensorToTileOps —— 将张量操作转换为 tile 操作
  9. FlattenTileNdTo2D —— 将 ND tile 操作规范化为 2D
  10. InferTileMemorySpace —— 推断 tile 内存空间
  11. ResolveBackendOpLayouts —— 修复 backend 受限的 tile 布局
  12. ExpandMixedKernel —— 在需要时拆分 mixed kernel
  13. InitMemRef —— 分配内存空间并插入缓冲区分配
  14. MemoryReuse —— 共享生命周期不重叠的缓冲区
  15. AllocateMemoryAddr —— 分配具体内存地址

JITFunction.compile()(用于 @pl.jit 内核)

@pl.jit 内核在 kernel(*args) 一次调用中完成 specialize + compile + dispatch 三步。当你需要把 compile 与 runtime 阶段分开——自己用 ChipWorker.run / ChipWorker.register 驱动执行,查看 compiled.output_dir 下的产物,或者做 AOT codegen 验证——调用 JITFunction.compile(*sample_args) 即可拿到底层 CompiledProgram,不会触发 device dispatch:

@pl.jit
def my_kernel(x, w, out): ...

# Stage 1:只编译,不上设备
compiled = my_kernel.compile(sample_x, sample_w, sample_out)
print("artifacts in:", compiled.output_dir)

# Stage 2:用新的 worker API 显式驱动 runtime
from pypto.runtime import ChipWorker, RunConfig

worker = ChipWorker(config=RunConfig(platform="a2a3sim"))
w_dev = worker.alloc_tensor(sample_w.shape, sample_w.dtype, init=sample_w)
handle = worker.register(compiled)
for batch in stream:
    handle(batch.x, w_dev, batch.out)
worker.free_tensor(w_dev)
worker.close()
  • compile()__call__config=RunConfig(...) 的处理一致: compile-side 参数(strategydump_passes、diagnostics 等)会转发到 ir.compile(),runtime-side 字段(device_id、DFX flag)在这里不生效 ——它们只影响 dispatch,不影响 compile 产物。
  • 返回的 CompiledProgram 就是 JIT 缓存里那一个;同一 specialization key 的 kernel(*args)kernel.compile(*args) 后续调用都会命中缓存,返回完全相同 的对象。
  • CompiledProgram 暴露完整的抽取接口 —— chip_callableruntime_nameruntime_configbuild_orch_argsbuild_call_configoutput_dirplatformoutput_indices —— 因此需要直接操纵 simpler.worker.Worker 的 harness 也可以直接用 JIT 内核,无需再包一个 @pl.program wrapper。

调试

使用 node.as_python() 查看函数或程序的 IR。传入 concise=True 可省略中间类型标注以获得更清晰的输出。编译时设置 dump_passes=True,可在输出目录下的 passes_dump/ 中得到各优化阶段的 IR 快照。