Spaces:
Running
Running
File size: 6,161 Bytes
e313085 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 | """CPU validation with trusted controls; no model-generated source or GPU kernel is executed."""
import math
import pytest
torch = pytest.importorskip("torch")
from kernel_env.server.worker import FROBENIUS, SOFTMAX, adversarial_inputs, numerical_error
def error(ref, actual, task):
return numerical_error(torch, ref, actual, task, atol=1e-2, rtol=1e-2)
def softmax_input():
return torch.rand(4, 8192, generator=torch.Generator().manual_seed(7))
@pytest.mark.parametrize("failure", ["zeros", "chunks", "wrong_stride", "incomplete", "wrong_rows"])
def test_softmax_old_tolerance_accepts_wrong_results(failure):
x = softmax_input()
ref = torch.softmax(x, dim=1)
if failure == "zeros":
actual = torch.zeros_like(x)
elif failure == "chunks":
actual = torch.softmax(x.reshape(4, -1, 1024), dim=-1).reshape_as(x)
elif failure == "wrong_stride":
# The audited launch used stride(1) for rows, leaving most output unwritten.
actual = torch.zeros_like(x)
for row in range(x.shape[0]):
actual.view(-1)[row:row + 1024] = torch.softmax(x.view(-1)[row:row + 1024], dim=0)
elif failure == "incomplete":
actual = ref.clone()
actual[:, 1024:] = 0
else:
actual = ref.roll(1, dims=0) # sums still equal 1; invariant alone is insufficient
assert torch.allclose(actual, ref, atol=1e-2, rtol=1e-2) # reproduce the old false positive
assert error(ref, actual, SOFTMAX) is not None
def test_full_size_softmax_probabilities_are_below_old_absolute_tolerance():
# rand() lies in [0,1], so every reference probability is bounded by e / width.
assert math.e / 393216 < 1e-2
@pytest.mark.parametrize("trial", range(5))
def test_softmax_valid_control_on_adversarial_inputs(trial):
x = adversarial_inputs(torch, [softmax_input()], SOFTMAX, trial)[0]
ref = torch.softmax(x, dim=1)
exp = torch.exp(x - x.amax(dim=1, keepdim=True))
actual = exp / exp.sum(dim=1, keepdim=True)
assert error(ref, actual, SOFTMAX) is None
if trial == 2:
assert ref.max() > 0.99
assert len(set(ref.argmax(dim=1).tolist())) == x.shape[0]
@pytest.mark.parametrize("failure", ["zeros", "per_slice", "incomplete", "wrong_norm"])
def test_frobenius_old_tolerance_accepts_wrong_results(failure):
x = torch.rand(4, 3, 256, 256, generator=torch.Generator().manual_seed(11))
ref = x / torch.linalg.vector_norm(x)
if failure == "zeros":
actual = torch.zeros_like(x)
elif failure == "per_slice":
actual = x / torch.linalg.vector_norm(x, dim=(-2, -1), keepdim=True)
elif failure == "incomplete":
actual = ref.clone()
actual[1:] = 0
else:
actual = ref * 0.5
assert torch.allclose(actual, ref, atol=1e-2, rtol=1e-2)
assert error(ref, actual, FROBENIUS) is not None
@pytest.mark.parametrize("trial", range(5))
def test_frobenius_valid_control_on_adversarial_inputs(trial):
x = torch.rand(4, 3, 16, 32, generator=torch.Generator().manual_seed(11))
x = adversarial_inputs(torch, [x], FROBENIUS, trial)[0]
ref = x / torch.linalg.vector_norm(x)
actual = x / x.square().sum().sqrt()
assert error(ref, actual, FROBENIUS) is None
if trial == 3:
assert (ref != 0).sum() == 2
assert ref.flatten()[0] == pytest.approx(0.6)
assert ref.flatten()[-1] == pytest.approx(-0.8)
@pytest.mark.parametrize("task", [SOFTMAX, FROBENIUS])
@pytest.mark.parametrize("value", [float("nan"), float("inf"), -float("inf")])
def test_nonfinite_or_unwritten_normalization_output_fails(task, value):
ref = torch.full((2, 8192), 1 / 8192) if task == SOFTMAX else torch.full((2, 8192), 1 / 128)
actual = ref.clone()
actual[0, 0] = value
assert "finite" in error(ref, actual, task)
@pytest.mark.parametrize("task", [SOFTMAX, FROBENIUS])
def test_normalization_retains_relative_error_allowance(task):
ref = torch.full((2, 8192), 1 / 8192) if task == SOFTMAX else torch.full((2, 8192), 1 / 128)
assert error(ref, ref * 1.005, task) is None
assert error(ref, -ref, task) is not None
def test_unchanged_task_keeps_configured_tolerance_and_inputs():
x = torch.rand(3, 4)
for trial in range(5):
assert adversarial_inputs(torch, [x], "level1/1_Square_matrix_multiplication_", trial)[0] is x
ref = torch.zeros(2, 3)
actual = torch.full_like(ref, 0.009)
assert error(ref, actual, "level1/1_Square_matrix_multiplication_") is None
assert numerical_error(torch, ref, actual, "other", atol=0.001, rtol=0.01)
def test_worker_applies_task_specific_validation(tmp_path, monkeypatch):
from kernel_env.server.worker import _evaluate_candidate
# Trusted control module: zeros pass the old Softmax comparison, but must never be timed now.
path = tmp_path / "candidate.py"
path.write_text("""import torch
class ModelNew(torch.nn.Module):
def forward(self, x):
return torch.zeros_like(x)
""")
monkeypatch.setattr(torch.cuda, "synchronize", lambda: None)
def bench(*args):
pytest.fail("invalid output reached timing")
result = _evaluate_candidate(
{"code_path": str(path), "seeds": [1, 2, 3, 4, 5], "atol": 0.01, "rtol": 0.01, "task_id": SOFTMAX},
torch, torch.device("cpu"), lambda x: torch.softmax(x, dim=1), [],
lambda seed: [softmax_input()], bench, {"t_ref_ms": 1.0, "gpu": "cpu-test"},
)
assert result["stage"] == "correctness" and result["correct"] is False
assert "row must sum to 1" in result["feedback"]
@pytest.mark.parametrize('task,activation',[
('level1/19_ReLU',torch.relu),
('level1/20_LeakyReLU',torch.nn.functional.leaky_relu),
('level1/31_ELU',torch.nn.functional.elu),
])
def test_activation_copy_passes_old_positive_inputs_but_fails_signed_inputs(task,activation):
x=torch.rand(4,8192)
assert torch.allclose(x,activation(x),atol=.01,rtol=.01)
x=adversarial_inputs(torch,[x],task,1)[0]
assert x.min() < -16 and x.max()>16
assert numerical_error(torch,activation(x),x.clone(),task,.01,.01)
assert numerical_error(torch,activation(x),activation(x),task,.01,.01) is None
|