From 52d5baecc80a09d79e89a1a91806089a07bc40c6 Mon Sep 17 00:00:00 2001 From: Radostin Stoyanov Date: Fri, 10 Jul 2026 17:08:39 +0100 Subject: [PATCH] contrib: Benchmark SGLang compression Process-only measurements do not show how compressed pages affect a containerized inference server, where Podman configuration, model startup, and service readiness are part of the restore path. SGLang also needs content validation so a quick restore cannot hide damaged state. Add a CPU and GPU SGLang benchmark that checkpoints through Podman. Compare uncompressed, per-page LZ4, and region LZ4 images. Send the same deterministic chat request before and after restore, compare response hashes, measure archive and operation costs, and verify the compression mode and region size recorded in the exported inventory. Derive the local serving URL from the selected port unless one is supplied. Podman passes CRIU options through the host-wide runc.conf. Serialize updates with a canonical lock and crash-recovery journal, preserve file metadata and xattrs, refuse to overwrite external changes, and restore the original configuration on success and during error or signal cleanup. Isolate CRIU from ambient defaults, keep access tokens out of arguments and diagnostics, detect stopped containers during health checks, and preserve failed artifacts. Add a bounded CPU test runner. Assisted-by: Codex:GPT-5 Assisted-by: Claude:claude-fable-5 Signed-off-by: Radostin Stoyanov --- .../compression-benchmark/podman-sglang.py | 172 +++ .../compression-benchmark/podman_common.py | 1347 +++++++++++++++++ test/others/compression/sglang/Makefile | 3 + test/others/compression/sglang/run.sh | 30 + 4 files changed, 1552 insertions(+) create mode 100755 contrib/compression-benchmark/podman-sglang.py create mode 100644 contrib/compression-benchmark/podman_common.py create mode 100644 test/others/compression/sglang/Makefile create mode 100755 test/others/compression/sglang/run.sh diff --git a/contrib/compression-benchmark/podman-sglang.py b/contrib/compression-benchmark/podman-sglang.py new file mode 100755 index 000000000..60a094f8a --- /dev/null +++ b/contrib/compression-benchmark/podman-sglang.py @@ -0,0 +1,172 @@ +#!/usr/bin/env python3 +""" +Podman SGLang Checkpoint/Restore Benchmark + +Starts an SGLang container with Podman, validates inference, checkpoints it with +Podman, removes it, restores it with Podman, and validates inference again. + +This benchmarks Podman's container checkpoint/restore path while varying CRIU +memory-page compression through /etc/criu/runc.conf. Podman's own checkpoint +archive compression is kept at "none" by default so the reported archive size +reflects CRIU image size rather than tar-level gzip/zstd compression. + +Example: + sudo HF_TOKEN=... python3 contrib/compression-benchmark/podman-sglang.py \\ + --model Qwen/Qwen3-0.6B -n 3 \\ + --modes uncompressed lz4-page lz4-region + +CPU-only example (requires an SGLang CPU image): + sudo python3 contrib/compression-benchmark/podman-sglang.py \\ + --accelerator cpu --image sglang-cpu:latest +""" + +import os +import sys + +_BENCHMARK_DIR = os.path.dirname(os.path.abspath(__file__)) +if _BENCHMARK_DIR not in sys.path: + sys.path.insert(0, _BENCHMARK_DIR) + +import podman_common as common # noqa: E402 + + +class SglangAdapter: + key = "sglang" + display_name = "SGLang" + heading = "SGLANG" + default_container_name = "sglang-criu-bench" + temp_prefix = "podman-sglang-bench-" + + @staticmethod + def add_image_arguments(parser): + parser.add_argument( + "--image", + help="Container image (default: lmsysorg/sglang:latest for GPU; " + "required for CPU)", + ) + + @staticmethod + def add_model_arguments(parser): + parser.add_argument( + "--sglang-model-arg", + choices=["model", "model-path"], + default="model-path", + help="SGLang launch flag used for the model path", + ) + + @staticmethod + def add_resource_arguments(parser): + parser.add_argument("--mem-fraction-static", type=float, default=0.35) + parser.add_argument("--max-total-tokens", type=int, default=8192) + parser.add_argument("--context-length", type=int, default=8192) + parser.add_argument( + "--tensor-parallel-size", + type=int, + default=1, + help="SGLang tensor-parallel ranks (default: 1)", + ) + + @staticmethod + def add_request_arguments(parser): + parser.add_argument( + "--enable-thinking", + action="store_true", + help="Do not add chat_template_kwargs.enable_thinking=false", + ) + + @staticmethod + def add_server_arguments(parser): + parser.add_argument( + "--sglang-arg", + action="append", + default=[], + help="Extra raw argument appended to sglang.launch_server", + ) + + @staticmethod + def normalize_args(parser, args): + if args.image is not None: + return + if args.accelerator == "cpu": + parser.error( + "--image is required for CPU mode; build the SGLang " + "docker/xeon.Dockerfile image first" + ) + args.image = "lmsysorg/sglang:latest" + + @staticmethod + def prepare_args(args): + if not args.enable_thinking: + extra = dict(args.chat_extra_json or {}) + extra.setdefault( + "chat_template_kwargs", {"enable_thinking": False} + ) + args.chat_extra_json = extra + + @staticmethod + def cpu_podman_args(args): + return [ + "--security-opt", "seccomp=unconfined", + "--cap-add", "SYS_NICE", + "--env", "SGLANG_USE_CPU_ENGINE=1", + ] + + @staticmethod + def server_argv(args): + command = [ + args.image, + "python3", "-m", "sglang.launch_server", + f"--{args.sglang_model_arg}", args.model, + "--host", "0.0.0.0", + "--port", str(args.port), + "--max-total-tokens", str(args.max_total_tokens), + "--context-length", str(args.context_length), + "--tp", str(args.tensor_parallel_size), + ] + if args.accelerator == "gpu": + command += ["--mem-fraction-static", str(args.mem_fraction_static)] + else: + command += ["--device", "cpu", "--disable-overlap-schedule"] + command += args.sglang_arg + return command + + @staticmethod + def server_summary(args): + return f"model-arg={args.sglang_model_arg}" + + +_benchmark = common.ServingBenchmark(SglangAdapter(), __doc__) + +# Keep useful import-level helpers available to callers and focused tests. +format_bytes = common.format_bytes +format_duration = common.format_duration +inventory_bytes_from_archive = common.inventory_bytes_from_archive +inventory_entry_from_archive = common.inventory_entry_from_archive +verify_archive_compression = common.verify_archive_compression +strip_compression_runc_options = common.strip_compression_runc_options +compression_config_lines = common.compression_config_lines +json_config = common.json_config +format_cmd = common.format_cmd +run_arg_sets_environment = common.run_arg_sets_environment +write_file = common.write_file +os = common.os +shutil = common.shutil +signal = common.signal +subprocess = common.subprocess +tempfile = common.tempfile +urllib = common.urllib + +main = _benchmark.main +podman_env = _benchmark.podman_env +set_runc_conf_for_cfg = _benchmark.set_runc_conf_for_cfg +restore_runc_conf = _benchmark.restore_runc_conf +build_container_cmd = _benchmark.build_container_cmd +run_trial = _benchmark.run_trial +cleanup = _benchmark.cleanup + + +if __name__ == "__main__": + try: + main() + except (OSError, RuntimeError) as error: + sys.exit(f"Error: {error}") diff --git a/contrib/compression-benchmark/podman_common.py b/contrib/compression-benchmark/podman_common.py new file mode 100644 index 000000000..ea092e4a7 --- /dev/null +++ b/contrib/compression-benchmark/podman_common.py @@ -0,0 +1,1347 @@ +#!/usr/bin/env python3 +"""Shared Podman serving checkpoint/restore benchmark implementation.""" + +import argparse +import base64 +import contextlib +import errno +import fcntl +import hashlib +import json +import os +import platform +import posixpath +import shlex +import shutil +import signal +import stat +import subprocess +import sys +import tempfile +import time +import urllib.error +import urllib.request + +# Sentinel distinguishing "runc.conf never touched" from "stashed, but the +# file did not exist" (which stashes None). +_RUNC_CONF_UNSET = object() + + +class RuntimeState: + """Mutable resources owned by this benchmark process.""" + + def __init__(self): + self.tempdirs = set() + self.started_containers = set() + self.cleanup_containers = True + self.original_runc_conf = _RUNC_CONF_UNSET + self.runc_conf_lock_fd = None + self.runc_conf_path = None + self.runc_conf_state_path = None + self.runc_conf_state = None + self.criu_wrapper_dir = None + self.cleanup_started = False + self.received_signal = None + + +PODMAN = "podman" +RUNC_CONF_BEGIN = "# BEGIN criu-compression-benchmark" +RUNC_CONF_END = "# END criu-compression-benchmark" +PAGE_SIZE = os.sysconf("SC_PAGE_SIZE") +MAX_REGION_SIZE = 4 * 1024 * 1024 +MAX_COMPRESSION_ACCELERATION = 65537 +MAX_DECOMPRESSION_THREADS = 1024 +COMPRESSION_OPTIONS = { + "compress", "compress-region", "compress-acceleration", + "decompress-threads", +} +HF_TOKEN_ENV_VARS = ("HF_TOKEN", "HUGGING_FACE_HUB_TOKEN") +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) +SIGNALS = (signal.SIGINT, signal.SIGHUP, signal.SIGTERM) + + +class ServingBenchmark: + """One framework driver with process-local mutable resource state.""" + + def __init__(self, adapter, description): + self.adapter = adapter + self.description = description + self.state = RuntimeState() + + def cleanup(self): + return cleanup(self) + + def signal_handler(self, signum, frame): + return signal_handler(self, signum, frame) + + def podman_env(self, args): + return podman_env(self, args) + + def ensure_no_default_config_wrapper(self): + return ensure_no_default_config_wrapper(self) + + def release_runc_conf_lock(self): + return release_runc_conf_lock(self) + + def set_runc_conf_for_cfg(self, path, cfg, acceleration, + decompress_threads=None): + return set_runc_conf_for_cfg( + self, path, cfg, acceleration, decompress_threads + ) + + def restore_runc_conf(self): + return restore_runc_conf(self) + + def build_container_cmd(self, name, args): + return build_container_cmd(self, name, args) + + def start_container(self, name, args): + return start_container(self, name, args) + + def checkpoint_container(self, name, archive, cfg, args): + return checkpoint_container(self, name, archive, cfg, args) + + def restore_container(self, name, archive, args): + return restore_container(self, name, archive, args) + + def chat_once(self, base_url, model, prompt, max_tokens, temperature, + seed, timeout, extra_body=None): + return chat_once( + base_url, model, prompt, max_tokens, temperature, seed, timeout, + extra_body, self.adapter.display_name + ) + + def run_trial(self, cfg, workdir, args, trial_id, keep_running=False): + return run_trial(self, cfg, workdir, args, trial_id, keep_running) + + def report(self, results_by_cfg, order): + return report(self, results_by_cfg, order) + + def main(self, argv=None): + return run_main(self, argv, self.description) + + +@contextlib.contextmanager +def _blocked_termination_signals(): + previous = signal.pthread_sigmask(signal.SIG_BLOCK, SIGNALS) + try: + yield + finally: + signal.pthread_sigmask(signal.SIG_SETMASK, previous) + + +def cleanup(benchmark): + runtime = benchmark.state + if runtime.cleanup_started: + return + with _blocked_termination_signals(): + runtime.cleanup_started = True + try: + restore_runc_conf(benchmark) + except (OSError, RuntimeError, ValueError) as e: + print(f"cleanup: failed to restore runc.conf: {e}", file=sys.stderr) + if runtime.cleanup_containers: + for name in list(runtime.started_containers): + try: + result = subprocess.run([PODMAN, "rm", "-f", name], + capture_output=True, text=True) + except OSError as e: + print(f"cleanup: failed to execute Podman for {name}: {e}", + file=sys.stderr) + continue + if result.returncode: + detail = (result.stderr or result.stdout).strip() + print(f"cleanup: failed to remove {name}: {detail[-1000:]}", + file=sys.stderr) + runtime.started_containers.clear() + for path in list(runtime.tempdirs): + shutil.rmtree(path, ignore_errors=True) + if os.path.exists(path): + print(f"cleanup: failed to remove temporary directory {path}", + file=sys.stderr) + runtime.tempdirs.clear() + + +def signal_handler(benchmark, signum, frame): + runtime = benchmark.state + if runtime.received_signal is not None: + return + runtime.received_signal = signum + # Defer cleanup until subprocess.run() has killed and waited for its + # active child and Python starts unwinding through atexit handlers. + for handled in SIGNALS: + signal.signal(handled, signal.SIG_IGN) + raise SystemExit(128 + signum) + + +def median(v): + s = sorted(v) + n = len(s) + if n == 0: + return 0 + return s[n // 2] if n % 2 else (s[n // 2 - 1] + s[n // 2]) / 2 + + +def format_bytes(byte_count): + if byte_count >= 1073741824: + return f"{byte_count / 1073741824:.2f} GB" + return f"{byte_count / 1048576:.1f} MB" + + +def format_duration(microseconds): + if microseconds >= 1e6: + return f"{microseconds / 1e6:.3f} s" + return f"{microseconds / 1000:.1f} ms" + + +def cfg_label(cfg): + if cfg["mode"] == "uncompressed": + return "Uncompressed" + if cfg["mode"] == "lz4-page": + return f"LZ4 per page ({PAGE_SIZE // 1024} KiB blocks)" + return f"LZ4 regions ({cfg['region_size'] // 1024} KiB blocks)" + + +def decompress_threads_label(threads): + """None uses CRIU's serial default; 0 is auto; positive values are explicit.""" + if threads is None: + return "default (serial)" + if threads == 0: + return "auto" + return str(threads) + + +def server_base_url(port, base_url): + if base_url is not None: + return base_url + return f"http://127.0.0.1:{port}" + + +def json_object(value): + try: + obj = json.loads(value) + except json.JSONDecodeError as e: + raise argparse.ArgumentTypeError(str(e)) + if not isinstance(obj, dict): + raise argparse.ArgumentTypeError("expected a JSON object") + return obj + + +def collect_system_info(): + info = {"kernel": platform.release(), "arch": platform.machine(), + "cpus": os.cpu_count()} + try: + with open("/proc/cpuinfo") as f: + for line in f: + if line.startswith("model name"): + info["cpu"] = line.split(":", 1)[1].strip() + break + except OSError: + info["cpu"] = "unknown" + try: + with open("/proc/meminfo") as f: + for line in f: + if line.startswith("MemTotal"): + info["memory_mb"] = int(line.split()[1]) // 1024 + break + except OSError: + info["memory_mb"] = 0 + for cmd, key in (([PODMAN, "--version"], "podman"), + (["criu", "--version"], "criu")): + try: + r = subprocess.run(cmd, capture_output=True, text=True) + if r.returncode == 0: + info[key] = r.stdout.strip() + except OSError: + info[key] = "unknown" + try: + r = subprocess.run(["nvidia-smi", "--query-gpu=name", + "--format=csv,noheader"], + capture_output=True, text=True) + if r.returncode == 0: + info["gpus"] = [line.strip() for line in r.stdout.splitlines() + if line.strip()] + except OSError: + info["gpus"] = [] + return info + + +def http_json(method, url, payload=None, timeout=120): + data = None + headers = {} + if payload is not None: + data = json.dumps(payload).encode() + headers["Content-Type"] = "application/json" + headers["Authorization"] = "Bearer EMPTY" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=timeout) as resp: + body = resp.read() + return json.loads(body.decode()) if body else {} + + +def podman_text(args): + try: + r = subprocess.run([PODMAN, *args], capture_output=True, text=True) + except OSError as e: + return f"unable to execute Podman: {e}" + return (r.stdout + r.stderr).strip() + + +def container_diagnostics(name): + parts = [f"container={name}"] + ps = podman_text(["ps", "-a", "--filter", f"name=^{name}$", + "--format", "{{.Names}} {{.Status}} {{.Ports}}"]) + if ps: + parts.append(f"podman ps:\n{ps}") + logs = podman_text(["logs", "--tail", "80", name]) + if logs: + parts.append(f"last logs:\n{logs[-6000:]}") + return "\n\n".join(parts) + + +def container_exit_code(name): + """Return an exit code for a stopped container, or None otherwise.""" + try: + result = subprocess.run( + [PODMAN, "inspect", "--format", + "{{.State.Running}} {{.State.ExitCode}}", name], + capture_output=True, text=True, + ) + except OSError: + return None + if result.returncode: + return None + fields = result.stdout.strip().split() + if len(fields) != 2 or fields[0].lower() != "false": + return None + try: + return int(fields[1]) + except ValueError: + return None + + +def wait_health(base_url, health_path, timeout, container_name=None, + framework_name="serving"): + deadline = time.monotonic() + timeout + next_state_check = 0 + last = None + url = f"{base_url.rstrip('/')}/{health_path.lstrip('/')}" + while time.monotonic() < deadline: + try: + urllib.request.urlopen(url, timeout=5).read() + return + except (OSError, urllib.error.URLError) as e: + last = e + now = time.monotonic() + if container_name and now >= next_state_check: + next_state_check = now + 5 + exit_code = container_exit_code(container_name) + if exit_code is not None: + raise RuntimeError( + f"{framework_name} container exited with status {exit_code} " + f"before becoming healthy\n\n" + f"{container_diagnostics(container_name)}" + ) from e + time.sleep(1) + detail = "" + if container_name: + detail = "\n\n" + container_diagnostics(container_name) + raise RuntimeError( + f"{framework_name} health check timed out after {timeout}s: {last}{detail}" + ) + + +def chat_once(base_url, model, prompt, max_tokens, temperature, seed, timeout, + extra_body=None, framework_name="serving"): + payload = { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "temperature": temperature, + "max_tokens": max_tokens, + "seed": seed, + } + if extra_body: + payload.update(extra_body) + t0 = time.monotonic() + data = http_json("POST", f"{base_url.rstrip('/')}/v1/chat/completions", + payload, timeout) + latency_us = int((time.monotonic() - t0) * 1e6) + content = data.get("choices", [{}])[0].get("message", {}).get("content", "") + if not content.strip(): + raise RuntimeError(f"{framework_name} validation returned an empty response") + return latency_us, content + + +def format_cmd(cmd): + """Format a command without printing values passed through --env.""" + out = [] + redact_next = False + for item in cmd: + if redact_next: + name = item.split("=", 1)[0] + out.append(f"{name}=" if "=" in item else item) + redact_next = False + continue + if item.startswith("--env=") or (item.startswith("-e") and + not item.startswith("--")): + if item.startswith("--env="): + option, value = item.split("=", 1) + else: + option, value = "-e", item[2:].lstrip("=") + name = value.split("=", 1)[0] + out.append(f"{option}{'=' if option == '--env' else ''}" + f"{name}=" if "=" in value + else item) + continue + out.append(item) + redact_next = item in ("--env", "-e") + return " ".join(out) + + +def run_arg_sets_environment(item): + """Return whether a raw Podman argument can carry an environment value.""" + return (item == "--env" or item.startswith("--env=") or + (item.startswith("-e") and not item.startswith("--"))) + + +def redact_run_arg(item): + if item.startswith("--env=") or (item.startswith("-e") and + not item.startswith("--")): + if item.startswith("--env="): + option, value = item.split("=", 1) + else: + option, value = "-e", item[2:].lstrip("=") + name = value.split("=", 1)[0] + separator = "=" if option == "--env" else "" + return f"{option}{separator}{name}=" if "=" in value else item + return item + + +def redact_run_args(items): + out = [] + redact_next = False + for item in items: + if redact_next: + name = item.split("=", 1)[0] + out.append(f"{name}=" if "=" in item else item) + redact_next = False + continue + out.append(redact_run_arg(item)) + redact_next = item in ("--env", "-e") + return out + + +def run_cmd(cmd, env=None, check=True): + try: + r = subprocess.run(cmd, capture_output=True, text=True, env=env) + except OSError as e: + raise RuntimeError(f"unable to execute {format_cmd(cmd)}: {e}") from e + if check and r.returncode: + msg = (r.stderr or r.stdout).strip() + raise RuntimeError(f"{format_cmd(cmd)} failed: {msg[-6000:]}") + return r + + +def podman_env(benchmark, args): + env = os.environ.copy() + # Start CRIU without global, user, or inherited configuration. runc still + # supplies the benchmark-owned runc.conf through the RPC request. + env.pop("CRIU_CONFIG_FILE", None) + wrapper_dir = benchmark.ensure_no_default_config_wrapper() + env["PATH"] = wrapper_dir + os.pathsep + env.get("PATH", os.defpath) + if args.criu_libdir: + env["CRIU_LIBS_DIR"] = args.criu_libdir + return env + + +def ensure_no_default_config_wrapper(benchmark): + runtime = benchmark.state + if runtime.criu_wrapper_dir is not None: + return runtime.criu_wrapper_dir + criu = shutil.which("criu") + if criu is None: + raise RuntimeError("criu was not found in PATH") + wrapper_dir = tempfile.mkdtemp(prefix="criu-no-default-config-") + runtime.tempdirs.add(wrapper_dir) + wrapper = os.path.join(wrapper_dir, "criu") + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + flags |= getattr(os, "O_NOFOLLOW", 0) + fd = os.open(wrapper, flags, 0o700) + try: + with os.fdopen(fd, "w") as output: + fd = -1 + output.write("#!/bin/sh\n") + output.write(f"exec {shlex.quote(criu)} --no-default-config \"$@\"\n") + finally: + if fd >= 0: + os.close(fd) + runtime.criu_wrapper_dir = wrapper_dir + return wrapper_dir + + +def read_file(path): + try: + with open(path) as f: + return f.read() + except FileNotFoundError: + return None + + +def _sync_directory(path): + directory = os.path.dirname(path) or "." + fd = os.open(directory, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)) + try: + os.fsync(fd) + finally: + os.close(fd) + + +def _read_xattrs(path): + if not hasattr(os, "listxattr"): + return None + try: + names = os.listxattr(path) + except OSError as e: + if e.errno in (errno.ENOTSUP, errno.ENOSYS): + return None + raise + return { + name: base64.b64encode(os.getxattr(path, name)).decode("ascii") + for name in names + } + + +def _apply_xattrs(path, encoded): + if encoded is None: + return + current = _read_xattrs(path) + if current is None: + if encoded: + raise OSError(errno.ENOTSUP, "extended attributes are unsupported", + path) + return + for name in current.keys() - encoded.keys(): + os.removexattr(path, name) + for name, value in encoded.items(): + decoded = base64.b64decode(value, validate=True) + if current.get(name) != value: + os.setxattr(path, name, decoded) + + +def _apply_metadata(path, metadata): + current = os.stat(path) + if current.st_uid != metadata["uid"] or current.st_gid != metadata["gid"]: + os.chown(path, metadata["uid"], metadata["gid"]) + os.chmod(path, metadata["mode"]) + _apply_xattrs(path, metadata.get("xattrs")) + os.utime(path, ns=(metadata["atime_ns"], metadata["mtime_ns"])) + + +def write_file(path, data, metadata=None): + directory = os.path.dirname(path) + if directory: + os.makedirs(directory, exist_ok=True) + tmp = f"{path}.tmp.{os.getpid()}.{time.monotonic_ns()}" + if metadata is None: + metadata = _runc_conf_metadata(path) + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + flags |= getattr(os, "O_NOFOLLOW", 0) + fd = os.open(tmp, flags, 0o600) + try: + with os.fdopen(fd, "w") as f: + fd = -1 + f.write(data) + f.flush() + os.fsync(f.fileno()) + if metadata is not None: + _apply_metadata(tmp, metadata) + os.fsync(f.fileno()) + os.replace(tmp, path) + _sync_directory(path) + finally: + if fd >= 0: + os.close(fd) + with contextlib.suppress(FileNotFoundError): + os.unlink(tmp) + + +def _runc_conf_metadata(path): + try: + metadata = os.stat(path) + except FileNotFoundError: + return None + return { + "mode": stat.S_IMODE(metadata.st_mode), + "uid": metadata.st_uid, + "gid": metadata.st_gid, + "atime_ns": metadata.st_atime_ns, + "mtime_ns": metadata.st_mtime_ns, + "xattrs": _read_xattrs(path), + } + + +def _file_state(path): + # Capture timestamps before opening the file: reading an old file can + # update its atime under relatime. + metadata = _runc_conf_metadata(path) + content = read_file(path) if metadata is not None else None + return { + "existed": content is not None, + "content": content, + "metadata": metadata, + } + + +def _metadata_identity(metadata): + if metadata is None: + return None + return {key: metadata.get(key) for key in + ("mode", "uid", "gid", "mtime_ns", "xattrs")} + + +def _same_file_state(actual, expected): + if actual["existed"] != expected["existed"]: + return False + if not actual["existed"]: + return True + if actual["content"] != expected["content"]: + return False + # Metadata for a newly created pending file is not known until after its + # atomic rename. Content still makes that narrow recovery window safe. + if expected.get("metadata") is None: + return True + return (_metadata_identity(actual["metadata"]) == + _metadata_identity(expected["metadata"])) + + +def _owned_runc_conf_state(state): + current = _file_state(state["path"]) + owned = [state["active"]] + if state.get("pending") is not None: + owned.append(state["pending"]) + if not any(_same_file_state(current, item) for item in owned): + raise RuntimeError( + f"{state['path']} changed outside the compression benchmark; " + "refusing to overwrite it (recovery state was preserved)" + ) + return current + + +def _restore_runc_conf_state(state, state_path): + _owned_runc_conf_state(state) + path = state["path"] + original = state["original"] + # Journal the restore before changing the target. If the process dies + # after the rename/unlink but before removing the journal, the next run + # can recognize the original file as benchmark-owned recovery work. + state["pending"] = original + _write_recovery_state(state_path, state) + if original["existed"]: + write_file(path, original["content"], original["metadata"]) + else: + with contextlib.suppress(FileNotFoundError): + os.unlink(path) + _sync_directory(path) + + +def _write_recovery_state(path, state): + tmp = f"{path}.tmp.{os.getpid()}.{time.monotonic_ns()}" + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + flags |= getattr(os, "O_NOFOLLOW", 0) + fd = os.open(tmp, flags, 0o600) + try: + with os.fdopen(fd, "w") as output: + fd = -1 + json.dump(state, output) + output.flush() + os.fsync(output.fileno()) + os.replace(tmp, path) + _sync_directory(path) + finally: + if fd >= 0: + os.close(fd) + with contextlib.suppress(FileNotFoundError): + os.unlink(tmp) + + +def _read_recovery_state(path): + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + fd = os.open(path, flags) + with os.fdopen(fd) as source: + metadata = os.fstat(source.fileno()) + if not stat.S_ISREG(metadata.st_mode): + raise RuntimeError(f"invalid runc.conf recovery file: {path}") + state = json.load(source) + if (not isinstance(state, dict) or + not all(key in state for key in ("path", "original", "active"))): + raise RuntimeError(f"invalid runc.conf recovery file: {path}") + return state + + +def acquire_runc_conf(benchmark, path): + runtime = benchmark.state + # Every spelling of the same target, including symlink aliases, must use + # one lock and one recovery journal. + target = os.path.realpath(path) + lock_path = target + ".compression-benchmark.lock" + directory = os.path.dirname(lock_path) + if directory: + os.makedirs(directory, exist_ok=True) + flags = os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) + lock_fd = os.open(lock_path, flags, 0o600) + try: + try: + fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + raise RuntimeError( + f"another compression benchmark is using {path}" + ) from exc + + state_path = lock_path + ".state" + if os.path.exists(state_path): + state = _read_recovery_state(state_path) + if state.get("path") != target: + raise RuntimeError( + f"runc.conf recovery state refers to {state.get('path')}, " + f"not {target}" + ) + _restore_runc_conf_state(state, state_path) + os.unlink(state_path) + _sync_directory(state_path) + + original = _file_state(target) + state = { + "path": target, + "original": original, + "active": original, + "pending": None, + } + _write_recovery_state(state_path, state) + runtime.runc_conf_lock_fd = lock_fd + runtime.runc_conf_path = target + runtime.runc_conf_state_path = state_path + runtime.runc_conf_state = state + return target, original["content"] + except BaseException: + os.close(lock_fd) + raise + + +def release_runc_conf_lock(benchmark): + runtime = benchmark.state + if runtime.runc_conf_lock_fd is not None: + fcntl.flock(runtime.runc_conf_lock_fd, fcntl.LOCK_UN) + os.close(runtime.runc_conf_lock_fd) + runtime.runc_conf_lock_fd = None + runtime.runc_conf_path = None + runtime.runc_conf_state_path = None + runtime.runc_conf_state = None + + +def strip_compression_runc_options(text): + """Remove prior benchmark blocks and any ambient compression options.""" + if not text: + return "" + lines = text.splitlines() + out = [] + skipping = False + for line in lines: + if line.strip() == RUNC_CONF_BEGIN: + skipping = True + continue + if line.strip() == RUNC_CONF_END: + skipping = False + continue + if skipping: + continue + statement = line.strip() + if not statement or statement.startswith("#"): + out.append(line) + continue + option = statement.split(None, 1)[0].split("=", 1)[0] + option = option.lstrip("-").replace("_", "-") + if option not in COMPRESSION_OPTIONS: + out.append(line) + return "\n".join(out).rstrip() + + +def compression_config_lines(cfg, acceleration, decompress_threads=None): + if cfg["mode"] == "uncompressed": + return [] + if cfg["mode"] == "lz4-page": + lines = ["compress"] + else: + lines = [f"compress-region {cfg['region_size']}"] + if acceleration != 1: + lines.append(f"compress-acceleration {acceleration}") + if decompress_threads is not None: + lines.append(f"decompress-threads {decompress_threads}") + return lines + + +def set_runc_conf_for_cfg(benchmark, path, cfg, acceleration, + decompress_threads=None): + runtime = benchmark.state + if runtime.original_runc_conf is _RUNC_CONF_UNSET: + path, original = acquire_runc_conf(benchmark, path) + runtime.original_runc_conf = original + else: + path = runtime.runc_conf_path + + base = strip_compression_runc_options(runtime.original_runc_conf) + lines = compression_config_lines(cfg, acceleration, decompress_threads) + if lines: + block = "\n".join([RUNC_CONF_BEGIN, *lines, RUNC_CONF_END]) + text = f"{base}\n\n{block}\n" if base else f"{block}\n" + else: + text = f"{base}\n" if base else "" + current = _owned_runc_conf_state(runtime.runc_conf_state) + runtime.runc_conf_state["pending"] = { + "existed": True, + "content": text, + "metadata": current["metadata"], + } + _write_recovery_state(runtime.runc_conf_state_path, + runtime.runc_conf_state) + write_file(path, text, current["metadata"]) + runtime.runc_conf_state["active"] = _file_state(path) + runtime.runc_conf_state["pending"] = None + _write_recovery_state(runtime.runc_conf_state_path, + runtime.runc_conf_state) + + +def restore_runc_conf(benchmark): + runtime = benchmark.state + if runtime.original_runc_conf is _RUNC_CONF_UNSET: + return + path = runtime.runc_conf_path + if not path: + return + with _blocked_termination_signals(): + try: + _restore_runc_conf_state(runtime.runc_conf_state, + runtime.runc_conf_state_path) + if runtime.runc_conf_state_path is not None: + os.unlink(runtime.runc_conf_state_path) + _sync_directory(runtime.runc_conf_state_path) + finally: + runtime.original_runc_conf = _RUNC_CONF_UNSET + release_runc_conf_lock(benchmark) + + +def build_container_cmd(benchmark, name, args): + cmd = [ + PODMAN, "run", "-d", + "--name", name, + "--security-opt", args.security_opt, + "--network", "host", + "--shm-size", args.shm_size, + "-v", f"{args.hf_cache}:/root/.cache/huggingface", + ] + for name in HF_TOKEN_ENV_VARS: + # Podman copies the value from its environment. It never appears in + # this process's command line or in an error message. + if name in os.environ: + cmd += ["--env", name] + if args.accelerator == "gpu": + cmd += [ + "--device", args.gpu_device, + "--env", f"CUDA_VISIBLE_DEVICES={args.cuda_visible_devices}", + "--env", "NCCL_P2P_DISABLE=1", + "--env", "NCCL_SHM_DISABLE=1", + "--env", "NCCL_IB_DISABLE=1", + "--env", "NCCL_CUMEM_ENABLE=0", + ] + else: + cmd += benchmark.adapter.cpu_podman_args(args) + for item in args.env: + cmd += ["--env", item] + for item in args.volume: + cmd += ["-v", item] + for item in args.run_arg: + cmd.append(item) + for item in args.ulimit: + cmd += ["--ulimit", item] + + cmd += benchmark.adapter.server_argv(args) + + return cmd + + +def start_container(benchmark, name, args): + os.makedirs(args.hf_cache, exist_ok=True) + cmd = build_container_cmd(benchmark, name, args) + + run_cmd([PODMAN, "rm", "-f", name], check=False) + benchmark.state.started_containers.add(name) + run_cmd(cmd, env=podman_env(benchmark, args)) + print(f" waiting for {name} health on {args.base_url}", flush=True) + wait_health(args.base_url, args.health_path, args.wait_seconds, name, + benchmark.adapter.display_name) + + +def checkpoint_container(benchmark, name, archive, cfg, args): + set_runc_conf_for_cfg(benchmark, args.runc_conf, cfg, + args.compress_acceleration, + args.decompress_threads) + cmd = [ + PODMAN, "container", "checkpoint", + "--export", archive, + "--compress", args.archive_compression, + "--ignore-volumes", + "--tcp-established", + ] + if args.print_stats: + cmd.append("--print-stats") + if args.keep_checkpoint_files: + cmd.append("--keep") + cmd.append(name) + + t0 = time.monotonic() + r = run_cmd(cmd, env=podman_env(benchmark, args)) + checkpoint_us = int((time.monotonic() - t0) * 1e6) + return checkpoint_us, r.stdout.strip() + + +def _tar_command(cmd, binary=False): + try: + result = subprocess.run(cmd, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=not binary) + except OSError as exc: + raise RuntimeError(f"unable to inspect checkpoint archive: {exc}") from exc + if result.returncode: + stderr = result.stderr + if binary: + stderr = stderr.decode(errors="replace") + raise RuntimeError( + "unable to inspect checkpoint archive: " + stderr.strip()[-2000:] + ) + return result.stdout + + +def inventory_bytes_from_archive(archive): + listing = _tar_command(["tar", "--list", "--file", archive]) + inventories = [ + member for member in listing.splitlines() + if posixpath.basename(member.rstrip("/")) == "inventory.img" + ] + if len(inventories) != 1: + raise RuntimeError( + "checkpoint archive must contain exactly one inventory.img; " + f"found {len(inventories)}" + ) + return _tar_command([ + "tar", "--extract", "--to-stdout", "--file", archive, + "--", inventories[0], + ], binary=True) + + +def inventory_entry_from_archive(archive): + if os.path.join(REPO_ROOT, "lib") not in sys.path: + sys.path.insert(0, os.path.join(REPO_ROOT, "lib")) + try: + from pycriu import images as pimg + inventory = pimg.loads(inventory_bytes_from_archive(archive)) + except Exception as exc: + raise RuntimeError( + "unable to decode checkpoint inventory; build the CRIU Python " + f"bindings with 'make -C {REPO_ROOT} lib': {exc}" + ) from exc + if inventory.get("magic") != "INVENTORY": + raise RuntimeError("checkpoint inventory has the wrong image magic") + entries = inventory.get("entries", []) + if len(entries) != 1: + raise RuntimeError( + f"checkpoint inventory must contain one entry; found {len(entries)}" + ) + return entries[0] + + +def verify_archive_compression(archive, cfg): + entry = inventory_entry_from_archive(archive) + expected = {"uncompressed": 0, "lz4-page": 1, "lz4-region": 2}[cfg["mode"]] + try: + actual = int(entry.get("compress", 0)) + except (TypeError, ValueError) as exc: + raise RuntimeError("checkpoint inventory has an invalid compression mode") from exc + if actual != expected: + raise RuntimeError( + "checkpoint compression mode does not match the benchmark " + f"configuration: expected {expected}, found {actual}" + ) + if expected == 2: + try: + region_size = int(entry.get("compress_region_size", 0)) + except (TypeError, ValueError) as exc: + raise RuntimeError( + "checkpoint inventory has an invalid compression region size" + ) from exc + if region_size != cfg["region_size"]: + raise RuntimeError( + "checkpoint compression region does not match the benchmark " + f"configuration: expected {cfg['region_size']}, found {region_size}" + ) + return actual + + +def restore_container(benchmark, name, archive, args): + cmd = [ + PODMAN, "container", "restore", + "--import", archive, + "--ignore-volumes", + "--tcp-established", + ] + if args.print_stats: + cmd.append("--print-stats") + + t0 = time.monotonic() + benchmark.state.started_containers.add(name) + r = run_cmd(cmd, env=podman_env(benchmark, args)) + restore_us = int((time.monotonic() - t0) * 1e6) + print(f" waiting for restored {name} health on {args.base_url}", flush=True) + wait_health(args.base_url, args.health_path, args.wait_seconds, name, + benchmark.adapter.display_name) + return restore_us, r.stdout.strip() + + +def run_trial(benchmark, cfg, workdir, args, trial_id, keep_running=False): + name = f"{args.container_name}-{os.getpid()}-{trial_id}" + archive = os.path.join(workdir, f"{name}.tar") + if args.archive_compression == "gzip": + archive += ".gz" + elif args.archive_compression == "zstd": + archive += ".zst" + + benchmark.start_container(name, args) + request_model = args.served_model_name or args.model + for _ in range(args.warmup_requests): + benchmark.chat_once(args.base_url, request_model, args.prompt, + args.max_tokens, args.temperature, args.seed, + args.request_timeout, args.chat_extra_json) + pre_us, pre_content = benchmark.chat_once( + args.base_url, request_model, args.prompt, args.max_tokens, + args.temperature, args.seed, args.request_timeout, + args.chat_extra_json, + ) + checkpoint_us, checkpoint_stats = benchmark.checkpoint_container( + name, archive, cfg, args + ) + inventory_compress_mode = verify_archive_compression(archive, cfg) + run_cmd([PODMAN, "rm", "-f", name]) + benchmark.state.started_containers.discard(name) + restore_us, restore_stats = benchmark.restore_container(name, archive, args) + post_us, post_content = benchmark.chat_once( + args.base_url, request_model, args.prompt, args.max_tokens, + args.temperature, args.seed, args.request_timeout, + args.chat_extra_json, + ) + pre_digest = hashlib.sha256(pre_content.encode()).hexdigest() + post_digest = hashlib.sha256(post_content.encode()).hexdigest() + valid = pre_digest == post_digest + if not valid: + raise RuntimeError( + "deterministic validation response changed after restore: " + f"before_sha256={pre_digest}, after_sha256={post_digest}" + ) + if keep_running: + # Exempt only the explicitly retained final container from atexit + # cleanup. Earlier trials must release the shared host-network port. + benchmark.state.started_containers.discard(name) + else: + run_cmd([PODMAN, "rm", "-f", name]) + benchmark.state.started_containers.discard(name) + + return { + "archive_size": os.path.getsize(archive), + "inventory_compress_mode": inventory_compress_mode, + "checkpoint_wall_us": checkpoint_us, + "restore_wall_us": restore_us, + "pre_request_us": pre_us, + "post_request_us": post_us, + "checkpoint_stats": checkpoint_stats, + "restore_stats": restore_stats, + "validation_response_sha256": post_digest, + "valid": valid, + "framework": benchmark.adapter.key, + "container_name": name if keep_running else None, + } + + +def json_config(args): + result = vars(args).copy() + result["env"] = [ + f"{item.split('=', 1)[0]}=" if "=" in item else item + for item in args.env + ] + result["run_arg"] = redact_run_args(getattr(args, "run_arg", [])) + return result + + +def report(benchmark, results_by_cfg, order): + label_width = max(20, max(len(label) for label in order)) + value_width = 14 + width = max(88, label_width + value_width * 3 + 12) + print() + print("=" * width) + print( + f" PODMAN {benchmark.adapter.heading} " + f"(n={len(next(iter(results_by_cfg.values())))})" + ) + print("=" * width) + ok = all(r["valid"] for trials in results_by_cfg.values() for r in trials) + print(f" Inference validation: {'PASS' if ok else 'FAIL'}") + print() + + baseline = "Uncompressed" if "Uncompressed" in results_by_cfg else order[0] + base = median([r["archive_size"] for r in results_by_cfg[baseline]]) + baseline_note = ( + "CRIU page-image compression disabled" + if baseline == "Uncompressed" + else "first selected configuration" + ) + print(f" Baseline: {baseline} ({baseline_note})") + storage_header = (f" {'configuration':<{label_width}} | " + f"{'archive':>{value_width}} | " + f"{'ratio':>{value_width}} | " + f"{'saved':>{value_width}}") + print(" STORAGE (relative to baseline)") + print(storage_header) + print(" " + "-" * (len(storage_header) - 2)) + for label in order: + size = median([r["archive_size"] for r in results_by_cfg[label]]) + ratio = size / base if base else 0 + print(f" {label:<{label_width}} | {format_bytes(size):>{value_width}} | " + f"{ratio:>{value_width - 1}.3f}x | " + f"{(1-ratio):>{value_width}.0%}") + print() + + print(" LATENCY (median)") + latency_header = (f" {'configuration':<{label_width}} | " + f"{'checkpoint':>{value_width}} | " + f"{'restore':>{value_width}} | " + f"{'request after':>{value_width}}") + print(latency_header) + print(" " + "-" * (len(latency_header) - 2)) + for label in order: + trials = results_by_cfg[label] + values = [ + format_duration(median([r[key] for r in trials])) + for key in ("checkpoint_wall_us", "restore_wall_us", + "post_request_us") + ] + print(f" {label:<{label_width}} | " + " | ".join( + f"{value:>{value_width}}" for value in values)) + + +def run_main(benchmark, argv=None, description=None): + adapter = benchmark.adapter + runtime = benchmark.state + ap = argparse.ArgumentParser(description=description, + formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--accelerator", choices=["gpu", "cpu"], default="gpu", + help="Inference accelerator (default: gpu)") + adapter.add_image_arguments(ap) + ap.add_argument("--model", default="Qwen/Qwen3-0.6B") + ap.add_argument("--served-model-name", + help="Model name to use in OpenAI API requests; defaults " + "to --model") + adapter.add_model_arguments(ap) + ap.add_argument("--port", type=int, default=30000) + ap.add_argument("--base-url", + help="Server URL (default: http://127.0.0.1:PORT)") + ap.add_argument("--health-path", default="/health") + ap.add_argument("--container-name", default=adapter.default_container_name) + ap.add_argument("-n", "--iterations", type=int, default=3) + ap.add_argument("--modes", nargs="+", + default=["uncompressed", "lz4-page", "lz4-region"], + choices=["uncompressed", "lz4-page", "lz4-region"], + help="CRIU memory page compression modes to compare") + ap.add_argument("--region-sizes", nargs="+", type=int, + default=[65536, 262144, 1048576], + help="Region sizes in bytes when --modes contains " + "lz4-region") + ap.add_argument("--compress-acceleration", type=int, default=1, + help="CRIU LZ4 acceleration level") + ap.add_argument("--decompress-threads", type=int, default=None, + help="CRIU LZ4/zero-fill worker concurrency " + "(default: unset = CRIU default; 0 = auto; " + "1 = serial per request; N > 1 = aggregate worker limit); " + "written to runc.conf") + ap.add_argument("--runc-conf", default="/etc/criu/runc.conf", + help="CRIU config file read by runc during Podman checkpoint") + ap.add_argument("--archive-compression", default="none", + choices=["none", "gzip", "zstd"], + help="Podman checkpoint archive compression") + ap.add_argument("--criu-libdir", default=os.environ.get("CRIU_LIBS_DIR"), + help="Directory containing CRIU plugin .so files") + ap.add_argument("--hf-cache", default=os.path.expanduser("~/.cache/huggingface")) + ap.add_argument("--gpu-device", default="nvidia.com/gpu=all", + help="GPU device selector passed to podman --device") + ap.add_argument("--security-opt", default="label=disable", + help="Security option passed to podman --security-opt") + ap.add_argument("--cuda-visible-devices", default="0") + ap.add_argument("--shm-size", default="32g") + adapter.add_resource_arguments(ap) + ap.add_argument("--prompt", default="Say hello in one short sentence.") + ap.add_argument("--prompt-file", + help="Read the validation/warmup prompt from a text file") + ap.add_argument("--max-tokens", type=int, default=32) + ap.add_argument("--temperature", type=float, default=0, + help="Chat completion temperature for validation requests") + ap.add_argument("--seed", type=int, default=42, + help="Chat completion seed used before and after restore") + ap.add_argument("--chat-extra-json", type=json_object, + help="JSON object merged into every chat completion request") + adapter.add_request_arguments(ap) + ap.add_argument("--warmup-requests", type=int, default=0, + help="Extra successful chat requests to issue before the " + "measured pre-checkpoint request") + ap.add_argument("--wait-seconds", type=int, default=900) + ap.add_argument("--request-timeout", type=int, default=180) + ap.add_argument("--env", action="append", default=[], + help="Extra container environment entry, e.g. KEY=VALUE") + ap.add_argument("--volume", action="append", default=[], + help="Extra container volume, e.g. /host:/container") + ap.add_argument("--run-arg", action="append", default=[], + help="Extra raw argument for podman run before the image") + ap.add_argument("--ulimit", action="append", default=[], + help="Extra Podman ulimit, e.g. nofile=65535:524288") + adapter.add_server_arguments(ap) + ap.add_argument("--print-stats", action="store_true", + help="Ask Podman to print checkpoint/restore stats") + ap.add_argument("--keep-checkpoint-files", action="store_true", + help="Pass --keep to podman checkpoint") + ap.add_argument("--keep-running", action="store_true", + help="Leave only the final measured restored container " + "running") + ap.add_argument("--json", metavar="FILE", help="Write raw results to JSON") + args = ap.parse_args(argv) + args.base_url = server_base_url(args.port, args.base_url) + + if args.iterations <= 0: + ap.error("--iterations must be greater than zero") + if not 1 <= args.compress_acceleration <= MAX_COMPRESSION_ACCELERATION: + ap.error("--compress-acceleration must be between 1 and " + f"{MAX_COMPRESSION_ACCELERATION}") + if args.decompress_threads is not None and \ + not 0 <= args.decompress_threads <= MAX_DECOMPRESSION_THREADS: + ap.error("--decompress-threads must be between 0 and " + f"{MAX_DECOMPRESSION_THREADS}") + if any(size <= 0 or size > MAX_REGION_SIZE or size % PAGE_SIZE + for size in args.region_sizes): + ap.error(f"--region-sizes must be positive multiples of {PAGE_SIZE} " + f"not exceeding {MAX_REGION_SIZE}") + if len(args.modes) != len(set(args.modes)): + ap.error("--modes must not contain duplicate values") + if len(args.region_sizes) != len(set(args.region_sizes)): + ap.error("--region-sizes must not contain duplicate values") + if any(item.split("=", 1)[0] in HF_TOKEN_ENV_VARS for item in args.env): + ap.error("set Hugging Face tokens in the host environment instead of " + "passing their values through --env") + if any(run_arg_sets_environment(item) for item in args.run_arg): + ap.error("pass environment entries through --env, not --run-arg") + adapter.normalize_args(ap, args) + if os.getuid() != 0: + sys.exit("Error: run as root so Podman/CRIU can checkpoint the container") + runtime.cleanup_containers = True + + import atexit + for handled in SIGNALS: + signal.signal( + handled, + lambda signum, frame: signal_handler(benchmark, signum, frame), + ) + atexit.register(cleanup, benchmark) + + info = collect_system_info() + print() + adapter.prepare_args(args) + if args.prompt_file: + prompt = read_file(args.prompt_file) + if prompt is None: + sys.exit(f"Error: prompt file not found: {args.prompt_file}") + args.prompt = prompt.rstrip("\n") + + print(f"Podman {adapter.display_name} Checkpoint/Restore Benchmark") + print(f" Kernel : {info.get('kernel', '?')}") + print(f" CPU : {info.get('cpu', '?')}") + print(f" Memory : {info.get('memory_mb', '?')} MB") + print(f" GPU : {', '.join(info.get('gpus', ['?']))}") + print(f" Accelerator: {args.accelerator}") + print(f" Podman : {info.get('podman', '?')}") + print(f" CRIU : {info.get('criu', '?')}") + print(f" Plugin : {args.criu_libdir or 'default CRIU plugin path'}") + print(f" Runc conf: {args.runc_conf}") + print(f" Archive: podman --compress={args.archive_compression}") + + cfgs = [] + for mode in args.modes: + if mode == "lz4-region": + for rs in args.region_sizes: + cfgs.append({"mode": "lz4-region", "region_size": rs}) + else: + cfgs.append({"mode": mode, "region_size": 0}) + labels = [cfg_label(cfg) for cfg in cfgs] + + print(f" Config : {args.iterations}+1 iterations, " + f"modes={','.join(labels)}") + print(" Restore: decompress-threads=" + f"{decompress_threads_label(args.decompress_threads)}") + print(f" Server : {args.base_url}, model={args.model}, " + f"{adapter.server_summary(args)}") + print(f" Request: max_tokens={args.max_tokens}, " + f"temperature={args.temperature:g}, " + f"warmup_requests={args.warmup_requests}") + if args.prompt_file: + print(f" Prompt : {args.prompt_file}") + if args.ulimit: + print(f" Ulimit : {','.join(args.ulimit)}") + + results = {label: [] for label in labels} + total = args.iterations + 1 + trial = 0 + for i in range(total): + warmup = (i == 0) + configurations = list(zip(cfgs, labels)) + offset = i % len(configurations) + configurations = configurations[offset:] + configurations[:offset] + for config_index, (cfg, label) in enumerate(configurations): + trial += 1 + workdir = tempfile.mkdtemp(prefix=adapter.temp_prefix) + runtime.tempdirs.add(workdir) + try: + retain = (args.keep_running and not warmup and + i == total - 1 and + config_index == len(configurations) - 1) + result = benchmark.run_trial(cfg, workdir, args, trial, retain) + if not warmup: + results[label].append(result) + except Exception as e: + runtime.tempdirs.discard(workdir) + print(f"\n ERROR: {label}: {e}", file=sys.stderr) + print(f" Artifacts preserved in {workdir}", file=sys.stderr) + print(container_diagnostics( + f"{args.container_name}-{os.getpid()}-{trial}"), + file=sys.stderr) + raise + else: + shutil.rmtree(workdir) + runtime.tempdirs.discard(workdir) + print(f" completed {'warmup' if warmup else f'{i}/{args.iterations}'}") + + benchmark.report(results, labels) + + if args.json: + with open(args.json, "w") as f: + json.dump({"system": info, + "framework": adapter.key, + "config": json_config(args), + "results": results}, f, indent=2) + print(f"\nResults written to {args.json}") + + # A successful benchmark must not report success until the host-wide + # CRIU configuration is restored. atexit remains a fallback for errors + # and signals, where cleanup diagnostics cannot change an existing status. + benchmark.restore_runc_conf() + print() diff --git a/test/others/compression/sglang/Makefile b/test/others/compression/sglang/Makefile new file mode 100644 index 000000000..594edc070 --- /dev/null +++ b/test/others/compression/sglang/Makefile @@ -0,0 +1,3 @@ +run: + ./run.sh +.PHONY: run diff --git a/test/others/compression/sglang/run.sh b/test/others/compression/sglang/run.sh new file mode 100755 index 000000000..cf7d7565b --- /dev/null +++ b/test/others/compression/sglang/run.sh @@ -0,0 +1,30 @@ +#!/bin/sh + +set -eu + +root=$(cd -- "$(dirname -- "$0")/../../../.." && pwd) + +: "${SGLANG_CPU_IMAGE:?SGLANG_CPU_IMAGE must name an image built from SGLang docker/xeon.Dockerfile}" +: "${SGLANG_CPU_MODEL:=Qwen/Qwen2.5-0.5B-Instruct}" + +if [ "$(id -u)" -ne 0 ]; then + echo "compression/sglang: must run as root" >&2 + exit 1 +fi + +timeout --foreground --kill-after=60s 3600s python3 \ + "$root/contrib/compression-benchmark/podman-sglang.py" \ + --accelerator cpu \ + --image "$SGLANG_CPU_IMAGE" \ + --model "$SGLANG_CPU_MODEL" \ + --iterations 1 \ + --modes uncompressed lz4-page lz4-region \ + --region-sizes 65536 \ + --decompress-threads 4 \ + --archive-compression none \ + --max-total-tokens 512 \ + --context-length 512 \ + --max-tokens 4 \ + --shm-size 4g \ + --wait-seconds 1200 \ + --request-timeout 300