GNN4Colliders / benchmarks /_common.py
ho22joshua's picture
rewriting codebase (#7)
916755e
Raw History Blame
2.06 kB
"""Small helpers shared by benchmark entry points."""
from __future__ import annotations
import argparse
import json
import platform
import statistics
import time
from collections.abc import Callable
from typing import Any
import torch
def common_parser(description: str) -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=description)
parser.add_argument("--iterations", type=int, default=20)
parser.add_argument("--warmup", type=int, default=5)
parser.add_argument("--device", default="cpu")
parser.add_argument("--seed", type=int, default=1234)
return parser
def synchronize(device: torch.device) -> None:
if device.type == "cuda":
torch.cuda.synchronize(device)
def measure(
operation: Callable[[], Any], *, iterations: int, warmup: int, device: torch.device
) -> dict[str, float]:
for _ in range(warmup):
operation()
synchronize(device)
durations = []
for _ in range(iterations):
start = time.perf_counter()
operation()
synchronize(device)
durations.append(time.perf_counter() - start)
return {
"mean_ms": statistics.mean(durations) * 1000,
"median_ms": statistics.median(durations) * 1000,
"min_ms": min(durations) * 1000,
"max_ms": max(durations) * 1000,
"stdev_ms": statistics.stdev(durations) * 1000 if len(durations) > 1 else 0.0,
}
def metadata(device: torch.device) -> dict[str, Any]:
result: dict[str, Any] = {
"python": platform.python_version(),
"torch": torch.__version__,
"device": str(device),
"cuda_available": torch.cuda.is_available(),
}
if device.type == "cuda" and torch.cuda.is_available():
result["gpu"] = torch.cuda.get_device_name(device)
try:
import dgl
result["dgl"] = dgl.__version__
except ImportError:
result["dgl"] = None
return result
def report(name: str, values: dict[str, Any]) -> None:
print(json.dumps({"benchmark": name, **values}, sort_keys=True))