use std::fs;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::atomic::AtomicBool;
use std::time::{Duration, Instant};
use znippy_plugin_git::archive_write::{ArchiveWrite, FastWriter, SafeWriter};
use znippy_plugin_git::indexer::PushPath;
use znippy_plugin_git::uring_write::UringWriter;
const ACCOUNT: &str = "bench-account";
fn rusage(who: libc::c_int) -> Duration {
let mut ru: libc::rusage = unsafe { std::mem::zeroed() };
unsafe { libc::getrusage(who, &mut ru) };
let d = |t: libc::timeval| {
Duration::from_secs(t.tv_sec as u64) + Duration::from_micros(t.tv_usec as u64)
};
d(ru.ru_utime) + d(ru.ru_stime)
}
fn cpu_self() -> Duration {
rusage(libc::RUSAGE_SELF)
}
fn cpu_children() -> Duration {
rusage(libc::RUSAGE_CHILDREN)
}
fn loadavg() -> String {
fs::read_to_string("/proc/loadavg")
.map(|s| s.split_whitespace().take(3).collect::<Vec<_>>().join(" "))
.unwrap_or_else(|_| "?".into())
}
fn random_bytes(seed: u64, n: usize) -> Vec<u8> {
let mut v = Vec::with_capacity(n + 8);
let mut s = seed | 1;
while v.len() < n {
s ^= s >> 12;
s ^= s << 25;
s ^= s >> 27;
v.extend_from_slice(&s.wrapping_mul(0x2545_F491_4F6C_DD1D).to_le_bytes());
}
v.truncate(n);
v
}
fn make_packs(repo: &Path, target: usize, iters: usize, seed0: u64) -> Vec<Vec<u8>> {
let payload = target.saturating_sub(45).max(1);
let mut out = Vec::with_capacity(iters);
for i in 0..iters {
let blob = random_bytes(seed0 ^ (i as u64) << 20 ^ target as u64, payload);
let oid = git_stdin(repo, &["hash-object", "-w", "--stdin"], &blob);
let oid = String::from_utf8(oid).unwrap().trim().to_string();
let pack = git_stdin(
repo,
&["pack-objects", "--stdout", "-q"],
format!("{oid}\n").as_bytes(),
);
out.push(pack);
}
out
}
fn git_stdin(cwd: &Path, args: &[&str], stdin: &[u8]) -> Vec<u8> {
let mut c = Command::new("git")
.args(args)
.current_dir(cwd)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.spawn()
.expect("spawn git");
c.stdin.take().unwrap().write_all(stdin).expect("write stdin");
let out = c.wait_with_output().expect("git");
assert!(out.status.success(), "git {args:?} failed");
out.stdout
}
struct Cell {
arm: String,
target: usize,
pack_bytes: u64,
iters: usize,
index_built: bool,
wall: Duration,
cpu: Duration,
load_before: String,
load_after: String,
durability: String,
artifact: String,
}
impl Cell {
fn bytes_per_s(&self) -> f64 {
(self.pack_bytes * self.iters as u64) as f64 / self.wall.as_secs_f64()
}
fn per_op_us(&self) -> f64 {
self.wall.as_secs_f64() * 1e6 / self.iters as f64
}
fn cpu_per_op_us(&self) -> f64 {
self.cpu.as_secs_f64() * 1e6 / self.iters as f64
}
}
fn human(b: f64) -> String {
const U: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"];
let mut v = b;
let mut i = 0;
while v >= 1024.0 && i < 4 {
v /= 1024.0;
i += 1;
}
format!("{v:.1} {}", U[i])
}
fn fresh(dir: &Path) -> PathBuf {
let _ = fs::remove_dir_all(dir);
fs::create_dir_all(dir).expect("mkdir");
dir.to_path_buf()
}
fn run_ack(name: &str, base: &Path, packs: &[Vec<u8>]) -> (Duration, Duration, String) {
let dir = fresh(&base.join(format!("{name}-ack")));
let archive = dir.join("repo.znippy");
let w: Box<dyn ArchiveWrite> = match name {
"FastWriter" => Box::new(FastWriter::create(&archive).unwrap()),
"SafeWriter" => Box::new(SafeWriter::create(&archive).unwrap()),
"UringWriter" => Box::new(UringWriter::create(&archive).unwrap()),
_ => unreachable!(),
};
let d = w.durability().to_string();
let c0 = cpu_self();
let t0 = Instant::now();
for p in packs {
w.append(p).unwrap();
}
(t0.elapsed(), cpu_self() - c0, d)
}
fn run_ack_plus_index(name: &str, base: &Path, packs: &[Vec<u8>]) -> (Duration, Duration) {
let dir = fresh(&base.join(format!("{name}-idx")));
let archive = dir.join("repo.znippy");
let (w, journal): (Box<dyn ArchiveWrite>, Option<PathBuf>) = match name {
"FastWriter" => (Box::new(FastWriter::create(&archive).unwrap()), None),
"SafeWriter" => {
let s = SafeWriter::create(&archive).unwrap();
let j = s.journal_file();
(Box::new(s), Some(j))
}
"UringWriter" => {
let u = UringWriter::create(&archive).unwrap();
let j = u.journal_file().to_path_buf();
(Box::new(u), Some(j))
}
_ => unreachable!(),
};
let path = PushPath::new(w, &archive, journal).unwrap();
let c0 = cpu_self();
let t0 = Instant::now();
for p in packs {
path.push_pack(ACCOUNT, p).unwrap();
}
path.pool().wait_caught_up();
let wall = t0.elapsed();
let cpu = cpu_self() - c0;
let idx = path.indexer(ACCOUNT);
assert_eq!(
idx.rows(),
packs.len(),
"index did not finish: {} of {} rows",
idx.rows(),
packs.len()
);
(wall, cpu)
}
fn run_git(base: &Path, packs: &[Vec<u8>]) -> (Duration, Duration) {
let dir = fresh(&base.join("git"));
let st = Command::new("git")
.args(["init", "-q", "--bare", "."])
.current_dir(&dir)
.status()
.expect("git init");
assert!(st.success());
let c0 = cpu_children();
let t0 = Instant::now();
for p in packs {
let mut c = Command::new("git")
.args(["index-pack", "--stdin"])
.current_dir(&dir)
.stdin(Stdio::piped())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.expect("spawn git index-pack");
c.stdin.take().unwrap().write_all(p).expect("feed pack");
let s = c.wait().expect("git index-pack");
assert!(s.success(), "git index-pack failed");
}
let wall = t0.elapsed();
let cpu = cpu_children() - c0;
let n = fs::read_dir(dir.join("objects/pack"))
.unwrap()
.filter(|e| {
e.as_ref()
.map(|e| e.path().extension().map(|x| x == "idx").unwrap_or(false))
.unwrap_or(false)
})
.count();
assert_eq!(n, packs.len(), "git wrote {n} .idx of {}", packs.len());
(wall, cpu)
}
fn run_gix(base: &Path, packs: &[Vec<u8>], threads: Option<usize>) -> (Duration, Duration) {
use gix_features::progress::Discard;
let dir = fresh(&base.join(match threads {
Some(n) => format!("gix-{n}t"),
None => "gix".to_string(),
}));
let stop = AtomicBool::new(false);
let opts = gix_pack::bundle::write::Options {
object_hash: gix_hash::Kind::Sha1,
thread_limit: threads,
..Default::default()
};
let c0 = cpu_self();
let t0 = Instant::now();
for p in packs {
let mut cur = std::io::Cursor::new(p.as_slice());
let out = gix_pack::Bundle::write_to_directory(
&mut cur,
Some(&dir),
&mut Discard,
&stop,
None::<gix_object::find::Never>,
opts.clone(),
)
.expect("gix write_to_directory");
assert!(out.index_path.is_some(), "gix produced no .idx");
}
let wall = t0.elapsed();
let cpu = cpu_self() - c0;
let n = fs::read_dir(&dir)
.unwrap()
.filter(|e| {
e.as_ref()
.map(|e| e.path().extension().map(|x| x == "idx").unwrap_or(false))
.unwrap_or(false)
})
.count();
assert_eq!(n, packs.len(), "gix wrote {n} .idx of {}", packs.len());
(wall, cpu)
}
fn main() {
let thp = znippy_zoomies::stree::thp_enable_for_process();
eprintln!("thp_enabled={}", thp);
let args: Vec<String> = std::env::args().skip(1).collect();
let opt = |k: &str| -> Option<String> {
args.iter()
.position(|a| a == k)
.and_then(|i| args.get(i + 1).cloned())
};
let base = PathBuf::from(
opt("--dir").unwrap_or_else(|| "/home/rickard/scratch/pushpath-bench".into()),
);
fs::create_dir_all(&base).expect("mkdir base");
let only = opt("--only");
let mut rungs: Vec<(usize, usize)> = vec![
(200, 200),
(8 * 1024, 100),
(512 * 1024, 32),
(32 * 1024 * 1024, 8),
];
if let (Some(s), Some(n)) = (opt("--size"), opt("--iters")) {
rungs = vec![(s.parse().unwrap(), n.parse().unwrap())];
}
let fixture_repo = fresh(&base.join("fixtures"));
let st = Command::new("git")
.args(["init", "-q", "."])
.current_dir(&fixture_repo)
.status()
.expect("git init fixtures");
assert!(st.success());
println!("# push-path bench — {} arms, one run", if only.is_some() { 1 } else { 6 });
println!();
println!("host: oden, /dev/md1 (raid0, 2× NVMe), kernel {}", kernel());
println!("loadavg at start: {}", loadavg());
println!();
println!(
"| rung | pack | n | arm | index built? | wall/op | CPU/op | bytes/s | loadavg (before → after) |"
);
println!("|---|---|---|---|---|---|---|---|---|");
let mut cells: Vec<Cell> = Vec::new();
for (target, iters) in rungs {
eprintln!("… building {iters} distinct packs of ~{target} B");
let packs = make_packs(&fixture_repo, target, iters, 0x5EED);
let mean: u64 = (packs.iter().map(|p| p.len()).sum::<usize>() / packs.len()) as u64;
let mut push = |arm: &str,
index: &str,
wall: Duration,
cpu: Duration,
lb: String,
la: String,
dur: &str,
art: &str| {
let c = Cell {
arm: arm.into(),
target,
pack_bytes: mean,
iters,
index_built: index.contains("yes"),
wall,
cpu,
load_before: lb,
load_after: la,
durability: dur.into(),
artifact: art.into(),
};
println!(
"| {} | {} | {} | {} | {} | {:.1} µs | {:.1} µs | {}/s | {} → {} |",
human(target as f64),
human(mean as f64),
iters,
c.arm,
index,
c.per_op_us(),
c.cpu_per_op_us(),
human(c.bytes_per_s()),
c.load_before,
c.load_after
);
cells.push(c);
};
for name in ["FastWriter", "SafeWriter", "UringWriter"] {
if only.as_deref().is_some_and(|o| !name.to_lowercase().starts_with(&o.to_lowercase())) {
continue;
}
let lb = loadavg();
let (w, c, dur) = run_ack(name, &base, &packs);
push(name, "no (deferred)", w, c, lb, loadavg(), &dur, "Arrow archive, pack verbatim");
let lb = loadavg();
let (w, c) = run_ack_plus_index(name, &base, &packs);
push(name, "**yes**", w, c, lb, loadavg(), &dur, "Arrow archive + Arrow index tables");
}
if only.as_deref().is_none_or(|o| o == "git") {
let lb = loadavg();
let (w, c) = run_git(&base, &packs);
push(
"git index-pack",
"**yes**",
w,
c,
lb,
loadavg(),
"full — 3 fsync/pack MEASURED (strace -c -f, 8 KiB rung, fixture baseline subtracted)",
".pack + .idx + .rev on the filesystem",
);
}
if only.as_deref().is_none_or(|o| o == "gix") {
for (label, threads) in [
("gix Bundle::write", None),
("gix Bundle::write 1thr", Some(1usize)),
] {
let lb = loadavg();
let (w, c) = run_gix(&base, &packs, threads);
push(
label,
"**yes**",
w,
c,
lb,
loadavg(),
"NONE — 0 fsync in 200 calls MEASURED (strace); persisted by renameat only, so a machine crash after return can lose the pack and the idx. Same class as FastWriter.",
".pack + .idx on the filesystem",
);
}
}
println!("| | | | | | | | | |");
}
println!();
println!("## Who won, per rung");
println!();
println!("Two separate races, because they are not the same work. **ack** is our");
println!("three arms with the index deferred; **index built** is every arm that has a");
println!("usable index when the clock stops — our `ack+index` rows against git and gix.");
println!();
println!("| rung | fastest with index built | runner-up | margin | fastest ack (ours) |");
println!("|---|---|---|---|---|");
let mut rungs_seen: Vec<usize> = Vec::new();
for c in &cells {
if !rungs_seen.contains(&c.target) {
rungs_seen.push(c.target);
}
}
for t in &rungs_seen {
let mut with: Vec<&Cell> = cells
.iter()
.filter(|c| c.target == *t && c.index_built)
.collect();
with.sort_by(|a, b| a.wall.partial_cmp(&b.wall).unwrap());
let mut ack: Vec<&Cell> = cells
.iter()
.filter(|c| c.target == *t && !c.index_built)
.collect();
ack.sort_by(|a, b| a.wall.partial_cmp(&b.wall).unwrap());
let w0 = with[0];
let w1 = with[1];
println!(
"| {} | **{}** {:.1} µs | {} {:.1} µs | **{:.1}×** | {} {:.1} µs |",
human(*t as f64),
w0.arm,
w0.per_op_us(),
w1.arm,
w1.per_op_us(),
w1.per_op_us() / w0.per_op_us(),
ack[0].arm,
ack[0].per_op_us()
);
}
println!();
println!("## What each arm actually bought");
println!();
println!("| arm | durability on return | artifact |");
println!("|---|---|---|");
let mut seen: Vec<&str> = Vec::new();
for c in &cells {
if seen.contains(&c.arm.as_str()) {
continue;
}
seen.push(&c.arm);
println!("| {} | {} | {} |", c.arm, c.durability, c.artifact);
}
println!();
println!("loadavg at end: {}", loadavg());
println!("child CPU total: {:?}", cpu_children());
println!("self CPU total: {:?}", cpu_self());
}
fn kernel() -> String {
fs::read_to_string("/proc/sys/kernel/osrelease")
.map(|s| s.trim().to_string())
.unwrap_or_default()
}