diff --git a/bench/repeat.sh b/bench/repeat.sh new file mode 100755 index 0000000..c8b4f1d --- /dev/null +++ b/bench/repeat.sh @@ -0,0 +1,131 @@ +#!/usr/bin/env bash +# Repeated onggi measurements and an ablation for the paper (#73), on one build set and one variant switch. +# +# bench/repeat.sh repeat [n] +# bench/repeat.sh ablation +# +# repeat, n times (default 5), interleaved in each round: +# push-set push the whole build dir into a new store: stored bytes (du, with indexes), time, peak RSS +# transfer-xet push A, then B, into a new store with --sub 0 (Xet-style): B's pack bytes and push time +# transfer-onggi the same with onggi's defaults +# pull-plain pull B into an empty dir from the second store (no cache): bytes read, time +# pull-delta pull A, then B with --base A --base-dir: bytes read, bytes left in place, time +# plain and delta pulls alternate which goes first. ablation runs push-set and the A → B transfer once for each of: +# default, --sub 0, --frame 0, --compress none. +# +# Prints one JSON object per measurement (run, method, flags, bytes, seconds, peak RSS KiB, and the onggi JSON). +# Pulled files are checked against the original's sha256. Needs GNU time at /usr/bin/time and about +# 2 × |A| + 3 × |B| + one store of the set free in the work dir. Everything runs under nice; nothing outside the +# work dir is written. The page cache cannot be dropped without root: the build set is read in every round. +set -euo pipefail +mode=$1 onggi=$2 src=$3 work=$4 va=$5 vb=$6 n=${7:-5} +N="nice -n 19" +mkdir -p "$work/tmp" +t0() { /usr/bin/time -v -o "$work/time.txt" $N "$@"; } +emit() { # emit [extra json] + python3 - "$work/time.txt" "$@" << 'EOF' +import json, sys +tf, run, method, flags, nbytes, jf = sys.argv[1:7] +extra = json.loads(sys.argv[7]) if len(sys.argv) > 7 else {} +wall = rss = None +for l in open(tf): + if "Elapsed (wall clock)" in l: + wall = 0.0 + for p in l.rsplit(" ", 1)[1].strip().split(":"): + wall = wall * 60 + float(p) + if "Maximum resident set size" in l: + rss = int(l.split(":")[1]) +d = {"run": int(run), "method": method, "flags": flags, "bytes": int(nbytes), "seconds": round(wall, 2), "peak_rss_kib": rss} +if jf != "-": + d["onggi"] = json.load(open(jf)) +d.update(extra) +print(json.dumps(d), flush=True) +EOF +} +jget() { python3 -c "import json,sys;d=json.load(open(sys.argv[1]));d=d.get('stats',d);print(d[sys.argv[2]])" "$@"; } +want=$(sha256sum "$src/$vb" | cut -c1-64) +same() { [ "$(sha256sum "$1" | cut -c1-64)" = "$want" ] && echo true || echo false; } + +# the two variants, each alone in a directory (copies: the build dir is only read) +for v in a b; do + f=$va + [ $v = b ] && f=$vb + if [ ! -f "$work/v$v/$f" ]; then + rm -rf "$work/v$v" && mkdir -p "$work/v$v" && cp "$src/$f" "$work/v$v/" + fi +done + +push_set() { # push_set [flags] + local run=$1 name=$2 + shift 2 + rm -rf "$work/set" + t0 "$onggi" push --store "$work/set" "$@" --quiet --json --tmp "$work/tmp" "$src" > "$work/push-set.json" + emit "$run" push-set "$name" "$(du -sb "$work/set" | cut -f1)" "$work/push-set.json" + rm -rf "$work/set" +} + +transfer() { # transfer [flags]: push A, then B; B's packs are what crosses the network + local run=$1 name=$2 st=$3 + shift 3 + rm -rf "$st" + $N "$onggi" push --store "$st" "$@" --ref a --quiet --json --tmp "$work/tmp" "$work/va" > /dev/null + t0 "$onggi" push --store "$st" "$@" --ref b --quiet --json --tmp "$work/tmp" "$work/vb" > "$work/push-b.json" + emit "$run" "transfer-$name" "$name" "$(jget "$work/push-b.json" pack_bytes)" "$work/push-b.json" +} + +pull_plain() { + local run=$1 st=$2 + rm -rf "$work/out" + t0 "$onggi" pull --store "$st" --json b "$work/out" > "$work/pull.json" + emit "$run" pull-plain default "$(jget "$work/pull.json" read_bytes)" "$work/pull.json" \ + "{\"identical\": $(same "$work/out/$vb")}" + rm -rf "$work/out" +} + +pull_delta() { + local run=$1 st=$2 + rm -rf "$work/base" "$work/out" + $N "$onggi" pull --store "$st" a "$work/base" > /dev/null + t0 "$onggi" pull --store "$st" --json --base a --base-dir "$work/base" b "$work/out" > "$work/pull.json" + emit "$run" pull-delta default "$(jget "$work/pull.json" read_bytes)" "$work/pull.json" \ + "{\"identical\": $(same "$work/out/$vb")}" + rm -rf "$work/base" "$work/out" +} + +case $mode in +repeat) + for i in $(seq 1 "$n"); do + push_set "$i" default + transfer "$i" xet "$work/t-xet" --sub 0 + rm -rf "$work/t-xet" + transfer "$i" onggi "$work/t-onggi" + if [ "$i" = 1 ]; then + $N "$onggi" diff --store "$work/t-onggi" --json --ranges a b > "$work/diff.json" + python3 -c "import json,sys;d=json.load(open(sys.argv[1]));print(json.dumps({'run':0,'method':'diff','diff':d}))" "$work/diff.json" + fi + if [ $((i % 2)) = 1 ]; then + pull_plain "$i" "$work/t-onggi" + pull_delta "$i" "$work/t-onggi" + else + pull_delta "$i" "$work/t-onggi" + pull_plain "$i" "$work/t-onggi" + fi + rm -rf "$work/t-onggi" + done + ;; +ablation) + for m in default: sub0:--sub=0 frame0:--frame=0 compress-none:--compress=none; do + name=${m%%:*} flag=${m#*:} + # shellcheck disable=SC2086 + push_set 1 "$name" $flag + # shellcheck disable=SC2086 + transfer 1 "$name" "$work/t-$name" $flag + rm -rf "$work/t-$name" + done + ;; +*) + echo "usage: bench/repeat.sh repeat|ablation [n]" >&2 + exit 2 + ;; +esac +rm -rf "$work/va" "$work/vb" "$work/tmp" diff --git a/bench/switch_baselines.sh b/bench/switch_baselines.sh new file mode 100755 index 0000000..b14dc08 --- /dev/null +++ b/bench/switch_baselines.sh @@ -0,0 +1,59 @@ +#!/usr/bin/env bash +# Switch baselines for the paper (#73): turning a node's copy of variant A into variant B with rsync and xdelta3. +# +# bench/switch_baselines.sh [xdelta3 binary] [window] +# +# rsync: the node's copy of A is placed under B's name and rsync --no-whole-file brings it to B (local transfer, +# so sender and receiver talk through a pipe; "Total bytes sent" is what would cross the network), once without +# and once with -z (zstd level 3). xdelta3: encode B against A with a source window of [window] bytes +# (default 2 GiB), decode it, compare. Every result is checked against B's sha256. Needs GNU time at +# /usr/bin/time and about 3 × the size of B free in the work dir. Prints TSV rows: +# tool, variant, bytes (crossing the network), seconds, peak RSS KiB, identical +# and keeps the logs in the work dir. Everything runs under nice; nothing outside the work dir is written. +set -euo pipefail +src=$1 work=$2 va=$3 vb=$4 xd=${5:-xdelta3} win=${6:-2147483648} +mkdir -p "$work" +N="nice -n 19" +row() { printf '%s\t%s\t%s\t%s\t%s\t%s\n' "$@"; } +tv() { # tv cmd...: run under GNU time; the log gets time's report + local log=$1 + shift + /usr/bin/time -v -o "$log.time" $N "$@" > "$log" 2>&1 +} +wall() { python3 - "$1" << 'EOF' +import sys +for l in open(sys.argv[1]): + if "Elapsed (wall clock)" in l: + t = l.rsplit(" ", 1)[1].strip().split(":") + s = 0.0 + for p in t: + s = s * 60 + float(p) + print(round(s, 2)) +EOF +} +rss() { awk -F': ' '/Maximum resident set size/ {print $2}' "$1"; } +want=$(sha256sum "$src/$vb" | cut -c1-64) +same() { [ "$(sha256sum "$1" | cut -c1-64)" = "$want" ] && echo yes || echo NO; } + +# rsync, without and with compression +for z in plain:"" zstd:"-z --compress-choice=zstd --compress-level=3"; do + name=${z%%:*} flags=${z#*:} + rm -rf "$work/rs" && mkdir -p "$work/rs" + cp "$src/$va" "$work/rs/$vb" + log=$work/rsync-$name.log + # shellcheck disable=SC2086 + tv "$log" rsync --no-whole-file --ignore-times --stats $flags "$src/$vb" "$work/rs/" + sent=$(awk -F': ' '/^Total bytes sent/ {gsub(/,/, "", $2); print $2}' "$log") + row "rsync-$name" "$vb" "$sent" "$(wall "$log.time")" "$(rss "$log.time")" "$(same "$work/rs/$vb")" + grep -E '^(Literal data|Matched data|Total bytes (sent|received))' "$log" | sed "s/^/# rsync-$name: /" + rm -rf "$work/rs" +done + +# xdelta3: encode against A with a large source window, then decode +log=$work/xdelta3-encode.log +tv "$log" "$xd" -e -f -B "$win" -s "$src/$va" "$src/$vb" "$work/delta.vcdiff" +row xdelta3-encode "$vb" "$(stat -c %s "$work/delta.vcdiff")" "$(wall "$log.time")" "$(rss "$log.time")" - +log=$work/xdelta3-decode.log +tv "$log" "$xd" -d -f -B "$win" -s "$src/$va" "$work/delta.vcdiff" "$work/out" +row xdelta3-decode "$vb" "$(stat -c %s "$work/delta.vcdiff")" "$(wall "$log.time")" "$(rss "$log.time")" "$(same "$work/out")" +rm -f "$work/delta.vcdiff" "$work/out" diff --git a/bench/tensor_dedup.py b/bench/tensor_dedup.py new file mode 100755 index 0000000..750d8a1 --- /dev/null +++ b/bench/tensor_dedup.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +"""Tensor-level dedup baseline (ZipLLM-style) for the paper (#73). + + bench/tensor_dedup.py [--workers N] [--level 3] [--follow-symlinks] ... > result.json + +safetensors files are split into tensors by their JSON header (8-byte little-endian length, then JSON); each +tensor is an object keyed by the sha256 of its bytes. Every other file is one opaque object keyed by the sha256 of +the whole file: compiled NPU artifacts (.rbln and the like) are not parsed (onggi ADR-0009, vendor formats stay +opaque), so on them this baseline is whole-file dedup. Each distinct object is stored once, compressed with +`zstd -` (the zstd CLI, one thread per object). safetensors headers are stored as they are. + +Stored bytes = compressed distinct objects + safetensors headers + the metadata the store would need to restore +every file (a manifest listing each file's objects and an object index), counted as compact JSON. Paths are walked in +sorted order. Needs Python 3.8+ and zstd on PATH; reads the input, writes nothing but stdout. +""" +import argparse +import hashlib +import json +import os +import struct +import subprocess +import sys +import threading +import time +from concurrent.futures import ThreadPoolExecutor + +BUF = 8 << 20 + + +def zstd_size(level, feed): + """Compress what feed(write) writes with the zstd CLI; return the compressed size.""" + p = subprocess.Popen(["zstd", "-q", "-%d" % level, "-T1", "-c"], stdin=subprocess.PIPE, stdout=subprocess.PIPE) + n = [0] + + def count(): + while True: + b = p.stdout.read(BUF) + if not b: + return + n[0] += len(b) + + t = threading.Thread(target=count) + t.start() + try: + feed(p.stdin.write) + finally: + p.stdin.close() + t.join() + if p.wait() != 0: + raise RuntimeError("zstd failed") + return n[0] + + +def read_range(f, off, length, write, hasher): + f.seek(off) + left = length + while left: + b = f.read(min(BUF, left)) + if not b: + raise IOError("short read") + hasher.update(b) + write(b) + left -= len(b) + + +class Store: + def __init__(self, level): + self.level = level + self.lock = threading.Lock() + self.objects = {} # sha256 hex -> [raw size, stored size] + self.dup_objects = 0 + self.dup_bytes = 0 + + def add(self, digest, raw, stored): + """Record an object; return True if it was new.""" + with self.lock: + if digest in self.objects: + self.dup_objects += 1 + self.dup_bytes += raw + return False + self.objects[digest] = [raw, stored] + return True + + def seen(self, digest): + with self.lock: + return digest in self.objects + + +def safetensors_header(path): + with open(path, "rb") as f: + head = f.read(8) + if len(head) != 8: + return None + (n,) = struct.unpack(" 100 << 20: + return None + raw = f.read(n) + try: + hdr = json.loads(raw.decode("utf-8")) + except ValueError: + return None + return 8 + n, hdr + + +def do_safetensors(path, store, parsed): + data_start, hdr = parsed + row = {"kind": "safetensors", "header_bytes": data_start, "tensors": 0, "objects": []} + tensors = sorted( + ((k, v) for k, v in hdr.items() if k != "__metadata__"), key=lambda kv: kv[1]["data_offsets"][0] + ) + with open(path, "rb") as f: + for name, info in tensors: + a, b = info["data_offsets"] + h = hashlib.sha256() + # hash first (cheap) so duplicates are not compressed + read_range(f, data_start + a, b - a, lambda _b: None, h) + d = h.hexdigest() + stored = 0 + if not store.seen(d): + stored = zstd_size(store.level, lambda w: read_range(f, data_start + a, b - a, w, hashlib.sha256())) + new = store.add(d, b - a, stored) + row["tensors"] += 1 + row["objects"].append(d) + row.setdefault("new_tensors", 0) + row["new_tensors"] += 1 if new else 0 + return row + + +def do_opaque(path, size, store): + h = hashlib.sha256() + # one pass: hash and compress together; a duplicate's compressed size is dropped + with open(path, "rb") as f: + stored = zstd_size(store.level, lambda w: read_range(f, 0, size, w, h)) + d = h.hexdigest() + store.add(d, size, stored) + return {"kind": "opaque", "objects": [d]} + + +def walk(paths, follow): + out = [] + for p in paths: + if os.path.isfile(p): + out.append((p, os.path.basename(p))) + continue + for root, dirs, files in os.walk(p, followlinks=follow): + dirs.sort() + for n in sorted(files): + full = os.path.join(root, n) + if os.path.islink(full) and not follow: + continue + out.append((full, os.path.relpath(full, p))) + return out + + +def main(): + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + ap.add_argument("paths", nargs="+") + ap.add_argument("--workers", type=int, default=8, help="files processed at once (default 8)") + ap.add_argument("--level", type=int, default=3, help="zstd level (default 3)") + ap.add_argument("--follow-symlinks", action="store_true", help="follow symlinks (e.g. a Hugging Face snapshot)") + args = ap.parse_args() + + files = walk(args.paths, args.follow_symlinks) + store = Store(args.level) + t0 = time.time() + + def one(item): + full, rel = item + size = os.path.getsize(full) + t = time.time() + parsed = safetensors_header(full) if full.endswith(".safetensors") else None + row = do_safetensors(full, store, parsed) if parsed else do_opaque(full, size, store) + row.update(path=full, name=rel, size=size, seconds=round(time.time() - t, 2)) + print("done %s %.1fs" % (rel, row["seconds"]), file=sys.stderr, flush=True) + return row + + with ThreadPoolExecutor(max_workers=args.workers) as ex: + rows = list(ex.map(one, files)) + + manifest = [{"name": r["name"], "size": r["size"], "objects": r["objects"]} for r in rows] + index = [[d, v[0], v[1]] for d, v in sorted(store.objects.items())] + meta = len(json.dumps(manifest, separators=(",", ":"))) + len(json.dumps(index, separators=(",", ":"))) + headers = sum(r.get("header_bytes", 0) for r in rows) + objects_stored = sum(v[1] for v in store.objects.values()) + raw = sum(r["size"] for r in rows) + for r in rows: + r["objects"] = len(r["objects"]) + res = { + "tool": "bench/tensor_dedup.py", + "zstd_level": args.level, + "files": len(rows), + "raw_bytes": raw, + "safetensors_files": sum(1 for r in rows if r["kind"] == "safetensors"), + "opaque_files": sum(1 for r in rows if r["kind"] == "opaque"), + "distinct_objects": len(store.objects), + "distinct_raw_bytes": sum(v[0] for v in store.objects.values()), + "duplicate_objects": store.dup_objects, + "duplicate_bytes": store.dup_bytes, + "objects_stored_bytes": objects_stored, + "safetensors_header_bytes": headers, + "metadata_bytes": meta, + "stored_bytes": objects_stored + headers + meta, + "ratio": round(raw / float(objects_stored + headers + meta), 3) if raw else None, + "seconds": round(time.time() - t0, 1), + "per_file": rows, + } + json.dump(res, sys.stdout, indent=1) + print() + + +if __name__ == "__main__": + main()