Skip to content

linalg/qr: regenerate timed inputs per repeat and mirror the stream rule #170

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
SamMausberg wants to merge 1 commit into
base: main
Choose a base branch
from
Open

linalg/qr: regenerate timed inputs per repeat and mirror the stream rule #170

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 98 additions & 16 deletions problems/linalg/qr_py/eval.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import dataclasses
import importlib.util
import math
import multiprocessing
import os
Expand All @@ -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):
Expand Down Expand Up @@ -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))
Expand All @@ -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]
Expand All @@ -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)
Expand All @@ -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
):
Expand All @@ -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))
Expand Down Expand Up @@ -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:
Expand All @@ -310,20 +379,33 @@ 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":
return run_testing(logger, pool, tests)
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))
Expand Down
114 changes: 98 additions & 16 deletions problems/linalg/qr_v2/eval.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import dataclasses
import importlib.util
import math
import multiprocessing
import os
Expand All @@ -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):
Expand Down Expand Up @@ -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))
Expand All @@ -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]
Expand All @@ -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)
Expand All @@ -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
):
Expand All @@ -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))
Expand Down Expand Up @@ -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:
Expand All @@ -310,20 +379,33 @@ 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":
return run_testing(logger, pool, tests)
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))
Expand Down