API reference

Everything on this page is generated from the source docstrings.

Top level

sns.measure

measure(fn: Callable[[], object], *, warmup: int = DEFAULT_WARMUP, iters: int = MIN_ITERS, device: int = 0, flush_l2: bool = True, policy: ClockPolicy | None = None, min_duration_us: float = MIN_DURATION_US) -> TimingResult

Time a callable honestly.

Returns an interval and a quality tier, never a bare number.

Known limitation: the reported CI is a bootstrap over samples within a single measurement window, so it captures sampling error inside that window but not run-to-run variability between windows. On unlocked hardware the cross-run spread can exceed the reported interval by orders of magnitude — scripts/validate_timing.py measures exactly this. Folding between-window variance into Tier B intervals is planned for Phase 1.

Source code in src/sns/timing.py

def measure(
    fn: Callable[[], object],
    *,
    warmup: int = DEFAULT_WARMUP,
    iters: int = MIN_ITERS,
    device: int = 0,
    flush_l2: bool = True,
    policy: ClockPolicy | None = None,
    min_duration_us: float = MIN_DURATION_US,
) -> TimingResult:
    """Time a callable honestly.

    Returns an interval and a quality tier, never a bare number.

    Known limitation: the reported CI is a bootstrap over samples within a
    single measurement window, so it captures sampling error inside that
    window but not run-to-run variability between windows. On unlocked
    hardware the cross-run spread can exceed the reported interval by orders
    of magnitude — scripts/validate_timing.py measures exactly this. Folding
    between-window variance into Tier B intervals is planned for Phase 1.
    """
    import torch

    if iters < MIN_ITERS:
        raise ValueError(f"iters must be at least {MIN_ITERS}, got {iters}")
    if not torch.cuda.is_available():
        raise RuntimeError("measure() requires a CUDA device")

    torch.cuda.set_device(device)
    policy = policy or UnlockedClockPolicy()
    dev = torch.device(f"cuda:{device}")
    scratch = l2_flush_buffer(dev) if flush_l2 else None
    sampler = ClockSampler(device)

    policy.apply()
    # restore() runs in the finally below and clears policy.locked, so capture
    # the lock state now. Reading it after restore would make Tier A unreachable.
    was_locked = policy.locked
    try:
        # Warmup. The default of 200 exists because do_bench's default of 25
        # yields two calls and underestimates by ~30% (triton#2306).
        for _ in range(warmup):
            fn()
        torch.cuda.synchronize()

        # Calibrate the inner loop against one timed iteration.
        probe_start = torch.cuda.Event(enable_timing=True)
        probe_end = torch.cuda.Event(enable_timing=True)
        probe_start.record()
        fn()
        probe_end.record()
        torch.cuda.synchronize()
        inner_reps = resolve_inner_reps(
            probe_start.elapsed_time(probe_end), min_duration_us
        )

        # Events are allocated up front so allocation never lands inside a
        # timed region.
        starts = [torch.cuda.Event(enable_timing=True) for _ in range(iters)]
        ends = [torch.cuda.Event(enable_timing=True) for _ in range(iters)]

        throttle_before = throttle_snapshot(device)
        clock_samples: list[float] = []
        throttled_during = False

        # NVML runs in-process at microsecond latency, unlike the nvidia-smi
        # subprocess this used to shell out to, so the old bounded-sample cap
        # (clock_sample_stride) is unnecessary here — sample every iteration.
        for i in range(iters):
            if scratch is not None:
                scratch.zero_()
            starts[i].record()
            for _ in range(inner_reps):
                fn()
            ends[i].record()
            sm_clock = sampler.sample_clock_mhz()
            if sm_clock is not None:
                clock_samples.append(sm_clock)
            if sampler.throttled_now():
                throttled_during = True

        torch.cuda.synchronize()
        throttle_after = throttle_snapshot(device)
    finally:
        policy.restore()
        sampler.shutdown()

    samples = [
        starts[i].elapsed_time(ends[i]) / inner_reps for i in range(iters)
    ]
    throttle_fired = (
        throttle_before != throttle_after
        or any(v == "Active" for v in {**throttle_before, **throttle_after}.values())
        or bool(throttled_during)
    )
    tier = assign_tier(was_locked, clock_samples, throttle_fired)
    ci_lo, ci_hi = bootstrap_ci(samples)

    return TimingResult(
        samples_ms=samples,
        median_ms=percentile(samples, 0.5),
        p10_ms=percentile(samples, 0.10),
        p90_ms=percentile(samples, 0.90),
        ci95_lo_ms=ci_lo,
        ci95_hi_ms=ci_hi,
        n=len(samples),
        tier=tier,
        warmup=warmup,
        inner_reps=inner_reps,
        throttle_fired=throttle_fired,
        clock_cv_pct=cv_percent(clock_samples) if clock_samples else None,
        clock_range_mhz=(
            max(clock_samples) - min(clock_samples) if clock_samples else None
        ),
    )

sns.compare

compare(candidate_fn: Callable[[], object], baseline_fn: Callable[[], object], **kwargs) -> ComparisonResult

Measure a candidate against a freshly measured baseline.

Both sides are timed in the same process, under the same clock policy, back to back. The baseline is never read from cache — a stale baseline makes it impossible to distinguish a kernel regression from an upstream improvement, which is the whole point of the tool.

Known limitation: the candidate is measured first and the baseline second, so on unlocked hardware the second measurement runs on a warmer, more boosted GPU. Comparing identical work on an RTX 3060 laptop yielded a speedup of 0.962 rather than 1.0 from this effect alone. Tier A pinning largely removes it; Tier B and C comparisons carry it. Interleaving the two sides is planned for Phase 1.

Source code in src/sns/timing.py

def compare(
    candidate_fn: Callable[[], object],
    baseline_fn: Callable[[], object],
    **kwargs,
) -> ComparisonResult:
    """Measure a candidate against a freshly measured baseline.

    Both sides are timed in the same process, under the same clock policy,
    back to back. The baseline is never read from cache — a stale baseline
    makes it impossible to distinguish a kernel regression from an upstream
    improvement, which is the whole point of the tool.

    Known limitation: the candidate is measured first and the baseline second,
    so on unlocked hardware the second measurement runs on a warmer, more
    boosted GPU. Comparing identical work on an RTX 3060 laptop yielded a
    speedup of 0.962 rather than 1.0 from this effect alone. Tier A pinning
    largely removes it; Tier B and C comparisons carry it. Interleaving the
    two sides is planned for Phase 1.
    """
    candidate = measure(candidate_fn, **kwargs)
    baseline = measure(baseline_fn, **kwargs)

    if candidate.median_ms <= 0:
        raise ValueError(
            "candidate measured 0 ms: the timed region fell below CUDA event "
            "resolution. Raise iters, or check that the callable does real work."
        )

    lo, hi = ratio_ci(candidate.samples_ms, baseline.samples_ms)

    return ComparisonResult(
        candidate=candidate,
        baseline=baseline,
        speedup=baseline.median_ms / candidate.median_ms,
        speedup_ci_lo=lo,
        speedup_ci_hi=hi,
    )

Result types

sns.TimingResult

Bases: BaseModel

A timing measurement.

Deliberately defines no float, int or index: a caller must not be able to collapse this into a bare number and lose the interval.

Source code in src/sns/types.py

class TimingResult(BaseModel):
    """A timing measurement.

    Deliberately defines no __float__, __int__ or __index__: a caller must
    not be able to collapse this into a bare number and lose the interval.
    """

    samples_ms: list[float]
    median_ms: float
    p10_ms: float
    p90_ms: float
    ci95_lo_ms: float
    ci95_hi_ms: float
    n: int
    tier: MeasurementTier
    warmup: int
    inner_reps: int = 1
    throttle_fired: bool = False
    clock_cv_pct: float | None = None
    clock_range_mhz: float | None = None

    @field_validator("samples_ms")
    @classmethod
    def _need_two_samples(cls, v: list[float]) -> list[float]:
        if len(v) < 2:
            raise ValueError("a timing result needs at least 2 samples")
        return v

    @property
    def is_performance_valid(self) -> bool:
        return self.tier is not MeasurementTier.C

sns.ComparisonResult

Bases: BaseModel

A candidate measured against a baseline timed in the same session.

The baseline is never cached across runs. Re-measuring it every time is what lets us tell "my kernel regressed" from "torch got faster", and it is what makes the ratio comparable across machines and architectures.

Source code in src/sns/types.py

class ComparisonResult(BaseModel):
    """A candidate measured against a baseline timed in the same session.

    The baseline is never cached across runs. Re-measuring it every time is
    what lets us tell "my kernel regressed" from "torch got faster", and it
    is what makes the ratio comparable across machines and architectures.
    """

    candidate: TimingResult
    baseline: TimingResult
    speedup: float
    speedup_ci_lo: float
    speedup_ci_hi: float

    @property
    def tier(self) -> MeasurementTier:
        """A comparison is only as trustworthy as its worse half."""
        order = [MeasurementTier.A, MeasurementTier.B, MeasurementTier.C]
        return max(
            (self.candidate.tier, self.baseline.tier), key=order.index
        )

    @property
    def is_performance_valid(self) -> bool:
        return self.tier is not MeasurementTier.C

tier property

tier: MeasurementTier

A comparison is only as trustworthy as its worse half.

sns.MeasurementTier

Bases: str, Enum

How much the numbers in a result can be trusted.

A: clocks locked and verified stable. Full verdicts, drift-eligible. B: clocks floating, variance measured and folded into the interval. C: unstable or throttled. No performance verdict is valid.

Source code in src/sns/types.py

class MeasurementTier(str, Enum):
    """How much the numbers in a result can be trusted.

    A: clocks locked and verified stable. Full verdicts, drift-eligible.
    B: clocks floating, variance measured and folded into the interval.
    C: unstable or throttled. No performance verdict is valid.
    """

    A = "A"
    B = "B"
    C = "C"

sns.EnvironmentFingerprint

Bases: BaseModel

Identifies the toolchain and device a run happened on.

Two runs are only comparable when their fingerprints match exactly.

Source code in src/sns/types.py

class EnvironmentFingerprint(BaseModel):
    """Identifies the toolchain and device a run happened on.

    Two runs are only comparable when their fingerprints match exactly.
    """

    torch_version: str
    triton_version: str | None = None
    cuda_version: str | None = None
    driver_version: str | None = None
    gpu_name: str | None = None
    compute_cap: str | None = None
    arch_family: str | None = None
    sm_count: int | None = None

    def matches(self, other: "EnvironmentFingerprint") -> bool:
        return self.model_dump() == other.model_dump()

Clock policies

sns.clocks.UnlockedClockPolicy

Measure without pinning. Honest, and the only option on most machines.

Source code in src/sns/clocks.py

class UnlockedClockPolicy:
    """Measure without pinning. Honest, and the only option on most machines."""

    locked = False

    def apply(self) -> None:
        return None

    def restore(self) -> None:
        return None

sns.clocks.LockedClockPolicy

Pin the SM clock, and optionally the power cap, verifying by readback.

Source code in src/sns/clocks.py

class LockedClockPolicy:
    """Pin the SM clock, and optionally the power cap, verifying by readback."""

    def __init__(self, target_sm_mhz: int, power_cap_w: int | None = None):
        self.target_sm_mhz = target_sm_mhz
        self.power_cap_w = power_cap_w
        self.locked = False

    def apply(self) -> None:
        _run_smi(["-lgc", f"{self.target_sm_mhz},{self.target_sm_mhz}"])
        try:
            # nvidia-smi exits 0 when it refuses the write, so the exit code
            # proves nothing. Only the readback does.
            observed = smi_query_float("clocks.sm")
            if (
                observed is None
                or abs(observed - self.target_sm_mhz) > LOCK_READBACK_TOLERANCE_MHZ
            ):
                raise ClockLockError(
                    f"requested {self.target_sm_mhz} MHz, device reports {observed}"
                )
            if self.power_cap_w is not None:
                _run_smi(["-pl", str(self.power_cap_w)])
                observed_w = smi_query_float("power.limit")
                if (
                    observed_w is None
                    or abs(observed_w - self.power_cap_w) > POWER_READBACK_TOLERANCE_W
                ):
                    raise ClockLockError(
                        f"requested {self.power_cap_w} W cap, device reports {observed_w}"
                    )
        except Exception:
            # The -lgc write already reached the device. Leaving it applied
            # after a loud failure silently pins the GPU for everything after.
            self.restore()
            raise
        self.locked = True

    def restore(self) -> None:
        _run_smi(["-rgc"])
        _run_smi(["-rmc"])
        self.locked = False

sns.clocks.ClockLockError

Bases: RuntimeError

Raised when a requested operating point could not be established.

Source code in src/sns/clocks.py

class ClockLockError(RuntimeError):
    """Raised when a requested operating point could not be established."""

sns.clocks.assign_tier

assign_tier(locked: bool, clock_samples: list[float], throttle_fired: bool) -> MeasurementTier

Classify a measurement window.

Throttle flags alone are not sufficient: on an RTX 3060 laptop the SM clock swung 495 MHz (5.1% CV) while the flags stayed silent, and two identical runs disagreed on whether throttling fired at all. Observed variance is the governing signal.

Source code in src/sns/clocks.py

def assign_tier(
    locked: bool, clock_samples: list[float], throttle_fired: bool
) -> MeasurementTier:
    """Classify a measurement window.

    Throttle flags alone are not sufficient: on an RTX 3060 laptop the SM
    clock swung 495 MHz (5.1% CV) while the flags stayed silent, and two
    identical runs disagreed on whether throttling fired at all. Observed
    variance is the governing signal.
    """
    if throttle_fired:
        return MeasurementTier.C
    if clock_samples and cv_percent(clock_samples) > TIER_C_CV_PCT:
        return MeasurementTier.C
    if locked and clock_samples:
        spread = max(clock_samples) - min(clock_samples)
        if spread <= TIER_A_RANGE_MHZ:
            return MeasurementTier.A
    return MeasurementTier.B

Statistics

sns.stats.bootstrap_ci

bootstrap_ci(samples: list[float], confidence: float = 0.95, n_resamples: int = 2000, seed: int = 12648430) -> tuple[float, float]

Percentile bootstrap CI of the median.

Seeded so a given sample set always yields the same interval — a reproducibility requirement, not a convenience.

Source code in src/sns/stats.py

def bootstrap_ci(
    samples: list[float],
    confidence: float = 0.95,
    n_resamples: int = 2000,
    seed: int = 0xC0FFEE,
) -> tuple[float, float]:
    """Percentile bootstrap CI of the median.

    Seeded so a given sample set always yields the same interval — a
    reproducibility requirement, not a convenience.
    """
    if len(samples) < 2:
        raise ValueError("bootstrap CI needs at least 2 samples")
    rng = random.Random(seed)
    k = len(samples)
    medians = sorted(
        statistics.median(rng.choices(samples, k=k)) for _ in range(n_resamples)
    )
    alpha = 1.0 - confidence
    lo_i = int((alpha / 2) * n_resamples)
    hi_i = min(n_resamples - 1, int((1 - alpha / 2) * n_resamples))
    return (float(medians[lo_i]), float(medians[hi_i]))

sns.stats.ratio_ci

ratio_ci(candidate: list[float], baseline: list[float], confidence: float = 0.95, n_resamples: int = 2000, seed: int = 12648430) -> tuple[float, float]

CI of speedup = median(baseline) / median(candidate).

Values above 1.0 mean the candidate is faster. Resampling both sides independently propagates uncertainty from each into the ratio, which a naive ratio-of-medians throws away.

Source code in src/sns/stats.py

def ratio_ci(
    candidate: list[float],
    baseline: list[float],
    confidence: float = 0.95,
    n_resamples: int = 2000,
    seed: int = 0xC0FFEE,
) -> tuple[float, float]:
    """CI of speedup = median(baseline) / median(candidate).

    Values above 1.0 mean the candidate is faster. Resampling both sides
    independently propagates uncertainty from each into the ratio, which a
    naive ratio-of-medians throws away.
    """
    if len(candidate) < 2 or len(baseline) < 2:
        raise ValueError("ratio CI needs at least 2 samples on each side")
    rng = random.Random(seed)
    ratios = []
    for _ in range(n_resamples):
        c = statistics.median(rng.choices(candidate, k=len(candidate)))
        b = statistics.median(rng.choices(baseline, k=len(baseline)))
        if c == 0:
            raise ValueError("candidate median resampled to zero")
        ratios.append(b / c)
    ratios.sort()
    alpha = 1.0 - confidence
    lo_i = int((alpha / 2) * n_resamples)
    hi_i = min(n_resamples - 1, int((1 - alpha / 2) * n_resamples))
    return (float(ratios[lo_i]), float(ratios[hi_i]))

sns.stats.percentile

percentile(samples: list[float], p: float) -> float

Linear-interpolated percentile. p is in [0, 1].

Source code in src/sns/stats.py

def percentile(samples: list[float], p: float) -> float:
    """Linear-interpolated percentile. p is in [0, 1]."""
    if not samples:
        raise ValueError("percentile of an empty sample set")
    if not 0.0 <= p <= 1.0:
        raise ValueError(f"p must be in [0, 1], got {p}")
    s = sorted(samples)
    if len(s) == 1:
        return float(s[0])
    idx = (len(s) - 1) * p
    lo = int(idx)
    hi = min(lo + 1, len(s) - 1)
    frac = idx - lo
    return float(s[lo] * (1 - frac) + s[hi] * frac)

sns.stats.cv_percent

cv_percent(samples: list[float]) -> float

Coefficient of variation as a percentage. 0.0 for a zero mean.

Source code in src/sns/stats.py

def cv_percent(samples: list[float]) -> float:
    """Coefficient of variation as a percentage. 0.0 for a zero mean."""
    if not samples:
        raise ValueError("cv of an empty sample set")
    if len(samples) == 1:
        return 0.0
    mean = statistics.mean(samples)
    if mean == 0:
        return 0.0
    return 100.0 * statistics.pstdev(samples) / abs(mean)

Environment

sns.env.capture_fingerprint

capture_fingerprint(device_index: int = 0) -> EnvironmentFingerprint

Source code in src/sns/env.py

def capture_fingerprint(device_index: int = 0) -> EnvironmentFingerprint:
    import torch

    triton_version = None
    try:
        import triton

        triton_version = triton.__version__
    except ImportError:
        pass

    compute_cap = smi_query("compute_cap", device_index)
    sm_count = None
    if torch.cuda.is_available():
        sm_count = torch.cuda.get_device_properties(device_index).multi_processor_count
        if compute_cap is None:
            major, minor = torch.cuda.get_device_capability(device_index)
            compute_cap = f"{major}.{minor}"

    return EnvironmentFingerprint(
        torch_version=torch.__version__,
        triton_version=triton_version,
        cuda_version=torch.version.cuda,
        driver_version=smi_query("driver_version", device_index),
        gpu_name=smi_query("name", device_index),
        compute_cap=compute_cap,
        arch_family=arch_family(compute_cap),
        sm_count=sm_count,
    )

sns.env.arch_family

arch_family(compute_cap: str | None) -> str | None

Source code in src/sns/env.py

def arch_family(compute_cap: str | None) -> str | None:
    if not compute_cap:
        return None
    try:
        major, minor = (int(x) for x in str(compute_cap).split("."))
    except ValueError:
        return None
    return _ARCH_BY_CAP.get((major, minor), f"unknown-sm{major}{minor}")

Telemetry

sns.telemetry.ClockSampler

Samples SM clock and throttle state cheaply, or not at all.

If NVML is unavailable we collect NO clock evidence rather than collecting bad evidence. Tier A requires clock evidence, so a host without NVML can never reach it. That is the intended, conservative degradation.

Source code in src/sns/telemetry.py

class ClockSampler:
    """Samples SM clock and throttle state cheaply, or not at all.

    If NVML is unavailable we collect NO clock evidence rather than collecting
    bad evidence. Tier A requires clock evidence, so a host without NVML can
    never reach it. That is the intended, conservative degradation.
    """

    def __init__(self, device: int = 0):
        self.device = device
        self.available = False
        self._handle = None
        self._nvml = None
        try:
            import pynvml

            pynvml.nvmlInit()
            self._handle = pynvml.nvmlDeviceGetHandleByIndex(device)
            self._nvml = pynvml
            self.available = True
        except Exception:
            self.available = False

    def sample_clock_mhz(self) -> float | None:
        if not self.available:
            return None
        try:
            return float(
                self._nvml.nvmlDeviceGetClockInfo(
                    self._handle, self._nvml.NVML_CLOCK_SM
                )
            )
        except Exception:
            return None

    def throttled_now(self) -> bool | None:
        """True if any throttle reason other than 'GpuIdle' is currently set."""
        if not self.available:
            return None
        try:
            reasons = self._nvml.nvmlDeviceGetCurrentClocksThrottleReasons(
                self._handle
            )
        except Exception:
            return None
        # Bit 0 is nvmlClocksThrottleReasonGpuIdle, which is not a throttle we
        # care about — an idle GPU between iterations is expected.
        idle_bit = getattr(self._nvml, "nvmlClocksThrottleReasonGpuIdle", 1)
        return bool(reasons & ~idle_bit & ~0x1)

    def shutdown(self) -> None:
        if self.available:
            try:
                self._nvml.nvmlShutdown()
            except Exception:
                pass

throttled_now

throttled_now() -> bool | None

True if any throttle reason other than 'GpuIdle' is currently set.

Source code in src/sns/telemetry.py

def throttled_now(self) -> bool | None:
    """True if any throttle reason other than 'GpuIdle' is currently set."""
    if not self.available:
        return None
    try:
        reasons = self._nvml.nvmlDeviceGetCurrentClocksThrottleReasons(
            self._handle
        )
    except Exception:
        return None
    # Bit 0 is nvmlClocksThrottleReasonGpuIdle, which is not a throttle we
    # care about — an idle GPU between iterations is expected.
    idle_bit = getattr(self._nvml, "nvmlClocksThrottleReasonGpuIdle", 1)
    return bool(reasons & ~idle_bit & ~0x1)