From f715eec102f256ff92b17510e4d1376352529822 Mon Sep 17 00:00:00 2001 From: Sam Date: Sun, 16 Aug 2026 10:56:52 -0700 Subject: [PATCH] linalg/qr: regenerate timed inputs per repeat and mirror the stream rule Close the remaining gaps from #148 in qr_py and qr_v2 (gap 1 was fixed in #150). The two eval.py files stay byte-identical. Gaps 2+3, timed-region output replay: the timed loop reused the same input objects on every repeat, and the geomean-dominant shapes have a data_list of length 1. A kernel could compute outputs during untimed calls (the warmup invocation, the pre-timing pass, earlier repeats), cache them by id() or content, and replay them inside the timed window. recheck=True does not catch this case, because a replayed output is still correct for that same input. Each timed repeat now regenerates fresh input content from a seed shift that is unique per (invocation, repeat, item), and warmup and timed invocations draw from disjoint seed_salt ranges. Untimed content therefore never reappears in a timed window, including for a cache persisted to disk across the test / benchmark / leaderboard runs of one submission. Generation and cloning happen outside the timed CUDA-event window, before clear_l2_cache(), and the warmup / pre-timing batch is bit-for-bit unchanged. The stability break (err/mean < 0.001) now requires a minimum of 10 timed samples. The regeneration and recheck work between repeats pushes wall time past the 0.1s arming threshold almost immediately on large shapes, and without a floor the break can fire on a 3-sample error estimate. The mean*runs and 120s budget caps are unchanged. An element-wise cross-check against torch.geqrf was considered and rejected: (H, tau) is not unique (sign conventions, blocking, inner precision), so exact-match checking would reject valid implementations that the tolerances are meant to admit. Behavior changes worth knowing about: a kernel that clobbers its input buffer in-place now passes benchmark/leaderboard (each repeat gets a fresh buffer; test mode already handed kernels a clone), and "mixed" benchmark lines draw a fresh profile assignment per repeat, so their mean ranks a distribution of instances rather than one fixed instance. Gap 4, the stream rule: KernelBot rejects any submission containing the substring "stream" (case-insensitive, comments included) at intake, so a stream-using kernel passes every local mode and only fails on its first remote submission. eval.py now applies the same substring test to the submission source before dispatching any mode and fails with a report entry plus the reason on stderr. Only the submission file is scanned, matching the remote rule. Validated on an RTX 5070 Ti (torch 2.8.0+cu128); this checks harness semantics, not B200 performance. old = main, new = this change, on a count=1 shape (batch 256, n 512), benchmark mode: torch.geqrf reference pass, 448.0 ms/input pass, 447.0 ms/input shape-keyed output cache pass, 5.3 us/input fail, first timed recheck content-keyed output cache pass, 203 us/input pass, 448.8 ms/input id()-keyed output cache pass, 3.0 ms/input fail, stale replay caught in-place input clobbering fail at repeat 2 pass side-stream kernel pass locally fail in every local mode Reference means agree to 0.2% between harnesses on that shape. On a large mixed shape the timed sample count went 44 (main) -> 3 (without the stability-break floor) -> 37 (with it). Reference and legitimate-optimization kernels (fp64 inner math whose factors differ element-wise from fp32 geqrf; in-place input clobbering) pass test and benchmark modes on the new harness, and the reference also passes leaderboard mode. Seed-shift uniqueness was checked exhaustively for all benchmark-line base seeds (1.25M derived seeds per line, distinct and disjoint from the warmup sequence); with a server seed the shifted values wrap mod 2^63 and disjointness is negligible-probability-of-collision rather than exact. Developed and validated in combination with Claude Fable 5. Closes #148. --- problems/linalg/qr_py/eval.py | 114 +++++++++++++++++++++++++++++----- problems/linalg/qr_v2/eval.py | 114 +++++++++++++++++++++++++++++----- 2 files changed, 196 insertions(+), 32 deletions(-) diff --git a/problems/linalg/qr_py/eval.py b/problems/linalg/qr_py/eval.py index 9b10212f0..b734cb07e 100644 --- a/problems/linalg/qr_py/eval.py +++ b/problems/linalg/qr_py/eval.py @@ -1,4 +1,5 @@ import dataclasses +import importlib.util import math import multiprocessing import os @@ -23,6 +24,23 @@ MAX_ITERATIONS_PER_BENCHMARK = 50 BENCHMARK_INPUT_BYTES_TARGET = 256 * 1024 * 1024 +# Seed-shift layout for the fresh inputs generated on every timed repeat: each +# (benchmark invocation, repeat, item) triple maps to a unique shift >= 1, so +# no timed input can repeat content from a warmup batch, an earlier repeat, or +# an earlier invocation. Exact for local task seeds; with a server-combined +# base seed the mod below can wrap, making collisions negligible-probability +# rather than impossible. _MAX_TIMED_REPEATS must exceed every max_repeats +# used below; _MAX_BATCH_ITEMS must exceed every _benchmark_batch_count value. +_MAX_TIMED_REPEATS = 1024 +_MAX_BATCH_ITEMS = MAX_ITERATIONS_PER_BENCHMARK + 1 +# Keep nested _combine results (quadratic in their inputs) inside int64 so +# torch.Generator.manual_seed accepts them. +_SEED_BOUND = 2**63 +# Minimum timed samples before the err/mean stability break may fire: the +# untimed per-repeat work (regeneration, recheck) arms the break almost +# immediately on large shapes, where a 3-sample error estimate is noise. +_MIN_STABLE_REPEATS = 10 + class PopcornOutput: def __init__(self, fd: int): @@ -165,6 +183,24 @@ def _make_data_batch(test: TestCase, count: int): return data_list +def _timed_repeat_batch(test: TestCase, count: int, seed_salt: int, repeat_index: int): + # Fresh input content for one timed repeat. The submission process survives + # across invocations, so a kernel could cache outputs from untimed calls + # (warmup, earlier repeats) and replay them inside the timed window; the + # checker would accept the replay because it is correct for that same + # input. Never reusing content in the timed loop denies every id()- or + # content-keyed replay: a stale output fails the recheck against the + # actual, fresh input. + args = dict(test.args) + data_list = [] + for item in range(count): + if "seed" in args: + shift = 1 + (seed_salt * _MAX_TIMED_REPEATS + repeat_index) * _MAX_BATCH_ITEMS + item + args["seed"] = _combine(int(test.args["seed"]), shift) % _SEED_BOUND + data_list.append(generate_input(**args)) + return data_list + + def _benchmark_batch_count(test: TestCase) -> int: batch = int(test.args.get("batch", 1)) n = int(test.args.get("n", 1)) @@ -181,10 +217,13 @@ def _run_single_benchmark( recheck: bool, max_repeats: int, max_time_ns: float, + seed_salt: int, ) -> Stats | Any: from submission import custom_kernel - data_list = _make_data_batch(test, _benchmark_batch_count(test)) + assert max_repeats < _MAX_TIMED_REPEATS + count = _benchmark_batch_count(test) + data_list = _make_data_batch(test, count) check_copy = _clone_data(data_list) outputs = [custom_kernel(_clone_data(data)) for data in data_list] @@ -196,6 +235,10 @@ def _run_single_benchmark( durations = [] bm_start_time = time.perf_counter_ns() for i in range(max_repeats): + # Regenerate inputs for every timed repeat (see _timed_repeat_batch); + # generation and cloning stay outside the timed window. + data_list = _timed_repeat_batch(test, count, seed_salt, i) + check_copy = _clone_data(data_list) if recheck else None torch.cuda.synchronize() clear_l2_cache() start_event = torch.cuda.Event(enable_timing=True) @@ -216,7 +259,7 @@ def _run_single_benchmark( if i > 1 and total_bm_duration > 1e8: stats = calculate_stats(durations) if ( - stats.err / stats.mean < 0.001 + (stats.runs >= _MIN_STABLE_REPEATS and stats.err / stats.mean < 0.001) or stats.mean * stats.runs > max_time_ns or total_bm_duration > 120e9 ): @@ -231,26 +274,27 @@ def run_single_benchmark( recheck: bool, max_repeats: int, max_time_ns: float, + seed_salt: int, ): - return pool.apply(_run_single_benchmark, (test, recheck, max_repeats, max_time_ns)) + return pool.apply(_run_single_benchmark, (test, recheck, max_repeats, max_time_ns, seed_salt)) def run_benchmarking(logger: PopcornOutput, pool: multiprocessing.Pool, tests: list[TestCase]): - run_single_benchmark(pool, tests[0], False, 200, 10e7) + # Every invocation gets a distinct seed_salt, so content the kernel saw + # while untimed (the warmup invocation) never reappears in a timed window. + run_single_benchmark(pool, tests[0], False, 200, 10e7, 0) passed = True logger.log("benchmark-count", len(tests)) for idx, test in enumerate(tests): logger.log(f"benchmark.{idx}.spec", test.spec) - # recheck=True: re-validate the output of every timed iteration, not just - # the pre-timing warmup. Without this, the timed loop (which for the - # low-`count` shapes reuses one input object across all repeats) never - # re-checks its outputs, so a kernel that diverges only inside the timed - # region -- e.g. one that caches and replays an output keyed on the - # reused input -- is scored as fast without ever being caught locally. - # `leaderboard` mode already rechecks; this brings `benchmark` mode in - # line so a wrong timed output fails here too. - result = run_single_benchmark(pool, test, True, 200, 10e9) + # recheck=True: re-validate every timed iteration against that + # iteration's freshly generated input, not just the pre-timing warmup. + # Combined with the per-repeat regeneration, a kernel that replays a + # stored output inside the timed region fails here: the timed content + # is never something it has seen before. `leaderboard` mode already + # rechecks; this keeps `benchmark` mode in line. + result = run_single_benchmark(pool, test, True, 200, 10e9, 1 + idx) if isinstance(result, Stats): for field in dataclasses.fields(Stats): logger.log(f"benchmark.{idx}.{field.name}", getattr(result, field.name)) @@ -295,6 +339,31 @@ def run_profiling(logger: PopcornOutput, pool: multiprocessing.Pool, tests: list return 0 +def _stream_rule_error() -> str | None: + # KernelBot rejects any submission whose source contains the substring + # "stream" (case-insensitive) before it ever runs, because non-default + # streams can escape the timed region. That gate lives in the submission + # intake, so locally a stream-using kernel passes every mode and is only + # rejected on its first remote submission. Mirror the rule here for parity. + spec = importlib.util.find_spec("submission") + if spec is not None and spec.origin: + submission_path = Path(spec.origin) + else: + submission_path = Path(__file__).resolve().with_name("submission.py") + try: + source = submission_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return None + if "stream" in source.lower(): + return ( + "submission.py contains the substring 'stream' (case-insensitive), " + "which the leaderboard rejects at submission time in any form, " + "including comments and identifiers. Work on non-default CUDA " + "streams is not allowed; remove every occurrence before submitting." + ) + return None + + def main(): fd = os.getenv("POPCORN_FD") if not fd: @@ -310,6 +379,17 @@ def main(): tests = get_test_cases(sys.argv[2], seed) with PopcornOutput(int(fd)) as logger: + stream_error = _stream_rule_error() + if stream_error is not None: + section = "test" if mode == "test" else "benchmark" + logger.log(f"{section}-count", 1) + logger.log(f"{section}.0.spec", "stream-rule") + logger.log(f"{section}.0.status", "fail") + logger.log(f"{section}.0.error", stream_error) + logger.log("check", "fail") + print(stream_error, file=sys.stderr) + return 112 + mp_context = multiprocessing.get_context("spawn") with mp_context.Pool(1) as pool: if mode == "test": @@ -317,13 +397,15 @@ def main(): if mode == "benchmark": return run_benchmarking(logger, pool, tests) if mode == "leaderboard": - for test in tests: - run_single_benchmark(pool, test, False, 1000, 5e8) + # Warmup salts 1..len(tests); timed salts start after them, so + # no timed repeat reuses content from any warmup invocation. + for idx, test in enumerate(tests): + run_single_benchmark(pool, test, False, 1000, 5e8, 1 + idx) logger.log("benchmark-count", len(tests)) passed = True for idx, test in enumerate(tests): logger.log(f"benchmark.{idx}.spec", test.spec) - result = run_single_benchmark(pool, test, True, 1000, 30e9) + result = run_single_benchmark(pool, test, True, 1000, 30e9, 1 + len(tests) + idx) if isinstance(result, Stats): for field in dataclasses.fields(Stats): logger.log(f"benchmark.{idx}.{field.name}", getattr(result, field.name)) diff --git a/problems/linalg/qr_v2/eval.py b/problems/linalg/qr_v2/eval.py index 9b10212f0..b734cb07e 100644 --- a/problems/linalg/qr_v2/eval.py +++ b/problems/linalg/qr_v2/eval.py @@ -1,4 +1,5 @@ import dataclasses +import importlib.util import math import multiprocessing import os @@ -23,6 +24,23 @@ MAX_ITERATIONS_PER_BENCHMARK = 50 BENCHMARK_INPUT_BYTES_TARGET = 256 * 1024 * 1024 +# Seed-shift layout for the fresh inputs generated on every timed repeat: each +# (benchmark invocation, repeat, item) triple maps to a unique shift >= 1, so +# no timed input can repeat content from a warmup batch, an earlier repeat, or +# an earlier invocation. Exact for local task seeds; with a server-combined +# base seed the mod below can wrap, making collisions negligible-probability +# rather than impossible. _MAX_TIMED_REPEATS must exceed every max_repeats +# used below; _MAX_BATCH_ITEMS must exceed every _benchmark_batch_count value. +_MAX_TIMED_REPEATS = 1024 +_MAX_BATCH_ITEMS = MAX_ITERATIONS_PER_BENCHMARK + 1 +# Keep nested _combine results (quadratic in their inputs) inside int64 so +# torch.Generator.manual_seed accepts them. +_SEED_BOUND = 2**63 +# Minimum timed samples before the err/mean stability break may fire: the +# untimed per-repeat work (regeneration, recheck) arms the break almost +# immediately on large shapes, where a 3-sample error estimate is noise. +_MIN_STABLE_REPEATS = 10 + class PopcornOutput: def __init__(self, fd: int): @@ -165,6 +183,24 @@ def _make_data_batch(test: TestCase, count: int): return data_list +def _timed_repeat_batch(test: TestCase, count: int, seed_salt: int, repeat_index: int): + # Fresh input content for one timed repeat. The submission process survives + # across invocations, so a kernel could cache outputs from untimed calls + # (warmup, earlier repeats) and replay them inside the timed window; the + # checker would accept the replay because it is correct for that same + # input. Never reusing content in the timed loop denies every id()- or + # content-keyed replay: a stale output fails the recheck against the + # actual, fresh input. + args = dict(test.args) + data_list = [] + for item in range(count): + if "seed" in args: + shift = 1 + (seed_salt * _MAX_TIMED_REPEATS + repeat_index) * _MAX_BATCH_ITEMS + item + args["seed"] = _combine(int(test.args["seed"]), shift) % _SEED_BOUND + data_list.append(generate_input(**args)) + return data_list + + def _benchmark_batch_count(test: TestCase) -> int: batch = int(test.args.get("batch", 1)) n = int(test.args.get("n", 1)) @@ -181,10 +217,13 @@ def _run_single_benchmark( recheck: bool, max_repeats: int, max_time_ns: float, + seed_salt: int, ) -> Stats | Any: from submission import custom_kernel - data_list = _make_data_batch(test, _benchmark_batch_count(test)) + assert max_repeats < _MAX_TIMED_REPEATS + count = _benchmark_batch_count(test) + data_list = _make_data_batch(test, count) check_copy = _clone_data(data_list) outputs = [custom_kernel(_clone_data(data)) for data in data_list] @@ -196,6 +235,10 @@ def _run_single_benchmark( durations = [] bm_start_time = time.perf_counter_ns() for i in range(max_repeats): + # Regenerate inputs for every timed repeat (see _timed_repeat_batch); + # generation and cloning stay outside the timed window. + data_list = _timed_repeat_batch(test, count, seed_salt, i) + check_copy = _clone_data(data_list) if recheck else None torch.cuda.synchronize() clear_l2_cache() start_event = torch.cuda.Event(enable_timing=True) @@ -216,7 +259,7 @@ def _run_single_benchmark( if i > 1 and total_bm_duration > 1e8: stats = calculate_stats(durations) if ( - stats.err / stats.mean < 0.001 + (stats.runs >= _MIN_STABLE_REPEATS and stats.err / stats.mean < 0.001) or stats.mean * stats.runs > max_time_ns or total_bm_duration > 120e9 ): @@ -231,26 +274,27 @@ def run_single_benchmark( recheck: bool, max_repeats: int, max_time_ns: float, + seed_salt: int, ): - return pool.apply(_run_single_benchmark, (test, recheck, max_repeats, max_time_ns)) + return pool.apply(_run_single_benchmark, (test, recheck, max_repeats, max_time_ns, seed_salt)) def run_benchmarking(logger: PopcornOutput, pool: multiprocessing.Pool, tests: list[TestCase]): - run_single_benchmark(pool, tests[0], False, 200, 10e7) + # Every invocation gets a distinct seed_salt, so content the kernel saw + # while untimed (the warmup invocation) never reappears in a timed window. + run_single_benchmark(pool, tests[0], False, 200, 10e7, 0) passed = True logger.log("benchmark-count", len(tests)) for idx, test in enumerate(tests): logger.log(f"benchmark.{idx}.spec", test.spec) - # recheck=True: re-validate the output of every timed iteration, not just - # the pre-timing warmup. Without this, the timed loop (which for the - # low-`count` shapes reuses one input object across all repeats) never - # re-checks its outputs, so a kernel that diverges only inside the timed - # region -- e.g. one that caches and replays an output keyed on the - # reused input -- is scored as fast without ever being caught locally. - # `leaderboard` mode already rechecks; this brings `benchmark` mode in - # line so a wrong timed output fails here too. - result = run_single_benchmark(pool, test, True, 200, 10e9) + # recheck=True: re-validate every timed iteration against that + # iteration's freshly generated input, not just the pre-timing warmup. + # Combined with the per-repeat regeneration, a kernel that replays a + # stored output inside the timed region fails here: the timed content + # is never something it has seen before. `leaderboard` mode already + # rechecks; this keeps `benchmark` mode in line. + result = run_single_benchmark(pool, test, True, 200, 10e9, 1 + idx) if isinstance(result, Stats): for field in dataclasses.fields(Stats): logger.log(f"benchmark.{idx}.{field.name}", getattr(result, field.name)) @@ -295,6 +339,31 @@ def run_profiling(logger: PopcornOutput, pool: multiprocessing.Pool, tests: list return 0 +def _stream_rule_error() -> str | None: + # KernelBot rejects any submission whose source contains the substring + # "stream" (case-insensitive) before it ever runs, because non-default + # streams can escape the timed region. That gate lives in the submission + # intake, so locally a stream-using kernel passes every mode and is only + # rejected on its first remote submission. Mirror the rule here for parity. + spec = importlib.util.find_spec("submission") + if spec is not None and spec.origin: + submission_path = Path(spec.origin) + else: + submission_path = Path(__file__).resolve().with_name("submission.py") + try: + source = submission_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return None + if "stream" in source.lower(): + return ( + "submission.py contains the substring 'stream' (case-insensitive), " + "which the leaderboard rejects at submission time in any form, " + "including comments and identifiers. Work on non-default CUDA " + "streams is not allowed; remove every occurrence before submitting." + ) + return None + + def main(): fd = os.getenv("POPCORN_FD") if not fd: @@ -310,6 +379,17 @@ def main(): tests = get_test_cases(sys.argv[2], seed) with PopcornOutput(int(fd)) as logger: + stream_error = _stream_rule_error() + if stream_error is not None: + section = "test" if mode == "test" else "benchmark" + logger.log(f"{section}-count", 1) + logger.log(f"{section}.0.spec", "stream-rule") + logger.log(f"{section}.0.status", "fail") + logger.log(f"{section}.0.error", stream_error) + logger.log("check", "fail") + print(stream_error, file=sys.stderr) + return 112 + mp_context = multiprocessing.get_context("spawn") with mp_context.Pool(1) as pool: if mode == "test": @@ -317,13 +397,15 @@ def main(): if mode == "benchmark": return run_benchmarking(logger, pool, tests) if mode == "leaderboard": - for test in tests: - run_single_benchmark(pool, test, False, 1000, 5e8) + # Warmup salts 1..len(tests); timed salts start after them, so + # no timed repeat reuses content from any warmup invocation. + for idx, test in enumerate(tests): + run_single_benchmark(pool, test, False, 1000, 5e8, 1 + idx) logger.log("benchmark-count", len(tests)) passed = True for idx, test in enumerate(tests): logger.log(f"benchmark.{idx}.spec", test.spec) - result = run_single_benchmark(pool, test, True, 1000, 30e9) + result = run_single_benchmark(pool, test, True, 1000, 30e9, 1 + len(tests) + idx) if isinstance(result, Stats): for field in dataclasses.fields(Stats): logger.log(f"benchmark.{idx}.{field.name}", getattr(result, field.name))