diff --git a/benchmark/examples/bench_all_gather.py b/benchmark/examples/bench_all_gather.py
new file mode 100644
index 000000000..f3a996eb8
--- /dev/null
+++ b/benchmark/examples/bench_all_gather.py
@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: MIT
+# Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved.
+
+"""Sample benchmark using iris.bench — all-gather collective."""
+
+import torch
+import iris.bench as bench
+from iris.ccl import Config
+
+
+@bench.register
+@bench.axis("num_ranks", [2, 4, 8])
+@bench.axis("M", [1024, 4096, 16384])
+@bench.axis("N", [1024, 4096])
+@bench.axis("dtype", [torch.float16, torch.bfloat16])
+def all_gather(state, ctx):
+    M, N, dtype = state["M"], state["N"], state["dtype"]
+    world_size = ctx.get_num_ranks()
+
+    inp = ctx.zeros((M, N), dtype=dtype)
+    out = ctx.zeros((world_size * M, N), dtype=dtype)
+    inp.fill_(float(ctx.get_rank() + 1))
+
+    total_bytes = (world_size - 1) * M * N * inp.element_size()
+    state.set_bytes(total_bytes)
+
+    config = Config(use_gluon=False)
+    state.exec(lambda: ctx.ccl.all_gather(out, inp, config=config))
+
+
+if __name__ == "__main__":
+    bench.main()
diff --git a/iris/__init__.py b/iris/__init__.py
index 02f78d428..3ec70efa8 100644
--- a/iris/__init__.py
+++ b/iris/__init__.py
@@ -76,6 +76,7 @@
 from . import experimental
 from . import ops
 from . import tensor_creation
+from . import bench
 from .logging import (
     set_logger_level,
     logger,
@@ -111,6 +112,7 @@
     "experimental",
     "ops",
     "tensor_creation",
+    "bench",
     "set_logger_level",
     "logger",
     "DEBUG",
diff --git a/iris/bench/__init__.py b/iris/bench/__init__.py
new file mode 100644
index 000000000..39a28a8e1
--- /dev/null
+++ b/iris/bench/__init__.py
@@ -0,0 +1,136 @@
+# SPDX-License-Identifier: MIT
+# Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved.
+
+"""
+iris.bench — GPU Benchmarking Framework
+
+A declarative benchmarking framework for iris that eliminates boilerplate.
+Write ~25 lines instead of ~350 to benchmark a GPU kernel.
+
+Execution Model
+---------------
+
+Every benchmark function has the signature ``fn(state, ctx)`` where *state*
+is a :class:`State` object and *ctx* is an :class:`~iris.Iris` context. The
+framework calls each function once per parameter combination. Inside the
+function you do three things:
+
+1. **Setup** — allocate tensors, build configs, fill data.  This code runs
+   **once** per parameter combination and is **not timed**.
+
+2. **Declare metrics** — call ``state.set_bytes(n)`` and/or
+   ``state.set_flops(n)`` so the framework can compute bandwidth / TFLOPS.
+
+3. **Register the kernel** — call ``state.exec(fn)`` with the callable to
+   time.  ``exec`` does **not** run the callable; it stores it. After your
+   function returns, the framework passes it to ``iris.do_bench()`` which
+   handles warmup, cache clearing, barrier synchronization, and CUDA-event
+   timing.
+
+The callable registered via ``state.exec()`` is invoked
+``1 + n_warmup + n_repeat`` times total.  Only the last ``n_repeat``
+invocations are timed.
+
+Per-Iteration Reset (``preamble_fn``)
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+If you need to reset state before **each** invocation (zero output buffers,
+reset locks, reinitialize a workspace), pass a ``preamble_fn``::
+
+    state.exec(
+        lambda: ctx.ccl.all_gather(out, inp, config=config),
+        preamble_fn=lambda: out.zero_(),
+    )
+
+``preamble_fn`` runs before every invocation (warmup and timed) but is
+**not timed** — it executes before the CUDA start event is recorded. It can
+be as heavyweight as needed without affecting measured results.
+
+The ``num_ranks`` Axis
+~~~~~~~~~~~~~~~~~~~~~~
+
+``num_ranks`` is a special axis. It controls how many GPU processes are
+spawned.  The framework collects all unique ``num_ranks`` values across
+registered benchmarks, then does a separate ``mp.spawn()`` for each value.
+Other axes are iterated inside the worker processes.
+
+If no ``num_ranks`` axis is declared, the benchmark runs with 8 ranks.
+
+Axes & Parameter Sweeps
+~~~~~~~~~~~~~~~~~~~~~~~
+
+Stack multiple ``@bench.axis`` decorators to define a sweep. The framework
+generates the Cartesian product of all axes. The outermost ``@axis``
+decorator is the slowest-varying axis in the output table.
+
+CLI Overrides
+~~~~~~~~~~~~~
+
+Any axis can be overridden or filtered from the command line:
+
+- ``--axis_M=1024,2048`` — replace the M axis with these values.
+- ``--axis_M=pow2:8:12`` — replace with ``[256, 512, 1024, 2048, 4096]``.
+- ``--axis_dtype=fp16`` — run only float16.
+- ``--skip_num_ranks=1,2`` — exclude 1- and 2-rank runs.
+- ``--benchmark_filter=all_gather`` — regex filter on benchmark name.
+
+Example
+-------
+
+::
+
+    import torch
+    import iris.bench as bench
+    from iris.ccl import Config
+
+    @bench.register
+    @bench.axis("num_ranks", [2, 4, 8])
+    @bench.axis("M", bench.power_of_two(8, 13))
+    @bench.axis("N", [256, 512, 1024])
+    @bench.axis("dtype", [torch.float16, torch.float32])
+    def all_gather(state, ctx):
+        M, N, dtype = state["M"], state["N"], state["dtype"]
+        world_size = ctx.get_num_ranks()
+
+        inp = ctx.zeros((M, N), dtype=dtype)
+        out = ctx.zeros((world_size * M, N), dtype=dtype)
+        inp.fill_(float(ctx.get_rank() + 1))
+
+        state.set_bytes((world_size - 1) * M * N * inp.element_size())
+
+        config = Config(use_gluon=False)
+        state.exec(lambda: ctx.ccl.all_gather(out, inp, config=config))
+
+    if __name__ == "__main__":
+        bench.main()
+
+Run::
+
+    python bench_all_gather.py
+    python bench_all_gather.py --skip_num_ranks=2
+    python bench_all_gather.py --axis_M=1024 --benchmark_format=json
+"""
+
+from ._core import (
+    AxisDef,
+    BenchmarkDef,
+    Result,
+    State,
+    axis,
+    linear_range,
+    power_of_two,
+    register,
+)
+from ._runner import main
+
+__all__ = [
+    "AxisDef",
+    "BenchmarkDef",
+    "Result",
+    "State",
+    "axis",
+    "linear_range",
+    "main",
+    "power_of_two",
+    "register",
+]
diff --git a/iris/bench/_core.py b/iris/bench/_core.py
new file mode 100644
index 000000000..c7d57f7d9
--- /dev/null
+++ b/iris/bench/_core.py
@@ -0,0 +1,232 @@
+# SPDX-License-Identifier: MIT
+# Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved.
+
+"""Core types, decorators, and range helpers for iris.bench."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Callable
+
+
+# Dataclasses
+@dataclass
+class AxisDef:
+    """A single sweep axis (name + list of values)."""
+
+    name: str
+    values: list[Any]
+
+
+@dataclass
+class BenchmarkDef:
+    """A registered benchmark: function + axes."""
+
+    name: str
+    fn: Callable
+    axes: list[AxisDef]
+
+
+@dataclass
+class Result:
+    """Stores results for one (benchmark x parameter-combination) run."""
+
+    benchmark_name: str
+    params: dict[str, Any]
+    gpu_time_ms: float
+    all_times_ms: list[float]
+    bandwidth_gbps: float | None = None
+    tflops: float | None = None
+    counters: dict[str, float] = field(default_factory=dict)
+    skipped: bool = False
+    skip_reason: str = ""
+    world_size: int = 1
+
+
+# Registry
+_registry: list[BenchmarkDef] = []
+
+
+# Skip sentinel
+class _SkipCombination(Exception):
+    """Raised by :meth:`State.skip` to skip the current parameter combo."""
+
+    def __init__(self, reason: str = ""):
+        self.reason = reason
+
+
+# State — passed into every benchmark function
+class State:
+    """Per-combination state object passed as the first argument to every
+    benchmark function.
+
+    The benchmark function body is the **setup phase** — it runs once per
+    parameter combination and is not timed.  Use ``State`` to:
+
+    - Read axis values: ``state["M"]``, ``state.get("dtype")``.
+    - Declare metrics: :meth:`set_bytes`, :meth:`set_flops`, :meth:`add_counter`.
+    - Register the callable to time: :meth:`exec`.
+    - Conditionally skip: :meth:`skip`.
+    - Override iteration counts: :meth:`set_warmup`, :meth:`set_repeat`.
+
+    After the benchmark function returns, the framework calls
+    ``iris.do_bench()`` with the callable registered via :meth:`exec`.
+    """
+
+    def __init__(self, params: dict[str, Any], n_warmup: int, n_repeat: int):
+        self._params = params
+        self._bytes: int | None = None
+        self._flops: int | None = None
+        self._counters: dict[str, float] = {}
+        self._exec_fn: Callable | None = None
+        self._preamble_fn: Callable = lambda: None
+        self._n_warmup = n_warmup
+        self._n_repeat = n_repeat
+
+    # -- axis access --------------------------------------------------------
+
+    def __getitem__(self, key: str) -> Any:
+        return self._params[key]
+
+    def get(self, key: str, default: Any = None) -> Any:
+        return self._params.get(key, default)
+
+    # -- metric declarations ------------------------------------------------
+
+    def set_bytes(self, n: int) -> None:
+        """Declare bytes transferred so the framework can report bandwidth.
+
+        The output table will include a **BW (GB/s)** column computed as
+        ``n / 1e9 / (gpu_time_ms * 1e-3)``.
+        """
+        self._bytes = n
+
+    def set_flops(self, n: int) -> None:
+        """Declare FLOPs so the framework can report throughput.
+
+        The output table will include a **TFLOPS** column computed as
+        ``n / 1e12 / (gpu_time_ms * 1e-3)``.
+        """
+        self._flops = n
+
+    def add_counter(self, name: str, value: float) -> None:
+        """Add a custom metric column to the output table.
+
+        Call multiple times with different names to add multiple columns.
+        """
+        self._counters[name] = value
+
+    # -- timing control -----------------------------------------------------
+
+    def set_warmup(self, n: int) -> None:
+        """Override the number of warmup iterations (default: 25, or ``--n_warmup``)."""
+        self._n_warmup = n
+
+    def set_repeat(self, n: int) -> None:
+        """Override the number of timed iterations (default: 100, or ``--n_repeat``)."""
+        self._n_repeat = n
+
+    def exec(self, fn: Callable, *, preamble_fn: Callable | None = None) -> None:
+        """Register the callable to time.
+
+        This does **not** call *fn* immediately.  After the benchmark
+        function returns, the framework passes *fn* to ``iris.do_bench()``
+        which runs it ``1 + n_warmup + n_repeat`` times (1 initial call,
+        warmup iterations, then timed iterations).
+
+        Parameters
+        ----------
+        fn:
+            The kernel / operation to benchmark.  Only this callable is
+            inside the timed region (between CUDA start/end events).
+        preamble_fn:
+            Optional callable executed before **every** invocation of *fn*
+            (warmup and timed).  Runs **outside** the timed region — before
+            the CUDA start event is recorded — so it can be arbitrarily
+            expensive without affecting results.  Use it to reset output
+            buffers, reinitialize locks, rebuild workspaces, etc.
+
+        Example::
+
+            # Zero the output buffer before each iteration
+            state.exec(
+                lambda: ctx.ccl.all_gather(out, inp, config=config),
+                preamble_fn=lambda: out.zero_(),
+            )
+        """
+        self._exec_fn = fn
+        if preamble_fn is not None:
+            self._preamble_fn = preamble_fn
+
+    # -- skip ---------------------------------------------------------------
+
+    def skip(self, reason: str = "") -> None:
+        """Skip this parameter combination.
+
+        Call this during setup to skip combinations that are invalid or
+        uninteresting.  The combination appears as ``(skipped)`` in the
+        output rather than being silently omitted.
+
+        Example::
+
+            if M < N:
+                state.skip("M must be >= N")
+        """
+        raise _SkipCombination(reason)
+
+
+# Range helpers
+def power_of_two(start_exp: int, end_exp: int) -> list[int]:
+    """Return ``[2**start_exp, ..., 2**end_exp]`` inclusive."""
+    return [1 << e for e in range(start_exp, end_exp + 1)]
+
+
+def linear_range(start: int, end: int, step: int) -> list[int]:
+    """Return ``[start, start+step, ..., end]`` inclusive."""
+    return list(range(start, end + 1, step))
+
+
+# Decorators
+def axis(name: str, values: list[Any]):
+    """Define a sweep axis for a benchmark.
+
+    Multiple ``@axis`` decorators stack; the framework generates the
+    Cartesian product of all axes at runtime.  The outermost ``@axis``
+    is the slowest-varying in the output.
+
+    The axis named ``"num_ranks"`` is special: it controls how many GPU
+    processes are spawned rather than being iterated inside a worker.
+
+    Any axis can be overridden (``--axis_M=1024``) or filtered
+    (``--skip_dtype=fp32``) from the command line.
+
+    Parameters
+    ----------
+    name:
+        Axis name.  Accessible in the benchmark via ``state["name"]``.
+    values:
+        List of values to sweep.  Use :func:`power_of_two` or
+        :func:`linear_range` for common patterns.
+    """
+
+    def decorator(fn: Callable) -> Callable:
+        if not hasattr(fn, "_bench_axes"):
+            fn._bench_axes = []
+        # Prepend so that declaration order matches iteration order
+        # (outermost decorator = slowest-varying axis).
+        fn._bench_axes.insert(0, AxisDef(name, list(values)))
+        return fn
+
+    return decorator
+
+
+def register(fn: Callable) -> Callable:
+    """Register *fn* as a benchmark.  Must be the **outermost** decorator.
+
+    The function must have the signature ``fn(state, ctx)`` where *state*
+    is a :class:`State` and *ctx* is an :class:`~iris.Iris` context
+    (or ``iris_gluon`` context if ``--use_gluon`` is passed).
+    """
+    axes: list[AxisDef] = getattr(fn, "_bench_axes", [])
+    _registry.append(BenchmarkDef(name=fn.__name__, fn=fn, axes=axes))
+    return fn
diff --git a/iris/bench/_runner.py b/iris/bench/_runner.py
new file mode 100644
index 000000000..44eba6750
--- /dev/null
+++ b/iris/bench/_runner.py
@@ -0,0 +1,569 @@
+# SPDX-License-Identifier: MIT
+# Copyright (c) 2026 Advanced Micro Devices, Inc. All rights reserved.
+
+"""Distributed runner, output formatters, and CLI entry point for iris.bench."""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import io
+import itertools
+import json
+import os
+import re
+import statistics
+import sys
+from typing import Any, Callable
+
+import torch
+import torch.distributed as dist
+from torch.distributed.launcher.api import LaunchConfig, elastic_launch
+
+from ._core import (
+    AxisDef,
+    BenchmarkDef,
+    Result,
+    State,
+    _SkipCombination,
+    _registry,
+    power_of_two,
+    linear_range,
+)
+
+# Reserved axis name that controls process spawning.
+_NUM_RANKS_AXIS = "num_ranks"
+_DEFAULT_NUM_RANKS = 8
+
+
+# Axis override / skip parsing
+_DTYPE_MAP = {
+    "fp16": torch.float16,
+    "float16": torch.float16,
+    "fp32": torch.float32,
+    "float32": torch.float32,
+    "bf16": torch.bfloat16,
+    "bfloat16": torch.bfloat16,
+}
+
+
+def _dtype_str(v: Any) -> str:
+    """Short string for a torch dtype, passthrough for anything else."""
+    if isinstance(v, torch.dtype):
+        return {
+            torch.float16: "float16",
+            torch.float32: "float32",
+            torch.bfloat16: "bfloat16",
+            torch.float64: "float64",
+            torch.int8: "int8",
+            torch.int16: "int16",
+            torch.int32: "int32",
+            torch.int64: "int64",
+        }.get(v, str(v))
+    return str(v)
+
+
+def _parse_axis_values(raw: str, axis_name: str) -> list[Any]:
+    """Parse a CLI ``--axis_<name>=<value>`` or ``--skip_<name>=<value>`` string.
+
+    Formats:
+    - ``1024,2048`` — explicit list
+    - ``pow2:8:13``  — ``power_of_two(8, 13)``
+    - ``lin:64:256:64`` — ``linear_range(64, 256, 64)``
+    - dtype names: ``fp16``, ``fp32``, ``bf16``
+    """
+    raw = raw.strip()
+
+    if raw.startswith("pow2:"):
+        parts = raw.split(":")
+        return power_of_two(int(parts[1]), int(parts[2]))
+
+    if raw.startswith("lin:"):
+        parts = raw.split(":")
+        return linear_range(int(parts[1]), int(parts[2]), int(parts[3]))
+
+    tokens = [t.strip() for t in raw.split(",")]
+
+    # Check if they look like dtype names
+    if axis_name == "dtype" or all(t.lower() in _DTYPE_MAP for t in tokens):
+        return [_DTYPE_MAP[t.lower()] for t in tokens]
+
+    # Try integers
+    try:
+        return [int(t) for t in tokens]
+    except ValueError:
+        pass
+
+    # Try floats
+    try:
+        return [float(t) for t in tokens]
+    except ValueError:
+        pass
+
+    # Fall back to strings
+    return tokens
+
+
+def _effective_values(
+    ax: AxisDef,
+    axis_overrides: dict[str, list[Any]],
+    skip_overrides: dict[str, list[Any]],
+) -> list[Any]:
+    """Resolve effective values for an axis after overrides and skips."""
+    if ax.name in axis_overrides:
+        values = list(axis_overrides[ax.name])
+    else:
+        values = list(ax.values)
+
+    if ax.name in skip_overrides:
+        skip_set = set(skip_overrides[ax.name])
+        values = [v for v in values if v not in skip_set]
+
+    return values
+
+
+def _get_benchmark_num_ranks(
+    bdef: BenchmarkDef,
+    axis_overrides: dict[str, list[Any]],
+    skip_overrides: dict[str, list[Any]],
+) -> list[int]:
+    """Return the effective num_ranks values for a benchmark."""
+    # Check if benchmark declares a num_ranks axis
+    for ax in bdef.axes:
+        if ax.name == _NUM_RANKS_AXIS:
+            return _effective_values(ax, axis_overrides, skip_overrides)
+
+    # No declared axis — check if there's a global override
+    if _NUM_RANKS_AXIS in axis_overrides:
+        values = list(axis_overrides[_NUM_RANKS_AXIS])
+        if _NUM_RANKS_AXIS in skip_overrides:
+            skip_set = set(skip_overrides[_NUM_RANKS_AXIS])
+            values = [v for v in values if v not in skip_set]
+        return values
+
+    return [_DEFAULT_NUM_RANKS]
+
+
+# Output formatters
+def _format_console(results: list[Result]) -> str:
+    """Render results as an aligned console table."""
+    if not results:
+        return ""
+
+    # Group by benchmark name
+    by_bench: dict[str, list[Result]] = {}
+    for r in results:
+        by_bench.setdefault(r.benchmark_name, []).append(r)
+
+    lines: list[str] = []
+    for bench_name, bench_results in by_bench.items():
+        lines.append(f"\n{bench_name}")
+
+        # Build column specs
+        param_names = list(bench_results[0].params.keys())
+        cols: list[tuple[str, Callable[[Result], str]]] = []
+        for pn in param_names:
+            cols.append((pn, lambda r, _pn=pn: _dtype_str(r.params[_pn])))
+        cols.append(("GPU Time (ms)", lambda r: f"{r.gpu_time_ms:.3f}"))
+        if any(r.bandwidth_gbps is not None for r in bench_results):
+            cols.append(("BW (GB/s)", lambda r: f"{r.bandwidth_gbps:.1f}" if r.bandwidth_gbps is not None else ""))
+        if any(r.tflops is not None for r in bench_results):
+            cols.append(("TFLOPS", lambda r: f"{r.tflops:.1f}" if r.tflops is not None else ""))
+
+        # Gather counter names across all results
+        counter_names: list[str] = []
+        seen: set[str] = set()
+        for r in bench_results:
+            for cn in r.counters:
+                if cn not in seen:
+                    counter_names.append(cn)
+                    seen.add(cn)
+        for cn in counter_names:
+            cols.append((cn, lambda r, _cn=cn: f"{r.counters[_cn]:.3f}" if _cn in r.counters else ""))
+
+        # Compute column widths
+        header_strs = [c[0] for c in cols]
+        row_strs: list[list[str]] = []
+        for r in bench_results:
+            if r.skipped:
+                row = [cols[0][1](r)] + ["(skipped)" + (f" {r.skip_reason}" if r.skip_reason else "")]
+                row += [""] * (len(cols) - 2)
+                row_strs.append(row)
+            else:
+                row_strs.append([c[1](r) for c in cols])
+
+        widths = [len(h) for h in header_strs]
+        for row in row_strs:
+            for i, cell in enumerate(row):
+                if i < len(widths):
+                    widths[i] = max(widths[i], len(cell))
+
+        fmt = "  ".join(f"{{:>{w}}}" for w in widths)
+        lines.append(fmt.format(*header_strs))
+        for row in row_strs:
+            while len(row) < len(widths):
+                row.append("")
+            lines.append(fmt.format(*row))
+
+    return "\n".join(lines) + "\n"
+
+
+def _format_json(results: list[Result]) -> str:
+    """Structured JSON output for CI."""
+    records = []
+    for r in results:
+        rec: dict[str, Any] = {
+            "benchmark": r.benchmark_name,
+            "world_size": r.world_size,
+            "params": {k: _dtype_str(v) for k, v in r.params.items()},
+            "gpu_time_ms": r.gpu_time_ms,
+            "all_times_ms": r.all_times_ms,
+        }
+        if r.bandwidth_gbps is not None:
+            rec["bandwidth_gbps"] = r.bandwidth_gbps
+        if r.tflops is not None:
+            rec["tflops"] = r.tflops
+        if r.counters:
+            rec["counters"] = r.counters
+        if r.skipped:
+            rec["skipped"] = True
+            rec["skip_reason"] = r.skip_reason
+        records.append(rec)
+    return json.dumps(records, indent=2) + "\n"
+
+
+def _format_csv(results: list[Result]) -> str:
+    """Flat CSV output."""
+    if not results:
+        return ""
+    buf = io.StringIO()
+
+    # Collect all param names and counter names
+    param_names: list[str] = []
+    counter_names: list[str] = []
+    seen_p: set[str] = set()
+    seen_c: set[str] = set()
+    for r in results:
+        for k in r.params:
+            if k not in seen_p:
+                param_names.append(k)
+                seen_p.add(k)
+        for k in r.counters:
+            if k not in seen_c:
+                counter_names.append(k)
+                seen_c.add(k)
+
+    fieldnames = (
+        ["benchmark", "world_size"]
+        + param_names
+        + ["gpu_time_ms", "bandwidth_gbps", "tflops"]
+        + counter_names
+        + ["skipped", "skip_reason"]
+    )
+    writer = csv.DictWriter(buf, fieldnames=fieldnames)
+    writer.writeheader()
+    for r in results:
+        row: dict[str, Any] = {
+            "benchmark": r.benchmark_name,
+            "world_size": r.world_size,
+            "gpu_time_ms": f"{r.gpu_time_ms:.4f}" if not r.skipped else "",
+            "bandwidth_gbps": f"{r.bandwidth_gbps:.2f}" if r.bandwidth_gbps is not None else "",
+            "tflops": f"{r.tflops:.2f}" if r.tflops is not None else "",
+            "skipped": r.skipped,
+            "skip_reason": r.skip_reason,
+        }
+        for pn in param_names:
+            row[pn] = _dtype_str(r.params.get(pn, ""))
+        for cn in counter_names:
+            row[cn] = f"{r.counters[cn]:.4f}" if cn in r.counters else ""
+        writer.writerow(row)
+    return buf.getvalue()
+
+
+# Distributed worker
+def _run_benchmarks_worker(
+    benchmarks: list[BenchmarkDef],
+    axis_overrides: dict[str, list[Any]],
+    skip_overrides: dict[str, list[Any]],
+    heap_size: int,
+    use_gluon: bool,
+    n_warmup: int,
+    n_repeat: int,
+    benchmark_filter: str | None,
+) -> list[Result]:
+    """Worker that runs inside each rank via ``elastic_launch``.
+
+    Returns results on rank 0; empty list on other ranks.
+    """
+    import iris as _iris
+
+    local_rank = int(os.environ["LOCAL_RANK"])
+    world_size = int(os.environ["WORLD_SIZE"])
+
+    torch.cuda.set_device(local_rank)
+    backend = "nccl" if torch.cuda.is_available() else "gloo"
+    dist.init_process_group(backend=backend)
+
+    # Create iris context
+    if use_gluon:
+        import iris.experimental.iris_gluon as iris_gluon
+
+        ctx = iris_gluon.iris(heap_size)
+    else:
+        ctx = _iris.iris(heap_size)
+
+    rank = ctx.get_rank()
+
+    all_results: list[Result] = []
+
+    for bdef in benchmarks:
+        # Filter by name
+        if benchmark_filter and not re.search(benchmark_filter, bdef.name):
+            continue
+
+        # Check if this benchmark should run at this world_size
+        has_nr_axis = any(ax.name == _NUM_RANKS_AXIS for ax in bdef.axes)
+        if not has_nr_axis and _NUM_RANKS_AXIS not in axis_overrides:
+            # No num_ranks axis declared and no global override —
+            # only run at the default
+            if world_size != _DEFAULT_NUM_RANKS:
+                continue
+
+        # Build non-num_ranks axes with overrides/skips applied
+        axes: list[AxisDef] = []
+        for ax in bdef.axes:
+            if ax.name == _NUM_RANKS_AXIS:
+                continue  # handled externally by the spawning loop
+            values = _effective_values(ax, axis_overrides, skip_overrides)
+            if not values:
+                break  # entire benchmark skipped if an axis is empty
+            axes.append(AxisDef(ax.name, values))
+        else:
+            # Generate Cartesian product of non-num_ranks axes
+            if axes:
+                axis_names = [a.name for a in axes]
+                axis_values = [a.values for a in axes]
+                combos = list(itertools.product(*axis_values))
+            else:
+                axis_names = []
+                combos = [()]
+
+            for combo in combos:
+                params: dict[str, Any] = {}
+                # Include num_ranks in params so it appears in output
+                if has_nr_axis or _NUM_RANKS_AXIS in axis_overrides:
+                    params[_NUM_RANKS_AXIS] = world_size
+                params.update(zip(axis_names, combo))
+
+                state = State(params, n_warmup=n_warmup, n_repeat=n_repeat)
+
+                skipped = False
+                skip_reason = ""
+                try:
+                    bdef.fn(state, ctx)
+                except _SkipCombination as exc:
+                    skipped = True
+                    skip_reason = exc.reason
+
+                if skipped:
+                    all_results.append(
+                        Result(
+                            benchmark_name=bdef.name,
+                            params=params,
+                            gpu_time_ms=0.0,
+                            all_times_ms=[],
+                            skipped=True,
+                            skip_reason=skip_reason,
+                            world_size=world_size,
+                        )
+                    )
+                    continue
+
+                if state._exec_fn is None:
+                    raise RuntimeError(
+                        f"Benchmark '{bdef.name}' with params {params} "
+                        f"did not call state.exec(fn). Every benchmark must "
+                        f"register a callable to time."
+                    )
+
+                # Time with do_bench
+                times = _iris.do_bench(
+                    state._exec_fn,
+                    barrier_fn=ctx.barrier,
+                    preamble_fn=state._preamble_fn,
+                    n_warmup=state._n_warmup,
+                    n_repeat=state._n_repeat,
+                    return_mode="all",
+                )
+
+                mean_ms = statistics.mean(times)
+
+                bw = None
+                if state._bytes is not None and mean_ms > 0:
+                    bw = (state._bytes / 1e9) / (mean_ms * 1e-3)
+
+                tflops = None
+                if state._flops is not None and mean_ms > 0:
+                    tflops = (state._flops / 1e12) / (mean_ms * 1e-3)
+
+                all_results.append(
+                    Result(
+                        benchmark_name=bdef.name,
+                        params=params,
+                        gpu_time_ms=mean_ms,
+                        all_times_ms=times,
+                        bandwidth_gbps=bw,
+                        tflops=tflops,
+                        counters=dict(state._counters),
+                        world_size=world_size,
+                    )
+                )
+
+    ctx.barrier()
+    dist.destroy_process_group()
+
+    return all_results if rank == 0 else []
+
+
+# CLI entry point
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="iris.bench — GPU benchmarking framework",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+    parser.add_argument(
+        "--benchmark_filter",
+        type=str,
+        default=None,
+        help="Regex filter for benchmark names",
+    )
+    parser.add_argument(
+        "--benchmark_format",
+        type=str,
+        default="console",
+        choices=["console", "json", "csv"],
+        help="Output format",
+    )
+    parser.add_argument(
+        "--benchmark_out",
+        type=str,
+        default=None,
+        help="Write results to this file",
+    )
+    parser.add_argument(
+        "--heap_size",
+        type=int,
+        default=1 << 34,
+        help="Iris symmetric heap size in bytes",
+    )
+    parser.add_argument(
+        "--use_gluon",
+        action="store_true",
+        help="Use Gluon backend",
+    )
+    parser.add_argument(
+        "--n_warmup",
+        type=int,
+        default=25,
+        help="Number of warmup iterations",
+    )
+    parser.add_argument(
+        "--n_repeat",
+        type=int,
+        default=100,
+        help="Number of timed iterations",
+    )
+    return parser
+
+
+def main(argv: list[str] | None = None) -> None:
+    """CLI entry point.  Call from ``if __name__ == '__main__': bench.main()``.
+
+    Collects all ``@bench.register``-ed benchmarks in the current module,
+    resolves ``num_ranks`` values, and launches one process group per unique
+    ``num_ranks`` via ``elastic_launch`` (the programmatic ``torchrun``
+    API).  Results are merged and formatted to stdout (and optionally a
+    file).
+
+    In addition to the flags shown by ``--help``, two families of dynamic
+    flags are supported:
+
+    - ``--axis_<name>=<values>`` — override an axis (replaces declared values).
+    - ``--skip_<name>=<values>`` — exclude specific values from an axis.
+
+    Value formats: ``1024,2048`` (explicit), ``pow2:8:13`` (power-of-two
+    range), ``lin:64:256:64`` (linear range), ``fp16``/``fp32``/``bf16``
+    (dtype shorthand).
+    """
+    parser = _build_parser()
+    args, remaining = parser.parse_known_args(argv)
+
+    # Parse --axis_<name>=<value> and --skip_<name>=<value> from remaining args
+    axis_overrides: dict[str, list[Any]] = {}
+    skip_overrides: dict[str, list[Any]] = {}
+    for token in remaining:
+        m = re.match(r"^--axis_(\w+)=(.+)$", token)
+        if m:
+            axis_overrides[m.group(1)] = _parse_axis_values(m.group(2), m.group(1))
+            continue
+        m = re.match(r"^--skip_(\w+)=(.+)$", token)
+        if m:
+            skip_overrides[m.group(1)] = _parse_axis_values(m.group(2), m.group(1))
+            continue
+        parser.error(f"Unrecognized argument: {token}")
+
+    benchmarks = list(_registry)
+    if not benchmarks:
+        print("No benchmarks registered.", file=sys.stderr)
+        sys.exit(1)
+
+    # Collect the union of all num_ranks values across registered benchmarks
+    all_num_ranks: set[int] = set()
+    for bdef in benchmarks:
+        # Skip benchmarks that don't match the filter before collecting num_ranks
+        if args.benchmark_filter and not re.search(args.benchmark_filter, bdef.name):
+            continue
+        all_num_ranks.update(_get_benchmark_num_ranks(bdef, axis_overrides, skip_overrides))
+
+    if not all_num_ranks:
+        print("No benchmark configurations to run after applying filters/skips.", file=sys.stderr)
+        sys.exit(1)
+
+    # Launch once per unique num_ranks, collecting results across runs
+    all_results: list[Result] = []
+
+    for num_ranks in sorted(all_num_ranks):
+        config = LaunchConfig(
+            min_nodes=1,
+            max_nodes=1,
+            nproc_per_node=num_ranks,
+            rdzv_backend="c10d",
+            rdzv_endpoint="localhost:0",
+            max_restarts=0,
+        )
+        results_by_rank = elastic_launch(config, _run_benchmarks_worker)(
+            benchmarks,
+            axis_overrides,
+            skip_overrides,
+            args.heap_size,
+            args.use_gluon,
+            args.n_warmup,
+            args.n_repeat,
+            args.benchmark_filter,
+        )
+        # Rank 0 returns the results; other ranks return []
+        all_results.extend(results_by_rank[0])
+
+    # Format and output (runs in the main process)
+    if args.benchmark_format == "json":
+        output = _format_json(all_results)
+    elif args.benchmark_format == "csv":
+        output = _format_csv(all_results)
+    else:
+        output = _format_console(all_results)
+
+    print(output, end="")
+
+    if args.benchmark_out:
+        with open(args.benchmark_out, "w") as f:
+            f.write(output)