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)