Measure what opening the catalog, a sync pass and a scan cost on a real library
Two benches for reading side by side before and after a change, against a copy of a real catalog, in the manner of identity_bench: - `dr-catalog --example catalog_bench CATALOG [FACES_DIR]` times `Catalog::open` and the backfill inside it step by step, the upload snapshot, a merge of the catalog with a copy of itself, and the face shard export and import in the steady state where nothing is new. - `persist_bench`, an ignored test in dr-ui's scan module because `persist` and `apply_judgement` are private to it, replays the catalog's own rows through `persist` (the largest folder, and the whole library) and looks up every `.drsc` sidecar the catalog has read. It works on a scratch copy and prints a fingerprint of what `persist` left, so two builds can be shown to agree. Both print best, median and CPU time; the CPU figure is the one to compare while other builds share the machine.
This commit is contained in:
@@ -0,0 +1,117 @@
|
|||||||
|
//! What the catalog's routine reads cost on a real library, off the GUI.
|
||||||
|
//!
|
||||||
|
//! cargo run --release -p dr-catalog --example catalog_bench -- CATALOG.sqlite [FACES_DIR]
|
||||||
|
//!
|
||||||
|
//! Times `Catalog::open` — which every worker thread pays, including the
|
||||||
|
//! develop view's fetch of each original and each neighbour it prefetches —
|
||||||
|
//! and the backfill that runs inside it, step by step. Run it against a
|
||||||
|
//! *copy* of a real catalog: opening migrates and backfills, which write.
|
||||||
|
//!
|
||||||
|
//! The figures are for reading side by side before and after a change; they
|
||||||
|
//! are not a gate. Compare the `cpu` column when the machine is busy.
|
||||||
|
|
||||||
|
use std::path::PathBuf;
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
|
use dr_catalog::{keywords, rating, schema, Catalog};
|
||||||
|
|
||||||
|
fn main() {
|
||||||
|
let args: Vec<String> = std::env::args().skip(1).collect();
|
||||||
|
let Some(path) = args.first().map(PathBuf::from) else {
|
||||||
|
eprintln!("usage: catalog_bench CATALOG.sqlite");
|
||||||
|
std::process::exit(2);
|
||||||
|
};
|
||||||
|
|
||||||
|
// Once untimed, so a migration or a first backfill is not in the figures.
|
||||||
|
drop(Catalog::open(&path).expect("catalog"));
|
||||||
|
|
||||||
|
time("Catalog::open", 20, || {
|
||||||
|
drop(Catalog::open(&path).unwrap());
|
||||||
|
});
|
||||||
|
|
||||||
|
let catalog = Catalog::open(&path).unwrap();
|
||||||
|
let conn = catalog.connection();
|
||||||
|
time("schema::backfill (all steps)", 20, || {
|
||||||
|
schema::backfill(conn).unwrap();
|
||||||
|
});
|
||||||
|
time(" rating::ensure_default_versions", 20, || {
|
||||||
|
rating::ensure_default_versions(conn).unwrap();
|
||||||
|
});
|
||||||
|
time(" rating::align_default_version_uuids", 20, || {
|
||||||
|
rating::align_default_version_uuids(conn).unwrap();
|
||||||
|
});
|
||||||
|
time(" keywords::adopt_orphan_terms", 20, || {
|
||||||
|
keywords::adopt_orphan_terms(conn).unwrap();
|
||||||
|
});
|
||||||
|
|
||||||
|
// A sync pass: the upload snapshot, then a merge of the catalog with a
|
||||||
|
// copy of itself — every row a match, which is the steady state.
|
||||||
|
let scratch = path.with_extension("bench-snapshot");
|
||||||
|
time("snapshot_for_upload", 3, || {
|
||||||
|
let _ = std::fs::remove_file(&scratch);
|
||||||
|
catalog.snapshot_for_upload(&scratch).unwrap();
|
||||||
|
});
|
||||||
|
println!(
|
||||||
|
" snapshot size {:.1} MB",
|
||||||
|
std::fs::metadata(&scratch).map(|m| m.len()).unwrap_or(0) as f64 / 1e6
|
||||||
|
);
|
||||||
|
let remote = path.with_extension("bench-remote");
|
||||||
|
let _ = std::fs::remove_file(&remote);
|
||||||
|
conn.execute("VACUUM INTO ?1", [remote.to_string_lossy().as_ref()])
|
||||||
|
.unwrap();
|
||||||
|
time("merge_remote_catalog (self)", 5, || {
|
||||||
|
catalog.merge_remote_catalog(&remote).unwrap();
|
||||||
|
});
|
||||||
|
let _ = std::fs::remove_file(&scratch);
|
||||||
|
let _ = std::fs::remove_file(&remote);
|
||||||
|
|
||||||
|
// The face half of a sync pass, against a copy of the face store: both
|
||||||
|
// directions in the steady state, where nothing is new either way.
|
||||||
|
if let Some(faces) = args.get(1).map(PathBuf::from) {
|
||||||
|
let model = "scrfd_10g+w600k_mbf";
|
||||||
|
let mut store = dr_catalog::FaceShardStore::open(&faces).unwrap();
|
||||||
|
println!(
|
||||||
|
" first export sent {}, first import adopted {}",
|
||||||
|
dr_catalog::face_shard::export_to_shards(conn, &mut store, model).unwrap(),
|
||||||
|
dr_catalog::face_shard::import_from_shards(conn, &store, model).unwrap()
|
||||||
|
);
|
||||||
|
time("face_shard::export_to_shards (steady)", 5, || {
|
||||||
|
dr_catalog::face_shard::export_to_shards(conn, &mut store, model).unwrap();
|
||||||
|
});
|
||||||
|
time("face_shard::import_from_shards (steady)", 5, || {
|
||||||
|
dr_catalog::face_shard::import_from_shards(conn, &store, model).unwrap();
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Run `f` a few times and print the best wall-clock, the median, and the
|
||||||
|
/// best CPU time — the figure to compare across runs on a busy machine.
|
||||||
|
fn time(label: &str, runs: usize, mut f: impl FnMut()) {
|
||||||
|
let mut wall: Vec<Duration> = Vec::with_capacity(runs);
|
||||||
|
let mut cpu: Vec<Duration> = Vec::with_capacity(runs);
|
||||||
|
for _ in 0..runs {
|
||||||
|
let c = cpu_now();
|
||||||
|
let t = Instant::now();
|
||||||
|
f();
|
||||||
|
wall.push(t.elapsed());
|
||||||
|
cpu.push(cpu_now().saturating_sub(c));
|
||||||
|
}
|
||||||
|
wall.sort();
|
||||||
|
cpu.sort();
|
||||||
|
println!(
|
||||||
|
"{label:42} best {:8.2} ms median {:8.2} ms cpu {:8.2} ms",
|
||||||
|
wall[0].as_secs_f64() * 1e3,
|
||||||
|
wall[runs / 2].as_secs_f64() * 1e3,
|
||||||
|
cpu[0].as_secs_f64() * 1e3
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// This thread's time on a CPU so far, from `/proc/self/schedstat`; zero where
|
||||||
|
/// the file is missing, which only makes the CPU column useless.
|
||||||
|
fn cpu_now() -> Duration {
|
||||||
|
std::fs::read_to_string("/proc/self/schedstat")
|
||||||
|
.ok()
|
||||||
|
.and_then(|s| s.split_whitespace().next()?.parse::<u64>().ok())
|
||||||
|
.map(Duration::from_nanos)
|
||||||
|
.unwrap_or_default()
|
||||||
|
}
|
||||||
+23
-23
File diff suppressed because one or more lines are too long
@@ -1051,4 +1051,200 @@ mod tests {
|
|||||||
0
|
0
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// What a scan pass and a sidecar pull cost on a real library.
|
||||||
|
///
|
||||||
|
/// DR_BENCH_CATALOG=/path/to/copy.sqlite cargo test --release -p dr-ui \
|
||||||
|
/// --lib persist_bench -- --ignored --nocapture --test-threads=1
|
||||||
|
///
|
||||||
|
/// Replays the catalog's own `remote` rows through [`persist`] — the
|
||||||
|
/// largest folder, which is what a pass relists after one sidecar write
|
||||||
|
/// there, and then the whole library, which is a first scan — and looks
|
||||||
|
/// up every `.drsc` sidecar the catalog has read, as [`apply_judgement`]
|
||||||
|
/// does on a pull (with no judgement, so nothing is written). Works on a
|
||||||
|
/// scratch copy beside the catalog it is given, and prints a fingerprint
|
||||||
|
/// of what `persist` left so two builds can be shown to agree.
|
||||||
|
#[test]
|
||||||
|
#[ignore]
|
||||||
|
fn persist_bench() {
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
let Ok(source) = std::env::var("DR_BENCH_CATALOG") else {
|
||||||
|
eprintln!("DR_BENCH_CATALOG not set; nothing to measure");
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
let scratch = format!("{source}.persist-bench");
|
||||||
|
let _ = std::fs::remove_file(&scratch);
|
||||||
|
{
|
||||||
|
let from = rusqlite::Connection::open(&source).unwrap();
|
||||||
|
from.execute("VACUUM INTO ?1", [&scratch]).unwrap();
|
||||||
|
}
|
||||||
|
let catalog = Catalog::open(std::path::Path::new(&scratch)).unwrap();
|
||||||
|
let conn = catalog.connection();
|
||||||
|
let root: String = conn
|
||||||
|
.query_row(
|
||||||
|
"SELECT label FROM roots WHERE kind = 'remote' LIMIT 1",
|
||||||
|
[],
|
||||||
|
|r| r.get(0),
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
let entries = |filter: &str| -> dr_sync::ScanResult {
|
||||||
|
let mut q = conn
|
||||||
|
.prepare(&format!(
|
||||||
|
"SELECT i.source_ref, r.file_id, r.etag, coalesce(i.file_size, 0)
|
||||||
|
FROM remote r JOIN images i ON i.id = r.image_id
|
||||||
|
WHERE 1 {filter}
|
||||||
|
ORDER BY i.source_ref"
|
||||||
|
))
|
||||||
|
.unwrap();
|
||||||
|
let images: Vec<dr_sync::RemoteEntry> = q
|
||||||
|
.query_map([], |r| {
|
||||||
|
Ok(dr_sync::RemoteEntry {
|
||||||
|
id: RemoteId::Stable(r.get::<_, i64>(1)? as u64),
|
||||||
|
path: RemotePath::new(r.get::<_, String>(0)?),
|
||||||
|
kind: dr_sync::EntryKind::File,
|
||||||
|
validator: dr_sync::Validator::new(
|
||||||
|
r.get::<_, Option<String>>(2)?.unwrap_or_default(),
|
||||||
|
),
|
||||||
|
size: r.get::<_, i64>(3)? as u64,
|
||||||
|
modified: None,
|
||||||
|
has_preview: false,
|
||||||
|
materialised: true,
|
||||||
|
})
|
||||||
|
})
|
||||||
|
.unwrap()
|
||||||
|
.collect::<Result<_, _>>()
|
||||||
|
.unwrap();
|
||||||
|
let mut dirs: Vec<String> = images
|
||||||
|
.iter()
|
||||||
|
.filter_map(|e| e.path.parent().map(|p| p.as_str().to_string()))
|
||||||
|
.collect();
|
||||||
|
dirs.sort();
|
||||||
|
dirs.dedup();
|
||||||
|
let directories = dirs
|
||||||
|
.into_iter()
|
||||||
|
.map(|d| {
|
||||||
|
let etag: Option<String> = conn
|
||||||
|
.query_row("SELECT etag FROM folders WHERE path = ?1", [&d], |r| {
|
||||||
|
r.get(0)
|
||||||
|
})
|
||||||
|
.ok()
|
||||||
|
.flatten();
|
||||||
|
(
|
||||||
|
RemotePath::new(d),
|
||||||
|
dr_sync::Validator::new(etag.unwrap_or_default()),
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.collect();
|
||||||
|
dr_sync::ScanResult {
|
||||||
|
images,
|
||||||
|
directories,
|
||||||
|
progress: Default::default(),
|
||||||
|
sidecars: Vec::new(),
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let biggest: i64 = conn
|
||||||
|
.query_row(
|
||||||
|
"SELECT folder_id FROM images GROUP BY folder_id ORDER BY count(*) DESC LIMIT 1",
|
||||||
|
[],
|
||||||
|
|r| r.get(0),
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
let folder = entries(&format!("AND i.folder_id = {biggest}"));
|
||||||
|
let all = entries("");
|
||||||
|
|
||||||
|
let cpu_now = || {
|
||||||
|
// The test's own thread: the harness runs it off the main one.
|
||||||
|
std::fs::read_to_string("/proc/thread-self/schedstat")
|
||||||
|
.ok()
|
||||||
|
.and_then(|s| s.split_whitespace().next()?.parse::<u64>().ok())
|
||||||
|
.map(Duration::from_nanos)
|
||||||
|
.unwrap_or_default()
|
||||||
|
};
|
||||||
|
let time = |label: &str, runs: usize, f: &mut dyn FnMut()| {
|
||||||
|
let mut wall = Vec::new();
|
||||||
|
let mut cpu = Vec::new();
|
||||||
|
for _ in 0..runs {
|
||||||
|
let (c, t) = (cpu_now(), Instant::now());
|
||||||
|
f();
|
||||||
|
wall.push(t.elapsed());
|
||||||
|
cpu.push(cpu_now().saturating_sub(c));
|
||||||
|
}
|
||||||
|
wall.sort();
|
||||||
|
cpu.sort();
|
||||||
|
println!(
|
||||||
|
"{label:44} best {:9.2} ms median {:9.2} ms cpu {:9.2} ms",
|
||||||
|
wall[0].as_secs_f64() * 1e3,
|
||||||
|
wall[runs / 2].as_secs_f64() * 1e3,
|
||||||
|
cpu[0].as_secs_f64() * 1e3
|
||||||
|
);
|
||||||
|
};
|
||||||
|
|
||||||
|
time(
|
||||||
|
&format!("persist, largest folder ({} images)", folder.images.len()),
|
||||||
|
5,
|
||||||
|
&mut || persist(&catalog, &root, &folder).unwrap(),
|
||||||
|
);
|
||||||
|
time(
|
||||||
|
&format!("persist, whole library ({} images)", all.images.len()),
|
||||||
|
3,
|
||||||
|
&mut || persist(&catalog, &root, &all).unwrap(),
|
||||||
|
);
|
||||||
|
|
||||||
|
let root_id: i64 = conn
|
||||||
|
.query_row("SELECT id FROM roots WHERE label = ?1", [&root], |r| {
|
||||||
|
r.get(0)
|
||||||
|
})
|
||||||
|
.unwrap();
|
||||||
|
let sidecars: Vec<String> = {
|
||||||
|
let mut q = conn
|
||||||
|
.prepare("SELECT path FROM sidecars WHERE path LIKE '%.drsc' ORDER BY path")
|
||||||
|
.unwrap();
|
||||||
|
q.query_map([], |r| r.get(0))
|
||||||
|
.unwrap()
|
||||||
|
.collect::<Result<_, _>>()
|
||||||
|
.unwrap()
|
||||||
|
};
|
||||||
|
let mut matched = 0usize;
|
||||||
|
time(
|
||||||
|
&format!("apply_judgement lookup x{}", sidecars.len()),
|
||||||
|
3,
|
||||||
|
&mut || {
|
||||||
|
for s in &sidecars {
|
||||||
|
matched += apply_judgement(conn, root_id, s, 0, 0, 0).unwrap();
|
||||||
|
}
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
let fingerprint: (i64, i64, i64, i64) = conn
|
||||||
|
.query_row(
|
||||||
|
"SELECT (SELECT count(*) || ':' || total(id * 31 + coalesce(folder_id, 0) * 7
|
||||||
|
+ file_size % 1000003 + availability * 3
|
||||||
|
+ metadata_state + length(source_ref))
|
||||||
|
FROM images),
|
||||||
|
(SELECT total(image_id * 13 + file_id % 1000003 + length(etag)
|
||||||
|
+ length(remote_path)) FROM remote),
|
||||||
|
(SELECT count(*) || ':' || total(kind * 17 + coalesce(subject_id, 0) * 5
|
||||||
|
+ priority * 3 + state + attempts) FROM jobs),
|
||||||
|
(SELECT count(*) || ':' || total(length(path) + length(etag)) FROM folders)",
|
||||||
|
[],
|
||||||
|
|r| {
|
||||||
|
let h = |s: String| {
|
||||||
|
s.bytes()
|
||||||
|
.fold(0i64, |a, b| a.wrapping_mul(131).wrapping_add(b as i64))
|
||||||
|
};
|
||||||
|
Ok((
|
||||||
|
h(r.get::<_, String>(0)?),
|
||||||
|
h(r.get::<_, f64>(1)?.to_string()),
|
||||||
|
h(r.get::<_, String>(2)?),
|
||||||
|
h(r.get::<_, String>(3)?),
|
||||||
|
))
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
println!("fingerprint after persist: {fingerprint:?}, judgements matched {matched}");
|
||||||
|
drop(catalog);
|
||||||
|
let _ = std::fs::remove_file(&scratch);
|
||||||
|
let _ = std::fs::remove_file(format!("{scratch}-wal"));
|
||||||
|
let _ = std::fs::remove_file(format!("{scratch}-shm"));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user