tilegym-adding-cutile-kernel
NVIDIA/skills
在 TileGym 中新增一個 cuTile GPU 核心運算子,涵蓋調度註冊、後端實作、匯出、測試及效能測試。
...展開全部將 cuTile 核心新增至 TileGym
使用 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([...]); 在 benchmark 函式內部:使用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
```
所有檔案
5 個檔案安裝 tilegym-adding-cutile-kernel
請下載並將技能檔案解壓縮至您的 .claude/skills/ 目錄中。
下載 ZIP複製儲存庫並將技能檔案複製到您的專案中。
git clone https://github.com/NVIDIA/skills/tree/main/skills/tilegym-adding-cutile-kernel # Copy SKILL.md to your .claude/skills/ directory
複製





首頁
