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))