PyPTO 循环边界控制pypto.is_loop_begin 判断循环首迭代的编程范式与实践【免费下载链接】pyptoPyPTO发音: pai p-t-oParallel Tensor/Tile Operation编程范式。项目地址: https://gitcode.com/cann/pypto导读pypto.is_loop_begin是 PyPTOParallel Tensor/Tile Operation 编程范式提供的循环控制流 API用于在张量/分块Tile算子内核中判断当前迭代是否为循环的开始从而在循环体内按首迭代/其余迭代分支执行不同计算逻辑。本文基于官方 API 文档结合仓库源码 python/pypto/_controller.py 与测试用例完整讲解该接口的函数原型、参数约束、pypto.cond包装规则、底层实现原理及可复现的实战示例帮助读者在编写动态循环内核如分块累加、首块初始化等场景时正确使用循环边界判断。产品支持情况pypto.is_loop_begin在以下产品上受支持产品系列支持情况Ascend 950PR / Ascend 950DT支持Atlas A3 训练系列产品 / Atlas A3 推理系列产品支持Atlas A2 训练系列产品 / Atlas A2 推理系列产品支持该支持范围与同目录下的 pypto-is_loop_end 一致两者共同构成循环边界判断能力。功能说明pypto.is_loop_begin用于判断当前迭代是否为循环的开始。它接收当前循环的索引index作为输入返回一个表示是否为循环首迭代的布尔符号标量表达式。典型应用场景包括在循环体内对第一次迭代做额外初始化例如首块清零、首块写入特殊值在分块tiling计算中第一个分块承担不同于后续分块的计算路径配合pypto.is_loop_end对首迭代与末迭代分别施加不同处理中间迭代走常规路径。函数原型is_loop_begin(scalar: SymInt) - SymbolicScalar对应实现位于 python/pypto/_controller.pydef is_loop_begin(scalar: SymbolicScalar) - SymbolicScalar: Determines if the current iteration is the start of loop if not hasattr(scalar, _loop_begin): raise FeError(ValueError(not loop index)) return pypto_impl.IsLoopBegin(scalar, getattr(scalar, _loop_begin))参数说明参数名输入/输出说明scalar输入当前循环的 index即循环迭代器返回的符号标量类型说明scalar的类型为SymInt符号整数通常由pypto.loop/pypto.loop_unroll迭代时产生。从源码看is_loop_begin内部通过hasattr(scalar, _loop_begin)校验该标量是否携带循环边界标记只有由循环迭代器产出的符号标量才具备该属性详见下文底层实现原理。返回值说明返回一个符号标量表达式SymbolicScalar其求值结果为布尔值当前迭代是循环的开始即 index 等于循环起始值时为真True否则为假False。该返回值通常直接用作if分支条件或经 pypto.cond 包装后作为条件表达式使用。约束说明使用pypto.is_loop_begin必须满足以下约束scalar必须是循环迭代器返回的符号标量即通过for idx in pypto.loop(...)或pypto.loop_unroll(...)迭代产出的索引变量。源码中循环迭代器在_LoopFunction.Iterator.__next__中会给索引标量动态挂载_loop_begin属性见 python/pypto/_controller.py。如果不是循环索引将抛出ValueError异常源码中通过if not hasattr(scalar, _loop_begin): raise FeError(ValueError(not loop index))实现。传入任意不携带该标记的普通符号标量或常量都会触发此异常。未使用装饰器时条件表达式需要用pypto.cond包装当函数未使用pypto.frontend.jit或pypto.frontend.function装饰器修饰时is_loop_begin的结果必须放在pypto.cond(...)中才能作为if分支的条件使用装饰器后则无需包装可直接书写if pypto.is_loop_begin(idx):。这一规则与 pypto-cond 文档中说明的条件表达式使用方式一致。调用示例以下示例完整继承自官方文档分别展示两种调用形态。未使用装饰器需用 pypto.cond 包装# 未使用装饰器需要用pypto.cond包装条件表达式 def kernel(): ... for idx in pypto.loop(0, 10, 1): if pypto.cond(pypto.is_loop_begin(idx)): ...使用装饰器无需 pypto.cond 包装# 使用装饰器无需pypto.cond包装 pypto.frontend.jit def kernel(): ... for idx in pypto.loop(0, 10, 1): if pypto.is_loop_begin(idx): ...使用 pypto.frontend.function 的等价写法在单元测试 python/tests/ut/interface/test_pto_loop.py 中可以看到在pypto.function(MAIN, a, b)上下文内使用pypto.cond包装is_loop_begin/is_loop_end的典型写法with pypto.function(MAIN, a, b): pypto.set_vec_tile_shapes(64, 64) for idx in pypto.loop(128, unroll_list[1, 4]): a_tile a[idx * 64:(idx 1) * 64, :] if pypto.cond(pypto.is_loop_begin(idx)): a_tile a_tile 1 elif pypto.cond(pypto.is_loop_end(idx)): a_tile a_tile 2 b[idx * 64:, 0:] a_tile 1端到端可运行示例首迭代分支计算仓库的 ST 测试 python/tests/st/interface/test_is_loop_begin_new_front.py 提供了一个完整的可运行内核在双重循环中当外层索引b_idx处于循环首迭代时执行add否则执行mul并将结果写回输出张量pypto.frontend.jit() def dyn_loop_with_loop_begin( in_tensor: pypto.Tensor([pypto.STATIC, pypto.STATIC, pypto.STATIC, pypto.STATIC], pypto.DT_FP32), out_tensor: pypto.Tensor([pypto.STATIC, pypto.STATIC, pypto.STATIC, pypto.STATIC], pypto.DT_FP32), ): pypto.set_vec_tile_shapes(1, 1, 64, 64) for b_idx in pypto.loop(B, nameb_loop, idx_nameb_idx): for s_idx in pypto.loop(S, names_loop, idx_names_idx): a0 pypto.view(in_tensor, [1, 1, N1, D], [b_idx, s_idx, 0, 0]) if pypto.is_loop_begin(b_idx): a1 pypto.add(a0, 1.0) pypto.assemble(a1, [b_idx, s_idx, 0, 0], out_tensor) else: a1 pypto.mul(a0, 1.0) pypto.assemble(a1, [b_idx, s_idx, 0, 0], out_tensor)测试中的 golden 校验逻辑直观说明了is_loop_begin的语义仅第一个 batch 的数据被加 1其余 batch 保持不变output_golden input_torch.clone().cpu() output_golden[0:1, :, :, :] output_golden[0:1, :, :, :] 1 assert torch.allclose(output_result, output_golden, atol1e-5)运行该测试时需准备 NPU 环境测试通过TILE_FWK_DEVICE_ID环境变量指定设备号默认 0并依赖torch、torch_npu与pypto运行时。底层实现原理循环索引的边界标记机制is_loop_begin之所以能校验scalar 必须是循环迭代器返回的符号标量源于_LoopFunction.Iterator.__next__在产出每个索引标量时动态挂载边界属性的机制python/pypto/_controller.pyclass _LoopFunction: class Iterator: def __next__(self): scalar self._iter.__next__() setattr(scalar, _loop_begin, self._begin) setattr(scalar, _loop_end, self._end) CompileState.bump_atomic_scope_iter() return scalar def __init__(self, name, loop_name, loop_range, unroll_list, submit_before_loop, parallel): loop_range loop_range.base() self._base pypto_impl.RecordLoopFunc(...) self._begin loop_range.Begin() self._end loop_range.End()每次for idx in pypto.loop(...)迭代时索引标量都会被附加上_loop_begin循环起始值和_loop_end循环结束值两个属性。因此is_loop_begin只需检查hasattr(scalar, _loop_begin)即可判断传入参数是否为合法的循环索引——这也解释了为什么传入普通标量会抛出ValueError(not loop index)。符号表达式构造is_loop_begin最终通过pypto_impl.IsLoopBegin(scalar, getattr(scalar, _loop_begin))构造一个符号标量表达式将当前迭代 index与循环起始值绑定为比较关系该表达式在编译/求值阶段解析为布尔结果。PIL 解释器中的语义实现在前端 PILProgram 级中间表示解释器 python/pypto/pil/ops.py 中is_loop_begin与is_loop_end的语义被实现为基于loop_stack的标量比较impl(pypto.is_loop_begin) def is_loop_begin_impl(ctx: BuildContext, scalar: SymbolicScalar): start, _, _ ctx.loop_stack[-1] assert isinstance(start, (SymbolicScalar, int)), is_loop_begin() must be called in a pypto.loop return scalar start impl(pypto.is_loop_end) def is_loop_end_impl(ctx: BuildContext, scalar: SymbolicScalar): _, end, step ctx.loop_stack[-1] assert isinstance(end, (SymbolicScalar, int)), is_loop_end() must be called in a pypto.loop assert isinstance(step, (SymbolicScalar, int)), is_loop_end() must be called in a pypto.loop return scalar step end可见is_loop_begin(idx)等价于比较idx start循环起始值对称地is_loop_end(idx)等价于比较idx step end逼近循环结束值实现细节可参考 pypto-is_loop_end若在pypto.loop之外调用会触发断言失败。loop_stack在解释器上下文 python/pypto/pil/pir.py 中维护用于记录嵌套循环的(start, end, step)元组。pypto.cond 的配合机制pypto.cond 在 python/pypto/_controller.py 中实现为pypto_impl.RecordIfBranch(to_sym(scalar), filename, line)它将条件表达式记录为计算图中的一个 if 分支节点同时记录源码位置用于调试。当函数未经过 JIT/function 装饰器时Python 原生if无法被框架捕获为图分支因此必须显式调用pypto.cond来告诉框架这是一个需要记录的条件分支而经过pypto.frontend.jit或pypto.frontend.function装饰后前端编译器会重写函数体、自动识别if后的符号条件从而可以直接书写if pypto.is_loop_begin(idx):。与其他控制流 API 的关系pypto.is_loop_begin属于 PyPTO 控制流 API 家族与其配套使用的接口包括API功能文档位置pypto.loop创建动态循环产出循环索引符号标量docs/zh/api/tensor_api/controlflow/pypto-loop.mdpypto.loop_unroll创建带展开因子的循环docs/zh/api/tensor_api/controlflow/pypto-loop_unroll.mdpypto.is_loop_end判断当前迭代是否为循环结束docs/zh/api/tensor_api/controlflow/pypto-is_loop_end.mdpypto.cond包装条件表达式以记录 if 分支docs/zh/api/tensor_api/controlflow/pypto-cond.mdpypto.function定义函数上下文以录制计算图docs/zh/api/tensor_api/controlflow/pypto-function.md在分块算子中is_loop_begin与is_loop_end常成对出现首迭代做初始化/前导prolog计算末迭代做收尾epilog计算中间迭代执行常规主体。例如在注意力类内核的 UT 实现python/tests/ut/ir/test_fa_score.py、python/tests/ut/ir/test_fa_score_grad.py以及 FlashAttention 相关实现python/tests/ut/interpreter/_ops/flash_attention_mha_impl.py中均能看到利用is_loop_begin在首个 K/V 分块执行初始化操作的典型模式。总结pypto.is_loop_begin通过循环索引标量携带边界标记 符号比较表达式的设计为分块算子提供了声明式的循环首迭代判断能力。使用时需牢记三点参数必须是pypto.loop/pypto.loop_unroll产出的索引标量非循环索引会触发ValueError未使用 JIT/function 装饰器时务必用pypto.cond包装条件。结合仓库中的 test_is_loop_begin_new_front.py 测试用例开发者可以快速验证并复用到自己的动态循环内核中。【免费下载链接】pyptoPyPTO发音: pai p-t-oParallel Tensor/Tile Operation编程范式。项目地址: https://gitcode.com/cann/pypto创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 PyPTO 循环边界控制：pypto.is_loop_begin 判断循环首迭代的编程范式与实践
PyPTO 循环边界控制pypto.is_loop_begin 判断循环首迭代的编程范式与实践【免费下载链接】pyptoPyPTO发音: pai p-t-oParallel Tensor/Tile Operation编程范式。项目地址: https://gitcode.com/cann/pypto导读pypto.is_loop_begin是 PyPTOParallel Tensor/Tile Operation 编程范式提供的循环控制流 API用于在张量/分块Tile算子内核中判断当前迭代是否为循环的开始从而在循环体内按首迭代/其余迭代分支执行不同计算逻辑。本文基于官方 API 文档结合仓库源码 python/pypto/_controller.py 与测试用例完整讲解该接口的函数原型、参数约束、pypto.cond包装规则、底层实现原理及可复现的实战示例帮助读者在编写动态循环内核如分块累加、首块初始化等场景时正确使用循环边界判断。产品支持情况pypto.is_loop_begin在以下产品上受支持产品系列支持情况Ascend 950PR / Ascend 950DT支持Atlas A3 训练系列产品 / Atlas A3 推理系列产品支持Atlas A2 训练系列产品 / Atlas A2 推理系列产品支持该支持范围与同目录下的 pypto-is_loop_end 一致两者共同构成循环边界判断能力。功能说明pypto.is_loop_begin用于判断当前迭代是否为循环的开始。它接收当前循环的索引index作为输入返回一个表示是否为循环首迭代的布尔符号标量表达式。典型应用场景包括在循环体内对第一次迭代做额外初始化例如首块清零、首块写入特殊值在分块tiling计算中第一个分块承担不同于后续分块的计算路径配合pypto.is_loop_end对首迭代与末迭代分别施加不同处理中间迭代走常规路径。函数原型is_loop_begin(scalar: SymInt) - SymbolicScalar对应实现位于 python/pypto/_controller.pydef is_loop_begin(scalar: SymbolicScalar) - SymbolicScalar: Determines if the current iteration is the start of loop if not hasattr(scalar, _loop_begin): raise FeError(ValueError(not loop index)) return pypto_impl.IsLoopBegin(scalar, getattr(scalar, _loop_begin))参数说明参数名输入/输出说明scalar输入当前循环的 index即循环迭代器返回的符号标量类型说明scalar的类型为SymInt符号整数通常由pypto.loop/pypto.loop_unroll迭代时产生。从源码看is_loop_begin内部通过hasattr(scalar, _loop_begin)校验该标量是否携带循环边界标记只有由循环迭代器产出的符号标量才具备该属性详见下文底层实现原理。返回值说明返回一个符号标量表达式SymbolicScalar其求值结果为布尔值当前迭代是循环的开始即 index 等于循环起始值时为真True否则为假False。该返回值通常直接用作if分支条件或经 pypto.cond 包装后作为条件表达式使用。约束说明使用pypto.is_loop_begin必须满足以下约束scalar必须是循环迭代器返回的符号标量即通过for idx in pypto.loop(...)或pypto.loop_unroll(...)迭代产出的索引变量。源码中循环迭代器在_LoopFunction.Iterator.__next__中会给索引标量动态挂载_loop_begin属性见 python/pypto/_controller.py。如果不是循环索引将抛出ValueError异常源码中通过if not hasattr(scalar, _loop_begin): raise FeError(ValueError(not loop index))实现。传入任意不携带该标记的普通符号标量或常量都会触发此异常。未使用装饰器时条件表达式需要用pypto.cond包装当函数未使用pypto.frontend.jit或pypto.frontend.function装饰器修饰时is_loop_begin的结果必须放在pypto.cond(...)中才能作为if分支的条件使用装饰器后则无需包装可直接书写if pypto.is_loop_begin(idx):。这一规则与 pypto-cond 文档中说明的条件表达式使用方式一致。调用示例以下示例完整继承自官方文档分别展示两种调用形态。未使用装饰器需用 pypto.cond 包装# 未使用装饰器需要用pypto.cond包装条件表达式 def kernel(): ... for idx in pypto.loop(0, 10, 1): if pypto.cond(pypto.is_loop_begin(idx)): ...使用装饰器无需 pypto.cond 包装# 使用装饰器无需pypto.cond包装 pypto.frontend.jit def kernel(): ... for idx in pypto.loop(0, 10, 1): if pypto.is_loop_begin(idx): ...使用 pypto.frontend.function 的等价写法在单元测试 python/tests/ut/interface/test_pto_loop.py 中可以看到在pypto.function(MAIN, a, b)上下文内使用pypto.cond包装is_loop_begin/is_loop_end的典型写法with pypto.function(MAIN, a, b): pypto.set_vec_tile_shapes(64, 64) for idx in pypto.loop(128, unroll_list[1, 4]): a_tile a[idx * 64:(idx 1) * 64, :] if pypto.cond(pypto.is_loop_begin(idx)): a_tile a_tile 1 elif pypto.cond(pypto.is_loop_end(idx)): a_tile a_tile 2 b[idx * 64:, 0:] a_tile 1端到端可运行示例首迭代分支计算仓库的 ST 测试 python/tests/st/interface/test_is_loop_begin_new_front.py 提供了一个完整的可运行内核在双重循环中当外层索引b_idx处于循环首迭代时执行add否则执行mul并将结果写回输出张量pypto.frontend.jit() def dyn_loop_with_loop_begin( in_tensor: pypto.Tensor([pypto.STATIC, pypto.STATIC, pypto.STATIC, pypto.STATIC], pypto.DT_FP32), out_tensor: pypto.Tensor([pypto.STATIC, pypto.STATIC, pypto.STATIC, pypto.STATIC], pypto.DT_FP32), ): pypto.set_vec_tile_shapes(1, 1, 64, 64) for b_idx in pypto.loop(B, nameb_loop, idx_nameb_idx): for s_idx in pypto.loop(S, names_loop, idx_names_idx): a0 pypto.view(in_tensor, [1, 1, N1, D], [b_idx, s_idx, 0, 0]) if pypto.is_loop_begin(b_idx): a1 pypto.add(a0, 1.0) pypto.assemble(a1, [b_idx, s_idx, 0, 0], out_tensor) else: a1 pypto.mul(a0, 1.0) pypto.assemble(a1, [b_idx, s_idx, 0, 0], out_tensor)测试中的 golden 校验逻辑直观说明了is_loop_begin的语义仅第一个 batch 的数据被加 1其余 batch 保持不变output_golden input_torch.clone().cpu() output_golden[0:1, :, :, :] output_golden[0:1, :, :, :] 1 assert torch.allclose(output_result, output_golden, atol1e-5)运行该测试时需准备 NPU 环境测试通过TILE_FWK_DEVICE_ID环境变量指定设备号默认 0并依赖torch、torch_npu与pypto运行时。底层实现原理循环索引的边界标记机制is_loop_begin之所以能校验scalar 必须是循环迭代器返回的符号标量源于_LoopFunction.Iterator.__next__在产出每个索引标量时动态挂载边界属性的机制python/pypto/_controller.pyclass _LoopFunction: class Iterator: def __next__(self): scalar self._iter.__next__() setattr(scalar, _loop_begin, self._begin) setattr(scalar, _loop_end, self._end) CompileState.bump_atomic_scope_iter() return scalar def __init__(self, name, loop_name, loop_range, unroll_list, submit_before_loop, parallel): loop_range loop_range.base() self._base pypto_impl.RecordLoopFunc(...) self._begin loop_range.Begin() self._end loop_range.End()每次for idx in pypto.loop(...)迭代时索引标量都会被附加上_loop_begin循环起始值和_loop_end循环结束值两个属性。因此is_loop_begin只需检查hasattr(scalar, _loop_begin)即可判断传入参数是否为合法的循环索引——这也解释了为什么传入普通标量会抛出ValueError(not loop index)。符号表达式构造is_loop_begin最终通过pypto_impl.IsLoopBegin(scalar, getattr(scalar, _loop_begin))构造一个符号标量表达式将当前迭代 index与循环起始值绑定为比较关系该表达式在编译/求值阶段解析为布尔结果。PIL 解释器中的语义实现在前端 PILProgram 级中间表示解释器 python/pypto/pil/ops.py 中is_loop_begin与is_loop_end的语义被实现为基于loop_stack的标量比较impl(pypto.is_loop_begin) def is_loop_begin_impl(ctx: BuildContext, scalar: SymbolicScalar): start, _, _ ctx.loop_stack[-1] assert isinstance(start, (SymbolicScalar, int)), is_loop_begin() must be called in a pypto.loop return scalar start impl(pypto.is_loop_end) def is_loop_end_impl(ctx: BuildContext, scalar: SymbolicScalar): _, end, step ctx.loop_stack[-1] assert isinstance(end, (SymbolicScalar, int)), is_loop_end() must be called in a pypto.loop assert isinstance(step, (SymbolicScalar, int)), is_loop_end() must be called in a pypto.loop return scalar step end可见is_loop_begin(idx)等价于比较idx start循环起始值对称地is_loop_end(idx)等价于比较idx step end逼近循环结束值实现细节可参考 pypto-is_loop_end若在pypto.loop之外调用会触发断言失败。loop_stack在解释器上下文 python/pypto/pil/pir.py 中维护用于记录嵌套循环的(start, end, step)元组。pypto.cond 的配合机制pypto.cond 在 python/pypto/_controller.py 中实现为pypto_impl.RecordIfBranch(to_sym(scalar), filename, line)它将条件表达式记录为计算图中的一个 if 分支节点同时记录源码位置用于调试。当函数未经过 JIT/function 装饰器时Python 原生if无法被框架捕获为图分支因此必须显式调用pypto.cond来告诉框架这是一个需要记录的条件分支而经过pypto.frontend.jit或pypto.frontend.function装饰后前端编译器会重写函数体、自动识别if后的符号条件从而可以直接书写if pypto.is_loop_begin(idx):。与其他控制流 API 的关系pypto.is_loop_begin属于 PyPTO 控制流 API 家族与其配套使用的接口包括API功能文档位置pypto.loop创建动态循环产出循环索引符号标量docs/zh/api/tensor_api/controlflow/pypto-loop.mdpypto.loop_unroll创建带展开因子的循环docs/zh/api/tensor_api/controlflow/pypto-loop_unroll.mdpypto.is_loop_end判断当前迭代是否为循环结束docs/zh/api/tensor_api/controlflow/pypto-is_loop_end.mdpypto.cond包装条件表达式以记录 if 分支docs/zh/api/tensor_api/controlflow/pypto-cond.mdpypto.function定义函数上下文以录制计算图docs/zh/api/tensor_api/controlflow/pypto-function.md在分块算子中is_loop_begin与is_loop_end常成对出现首迭代做初始化/前导prolog计算末迭代做收尾epilog计算中间迭代执行常规主体。例如在注意力类内核的 UT 实现python/tests/ut/ir/test_fa_score.py、python/tests/ut/ir/test_fa_score_grad.py以及 FlashAttention 相关实现python/tests/ut/interpreter/_ops/flash_attention_mha_impl.py中均能看到利用is_loop_begin在首个 K/V 分块执行初始化操作的典型模式。总结pypto.is_loop_begin通过循环索引标量携带边界标记 符号比较表达式的设计为分块算子提供了声明式的循环首迭代判断能力。使用时需牢记三点参数必须是pypto.loop/pypto.loop_unroll产出的索引标量非循环索引会触发ValueError未使用 JIT/function 装饰器时务必用pypto.cond包装条件。结合仓库中的 test_is_loop_begin_new_front.py 测试用例开发者可以快速验证并复用到自己的动态循环内核中。【免费下载链接】pyptoPyPTO发音: pai p-t-oParallel Tensor/Tile Operation编程范式。项目地址: https://gitcode.com/cann/pypto创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考