mreschke/speedtest-hd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
1fde37bb914c853c40514eaf8b722164f5ed88d0
speedtest-hd/speedtest-hd.py
T

732 lines
25 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""speedtest-hd — a CrystalDiskMark-style storage benchmark for Linux, on fio.
This is the Python successor to ``speedtest-hd.sh``. It measures storage the way
CrystalDiskMark does, and adds a dedicated **SLOG / sync-write latency profile**
for diagnosing ZFS ZIL performance (NFS / iSCSI / VM sync workloads).
Three profiles
--------------
* **cdm** — the four CrystalDiskMark default tests (``SEQ1M Q8T1``, ``SEQ1M
Q1T1``, ``RND4K Q32T16``, ``RND4K Q1T1``), each measured for read *and*
write, reported in MB/s and IOPS.
* **slog** — synchronous 4K random writes swept across thread counts, reporting
IOPS, MB/s and p50/p99 *commit latency*. This is the load a ZFS SLOG sees.
* **dd** — a dependency-free fallback when ``fio`` isn't installed.
Why fio JSON?
-------------
Every measurement asks fio for ``--output-format=json`` and parses it with the
standard library. That's the whole reason this is Python and not bash: robust,
unit-safe parsing of bandwidth / IOPS / latency percentiles with no fragile
text scraping.
See README.md for the full case study (diagnosing a "slow" Optane SLOG that
turned out to be CPU power management, not the disk).
"""
from __future__ import annotations
import argparse
import glob
import json
import os
import shutil
import statistics
import subprocess
import sys
import time
from dataclasses import dataclass, field
from typing import Callable, Optional, Sequence
# --------------------------------------------------------------------------- #
# Color / styling (standard library only -- raw ANSI escape codes)
# --------------------------------------------------------------------------- #
# SGR escape sequences. We stick to the basic 16-color set so output respects
# the user's terminal theme instead of hard-coding RGB.
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
ITALIC = "\033[3m"
RED = "\033[31m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
BLUE = "\033[34m"
MAGENTA = "\033[35m"
CYAN = "\033[36m"
WHITE = "\033[37m"
BRIGHT_CYAN = "\033[96m"
def supports_color(stream) -> bool:
"""Decide per-stream whether to emit ANSI codes.
Honors the de-facto ``NO_COLOR`` / ``FORCE_COLOR`` conventions and otherwise
colors only when the stream is an interactive terminal.
"""
if os.environ.get("NO_COLOR"):
return False
if os.environ.get("FORCE_COLOR"):
return True
if os.environ.get("TERM") == "dumb":
return False
return bool(getattr(stream, "isatty", None)) and stream.isatty()
class Painter:
"""Wraps text in SGR codes, or returns it untouched when color is off."""
def __init__(self, enabled: bool):
self.enabled = enabled
def paint(self, text: str, *codes: str) -> str:
if not self.enabled or not codes:
return text
return "".join(codes) + text + RESET
# stdout carries results (banner + tables); stderr carries progress + verbose
# fio dumps. Each gets its own enable flag so piping one but not the other works.
OUT = Painter(supports_color(sys.stdout))
ERR = Painter(supports_color(sys.stderr))
# --------------------------------------------------------------------------- #
# Constants
# --------------------------------------------------------------------------- #
# Preference order for the IO engine. io_uring is the modern, lowest-overhead
# Linux engine; libaio is older (and only truly async with O_DIRECT); posixaio
# and sync are the portable fallbacks (e.g. some NFS mounts).
ENGINE_CANDIDATES = ("io_uring", "libaio", "posixaio", "sync")
# Binary unit multipliers (fio treats "1g" as 1 GiB, so we match that).
_SIZE_UNITS = {"k": 1024, "m": 1024**2, "g": 1024**3, "t": 1024**4}
@dataclass(frozen=True)
class CdmTest:
"""One CrystalDiskMark test. ``seq`` picks sequential vs random patterns."""
label: str
bs: str
iodepth: int
numjobs: int
seq: bool
@property
def read_rw(self) -> str:
return "read" if self.seq else "randread"
@property
def write_rw(self) -> str:
return "write" if self.seq else "randwrite"
# The four CrystalDiskMark default tests, as data, in CrystalDiskMark's own
# display order. Q = queue depth (iodepth), T = threads (numjobs).
CDM_TESTS: tuple[CdmTest, ...] = (
CdmTest("SEQ1M Q8T1", bs="1m", iodepth=8, numjobs=1, seq=True),
CdmTest("SEQ1M Q1T1", bs="1m", iodepth=1, numjobs=1, seq=True),
CdmTest("RND4K Q32T16", bs="4k", iodepth=32, numjobs=16, seq=False),
CdmTest("RND4K Q1T1", bs="4k", iodepth=1, numjobs=1, seq=False),
)
# Thread counts for the SLOG sweep. T1 is the headline single-stream latency;
# the higher counts show how the SLOG scales as concurrent sync writers pile on.
SLOG_THREADS: tuple[int, ...] = (1, 4, 8, 16)
# --------------------------------------------------------------------------- #
# Configuration
# --------------------------------------------------------------------------- #
@dataclass
class Config:
"""Resolved run settings. ``engine`` and ``direct`` may start unset and get
filled in by :func:`detect_io_settings`."""
path: str
mode: Optional[str] # "cdm" | "slog" | "dd" | None (auto)
engine: Optional[str] # None => auto-detect
direct: Optional[bool] # None => auto-detect
runtime: int
size: str
verbose: bool
assume_yes: bool
benchfile: str = field(default="")
@property
def direct_flag(self) -> int:
"""fio's --direct takes 0/1; default to 0 until detection runs."""
return 1 if self.direct else 0
# --------------------------------------------------------------------------- #
# Small helpers
# --------------------------------------------------------------------------- #
def log(message: str) -> None:
"""Progress/diagnostic line -> stderr, so stdout stays pure results."""
print(message, file=sys.stderr, flush=True)
def step(message: str) -> None:
"""A single colored progress line ('measuring ...') -> stderr."""
log(f" {ERR.paint('▶', BOLD, CYAN)} {ERR.paint(message, DIM)}")
def vsection(title: str) -> None:
"""A clear, ruled header for one --verbose fio section -> stderr."""
width = 78
head = f"─── {title} "
head += "─" * max(3, width - len(head))
log("")
log(ERR.paint(head, BOLD, CYAN))
def parse_size_bytes(size: str) -> int:
"""Turn an fio-style size ("4k", "1g", "512m", "2048") into bytes."""
s = size.strip().lower()
if s and s[-1] in _SIZE_UNITS:
return int(float(s[:-1]) * _SIZE_UNITS[s[-1]])
return int(s)
def render_table(
headers: Sequence[str],
rows: Sequence[Sequence[str]],
aligns: Optional[Sequence[str]] = None,
col_styles: Optional[Sequence[Optional[str]]] = None,
) -> str:
"""Render a bordered ASCII table whose columns auto-size to their content.
``aligns`` is a per-column "<" (left) or ">" (right); by default the first
column is left-aligned (the label) and the rest are right-aligned (numbers).
``col_styles`` is an optional per-column ANSI style applied to body cells
(header is always bold, borders dim). Widths are computed on the *plain*
text and color is applied after padding, so escape codes never skew layout.
"""
ncols = len(headers)
if aligns is None:
aligns = ["<"] + [">"] * (ncols - 1)
if col_styles is None:
col_styles = [None] * ncols
# Widest cell (header or any row) sets each column's width.
widths = [
max([len(str(headers[c]))] + [len(str(r[c])) for r in rows])
for c in range(ncols)
]
bar = OUT.paint("|", DIM)
sep = OUT.paint(" | ", DIM)
def rule() -> str:
return OUT.paint("+" + "+".join("-" * (w + 2) for w in widths) + "+", DIM)
def line(cells: Sequence[str], *, header: bool = False) -> str:
parts = []
for c in range(ncols):
padded = f"{str(cells[c]):{aligns[c]}{widths[c]}}"
if header:
parts.append(OUT.paint(padded, BOLD))
elif col_styles[c]:
parts.append(OUT.paint(padded, col_styles[c]))
else:
parts.append(padded)
return f"{bar} " + sep.join(parts) + f" {bar}"
out = [rule(), line(headers, header=True), rule()]
out += [line(r) for r in rows]
out.append(rule())
return "\n".join(out)
# --------------------------------------------------------------------------- #
# fio: results, invocation, parsing
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class FioResult:
"""Parsed metrics for one direction (read or write) of an fio run.
Latencies are completion latency (clat) in microseconds; for synchronous
writes that is effectively the durable-commit latency.
"""
bw_mbps: float # decimal MB/s, matching CrystalDiskMark's SI figure
iops: float
p50_us: float
p99_us: float
mean_us: float
@classmethod
def zero(cls) -> "FioResult":
return cls(0.0, 0.0, 0.0, 0.0, 0.0)
def _direction_of(rw: str) -> str:
"""fio reports 'read' and 'write' sub-objects; pick the active one."""
return "read" if "read" in rw else "write"
def _aggregate(jobs: list[dict], direction: str) -> FioResult:
"""Combine per-job fio JSON stats into a single :class:`FioResult`.
We deliberately run fio *without* --group_reporting and aggregate ourselves,
which sidesteps fio's group-merge quirks: sum throughput, average the median
latency, and take the worst-case tail (p99) across jobs.
"""
def section(job: dict) -> dict:
return job.get(direction, {})
bw_mbps = sum(section(j).get("bw_bytes", 0) for j in jobs) / 1e6
iops = sum(section(j).get("iops", 0.0) for j in jobs)
def percentiles(key: str) -> list[float]:
vals = []
for j in jobs:
pct = section(j).get("clat_ns", {}).get("percentile", {})
v = pct.get(key)
if v is not None:
vals.append(v)
return vals
p50_ns = percentiles("50.000000")
p99_ns = percentiles("99.000000")
mean_ns = [section(j).get("clat_ns", {}).get("mean", 0.0) for j in jobs]
return FioResult(
bw_mbps=bw_mbps,
iops=iops,
p50_us=(statistics.mean(p50_ns) / 1000.0) if p50_ns else 0.0, # avg of medians
p99_us=(max(p99_ns) / 1000.0) if p99_ns else 0.0, # worst tail
mean_us=(statistics.mean(mean_ns) / 1000.0) if mean_ns else 0.0,
)
def run_fio(
cfg: Config,
*,
rw: str,
bs: str,
iodepth: int,
numjobs: int,
engine: Optional[str] = None,
direct: Optional[bool] = None,
sync: bool = False,
end_fsync: bool = False,
) -> FioResult:
"""Run a single fio job against the shared bench file and parse its JSON.
``engine``/``direct`` default to the detected config values; the SLOG path
overrides them (psync + buffered + O_SYNC). ``end_fsync`` flushes the device
cache at the end of a write run so cached writes can't inflate the result.
"""
engine = engine if engine is not None else cfg.engine
direct = direct if direct is not None else cfg.direct
cmd = [
"sudo", "fio",
"--name=speedtest",
f"--filename={cfg.benchfile}",
f"--ioengine={engine}",
f"--direct={1 if direct else 0}",
f"--rw={rw}",
f"--bs={bs}",
f"--size={cfg.size}",
f"--numjobs={numjobs}",
f"--iodepth={iodepth}",
"--time_based",
f"--runtime={cfg.runtime}",
"--output-format=json",
]
if sync:
cmd.append("--sync=1") # O_SYNC: every write is a durable commit
if end_fsync:
cmd.append("--end_fsync=1") # flush device cache once at the end
proc = subprocess.run(cmd, capture_output=True, text=True)
if cfg.verbose:
sync_tag = " sync=1" if sync else ""
vsection(f"fio · {rw} bs={bs} iodepth={iodepth} numjobs={numjobs}{sync_tag}")
log(" " + ERR.paint("$ " + " ".join(cmd[1:]), DIM, ITALIC))
log(ERR.paint(proc.stdout.strip() or "(no stdout)", DIM))
if proc.stderr.strip():
log(ERR.paint(proc.stderr.strip(), YELLOW))
try:
data = json.loads(proc.stdout)
return _aggregate(data["jobs"], _direction_of(rw))
except (json.JSONDecodeError, KeyError, ValueError):
# fio failed or produced no parseable JSON; report zeros rather than
# crashing the whole run (verbose mode above shows what went wrong).
return FioResult.zero()
def fio_probe(path: str, engine: str, direct: bool) -> bool:
"""Throwaway 1s fio job to see if an (engine, O_DIRECT) combo works here.
This is how we stay accurate *and* portable across ext4/xfs/btrfs/ZFS/NFS
without the caller needing to know what each filesystem supports.
"""
cmd = [
"sudo", "fio", "--name=probe",
f"--directory={path}",
f"--ioengine={engine}",
"--rw=write", "--bs=4k", "--size=1m",
f"--direct={1 if direct else 0}",
"--time_based", "--runtime=1",
]
proc = subprocess.run(cmd, capture_output=True, text=True)
for leftover in glob.glob(os.path.join(path, "probe*")):
try:
os.remove(leftover)
except OSError:
pass
return proc.returncode == 0
def detect_io_settings(cfg: Config) -> None:
"""Fill in ``cfg.engine`` and ``cfg.direct`` if the user didn't force them.
O_DIRECT bypasses the OS page cache so we measure the device, not RAM. Not
every filesystem supports it (older OpenZFS, some NFS), so we probe and fall
back to buffered rather than erroring out.
"""
if cfg.engine is None:
for engine in ENGINE_CANDIDATES:
if fio_probe(cfg.path, engine, direct=False):
cfg.engine = engine
break
else:
cfg.engine = "sync"
if cfg.direct is None:
cfg.direct = fio_probe(cfg.path, cfg.engine, direct=True)
# --------------------------------------------------------------------------- #
# Profiles
# --------------------------------------------------------------------------- #
def _meta_line(text: str) -> str:
"""Color the 'Label :' prefix of a banner metadata line, leave value as-is.
The value may already contain its own ANSI codes (e.g. a red O_DIRECT
warning); since these lines aren't width-aligned that's harmless.
"""
key, sep, val = text.partition(":")
if not sep:
return " " + text
return " " + OUT.paint(key + ":", BOLD, BRIGHT_CYAN) + val
def banner(title: str, cfg: Config, extra: Sequence[str] = ()) -> None:
full = f"speedtest-hd · {title}"
inner = len(full) + 4
print()
print(OUT.paint("╭" + "─" * inner + "╮", BOLD, CYAN))
print(
OUT.paint("│", BOLD, CYAN)
+ " " + OUT.paint(full, BOLD, WHITE) + " "
+ OUT.paint("│", BOLD, CYAN)
)
print(OUT.paint("╰" + "─" * inner + "╯", BOLD, CYAN))
print(_meta_line(f"Target : {OUT.paint(cfg.path, CYAN)}"))
for line_ in extra:
print(_meta_line(line_))
print()
def cdm_profile(cfg: Config) -> None:
"""The CrystalDiskMark-style profile: 4 tests x (read, write), MB/s + IOPS."""
detect_io_settings(cfg)
if cfg.direct:
direct_label = OUT.paint("enabled (device)", GREEN)
else:
direct_label = OUT.paint(
"DISABLED (buffered -- may reflect RAM cache!)", BOLD, RED
)
banner(
"CrystalDiskMark-style storage benchmark",
cfg,
extra=[
f"Engine : {cfg.engine} O_DIRECT: {direct_label}",
f"Profile : size={cfg.size.upper()} runtime={cfg.runtime}s/run (8 runs)",
],
)
rows: list[list[str]] = []
for test in CDM_TESTS:
step(f"measuring {test.label.strip()} ...")
r = run_fio(cfg, rw=test.read_rw, bs=test.bs,
iodepth=test.iodepth, numjobs=test.numjobs)
w = run_fio(cfg, rw=test.write_rw, bs=test.bs,
iodepth=test.iodepth, numjobs=test.numjobs, end_fsync=True)
rows.append([
test.label,
f"{r.bw_mbps:.2f}", f"{w.bw_mbps:.2f}",
f"{r.iops:.0f}", f"{w.iops:.0f}",
])
_cleanup(cfg)
print()
print(render_table(
["Test", "Read (MB/s)", "Write (MB/s)", "Read (IOPS)", "Write (IOPS)"],
rows,
col_styles=[CYAN, GREEN, YELLOW, GREEN, YELLOW],
))
print()
def slog_profile(cfg: Config) -> None:
"""SLOG / sync-write latency profile.
Forces synchronous 4K random writes (O_SYNC) via the portable psync engine,
so every write is a ZIL commit -- exercising a ZFS SLOG exactly the way a
sync=always dataset (NFS/iSCSI/VM) does, regardless of the dataset's own
sync property. Engine/direct detection is only for the banner here.
"""
detect_io_settings(cfg)
banner(
"SLOG / sync-write latency profile",
cfg,
extra=[
"Method : fio randwrite bs=4k --sync=1 (O_SYNC), psync engine",
f"Profile : runtime={cfg.runtime}s/run size={cfg.size.upper()}",
"Note : every write is a synchronous ZIL commit -- the load your",
" SLOG actually sees. Watch it live in another shell with:",
" zpool iostat -vl <pool> 1",
],
)
# Measure everything first (progress -> stderr), then draw the whole table,
# so the "measuring..." lines can't interleave into the middle of it.
rows: list[list[str]] = []
for threads in SLOG_THREADS:
step(f"measuring 4K sync randwrite T{threads} ...")
res = run_fio(
cfg, rw="randwrite", bs="4k", iodepth=1, numjobs=threads,
engine="psync", direct=False, sync=True,
)
rows.append([
f"4K sync T{threads}",
f"{res.iops:.0f}", f"{res.bw_mbps:.2f}",
f"{res.p50_us:.1f}", f"{res.p99_us:.1f}",
])
_cleanup(cfg)
print()
print(render_table(
["Test", "IOPS", "MB/s", "p50 lat(us)", "p99 lat(us)"],
rows,
col_styles=[CYAN, GREEN, GREEN, YELLOW, MAGENTA],
))
print()
print(OUT.paint(" Healthy Optane SLOG (eg P1600X) single-stream (T1) target:", BOLD))
print(OUT.paint(" ~15-25k IOPS, p50 latency ~40-65us.", GREEN)
+ OUT.paint(" Much higher latency usually means", DIM))
print(OUT.paint(" CPU C-states / PCIe ASPM / BIOS power profile (eg Dell DAPC) throttling.", DIM))
print()
def dd_profile(cfg: Config) -> None:
"""Dependency-free fallback when fio isn't installed.
Far cruder than fio: a single sequential stream, and the cached-read figure
will reflect RAM. Use it only as a rough sanity check.
"""
count_mib = max(1, parse_size_bytes(cfg.size) // (1024**2))
nbytes = count_mib * 1024**2
banner(
"basic dd benchmark (fio not installed)",
cfg,
extra=[f"Profile : {count_mib} MiB sequential stream"],
)
def timed_dd(args: list[str]) -> float:
start = time.monotonic()
subprocess.run(args, capture_output=True, text=True)
elapsed = time.monotonic() - start
return (nbytes / elapsed / 1e6) if elapsed > 0 else 0.0 # decimal MB/s
step("measuring sequential write ...")
write_mbps = timed_dd([
"dd", "if=/dev/zero", f"of={cfg.benchfile}",
"bs=1M", f"count={count_mib}", "conv=fdatasync,notrunc",
])
step("dropping caches for uncached read ...")
subprocess.run(["sudo", "sh", "-c", "echo 3 > /proc/sys/vm/drop_caches"],
capture_output=True, text=True)
step("measuring uncached read ...")
uncached_mbps = timed_dd([
"dd", f"if={cfg.benchfile}", "of=/dev/null", "bs=1M", f"count={count_mib}",
])
step("measuring cached read ...")
cached_mbps = timed_dd([
"dd", f"if={cfg.benchfile}", "of=/dev/null", "bs=1M", f"count={count_mib}",
])
_cleanup(cfg)
print()
print(render_table(
["Test", "MB/s"],
[
["Sequential write", f"{write_mbps:.2f}"],
["Uncached read", f"{uncached_mbps:.2f}"],
["Cached read (RAM)", f"{cached_mbps:.2f}"],
],
col_styles=[CYAN, GREEN],
))
print()
def _cleanup(cfg: Config) -> None:
"""Remove the shared bench file."""
try:
os.remove(cfg.benchfile)
except OSError:
pass
# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="speedtest-hd.py",
description="CrystalDiskMark-style storage benchmark built on fio.",
epilog=(
"Examples:\n"
" speedtest-hd.py .\n"
" speedtest-hd.py /mnt/nvmepool --runtime=10 --size=4g\n"
" speedtest-hd.py /mnt/nfsshare --buffered\n"
" speedtest-hd.py /mnt/nvme-ultra-r10/vm-root --slog --runtime=30\n"
),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
p.add_argument("path", help="directory/mount to benchmark ('.' for cwd)")
mode = p.add_mutually_exclusive_group()
mode.add_argument("--fio", action="store_const", const="cdm", dest="mode",
help="force the fio CrystalDiskMark-style profile")
mode.add_argument("--dd", action="store_const", const="dd", dest="mode",
help="force the basic dd fallback test")
mode.add_argument("--slog", action="store_const", const="slog", dest="mode",
help="SLOG / sync-write latency profile (ZFS ZIL)")
direct = p.add_mutually_exclusive_group()
direct.add_argument("--direct", action="store_const", const=True, dest="direct",
help="force O_DIRECT (bypass page cache)")
direct.add_argument("--buffered", action="store_const", const=False, dest="direct",
help="force buffered IO (e.g. if O_DIRECT unsupported)")
p.add_argument("--engine", choices=ENGINE_CANDIDATES,
help="force a specific IO engine (default: auto-detect)")
p.add_argument("--runtime", type=int, default=5, metavar="SEC",
help="seconds per run (default: 5, like CrystalDiskMark)")
p.add_argument("--size", default="1g", metavar="SIZE",
help="test file size (default: 1g)")
p.add_argument("--verbose", action="store_true",
help="also print the full fio output for every run")
p.add_argument("-y", "--yes", action="store_true", dest="assume_yes",
help="skip the confirmation prompt")
p.set_defaults(mode=None, direct=None)
return p
def confirm(cfg: Config) -> None:
"""Guard prompt -- we're about to write a multi-GB file to the target."""
if cfg.assume_yes:
return
print(OUT.paint("NOTICE:", BOLD, YELLOW)
+ f" {cfg.size.upper()} free space on "
+ OUT.paint(f"'{cfg.path}'", CYAN)
+ " is required to perform the benchmark.")
answer = input(f"Are you ready to start a storage benchmark against "
f"'{cfg.path}' ? ")
if not answer.strip().lower().startswith("y"):
print(OUT.paint("Ok, cancelled!", YELLOW))
sys.exit(0)
print(OUT.paint("Great! Starting benchmark now!", BOLD, GREEN))
def main(argv: Optional[Sequence[str]] = None) -> int:
args = build_parser().parse_args(argv)
path = os.getcwd() if args.path == "." else args.path
if not os.path.exists(path):
print(f"Path {path} does not exist", file=sys.stderr)
return 1
cfg = Config(
path=path,
mode=args.mode,
engine=args.engine,
direct=args.direct,
runtime=args.runtime,
size=args.size,
verbose=args.verbose,
assume_yes=args.assume_yes,
benchfile=os.path.join(path, "speedtest-hd.bench"),
)
have_fio = shutil.which("fio") is not None
# Resolve the mode: explicit flag wins; otherwise fio if available, else dd.
mode = cfg.mode
if mode in ("cdm", "slog") and not have_fio:
print(ERR.paint("ERROR:", BOLD, RED)
+ " --fio/--slog require fio (apt install fio / pacman -S fio).",
file=sys.stderr)
return 1
if mode is None:
if have_fio:
mode = "cdm"
else:
print(OUT.paint("\nfio is not installed -- falling back to basic dd test.",
YELLOW))
print(OUT.paint("Install fio for the full CrystalDiskMark-style benchmark.",
DIM))
mode = "dd"
confirm(cfg)
# Dictionary dispatch: map each mode name to its handler function (the values
# are the functions themselves -- no parentheses), look up the one for `mode`,
# then call it with (cfg). Equivalent to an if/elif chain over `mode`. `mode`
# is always one of these three keys by now, so the lookup can't KeyError.
{"cdm": cdm_profile, "slog": slog_profile, "dd": dd_profile}[mode](cfg)
return 0
if __name__ == "__main__":
try:
sys.exit(main())
except KeyboardInterrupt:
print(ERR.paint("\nInterrupted.", YELLOW), file=sys.stderr)
sys.exit(130)
Reference in New Issue View Git Blame Copy Permalink