JustinTX's picture
Add files using upload-large-folder tool
2facf1f verified
Raw
History Blame Contribute Delete
28.5 kB
import base64
import copy
import dataclasses
import datetime
import functools
import json
import os
import shlex
import shutil
import socket
import subprocess
import tempfile
import time
from pathlib import Path
from types import NoneType
from typing import Optional, Protocol, Union
from libkernelbot.consts import CUDA_FLAGS, ExitCode, Timeout
@dataclasses.dataclass
class ProfileResult:
# fmt: off
profiler: str # The profiler used to gather this data
# Profiler trace. May be empty, in which case `download_url`
# should point to the trace file.
trace: str
# Public download URL of all files created by the profiler
# This may also be configured later
download_url: Optional[str]
# fmt: on
@dataclasses.dataclass
class CompileResult:
# fmt: off
nvcc_found: bool # did we find nvcc?
nvcc_version: str # the result of nvcc --version
success: bool # did it compile successfully
command: str # the command that was run to compile the code
stdout: str # standard output produced by the compiler
stderr: str # standard error produced by the compiler
exit_code: int # exit code produced by the compiler
# fmt: on
@dataclasses.dataclass
class RunResult:
# fmt: off
success: bool # did the compiled executable run successfully
passed: bool # did it pass all tests
command: str # the command that was run to compile the code
stdout: str # standard output produced by the compiler
stderr: str # standard error produced by the compiler
exit_code: int # exit code produced by the compiler
duration: float # execution time (NOT kernel duration)
result: dict # dictionary with the results generated by the tester
# fmt: on
@dataclasses.dataclass
class SystemInfo:
# fmt: off
gpu: str = '' # Model name of the GPU
device_count: int = 1 # Number of GPUs
cpu: str = '' # Model name of the CPU
runtime: str = '' # Whether CUDA or ROCm
platform: str = '' # Platform string of the machine
torch: str = '' # Torch version
hostname: str = '' # Hostname of the machine
requeues: int = 0 # How many Modal requeues/retries were used before running
# fmt: on
@dataclasses.dataclass
class EvalResult:
# fmt: off
start: datetime.datetime # when did this run start (excluding container setup time)
end: datetime.datetime # and when did it finish
compilation: CompileResult | None # results of compilation
run: RunResult | None # result of actually running the executable/script
profile: ProfileResult | None # result of profiling the executable
# fmt: on
@dataclasses.dataclass
class FullResult:
# fmt: off
success: bool # did the runner (github/modal) execute successfully
error: str # if not success, an error message
system: SystemInfo # specs of the system this was run on
# results of running. There can be multiple runs in one submission, using separate
# 'test' and 'benchmark' keys, for example
runs: dict[str, EvalResult] = dataclasses.field(default_factory=dict)
# fmt: on
def _make_cmd(args: list[str]):
return " ".join(map(shlex.quote, args))
def _limit_length(text: Union[NoneType, str, bytes], max_len: int = 16384):
if text is None:
return ""
if isinstance(text, bytes):
text = text.decode("utf-8")
lines = text.split("\n")
size = 0
for i, line in enumerate(lines):
size += len(line) + 1
if size + 100 > max_len:
lines = lines[:i] + [f"[...] {len(lines) - i} lines omitted"]
return "\n".join(lines)
return text
def _create_files(files: Optional[dict[str, str]]):
"""
Create text files
Args:
files: A dictionary mapping file names to their contents.
Raises:
AssertionError, if the file is not within the current working directory.
"""
if files is None:
return
for name, content in files.items():
assert Path(name).resolve().is_relative_to(Path.cwd())
Path(name).write_text(content)
def _directory_to_zip_bytes(directory_path) -> str:
"""Create a zip archive and return as base64 encoded bytes."""
with tempfile.TemporaryDirectory() as temp_dir:
archive_path = os.path.join(temp_dir, "archive")
shutil.make_archive(archive_path, "zip", directory_path)
with open(archive_path + ".zip", "rb") as f:
data = f.read()
return base64.b64encode(data).decode("utf-8")
def _filter_ncu_report(report: str, tables: list): # noqa: C901
"""
Extract the Speed-of-light section from the full ncu terminal report.
For expert users, we just attach the full ncu profile to the result,
and they can view whichever metrics they are interested in. But to
encourage novice users to try out profiling, we want to have a
*simple* set of things to display automatically, short enough to fit
in a *single* discord message.
"""
result = ""
n_kernels = 0
collect = False
length = 0
for line in report.splitlines():
if len(line) >= 3 and line[1] == " " and line[2] != " ":
if n_kernels != 0:
result += "\n"
n_kernels += 1
if n_kernels == 3:
result += "\nAdditional kernel launches follow. Please check the .ncu-rep file for more details.\n" # noqa: E501
result += line + "\n"
if n_kernels > 2:
continue
if "Table Name : " in line:
table = line[line.find("Table Name :") + len("Table Name :") :].strip()
if table in tables:
result += "\n"
collect = True
else:
collect = False
if len(line.strip()) == 0:
collect = False
if collect:
result += line + "\n"
length += 1
# just as a precaution, also limit lines directly
if length > 100:
result += "\n[...]\nReport has been truncated. Please check the .ncu-rep file for more details.\n" # noqa: E501
break
return result
def compile_cuda_script( # # noqa: C901
files: list[str],
arch: Optional[int] = None,
include_dirs: Optional[list[str]] = None,
defines: Optional[dict[str, str]] = None,
libraries: Optional[list[str]] = None,
flags: Optional[list[str]] = None,
verbose: bool = False,
) -> CompileResult:
"""
Compiles a set of cuda files with nvcc.
Args:
files: List of files to compile.
arch: Architecture to compile for. If None, uses `native`
include_dirs: additional include directories to supply to nvcc
defines: Additional defines for the preprocessor
libraries: Additional libraries to link to
flags: Other compiler flags
verbose: whether to print progress or be silent
Returns:
A `CompileResult` that summarizes the compilation process.
"""
if flags is None:
flags = CUDA_FLAGS
if include_dirs is not None:
flags += [f"-I{d}" for d in include_dirs]
# validate include directories
for directory in include_dirs:
if not Path(directory).exists():
raise FileNotFoundError(f"Directory `{directory}` does not exist")
elif not Path(directory).is_dir():
raise NotADirectoryError(f"`{directory}` is not a directory")
if libraries is not None:
flags += [f"-l{lib}" for lib in libraries]
if defines is not None:
for name, value in defines.items():
# restrict macro names to valid identifiers
if not name.isidentifier():
raise ValueError(f"Define key `{name}` contains invalid character")
if value is not None:
flags.append(f"-D{name}={value}")
else:
flags.append(f"-D{name}")
for flag in flags:
if not flag.startswith("-"):
raise ValueError(f"Flag `{flag}` should start with a dash.")
if verbose:
print_ = print
else:
print_ = lambda *args, **kwargs: None # noqa
# Check CUDA is available and installed correctly
print_("[CUDA Env Check]")
try:
# these check cuda compiler is also available
nvcc = subprocess.check_output(["which", "nvcc"], encoding="utf-8").strip()
nvcc_version = subprocess.check_output(["nvcc", "--version"], encoding="utf-8")
except subprocess.CalledProcessError as e:
return CompileResult(
nvcc_found=False,
success=False,
nvcc_version="",
command=_make_cmd(e.cmd),
stdout=_limit_length(e.stdout),
stderr=_limit_length(e.stderr),
exit_code=e.returncode,
)
if arch is None:
ARCH = "-arch=native"
else:
ARCH = f"-gencode=arch=compute_{arch},code=sm_{arch}"
command = [nvcc] + flags + files + [ARCH, "-o", "eval.out"]
print_("[Compiling]")
try:
compile_process = subprocess.run(
command, capture_output=True, text=True, check=True, timeout=Timeout.COMPILE
)
except subprocess.CalledProcessError as e:
return CompileResult(
nvcc_found=True,
success=False,
nvcc_version=nvcc_version,
command=_make_cmd(e.cmd),
stdout=_limit_length(e.stdout),
stderr=_limit_length(e.stderr),
exit_code=e.returncode,
)
return CompileResult(
nvcc_found=True,
success=True,
nvcc_version=nvcc_version,
command=_make_cmd(compile_process.args),
stdout=_limit_length(compile_process.stdout),
stderr=_limit_length(compile_process.stderr),
exit_code=compile_process.returncode,
)
def run_program(
args: list[str],
seed: Optional[int],
timeout: int,
multi_gpu: bool = False,
extra_env: Optional[dict[str, str]] = None,
) -> RunResult:
print("[Running]")
# set up a pipe so the tester can communicate its verdict with us
env = os.environ.copy()
if extra_env is not None:
env.update(extra_env)
pipe_read, pipe_write = os.pipe()
env["POPCORN_FD"] = str(pipe_write)
if seed is not None:
env["POPCORN_SEED"] = str(seed)
if multi_gpu:
import torch
env["POPCORN_GPUS"] = str(torch.cuda.device_count())
execution_start_time = time.perf_counter()
try:
run_process = subprocess.run(
args,
capture_output=True,
text=True,
check=False,
env=env,
pass_fds=[pipe_write],
timeout=timeout,
)
except subprocess.TimeoutExpired as e:
return RunResult(
success=False,
passed=False,
command=_make_cmd(e.cmd),
stdout=_limit_length(e.stdout),
stderr=_limit_length(e.stderr),
exit_code=ExitCode.TIMEOUT_EXPIRED,
duration=timeout,
result={},
)
execution_end_time = time.perf_counter()
# terminate output writing
os.close(pipe_write)
# and fetch pipe's content
result = os.fdopen(pipe_read, "r").read()
result_dict = {}
for line in result.splitlines():
key, _, value = line.partition(":")
if key != "" or value != "":
result_dict[key.strip()] = value.strip()
return RunResult(
success=(
run_process.returncode == ExitCode.SUCCESS
or run_process.returncode == ExitCode.VALIDATE_FAIL
),
passed=result_dict.get("check", None) == "pass",
command=_make_cmd(run_process.args),
stdout=_limit_length(run_process.stdout),
stderr=_limit_length(run_process.stderr),
exit_code=run_process.returncode,
duration=execution_end_time - execution_start_time,
result=result_dict,
)
def profile_program_roc(
call: list[str],
seed: Optional[int],
timeout: int,
multi_gpu: bool,
output_dir: Path,
) -> tuple[RunResult, Optional[ProfileResult]]:
# Wrap program in rocprof
call = [
"rocprofv3",
"--log-level",
"fatal",
"--hip-trace",
"--kernel-trace",
"--rccl-trace",
"--marker-trace",
"--hip-trace",
"--memory-copy-trace",
# New? Doesn't work in the runner
# "--memory-allocation-trace",
"--scratch-memory-trace",
# The HSA trace output is very large, so skip it for now
# "--hsa-trace",
"--output-format",
"pftrace",
"csv",
"-d",
str(output_dir),
# Just store the files as %pid%_tracename.ext instead of putting them in an
# additional directory named after the hostname.
"-o",
# Insert an extra path here so that the resulting zip has all files
# in the profile_data/ directory rather than the root.
"%pid%",
"--",
] + call
run_result = run_program(
call,
seed=seed,
timeout=timeout,
multi_gpu=multi_gpu,
extra_env={
"GPU_DUMP_CODE_OBJECT": "1",
},
)
profile_result = None
if run_result.success:
# Post-process trace data.
# rocPROF generates one trace for every process, but its more useful to
# have all traces be in the same file. Fortunately we can do that by
# concatenating.
traces = list(output_dir.glob("*.pftrace"))
with (output_dir / "combined.pftrace").open("wb") as combined:
for trace_path in traces:
with trace_path.open("rb") as trace:
shutil.copyfileobj(trace, combined)
# After we've created the combined trace, there is no point in
# keeping the individual traces around.
trace_path.unlink()
# Also move the code objects to the profiling output directory.
for code_obj in list(Path.cwd().glob("_code_object*.o")):
code_obj.rename(output_dir / code_obj.name)
profile_result = ProfileResult(
profiler="rocPROF",
trace=_directory_to_zip_bytes(output_dir),
download_url=None,
)
return run_result, profile_result
def profile_program_ncu(
call: list[str],
seed: Optional[int],
timeout: int,
multi_gpu: bool,
output_dir: Path,
) -> tuple[RunResult, Optional[ProfileResult]]:
assert not multi_gpu, "Multi-GPU profiling not supported for ncu."
# Wrap program in ncu
call = [
"ncu",
"--set",
"full",
"--nvtx",
"--nvtx-include",
"custom_kernel/",
"--import-source",
"1",
"-c",
"10",
"-o",
f"{str(output_dir / 'profile.ncu-rep')}",
"--",
] + call
run_result = run_program(
call, seed=seed, timeout=timeout, multi_gpu=multi_gpu, extra_env={"POPCORN_NCU": "1"}
)
profile_result = None
try:
get_tables = [
"GPU Throughput",
"Pipe Utilization (% of active cycles)",
"Warp State (All Cycles)",
]
ncu_cmd = [
"ncu",
"--import",
f"{str(output_dir / 'profile.ncu-rep')}",
"--print-details",
"body",
]
report = subprocess.check_output(ncu_cmd, text=True)
report = _filter_ncu_report(report, get_tables)
run_result.result["benchmark.0.report"] = base64.b64encode(report.encode("utf-8")).decode(
"utf-8"
)
except subprocess.CalledProcessError:
pass
if run_result.success:
profile_result = ProfileResult(
profiler="Nsight-Compute",
trace=_directory_to_zip_bytes(output_dir),
download_url=None,
)
return run_result, profile_result
def profile_program(
system: SystemInfo,
call: list[str],
seed: Optional[int],
timeout: int,
multi_gpu: bool,
) -> tuple[RunResult, Optional[ProfileResult]]:
# The runner-specific configuration should implement logic
# to fetch the data in this directory and return it as
# ProfileResult.download_url.
# Insert an extra nested path here so that the resulting zip has all files
# in the profile_data/ directory rather than directly in the root.
with tempfile.TemporaryDirectory(dir=".") as tmpdir:
output_dir = Path(tmpdir) / "profile_data"
output_dir.mkdir()
if system.runtime == "ROCm":
return profile_program_roc(call, seed, timeout, multi_gpu, output_dir)
elif system.runtime == "CUDA":
return profile_program_ncu(call, seed, timeout, multi_gpu, output_dir)
else:
raise ValueError(f"Unknown runtime {system.runtime}")
def run_single_evaluation(
call: list[str],
mode: str,
*,
system: SystemInfo,
multi_gpu: bool = False,
tests: Optional[str] = None,
benchmarks: Optional[str] = None,
test_timeout: int = Timeout.TEST,
benchmark_timeout: int = Timeout.BENCHMARK,
ranked_timeout: int = Timeout.RANKED,
ranking_by: str = "last",
seed: Optional[int] = None,
) -> tuple[RunResult, Optional[ProfileResult]]:
"""
A single runner run, either in the context of test files, or in the
context of benchmark files.
"""
with tempfile.NamedTemporaryFile("w") as cases:
if mode == "test":
timeout = test_timeout
cases.write(tests)
elif mode in ["benchmark", "profile", "leaderboard"]:
timeout = ranked_timeout if mode == "leaderboard" else benchmark_timeout
if ranking_by == "last":
cases.write(benchmarks.splitlines(keepends=True)[-1])
else:
cases.write(benchmarks)
else:
raise ValueError(f"Invalid mode {mode}")
cases.flush()
call = call + [mode, cases.name]
if mode == "profile":
return profile_program(system, call, seed=seed, timeout=timeout, multi_gpu=multi_gpu)
return run_program(call, seed=seed, timeout=timeout, multi_gpu=multi_gpu), None
def make_system_info() -> SystemInfo: # noqa: C901
info = SystemInfo()
try:
import torch
info.torch = torch.torch_version.internal_version
# Note: cuda.is_available() also covers HiP
# https://pytorch.org/docs/stable/notes/hip.html
if torch.cuda.is_available():
info.gpu = torch.cuda.get_device_name()
info.device_count = torch.cuda.device_count()
if torch.version.hip is not None:
info.runtime = "ROCm"
elif torch.version.cuda is not None:
info.runtime = "CUDA"
except ImportError:
# get GPU info manually
try:
info.gpu = subprocess.check_output(
["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"], encoding="utf-8"
)
info.device_count = info.gpu.count("\n")
info.runtime = "CUDA"
except subprocess.CalledProcessError:
# try again for HIP
try:
rocm_info = json.loads(
subprocess.check_output(
["rocm-smi", "--showproductname", "--json"], encoding="utf-8"
)
)
if len(rocm_info) > 0:
info.gpu = next(rocm_info.__iter__())["Card Series"]
info.device_count = len(rocm_info)
info.runtime = "ROCm"
except subprocess.CalledProcessError:
# OK, no GPU info available
pass
try:
cpu_info_str = Path("/proc/cpuinfo").read_text()
cpu_info_dict = {}
for line in cpu_info_str.splitlines():
key, _, val = line.partition(":")
cpu_info_dict[key.strip()] = val.strip()
info.cpu = cpu_info_dict.get("model name", "")
# on modal, we don't get to know the exact CPU model
# make due with the vendor in that case
if info.cpu == "unknown":
# ¯\_(ツ)_/¯
info.cpu = cpu_info_dict.get("vendor_id", "")
except PermissionError:
# nothing we can do here; we're not getting CPU info
pass
import platform
info.hostname = socket.gethostname()
info.platform = platform.platform()
return info
def run_cuda_script( # # noqa: C901
sources: dict[str, str],
headers: Optional[dict[str, str]] = None,
arch: Optional[int] = None,
defines: Optional[dict[str, str]] = None,
include_dirs: Optional[list[str]] = None,
libraries: Optional[list[str]] = None,
flags: Optional[list[str]] = None,
**kwargs,
) -> EvalResult:
"""
Executes the provided CUDA kernel in an isolated environment
Args:
sources: The source files to compile. Mapping file name to content.
headers: Additional header files to create for the compile run.
Mapping of file name to file contents. These files will _not_ be added to the
compile command.
arch: The arch code for the compute/sm versions. If None, native arch is used.
include_dirs: Additional include directories, e.g., for thunderkittens/cutlass etc
defines: Preprocessor defines
libraries: Additional libraries to link to
flags: Additional flags to give to the compiler
seed: Random seed to initialize the RNG for testing
Returns:
tuple[CompileResult, RunResult]: CUDA compile/eval result information
"""
start = datetime.datetime.now()
try:
# Write submission files to directory
_create_files(sources)
_create_files(headers)
compile_result = compile_cuda_script(
files=list(sources.keys()),
arch=arch,
include_dirs=include_dirs,
defines=defines,
libraries=libraries,
flags=flags,
verbose=True,
)
if not compile_result.success:
return EvalResult(
start=start,
end=datetime.datetime.now(),
compilation=compile_result,
run=None,
profile=None,
)
# cleaning up all source files _before_ we let the user code run, just in
# case there's something in there that the user isn't supposed to snoop
finally:
tmp_files = list(sources.keys()) + list((headers or {}).keys())
for f in tmp_files:
if os.path.exists(f):
os.remove(f)
run_result, profile_result = run_single_evaluation(["./eval.out"], **kwargs)
return EvalResult(
start=start,
end=datetime.datetime.now(),
compilation=compile_result,
run=run_result,
profile=profile_result,
)
def run_pytorch_script( # noqa: C901
sources: dict[str, str],
main: str,
**kwargs,
) -> EvalResult:
"""
Executes the provided PyTorch GPU kernel in an isolated environment
Args:
sources: Files to generate
main: Which file to run. Must be one of the keys in sources.
seed: Random seed to initialize the RNG for testing
Returns:
RunResult
"""
start = datetime.datetime.now()
try:
assert main in sources.keys()
# Write submission files to directory
_create_files(sources)
# "compile" step: execute the script once. Will populate
# `load_inline`'s compile cache, so the actual runs will be faster.
try:
compile_run = run_program(["python3", "submission.py"], seed=1, timeout=Timeout.COMPILE)
if "-DTORCH_EXTENSION_NAME" in compile_run.stdout:
comp = CompileResult(
nvcc_found=True,
nvcc_version="",
success=True,
command=compile_run.command,
stdout=compile_run.stdout,
stderr=compile_run.stderr,
exit_code=compile_run.exit_code,
)
else:
comp = None
except subprocess.CalledProcessError as e:
# This step is purely optional, so we just go on
# if it fails
comp = CompileResult(
nvcc_found=False,
nvcc_version="",
success=False,
command="python submission.py",
stdout=e.stdout,
stderr=e.stderr,
exit_code=e.returncode,
)
run, profile = run_single_evaluation(["python3", main], **kwargs)
return EvalResult(
start=start,
end=datetime.datetime.now(),
compilation=comp,
run=run,
profile=profile,
)
finally:
for f in sources.keys():
if os.path.exists(f):
os.remove(f)
class _EvalRunner(Protocol):
def __call__(self, mode: str, **kwargs) -> EvalResult: ...
def run_evaluation(
call: _EvalRunner,
mode: str,
common_args: dict,
) -> dict[str, EvalResult]:
"""
Given a "runner" function `call`, interprets the mode
and calls the runner with the right arguments.
Simple modes (test, benchmark, profile) just
invoke the runner once, but private/leaderboard
require multiple runner calls.
"""
results: dict[str, EvalResult] = {}
if mode == "profile":
benchmarks = copy.deepcopy(common_args["benchmarks"])
for i, benchmark in enumerate(benchmarks.splitlines()):
common_args["benchmarks"] = benchmark
results[f"{mode}.{i}"] = call(mode=mode, **common_args)
elif mode in ["test", "benchmark"]:
results[mode] = call(mode=mode, **common_args)
elif mode in ["private", "leaderboard"]:
# first, run the tests
results["test"] = call(mode="test", **common_args)
if not results["test"].run or not results["test"].run.passed:
return results
# Unnecessary
# results["benchmark"] = call(mode="benchmark", **common_args)
# if not results["benchmark"].run or not results["benchmark"].run.passed:
# return results
# if they pass, run the leaderboard validation
results["leaderboard"] = call(mode="leaderboard", **common_args)
else:
raise AssertionError("Invalid mode")
return results
def build_test_string(tests: list[dict]):
as_str = ""
for test in tests:
kvs = []
for k, v in test.items():
kvs.append(f"{k}: {v}")
as_str += "; ".join(kvs) + "\n"
return as_str
def run_config(config: dict):
system = make_system_info()
common_args = {
"system": system,
"tests": build_test_string(config.get("tests", [])),
"benchmarks": build_test_string(config.get("benchmarks", [])),
"seed": config.get("seed", None),
"ranking_by": config.get("ranking_by", "last"),
"ranked_timeout": config.get("ranked_timeout", Timeout.RANKED),
"benchmark_timeout": config.get("benchmark_timeout", Timeout.BENCHMARK),
"test_timeout": config.get("test_timeout", Timeout.TEST),
"multi_gpu": config.get("multi_gpu", False),
}
if config["lang"] == "py":
runner = functools.partial(
run_pytorch_script, sources=config["sources"], main=config["main"]
)
elif config["lang"] == "cu":
runner = functools.partial(
run_cuda_script,
sources=config["sources"],
headers=config.get("headers", {}),
arch=config.get("arch", None),
defines=config.get("defines", {}),
include_dirs=config.get("include_dirs", []),
libraries=config.get("libraries", []),
flags=CUDA_FLAGS,
)
else:
raise ValueError(f"Invalid language {config['lang']}")
results = run_evaluation(runner, config["mode"], common_args)
return FullResult(success=True, error="", runs=results, system=system)