PyPTO 语言指南¶
pypto.language(pl)模块的完整参考。
类型系统¶
数据类型(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] —— 片上内存缓冲区(默认统一缓冲区)。
pl.Scalar[dtype] —— 单个标量值。
张量布局(TensorLayout)¶
pl.Tensor[...] annotation 写 runtime 行优先 shape,不写 layout 标记。layout 是 IR 内部概念,由派生/消费视图的 op 推导,不需要在 annotation 上表达。
为什么弃用
pl.Tensor[..., pl.DN]。 layout-only 简写迫使用户脑子里同时持有两套坐标系(IR 逻辑后视图 shape 与 runtime 行优先 shape)—— 恰恰是 RFC #1300 想要消除的歧义。改用:去掉 layout 标记,写 runtime shape —— matmul B^T 场景给pl.matmul传b_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...>] = src 是 pl.assemble 的语法糖:
该语法糖仅在 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() 相同,但迭代可以并行执行:
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(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 执行,无需创建单独的函数:
如需为 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.Mem是pl.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.Default、DebugTileOptimization |
Default = 完整 tensor 导向流水线。DebugTileOptimization = 仅用于调试的 PTO tile 流水线,不包含 tensor-only pass |
backend_type |
BackendType.Ascend910B、BackendType.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 |
None、ir.VerificationLevel.NONE、BASIC |
None 表示使用默认(BASIC,或由环境变量 PYPTO_VERIFY_LEVEL 覆盖);否则显式指定校验级别 |
优化流水线¶
Default 策略按顺序运行以下 pass:
- UnrollLoops —— 展开循环迭代
- CtrlFlowTransform —— 将控制流改写为结构化 IR
- ConvertToSSA —— 转换为静态单赋值形式
- FlattenCallExpr —— 展平嵌套函数调用
- OutlineHierarchyScopes —— 提取 hierarchy 作用域
- OutlineIncoreScopes —— 将 InCore 作用域提取为独立函数
- OutlineClusterScopes —— 提取 cluster 作用域
- ConvertTensorToTileOps —— 将张量操作转换为 tile 操作
- FlattenTileNdTo2D —— 将 ND tile 操作规范化为 2D
- InferTileMemorySpace —— 推断 tile 内存空间
- ResolveBackendOpLayouts —— 修复 backend 受限的 tile 布局
- ExpandMixedKernel —— 在需要时拆分 mixed kernel
- InitMemRef —— 分配内存空间并插入缓冲区分配
- MemoryReuse —— 共享生命周期不重叠的缓冲区
- 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 参数(strategy、dump_passes、diagnostics 等)会转发到ir.compile(),runtime-side 字段(device_id、DFX flag)在这里不生效 ——它们只影响 dispatch,不影响 compile 产物。- 返回的
CompiledProgram就是 JIT 缓存里那一个;同一 specialization key 的kernel(*args)和kernel.compile(*args)后续调用都会命中缓存,返回完全相同 的对象。 - 该
CompiledProgram暴露完整的抽取接口 ——chip_callable、runtime_name、runtime_config、build_orch_args、build_call_config、output_dir、platform、output_indices—— 因此需要直接操纵simpler.worker.Worker的 harness 也可以直接用 JIT 内核,无需再包一个@pl.programwrapper。
调试¶
使用 node.as_python() 查看函数或程序的 IR。传入 concise=True 可省略中间类型标注以获得更清晰的输出。编译时设置 dump_passes=True,可在输出目录下的 passes_dump/ 中得到各优化阶段的 IR 快照。