main
py 90 lines 2.44 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4 import pytest
5 import torch
6
7 from vllm.utils.import_utils import has_helion
8
9 if not has_helion():
10 pytest.skip("Helion is not installed", allow_module_level=True)
11
12 from tools import benchmark_helion_kernels
13
14 check_correctness = benchmark_helion_kernels.check_correctness
15
16
17 class _FakeKernel:
18 helion_settings = None
19
20 def __init__(self, offset: float):
21 self.calls = 0
22 self.offset = offset
23
24 def __call__(self, output: torch.Tensor, input: torch.Tensor) -> torch.Tensor:
25 self.calls += 1
26 output.copy_(input + self.offset)
27 return input * 2
28
29
30 def test_check_correctness_runs_once_without_mutating_benchmark_inputs():
31 kernel = _FakeKernel(offset=1)
32 baseline_calls = 0
33
34 def baseline(output: torch.Tensor, input: torch.Tensor) -> torch.Tensor:
35 nonlocal baseline_calls
36 baseline_calls += 1
37 output.copy_(input + 1)
38 return input * 2
39
40 output = torch.zeros(4)
41 inputs = (output, torch.arange(4))
42
43 check_correctness(kernel, baseline, inputs, "matching")
44
45 assert kernel.calls == 1
46 assert baseline_calls == 1
47 torch.testing.assert_close(output, torch.zeros(4))
48
49
50 def test_check_correctness_reports_mutated_input_mismatch():
51 kernel = _FakeKernel(offset=2)
52
53 def baseline(output: torch.Tensor, input: torch.Tensor) -> torch.Tensor:
54 output.copy_(input + 1)
55 return input * 2
56
57 inputs = (torch.zeros(4), torch.arange(4))
58
59 with pytest.raises(AssertionError, match="Numerics check failed for case bad"):
60 check_correctness(kernel, baseline, inputs, "bad")
61
62
63 def test_check_correctness_reports_return_value_mismatch():
64 kernel = _FakeKernel(offset=1)
65
66 def baseline(output: torch.Tensor, input: torch.Tensor) -> torch.Tensor:
67 output.copy_(input + 1)
68 return input * 3
69
70 inputs = (torch.zeros(4), torch.arange(4))
71
72 with pytest.raises(AssertionError, match="Numerics check failed for case bad"):
73 check_correctness(kernel, baseline, inputs, "bad")
74
75
76 def test_log_versions(monkeypatch):
77 messages = []
78
79 def record(message: str, *args):
80 messages.append(message % args)
81
82 monkeypatch.setattr(benchmark_helion_kernels.logger, "info", record)
83
84 benchmark_helion_kernels.log_versions()
85
86 assert [message.split(":", 1)[0] for message in messages] == [
87 "torch",
88 "helion",
89 "triton",
90 ]