tilegym-adding-cutile-kernel
NVIDIA/skills
Добавить в TileGym новый оператор ядра cuTile для графического процессора, включая регистрацию диспетчеризации, реализацию бэкэнда, экспорт, тесты и тесты производительности.
...Расширить всеДобавление ядра cuTile в TileGym
Пошаговый рабочий процесс добавления нового оператора (например, my_op) с бэкендом cuTile.
Правила выполнения
НЕОБХОДИМО строго соблюдать следующие правила:
- Перед написанием любого кода используйте 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(например, по shape/dtype). - Используйте
@triton.testing.perf_report([...])вbench_my_op(...); внутри функции bench: проверьте корректность с помощью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 и размера задачи
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
Копировать





Дом
