tilegym-adding-cutile-kernel
NVIDIA/skills
向 TileGym 添加一个新的 cuTile GPU 内核运算符,内容涵盖调度注册、后端实现、导出、测试和基准测试。
...展开全部向 TileGym 添加 cuTile 内核
使用 cuTile 后端添加新运算符(例如my_op)的端到端工作流。
执行规则
必须严格遵守以下规则:
- 在编写任何代码之前,请使用 TodoWrite 创建以下检查清单
- 按顺序执行步骤——切勿跳过或合并步骤
- 完成每个待办事项后标记为
“已完成”,开始时标记为“进行中” - 如果某个步骤不适用(例如,没有 cuTile 实现),请在标记为
“已完成”时附上说明,切勿无声跳过 - 每个步骤必须导致文件写入或明确的跳过决定——不得默认省略
说明
必须在开始时将此检查清单复制到 TodoWrite 中:
- [ ] 步骤 1:在 ops.py 中注册调度接口
- [ ] 步骤 2:实现 cuTile 后端
- [ ] 步骤 3:在 __init__.py 中注册(cutile)
- [ ] 步骤 4:添加测试
- [ ] 步骤 5:将基准测试添加到 tests/benchmark
- [ ] 步骤 6:验证(运行 pytest + lint)
步骤 1:注册调度接口
文件:src/tilegym/ops/ops.py
添加一个@dispatch函数——这是所有后端的唯一入口点。
@dispatch(
"my_op",
)
def my_op(
input: torch.Tensor,
out: Optional[torch.Tensor] = None,
**kwargs: Any,
):
"""
my_op 的描述。
参数:
input:输入张量
out:可选的预分配输出张量
**kwargs:用于后端特定配置的附加参数
返回值:
torch.Tensor
"""
raise NotImplementedError(f"my_op 尚未在 {get_current_backend()} 上实现")
关键规则:
- 函数体仅抛出
NotImplementedError - 包含
**kwargs以指定后端特有的参数
参考:请参阅src/tilegym/ops/ops.py中的现有操作(例如silu_and_mul、softmax)
步骤 2:实现 cuTile 后端
文件:src/tilegym/ops/cutile/my_op.py
文件结构遵循以下模板:
import torch
import cuda.tile as ct
from tilegym.backend import register_impl
@ct.kernel
def my_op_kernel_ct(x, output, n_elements: ct.Constant[int], BLOCK_SIZE: ct.Constant[int]):
bid = ct.bid(0)
indices = bid * BLOCK_SIZE + ct.arange(0, BLOCK_SIZE)
x_val = ct.gather(x, indices)
# ... 计算 ...
ct.scatter(output, indices, result)
@register_impl("my_op", backend="cutile")
def my_op(input: torch.Tensor, out: torch.Tensor = None, **kwargs) -> torch.Tensor:
n = input.numel()
if out is None:
out = torch.empty_like(input)
grid = ((n + 1023) // 1024,)
ct.launch(stream, grid, kernel, (some args, ...))
return out
参考:src/tilegym/ops/cutile/silu_and_mul.py
步骤 3:在__init__.py中进行注册(至关重要)
如果漏掉这一步,cuTile 后端实现将无法被加载。
文件:src/tilegym/ops/cutile/__init__.py
在if is_backend_available("cutile"):代码块内添加(按字母顺序排列):
from . import my_op
并在函数导入部分:
from .my_op import my_op
并将“my_op”添加到__all__ 中。
步骤 4:添加测试
文件:tests/ops/test_my_op.py
重要提示:请务必从tilegym.ops 导入,切勿从tilegym.ops.cutile.my_op 导入。
import pytest
import torch
from tilegym.backend import is_backend_available, set_backend
from .. import common
_backends = ["cutile"]
class Test_MY_OP(common.PyTestCase):
@staticmethod
def reference(input):
"""使用 PyTorch 的参考实现。"""
return torch.some_reference(input)
@pytest.mark.parametrize("shape, dtype", [
((1024,), torch.float16),
((1024, 512), torch.float32),
((64, 64, 64), torch.bfloat16),
])
@pytest.mark.parametrize("backend", _backends)
def test_op(self, shape, dtype, backend, arch):
if backend == "cutile" and not is_backend_available("cutile"):
pytest.skip("Cutile 后端不可用")
try:
set_backend(backend)
except Exception as e:
pytest.skip(f"不支持该后端:{e}")
self.setUp()
from tilegym.ops import my_op
A = torch.randn(*shape, dtype=dtype, device="cuda")
self.assertCorrectness(
my_op, self.reference, {"input": A},
atol=1e-3, rtol=1e-3,
)
关键模式:
_backends = ["cutile"]test_op:使用set_backend(backend)并配合 try-except 语句,调用self.setUp()
参考:tests/ops/test_silu_and_mul.py
以下是常见错误。
1. 缺少 _backends 列表(在类内部)
2. test_op / test_op_xxx — 缺少 @pytest.mark.parametrize("backend", _backends)、backend 参数,以及 tilegym.is_backend_available / tilegym.set_backend 模式
步骤 5:将基准测试添加到 tests/benchmark 目录
文件:tests/benchmark/bench_my_op.py
来自 benchmark_rules.md 的关键规则:
- 通过
tilegym.ops.my_op(a, b, ..., backend=backend)调用运算 — 请勿使用set_backend。 - 定义
ALL_BACKENDS(至少包含cutile和torch),并通过get_supported_backends()进行筛选。 - 实现
reference_my_op(...)并进行注册:register_impl("my_op", "torch")(reference_my_op)。 - 使用
create_benchmark_config()构建triton.testing.Benchmark配置(例如按形状/数据类型)。 - 在
bench_my_op(...)中使用@triton.testing.perf_report([...]); 在基准测试函数内部:使用torch.testing.assert_close(fn(), ref(), ...)进行正确性检查,随后执行ms = triton.testing.do_bench(fn)(或do_bench_cudagraph),计算 GB/s 或 TFLOPS,并返回该指标。 - 入口点:
if __name__ == "__main__": bench_my_op.run(print_data=True)。
模板结构:
import torch
import triton
import triton.testing
import tilegym
from tilegym.backend import is_backend_available, register_impl
ALL_BACKENDS = [
("cutile", "cuTile", ("orange", "-")) if is_backend_available("cutile") else None,
("torch", "PyTorch", ("green", "-")),
]
def get_supported_backends():
return [p for p in ALL_BACKENDS if p is not None]
def reference_my_op(input: torch.Tensor, out: torch.Tensor = None, **kwargs):
"""使用 PyTorch 的参考实现。"""
...
register_impl("my_op", "torch")(reference_my_op)
def create_benchmark_config(datatype, ...):
available_backends = get_supported_backends()
if not available_backends:
return None
backends, names, styles = zip(*available_backends)
return triton.testing.Benchmark(
x_names=["M"], # 或其他维度名称
x_vals=[...],
line_arg="backend",
line_vals=list(backends),
line_names=list(names),
styles=list(styles),
ylabel="GB/s", # 或 TFLOPS
plot_name="my-op-...",
args={"datatype": datatype, ...},
)
@triton.testing.perf_report([
create_benchmark_config(datatype, ...)
for datatype in [torch.float16, torch.float32]
for ... in [...]
])
def bench_my_op(M, backend, datatype, ..., device="cuda"):
x = torch.randn(..., dtype=datatype, device=device)
fn = lambda: tilegym.ops.my_op(x, backend=backend)
ref = lambda: reference_my_op(x)
torch.testing.assert_close(fn(), ref(), rtol=1e-2, atol=1e-2)
ms = triton.testing.do_bench(fn) # 或 do_bench_cudagraph(fn)
# 根据 ms 和问题规模计算指标(例如 GB/s 或 TFLOPS)
return metric
if __name__ == "__main__":
bench_my_op.run(print_data=True)
基准测试图名称:必须包含-TFLOPS或-GBps后缀
- 示例:
plot_name=f"persistent-layer-norm-M{num_rows}-{dtype_name}-GBps"
步骤 6:验证
# 运行测试
pytest tests/ops/test_my_op.py -v
# 运行基准测试(可选)
python tests/benchmark/bench_my_op.py
# 代码检查
pre-commit run -a
---
name: tilegym-adding-cutile-kernel
description: Add a new cuTile GPU kernel operator to TileGym, covering dispatch registration, backend implementation, exports, tests, and benchmarks.
license: CC-BY-4.0 AND Apache-2.0
---
# Adding a cuTile Kernel to TileGym
End-to-end workflow for adding a new operator (e.g., `my_op`) with cuTile backend.
## Execution Rules
**MUST follow these rules strictly:**
1. Use TodoWrite to create the checklist below BEFORE writing any code
2. Execute steps **in order** — do NOT skip ahead or combine steps
3. Mark each todo as `completed` after finishing, `in_progress` when starting
4. If a step is not applicable (e.g., no cuTile impl), mark it `completed` with a note, do NOT silently skip
5. Each step MUST result in a file write or explicit skip decision — no silent omissions
## Instructions
MUST copy this checklist to TodoWrite at the start:
```
- [ ] Step 1: Register dispatch interface in ops.py
- [ ] Step 2: Implement cuTile backend
- [ ] Step 3: Register in __init__.py (cutile)
- [ ] Step 4: Add tests
- [ ] Step 5: Add benchmark to tests/benchmark
- [ ] Step 6: Verify (run pytest + lint)
```
## Step 1: Register dispatch interface
**File**: `src/tilegym/ops/ops.py`
Add a `@dispatch` function — this is the **single entry point** for all backends.
```python
@dispatch(
"my_op",
)
def my_op(
input: torch.Tensor,
out: Optional[torch.Tensor] = None,
**kwargs: Any,
):
"""
Description of my_op.
Args:
input: Input tensor
out: Optional preallocated output tensor
**kwargs: Additional arguments for backend-specific configurations
Returns:
torch.Tensor
"""
raise NotImplementedError(f"my_op is not implemented for {get_current_backend()}")
```
**Key rules:**
- Function body only raises `NotImplementedError`
- Include `**kwargs` for backend-specific parameters
**Reference**: See existing ops in `src/tilegym/ops/ops.py` (e.g., `silu_and_mul`, `softmax`)
## Step 2: Implement cuTile backend
**File**: `src/tilegym/ops/cutile/my_op.py`
The file structure follows this template:
```python
import torch
import cuda.tile as ct
from tilegym.backend import register_impl
@ct.kernel
def my_op_kernel_ct(x, output, n_elements: ct.Constant[int], BLOCK_SIZE: ct.Constant[int]):
bid = ct.bid(0)
indices = bid * BLOCK_SIZE + ct.arange(0, BLOCK_SIZE)
x_val = ct.gather(x, indices)
# ... compute ...
ct.scatter(output, indices, result)
@register_impl("my_op", backend="cutile")
def my_op(input: torch.Tensor, out: torch.Tensor = None, **kwargs) -> torch.Tensor:
n = input.numel()
if out is None:
out = torch.empty_like(input)
grid = ((n + 1023) // 1024,)
ct.launch(stream, grid, kernel, (some args, ...))
return out
```
**Reference**: `src/tilegym/ops/cutile/silu_and_mul.py`
## Step 3: Register in `__init__.py` (CRITICAL)
Missing this step means the cuTile backend implementation never gets loaded.
**File**: `src/tilegym/ops/cutile/__init__.py`
Add inside `if is_backend_available("cutile"):` block (alphabetically):
```python
from . import my_op
```
And in the function import section:
```python
from .my_op import my_op
```
And add `"my_op"` to `__all__`.
## Step 4: Add tests
**File**: `tests/ops/test_my_op.py`
**CRITICAL**: Always import from `tilegym.ops`, NEVER from `tilegym.ops.cutile.my_op`.
```python
import pytest
import torch
from tilegym.backend import is_backend_available, set_backend
from .. import common
_backends = ["cutile"]
class Test_MY_OP(common.PyTestCase):
@staticmethod
def reference(input):
"""Reference implementation using PyTorch."""
return torch.some_reference(input)
@pytest.mark.parametrize("shape, dtype", [
((1024,), torch.float16),
((1024, 512), torch.float32),
((64, 64, 64), torch.bfloat16),
])
@pytest.mark.parametrize("backend", _backends)
def test_op(self, shape, dtype, backend, arch):
if backend == "cutile" and not is_backend_available("cutile"):
pytest.skip("Cutile backend not available")
try:
set_backend(backend)
except Exception as e:
pytest.skip(f"Backend is not supported: {e}")
self.setUp()
from tilegym.ops import my_op
A = torch.randn(*shape, dtype=dtype, device="cuda")
self.assertCorrectness(
my_op, self.reference, {"input": A},
atol=1e-3, rtol=1e-3,
)
```
**Key patterns:**
- `_backends = ["cutile"]`
- `test_op`: use `set_backend(backend)` with try-except, call `self.setUp()`
**Reference**: `tests/ops/test_silu_and_mul.py`
Below is the common errors.
```
1. Missing _backends list (inside class)
2. test_op / test_op_xxx — missing @pytest.mark.parametrize("backend", _backends), backend parameter, and tilegym.is_backend_available / tilegym.set_backend pattern
```
## Step 5: Add benchmark to tests/benchmark
**File**: `tests/benchmark/bench_my_op.py`
**Key rules from benchmark_rules.md:**
- Call the op via `tilegym.ops.my_op(a, b, ..., backend=backend)` — do **not** use `set_backend`.
- Define `ALL_BACKENDS` (include at least `cutile` and `torch`), filter with `get_supported_backends()`.
- Implement `reference_my_op(...)` and register it: `register_impl("my_op", "torch")(reference_my_op)`.
- Use `create_benchmark_config()` to build `triton.testing.Benchmark` configs (e.g. by shape/dtype).
- Use `@triton.testing.perf_report([...])` on `bench_my_op(...)`; inside the bench function: correctness check with `torch.testing.assert_close(fn(), ref(), ...)`, then `ms = triton.testing.do_bench(fn)` (or `do_bench_cudagraph`), compute GB/s or TFLOPS, and return the metric.
- Entry point: `if __name__ == "__main__": bench_my_op.run(print_data=True)`.
Template structure:
```python
import torch
import triton
import triton.testing
import tilegym
from tilegym.backend import is_backend_available, register_impl
ALL_BACKENDS = [
("cutile", "cuTile", ("orange", "-")) if is_backend_available("cutile") else None,
("torch", "PyTorch", ("green", "-")),
]
def get_supported_backends():
return [p for p in ALL_BACKENDS if p is not None]
def reference_my_op(input: torch.Tensor, out: torch.Tensor = None, **kwargs):
"""Reference implementation using PyTorch."""
...
register_impl("my_op", "torch")(reference_my_op)
def create_benchmark_config(datatype, ...):
available_backends = get_supported_backends()
if not available_backends:
return None
backends, names, styles = zip(*available_backends)
return triton.testing.Benchmark(
x_names=["M"], # or other dimension names
x_vals=[...],
line_arg="backend",
line_vals=list(backends),
line_names=list(names),
styles=list(styles),
ylabel="GB/s", # or TFLOPS
plot_name="my-op-...",
args={"datatype": datatype, ...},
)
@triton.testing.perf_report([
create_benchmark_config(datatype, ...)
for datatype in [torch.float16, torch.float32]
for ... in [...]
])
def bench_my_op(M, backend, datatype, ..., device="cuda"):
x = torch.randn(..., dtype=datatype, device=device)
fn = lambda: tilegym.ops.my_op(x, backend=backend)
ref = lambda: reference_my_op(x)
torch.testing.assert_close(fn(), ref(), rtol=1e-2, atol=1e-2)
ms = triton.testing.do_bench(fn) # or do_bench_cudagraph(fn)
# Compute metric (e.g. GB/s or TFLOPS) from ms and problem size
return metric
if __name__ == "__main__":
bench_my_op.run(print_data=True)
```
**Benchmark Plot Names**: Must include `-TFLOPS` or `-GBps` suffix
- Example: `plot_name=f"persistent-layer-norm-M{num_rows}-{dtype_name}-GBps"`
## Step 6: Verify
```bash
# Run tests
pytest tests/ops/test_my_op.py -v
# Run benchmark (optional)
python tests/benchmark/bench_my_op.py
# Lint
pre-commit run -a
```





首页
