Skip to main content

max / audiofiles

34.2 KB · 918 lines History Blame Raw
1 //! Ingest and query benchmarks: the vault layer rather than the DSP layer.
2 //!
3 //! The analysis bench in `main.rs` measures per-file DSP cost. This measures
4 //! what happens to a vault as it fills up: import throughput, dedup, and the
5 //! query latency that backs the browser UI.
6 //!
7 //! Two properties of the store make scale worth measuring rather than
8 //! assuming. Blobs are sharded one level deep on the hash prefix
9 //! (`{root}/{ab}/{hash}.{ext}`, 256 leaves), which is what a run on this bench
10 //! bought: the flat layout it replaced lost about 90% of its import throughput
11 //! between an empty vault and a 40,000-entry one. Directory cost is filesystem
12 //! dependent either way, so it stays worth measuring. And the DB runs in WAL mode
13 //! with several worker connections, so insert cost moves with index depth.
14 //!
15 //! Reported per batch rather than as one average, because the number that
16 //! matters is whether throughput is flat or degrading as the vault grows.
17 //!
18 //! Take baselines with the corpus drive otherwise idle. Import is I/O bound, so
19 //! anything else touching the same spindle lands directly in the number: a run
20 //! taken while a dataset was downloading to the same drive read 40.7 files/s
21 //! against 204 for the same corpus on an idle one. The JSON output makes that
22 //! kind of contamination easy to mistake for a regression.
23
24 use std::path::{Path, PathBuf};
25 use std::time::Instant;
26
27 use audiofiles_core::analysis::{self, config::AnalysisConfig};
28 use audiofiles_core::db::Database;
29 use audiofiles_core::id_types::SampleHash;
30 use audiofiles_core::search::{self, SearchFilter, SearchScope};
31 use audiofiles_core::store::SampleStore;
32 use audiofiles_core::vfs;
33 use rayon::prelude::*;
34
35 use crate::report::{Report, peak_rss_mb};
36 use crate::storage;
37
38 /// How many samples to analyse after import.
39 ///
40 /// Bounded because analysis is orders of magnitude dearer than import and the
41 /// point here is to populate `audio_analysis` so the filter queries below
42 /// measure something, not to benchmark the DSP (the default bench mode does
43 /// that per file). At NSynth scale an unbounded pass would dominate the run.
44 fn analyze_budget() -> usize {
45 std::env::var("AF_BENCH_ANALYZE")
46 .ok()
47 .and_then(|v| v.parse().ok())
48 .unwrap_or(2000)
49 }
50
51 /// How many files to run the per-stage probe over at each checkpoint, or 0 for
52 /// no probe.
53 ///
54 /// Off by default, and that is deliberate rather than timid. The probe analyses
55 /// files in the middle of an import, so it competes for the drive and the page
56 /// cache with the thing being measured. Every ingest baseline saved before this
57 /// existed was recorded without it, and turning it on by default would make new
58 /// runs quietly incomparable with `benchmarks/ingest-2026-07-29-*.json`. Opt in
59 /// when the per-stage curve is the question; leave it off when the ingest curve
60 /// is.
61 fn stage_probe_size() -> usize {
62 std::env::var("AF_BENCH_STAGES")
63 .ok()
64 .and_then(|v| v.parse().ok())
65 .unwrap_or(0)
66 }
67
68 /// One per-stage probe: the DSP breakdown plus persistence, at a known vault size.
69 ///
70 /// `persist_ms` is the reason this exists. The DSP stages are pure CPU over one
71 /// decoded buffer and have no way to know how large the vault is, so a flat line
72 /// from them is the expected result and the useful one: it says the per-file
73 /// numbers in section 1 keep their meaning at scale. Persistence is the stage
74 /// that can degrade, because it writes into a growing database, and section 1
75 /// never touches the DB at all.
76 struct StageProbe {
77 samples: i64,
78 decode_ms: f64,
79 loudness_ms: f64,
80 spectral_ms: f64,
81 mfcc_ms: f64,
82 vector_ms: f64,
83 bpm_key_ms: f64,
84 loop_ms: f64,
85 fingerprint_ms: f64,
86 total_ms: f64,
87 persist_ms: f64,
88 files: usize,
89 }
90
91 /// Run the per-stage probe over `files` against a vault currently holding
92 /// `samples` rows.
93 ///
94 /// Medians, not means: one file that happens to be long drags a mean far enough
95 /// to invent a trend across checkpoints that is not there.
96 fn stage_probe(
97 db: &Database,
98 samples: i64,
99 files: &[(String, PathBuf)],
100 config: &AnalysisConfig,
101 ) -> Option<StageProbe> {
102 if files.is_empty() {
103 return None;
104 }
105 let mut decode = Vec::new();
106 let mut loud = Vec::new();
107 let mut spec = Vec::new();
108 let mut mfcc = Vec::new();
109 let mut vector = Vec::new();
110 let mut bpm_key = Vec::new();
111 let mut loops = Vec::new();
112 let mut fp = Vec::new();
113 let mut total = Vec::new();
114
115 for (_, path) in files {
116 let Some((t, _, _)) = crate::time_stages(path) else {
117 continue;
118 };
119 decode.push(t.decode_ms);
120 loud.push(t.loudness_ms);
121 spec.push(t.spectral_ms);
122 mfcc.push(t.mfcc_ms);
123 vector.push(t.vector_ms);
124 bpm_key.push(t.bpm_key_ms);
125 loops.push(t.loop_ms);
126 fp.push(t.fingerprint_ms);
127 total.push(t.total_ms);
128 }
129 if total.is_empty() {
130 return None;
131 }
132
133 // Persistence is measured on a real analysis of the same files, written into
134 // the live vault. Timing a fabricated row would measure the wrong statement.
135 let results: Vec<_> = files
136 .iter()
137 .filter_map(|(hash, path)| analysis::analyze_sample(hash, path, config).ok())
138 .collect();
139 let persist_start = Instant::now();
140 let persisted = analysis::save_analysis_batch(db, &results).is_ok();
141 let persist_total_ms = persist_start.elapsed().as_secs_f64() * 1000.0;
142 let persist_ms = if persisted && !results.is_empty() {
143 persist_total_ms / results.len() as f64
144 } else {
145 f64::NAN
146 };
147
148 let files_probed = total.len();
149 Some(StageProbe {
150 samples,
151 decode_ms: crate::percentile(&mut decode, 50.0),
152 loudness_ms: crate::percentile(&mut loud, 50.0),
153 spectral_ms: crate::percentile(&mut spec, 50.0),
154 mfcc_ms: crate::percentile(&mut mfcc, 50.0),
155 vector_ms: crate::percentile(&mut vector, 50.0),
156 bpm_key_ms: crate::percentile(&mut bpm_key, 50.0),
157 loop_ms: crate::percentile(&mut loops, 50.0),
158 fingerprint_ms: crate::percentile(&mut fp, 50.0),
159 total_ms: crate::percentile(&mut total, 50.0),
160 persist_ms,
161 files: files_probed,
162 })
163 }
164
165 /// The per-stage curve as a JSON array, one object per checkpoint.
166 fn stage_series(probes: &[StageProbe]) -> serde_json::Value {
167 serde_json::Value::Array(
168 probes
169 .iter()
170 .map(|p| {
171 let round = |v: f64| (v * 1000.0).round() / 1000.0;
172 serde_json::json!({
173 "samples": p.samples,
174 "files_probed": p.files,
175 "decode_ms": round(p.decode_ms),
176 "loudness_ms": round(p.loudness_ms),
177 "spectral_ms": round(p.spectral_ms),
178 "mfcc_ms": round(p.mfcc_ms),
179 "vector_ms": round(p.vector_ms),
180 "bpm_key_ms": round(p.bpm_key_ms),
181 "loop_ms": round(p.loop_ms),
182 "fingerprint_ms": round(p.fingerprint_ms),
183 "total_ms": round(p.total_ms),
184 // NaN is not representable in JSON, so a failed persist
185 // reads as null rather than as a plausible number.
186 "persist_ms": if p.persist_ms.is_finite() {
187 serde_json::json!(round(p.persist_ms))
188 } else {
189 serde_json::Value::Null
190 },
191 })
192 })
193 .collect(),
194 )
195 }
196
197 /// Print the per-stage curve, and say whether anything actually drifted.
198 fn report_stage_series(probes: &[StageProbe]) {
199 if probes.is_empty() {
200 return;
201 }
202 println!();
203 println!("━━━ PER-STAGE TIMING AT VAULT SCALE ━━━");
204 println!();
205 println!(" Medians in ms per file, probed mid-import at each vault size.");
206 println!(" DSP stages should be flat: they cannot see the vault. persist");
207 println!(" writes into the growing DB and is the one that can drift.");
208 println!();
209 println!(
210 " {:>8} {:>7} {:>7} {:>7} {:>7} {:>7} {:>7} {:>8}",
211 "samples", "decode", "spectral", "mfcc", "bpm/key", "loop", "total", "persist"
212 );
213 println!(" {}", "-".repeat(70));
214 // Anything under the display resolution prints as `<0.01`, never as `0.00`.
215 // The corpus is what makes this matter: `detect_bpm_key` and `is_loop` bail
216 // out early on a file too short to hold a beat, so a one-shot corpus drives
217 // both to microseconds. Those are real early returns, not free work, and
218 // `0.00` in a column somebody is scanning for a bottleneck reads as "this
219 // stage costs nothing" rather than "this stage declined to run here".
220 let cell = |v: f64| {
221 if v.abs() < 0.005 {
222 "<0.01".to_string()
223 } else {
224 format!("{v:.2}")
225 }
226 };
227 for p in probes {
228 let persist = if p.persist_ms.is_finite() {
229 format!("{:.3}", p.persist_ms)
230 } else {
231 "FAILED".to_string()
232 };
233 println!(
234 " {:>8} {:>7} {:>7} {:>7} {:>7} {:>7} {:>7} {:>8}",
235 p.samples,
236 cell(p.decode_ms),
237 cell(p.spectral_ms),
238 cell(p.mfcc_ms),
239 cell(p.bpm_key_ms),
240 cell(p.loop_ms),
241 cell(p.total_ms),
242 persist
243 );
244 }
245
246 println!();
247 if probes.len() < 3 {
248 // Deliberately no trend line here. The first probe carries the warmup
249 // this whole section has to exclude, so with fewer than three
250 // checkpoints there is nothing left to compare after dropping it.
251 println!(" fewer than 3 checkpoints, so no trend is reported: the first");
252 println!(" probe carries cold-cache warmup and there is nothing left to");
253 println!(" compare against once it is dropped. Raise the file count or");
254 println!(" lower AF_BENCH_BATCH.");
255 return;
256 }
257
258 // THE FIRST PROBE IS DISCARDED, and this is the correction that makes the
259 // section honest. An early version compared first to last and reported
260 // "DSP total -84.7%" on a run where nothing had degraded at all: at 15 files
261 // per probe, the first checkpoint is dominated by cold page cache and first
262 // decode, so it is 3-5x the steady-state cost. The import curve above can
263 // compare first to last because its batches are 500 files and amortise that
264 // away; a probe this small cannot. Reporting a warmup artifact as a scaling
265 // trend is precisely the failure mode the measurement-traps list exists for.
266 let steady = &probes[1..];
267 let spread = |mut v: Vec<f64>| {
268 let hi = crate::percentile(&mut v.clone(), 100.0);
269 let lo = crate::percentile(&mut v, 0.0);
270 (lo, hi)
271 };
272 let (dsp_lo, dsp_hi) = spread(steady.iter().map(|p| p.total_ms).collect());
273 println!(
274 " Steady state ({} checkpoints, first dropped as warmup):",
275 steady.len()
276 );
277 println!(
278 " DSP total {dsp_lo:.2} to {dsp_hi:.2} ms{}",
279 if dsp_lo > 1e-9 && dsp_hi / dsp_lo > 2.0 {
280 " <- spread over 2x, treat as noise not trend"
281 } else {
282 ""
283 }
284 );
285 let persists: Vec<f64> = steady
286 .iter()
287 .map(|p| p.persist_ms)
288 .filter(|v| v.is_finite())
289 .collect();
290 if persists.is_empty() {
291 println!(" persist not measured");
292 } else {
293 let (p_lo, p_hi) = spread(persists);
294 println!(" persist {p_lo:.3} to {p_hi:.3} ms");
295 if p_lo > 1e-9 && p_hi / p_lo > 2.0 {
296 println!(" ^ persistence moved more than 2x across the fill: the stage");
297 println!(" that writes into the growing DB is the one to look at.");
298 }
299 }
300 println!();
301 println!(" A range, not a delta: two endpoints cannot tell a trend from");
302 println!(" noise. The machine has to be idle for any of this to mean");
303 println!(" anything (wiki af-benchmarks, \"Measurement traps\").");
304 }
305
306 /// Deterministic reorder of the file list.
307 ///
308 /// Without this the list arrives sorted by path, which groups files by class,
309 /// and classes have very different mean file sizes. Batch N would then differ
310 /// from batch 1 in content as well as in vault size, so a files/s trend across
311 /// batches would measure file-size composition rather than scaling behaviour.
312 /// FNV-1a over the path keeps it deterministic without pulling in `rand`.
313 fn shuffle_deterministic(files: &mut [PathBuf]) {
314 fn fnv1a(s: &str) -> u64 {
315 let mut h: u64 = 0xcbf2_9ce4_8422_2325;
316 for b in s.as_bytes() {
317 h ^= u64::from(*b);
318 h = h.wrapping_mul(0x100_0000_01b3);
319 }
320 h
321 }
322 files.sort_by_key(|p| fnv1a(&p.to_string_lossy()));
323 }
324
325 /// One batch of imports.
326 struct BatchStat {
327 /// Cumulative sample count in the vault after this batch.
328 cumulative: usize,
329 files: usize,
330 bytes: u64,
331 elapsed_s: f64,
332 }
333
334 impl BatchStat {
335 fn files_per_sec(&self) -> f64 {
336 if self.elapsed_s <= 0.0 {
337 return 0.0;
338 }
339 self.files as f64 / self.elapsed_s
340 }
341
342 fn mb_per_sec(&self) -> f64 {
343 if self.elapsed_s <= 0.0 {
344 return 0.0;
345 }
346 (self.bytes as f64 / 1e6) / self.elapsed_s
347 }
348 }
349
350 /// The per-batch curve as JSON, one object per batch in order.
351 ///
352 /// The shape of the curve is what the import benchmark is actually asking
353 /// about (does throughput degrade as the vault grows), and it used to live only
354 /// in the stdout table. Serialising it means a killed run still leaves the
355 /// measurement behind.
356 fn batch_series(stats: &[BatchStat]) -> serde_json::Value {
357 serde_json::Value::Array(
358 stats
359 .iter()
360 .enumerate()
361 .map(|(i, s)| {
362 serde_json::json!({
363 "batch": i + 1,
364 "cumulative": s.cumulative,
365 "files": s.files,
366 "bytes": s.bytes,
367 "elapsed_s": (s.elapsed_s * 100.0).round() / 100.0,
368 "files_per_sec": (s.files_per_sec() * 10.0).round() / 10.0,
369 "mb_per_sec": (s.mb_per_sec() * 10.0).round() / 10.0,
370 })
371 })
372 .collect(),
373 )
374 }
375
376 /// Collect audio files under `dir`, recursively.
377 fn collect(dir: &Path, out: &mut Vec<PathBuf>) {
378 let Ok(entries) = std::fs::read_dir(dir) else {
379 return;
380 };
381 for entry in entries.flatten() {
382 let path = entry.path();
383 if path.is_dir() {
384 collect(&path, out);
385 } else if path.extension().and_then(|e| e.to_str()).is_some_and(|e| {
386 matches!(
387 e.to_lowercase().as_str(),
388 "wav" | "aif" | "aiff" | "flac" | "mp3" | "ogg"
389 )
390 }) {
391 out.push(path);
392 }
393 }
394 }
395
396 fn count_samples(db: &Database) -> i64 {
397 db.conn()
398 .query_row("SELECT count(*) FROM samples", [], |r| r.get(0))
399 .unwrap_or(-1)
400 }
401
402 /// Count blobs anywhere under the store root, descending into shard directories.
403 ///
404 /// Must recurse. Blobs live at `{root}/{ab}/{hash}.{ext}`, so counting the root's
405 /// own entries returns the number of shard directories (at most 256) rather than
406 /// the number of blobs, which would make the dedup check below report "unchanged"
407 /// no matter what the store did.
408 fn count_blobs(root: &Path) -> usize {
409 let Ok(entries) = std::fs::read_dir(root) else {
410 return 0;
411 };
412 entries
413 .filter_map(std::result::Result::ok)
414 .map(|e| {
415 let path = e.path();
416 if path.is_dir() { count_blobs(&path) } else { 1 }
417 })
418 .sum()
419 }
420
421 /// Time a query, returning milliseconds. Runs it `reps` times and takes the
422 /// median, since a single cold query mostly measures page-cache state.
423 fn time_query(reps: usize, mut f: impl FnMut()) -> f64 {
424 let mut times: Vec<f64> = Vec::with_capacity(reps);
425 for _ in 0..reps {
426 let t = Instant::now();
427 f();
428 times.push(t.elapsed().as_secs_f64() * 1000.0);
429 }
430 times.sort_by(f64::total_cmp);
431 times[times.len() / 2]
432 }
433
434 /// Measure the query paths the browser list and filter panel depend on.
435 fn report_query_latency(db: &Database, report: &mut Report) {
436 let n = count_samples(db);
437 let nodes: i64 = db
438 .conn()
439 .query_row("SELECT count(*) FROM vfs_nodes", [], |r| r.get(0))
440 .unwrap_or(-1);
441 // Printed together on purpose: a large sample count with zero nodes means
442 // the timings below are measuring empty result sets.
443 println!(" {n} samples / {nodes} vfs nodes (median of 5):");
444 let rows = search::search_global(
445 db,
446 &SearchFilter {
447 scope: SearchScope::Global,
448 ..Default::default()
449 },
450 )
451 .map_or(0, |r| r.len());
452 // search.rs caps every result set at SEARCH_RESULT_LIMIT (500), so list
453 // latency is bounded by design no matter how large the vault gets. The
454 // scan and sort underneath it are not bounded, which is what these numbers
455 // actually track.
456 println!(" unfiltered search returns {rows} rows (capped at 500 by SEARCH_RESULT_LIMIT)");
457
458 let analyzed: i64 = db
459 .conn()
460 .query_row("SELECT count(*) FROM audio_analysis", [], |r| r.get(0))
461 .unwrap_or(0);
462 if analyzed == 0 {
463 println!(" NOTE: audio_analysis is empty, so the class and bpm filters below");
464 println!(" match nothing and their timings are not meaningful. Run");
465 println!(" the analysis pipeline over this vault to benchmark them.");
466 }
467
468 let ms = time_query(5, || {
469 let _ = count_samples(db);
470 });
471 println!(" count(*) {ms:>8.2} ms");
472 report.set("count_star_ms", (ms * 100.0).round() / 100.0);
473
474 let mut filter = SearchFilter {
475 scope: SearchScope::Global,
476 ..Default::default()
477 };
478 let ms = time_query(5, || {
479 let _ = search::search_global(db, &filter);
480 });
481 println!(" search_global (no filter) {ms:>6.2} ms <- worst-case list load");
482 report.set("search_unfiltered_ms", (ms * 100.0).round() / 100.0);
483
484 filter.text_query = "kick".to_string();
485 let ms = time_query(5, || {
486 let _ = search::search_global(db, &filter);
487 });
488 println!(" search_global (text) {ms:>6.2} ms <- search box keystroke");
489 report.set("search_text_ms", (ms * 100.0).round() / 100.0);
490
491 filter.text_query.clear();
492 filter.bpm_min = Some(120.0);
493 filter.bpm_max = Some(130.0);
494 let ms = time_query(5, || {
495 let _ = search::search_global(db, &filter);
496 });
497 println!(" search_global (bpm range) {ms:>6.2} ms");
498 report.set("search_bpm_ms", (ms * 100.0).round() / 100.0);
499 }
500
501 /// Run the ingest benchmark against `corpus`, building a scratch vault at
502 /// `vault`. Any existing scratch vault is removed first so runs are comparable.
503 pub(crate) fn run(corpus: &Path, vault: &Path, batch: usize, limit: Option<usize>) {
504 println!("━━━ INGEST AT SCALE ━━━");
505 println!();
506 println!(" corpus: {}", corpus.display());
507 println!(" vault: {}", vault.display());
508
509 let mut report = Report::new("ingest");
510
511 let mut files = Vec::new();
512 collect(corpus, &mut files);
513 files.sort();
514 shuffle_deterministic(&mut files);
515 if let Some(lim) = limit {
516 files.truncate(lim);
517 }
518 if files.is_empty() {
519 eprintln!("no audio files under {}", corpus.display());
520 return;
521 }
522 println!(" files: {}", files.len());
523 println!();
524
525 if vault.exists()
526 && let Err(e) = std::fs::remove_dir_all(vault)
527 {
528 eprintln!("could not clear scratch vault: {e}");
529 return;
530 }
531 let samples_dir = vault.join("samples");
532 if let Err(e) = std::fs::create_dir_all(&samples_dir) {
533 eprintln!("could not create scratch vault: {e}");
534 return;
535 }
536
537 // Described after the vault exists so its path canonicalizes, and reported
538 // as two roles because corpus and vault are routinely on different drives:
539 // reading from the external disk while writing to the internal one is a
540 // third measurement, distinct from either drive on its own.
541 let corpus_storage = storage::describe(corpus);
542 let vault_storage = storage::describe(vault);
543 report.set_storage("corpus", &corpus_storage);
544 report.set_storage("vault", &vault_storage);
545 let corpus_bytes: u64 = files
546 .iter()
547 .filter_map(|p| std::fs::metadata(p).ok())
548 .map(|m| m.len())
549 .sum();
550 storage::print_conditions(
551 &[("corpus", &corpus_storage), ("vault", &vault_storage)],
552 Some(corpus_bytes),
553 );
554
555 let db = match Database::open(vault.join("audiofiles.db")) {
556 Ok(db) => db,
557 Err(e) => {
558 // Worth surfacing loudly: on a filesystem that cannot support WAL
559 // this is exactly where a vault fails, and the app surfaces it as a
560 // generic init error.
561 eprintln!("Database::open failed (WAL unsupported on this fs?): {e}");
562 return;
563 }
564 };
565 let store = match SampleStore::new(&samples_dir) {
566 Ok(s) => s,
567 Err(e) => {
568 eprintln!("SampleStore::new failed: {e}");
569 return;
570 }
571 };
572
573 // `store.import` writes the blob and the `samples` row but no VFS node.
574 // The browser's import workflow creates those separately, and every query
575 // the UI runs goes through `vfs_nodes`. Without them `search_global`
576 // returns an empty set instantly and the query numbers below would be
577 // measuring nothing.
578 let vfs_id = match vfs::create_vfs(&db, "bench") {
579 Ok(id) => id,
580 Err(e) => {
581 eprintln!("could not create bench vfs: {e}");
582 return;
583 }
584 };
585
586 println!(" batch cumulative files/s MB/s elapsed");
587 println!(" ---------------------------------------------------------");
588
589 let mut stats: Vec<BatchStat> = Vec::new();
590 let mut cumulative = 0usize;
591 let mut failures = 0usize;
592 let mut link_failures = 0usize;
593 // (hash, source path) for the analysis pass. The source file and the stored
594 // blob are byte-identical by construction, so analysing either is the same
595 // measurement and this avoids an extension lookup per sample.
596 let mut imported: Vec<(String, PathBuf)> = Vec::with_capacity(files.len());
597
598 let probe_size = stage_probe_size();
599 let mut probes: Vec<StageProbe> = Vec::new();
600 let probe_config = AnalysisConfig::default();
601 if probe_size > 0 {
602 println!(" (per-stage probe on, {probe_size} file(s) per batch: import numbers below");
603 println!(" are NOT comparable with baselines recorded without it)");
604 }
605
606 for chunk in files.chunks(batch) {
607 let mut bytes = 0u64;
608 let start = Instant::now();
609 for path in chunk {
610 match store.import(path, &db) {
611 Ok(hash) => {
612 // Name links by index: sample names must be unique among
613 // siblings, and the corpus has repeated basenames across
614 // packs.
615 let name = format!(
616 "{cumulative:06}_{}",
617 path.file_name().unwrap_or_default().to_string_lossy()
618 );
619 if vfs::create_sample_link(
620 &db,
621 vfs_id,
622 None,
623 &name,
624 &SampleHash::from_trusted(hash.clone()),
625 )
626 .is_err()
627 {
628 link_failures += 1;
629 }
630 imported.push((hash, path.clone()));
631 bytes += std::fs::metadata(path).map_or(0, |m| m.len());
632 cumulative += 1;
633 }
634 Err(_) => failures += 1,
635 }
636 }
637 let stat = BatchStat {
638 cumulative,
639 files: chunk.len(),
640 bytes,
641 elapsed_s: start.elapsed().as_secs_f64(),
642 };
643 println!(
644 " {:>5} {:>10} {:>10.1} {:>10.1} {:>7.2}s",
645 stats.len() + 1,
646 stat.cumulative,
647 stat.files_per_sec(),
648 stat.mb_per_sec(),
649 stat.elapsed_s,
650 );
651 stats.push(stat);
652
653 // Checkpoint the curve after every batch. A long import is the run most
654 // worth recording and the one most likely to be killed part way, and
655 // the summary metrics below only exist once the loop finishes.
656 report.set("import_batches", batch_series(&stats));
657 report.set("import_files", stats.iter().map(|s| s.files).sum::<usize>());
658 report.set("import_bytes", stats.iter().map(|s| s.bytes).sum::<u64>());
659 report.set("import_complete", false);
660
661 // Probe AFTER the batch timer has stopped, so the probe's own decode and
662 // DB work never lands inside a files/s figure. It still perturbs the
663 // drive and the page cache for the batches that follow, which is why the
664 // whole thing is opt-in.
665 if probe_size > 0 {
666 let sample_rows = count_samples(&db);
667 let recent: Vec<(String, PathBuf)> =
668 imported.iter().rev().take(probe_size).cloned().collect();
669 if let Some(p) = stage_probe(&db, sample_rows, &recent, &probe_config) {
670 probes.push(p);
671 report.set("stage_series", stage_series(&probes));
672 }
673 }
674
675 report.checkpoint();
676 }
677
678 println!();
679 if failures > 0 {
680 println!(" {failures} file(s) failed to import");
681 }
682 if link_failures > 0 {
683 println!(" {link_failures} vfs link(s) failed -- query numbers below undercount");
684 }
685
686 // Degradation is the actual question. Comparing first batch to last is the
687 // cheapest signal that the flat blob directory or an index has started to
688 // bite; a flat profile means it has not.
689 if stats.len() >= 2 {
690 let first = stats[0].files_per_sec();
691 let last = stats[stats.len() - 1].files_per_sec();
692 let delta = if first > 0.0 {
693 (last - first) / first * 100.0
694 } else {
695 0.0
696 };
697 println!(" first batch: {first:.1} files/s");
698 println!(" last batch: {last:.1} files/s ({delta:+.1}%)");
699 if delta < -25.0 {
700 println!(" ^ throughput degraded as the vault grew");
701 }
702 }
703
704 let total_files: usize = stats.iter().map(|s| s.files).sum();
705 let total_bytes: u64 = stats.iter().map(|s| s.bytes).sum();
706 let total_s: f64 = stats.iter().map(|s| s.elapsed_s).sum();
707 println!();
708 println!(
709 " total: {total_files} files, {:.2} GB in {total_s:.1}s ({:.1} files/s, {:.1} MB/s)",
710 total_bytes as f64 / 1e9,
711 total_files as f64 / total_s,
712 (total_bytes as f64 / 1e6) / total_s,
713 );
714 // `import_complete` separates a finished import from a checkpoint of one
715 // that was killed, so a comparison tool does not read a partial curve as a
716 // regression.
717 report.set("import_complete", true);
718 report.set("import_files", total_files);
719 report.set("import_bytes", total_bytes);
720 report.set(
721 "import_files_per_sec",
722 ((total_files as f64 / total_s) * 10.0).round() / 10.0,
723 );
724 report.set(
725 "import_mb_per_sec",
726 (((total_bytes as f64 / 1e6) / total_s) * 10.0).round() / 10.0,
727 );
728 if let (Some(first), Some(last)) = (stats.first(), stats.last()) {
729 // The scaling signal, not a throughput figure: negative means the vault
730 // got slower as it filled.
731 let delta = (last.files_per_sec() - first.files_per_sec()) / first.files_per_sec() * 100.0;
732 report.set("import_throughput_drift_pct", (delta * 10.0).round() / 10.0);
733 }
734
735 report_stage_series(&probes);
736
737 // Dedup: re-importing the same files must hit the content-addressed store
738 // and skip the copy. If this is not dramatically faster, dedup is not
739 // working and every duplicate costs a full hash-and-copy.
740 println!();
741 println!("━━━ DEDUP (re-import of identical content) ━━━");
742 println!();
743 let blobs_before = count_blobs(&samples_dir);
744 let rows_before = count_samples(&db);
745
746 let redo: Vec<&PathBuf> = files.iter().take(batch.min(files.len())).collect();
747 let start = Instant::now();
748 for path in &redo {
749 let _ = store.import(path, &db);
750 }
751 let redo_s = start.elapsed().as_secs_f64();
752
753 let blobs_after = count_blobs(&samples_dir);
754 let rows_after = count_samples(&db);
755
756 println!(
757 " re-imported {} files in {redo_s:.2}s ({:.1} files/s)",
758 redo.len(),
759 redo.len() as f64 / redo_s.max(1e-9)
760 );
761 println!(" blobs on disk: {blobs_before} -> {blobs_after} (want: unchanged)");
762 println!(" sample rows: {rows_before} -> {rows_after}");
763
764 // Distinguishing these two matters: equal blob counts with equal row counts
765 // means full dedup; equal blobs with more rows means the blob was reused but
766 // a duplicate row was still written.
767 if blobs_after == blobs_before {
768 println!(" blob dedup: OK (no new blobs written)");
769 } else {
770 println!(
771 " blob dedup: {} new blob(s) written",
772 blobs_after - blobs_before
773 );
774 }
775
776 // Analysis pass. Two purposes: it is a throughput number in its own right at
777 // vault scale, and without it `audio_analysis` stays empty and the class and
778 // bpm filter timings below measure nothing.
779 println!();
780 println!("━━━ ANALYSIS AT SCALE ━━━");
781 println!();
782 let budget = analyze_budget();
783 let to_analyze: Vec<(String, PathBuf)> = imported.into_iter().take(budget).collect();
784 if to_analyze.is_empty() {
785 println!(" skipped (AF_BENCH_ANALYZE=0)");
786 } else {
787 println!(
788 " analysing {} samples (AF_BENCH_ANALYZE={budget})",
789 to_analyze.len()
790 );
791 let config = AnalysisConfig::default();
792 let start = Instant::now();
793 let results: Vec<_> = to_analyze
794 .par_iter()
795 .filter_map(|(hash, path)| analysis::analyze_sample(hash, path, &config).ok())
796 .collect();
797 let analyze_s = start.elapsed().as_secs_f64();
798
799 let save_start = Instant::now();
800 let saved = analysis::save_analysis_batch(&db, &results).is_ok();
801 let save_s = save_start.elapsed().as_secs_f64();
802
803 let rate = results.len() as f64 / analyze_s.max(1e-9);
804 println!(
805 " analysed {} in {analyze_s:.1}s ({rate:.1} files/s)",
806 results.len()
807 );
808 println!(
809 " persisted {} rows in {save_s:.2}s{}",
810 results.len(),
811 if saved { "" } else { " (SAVE FAILED)" }
812 );
813 if results.len() < to_analyze.len() {
814 println!(
815 " {} file(s) failed analysis",
816 to_analyze.len() - results.len()
817 );
818 }
819 report.set("analysis_files", results.len());
820 report.set("analysis_files_per_sec", (rate * 10.0).round() / 10.0);
821 report.set("analysis_persist_s", (save_s * 100.0).round() / 100.0);
822 }
823
824 println!();
825 println!("━━━ QUERY LATENCY (backs the browser UI) ━━━");
826 println!();
827 report_query_latency(&db, &mut report);
828
829 // Left in place deliberately: the vault is the artifact to point the app at
830 // for eyeballing UI responsiveness at this size.
831 println!();
832 if let Some(rss) = peak_rss_mb() {
833 // High-water across the whole run, so it covers the import and analysis
834 // peaks rather than whatever happens to be resident at the end.
835 println!(" peak RSS: {rss:.1} MB");
836 }
837 println!(
838 " scratch vault left at {} for UI inspection",
839 vault.display()
840 );
841 report.write();
842 }
843
844 #[cfg(test)]
845 mod tests {
846 use super::*;
847
848 fn probe(samples: i64, total_ms: f64, persist_ms: f64) -> StageProbe {
849 StageProbe {
850 samples,
851 decode_ms: 1.0,
852 loudness_ms: 1.0,
853 spectral_ms: 1.0,
854 mfcc_ms: 1.0,
855 vector_ms: 1.0,
856 bpm_key_ms: 1.0,
857 loop_ms: 1.0,
858 fingerprint_ms: 1.0,
859 total_ms,
860 persist_ms,
861 files: 25,
862 }
863 }
864
865 #[test]
866 fn stage_probe_is_off_unless_asked_for() {
867 // Guards the default. Turning the probe on silently would make every
868 // new ingest run incomparable with the saved baselines.
869 // SAFETY: single-threaded test, no other thread reads the environment.
870 unsafe { std::env::remove_var("AF_BENCH_STAGES") };
871 assert_eq!(stage_probe_size(), 0);
872 }
873
874 #[test]
875 fn stage_series_carries_one_object_per_checkpoint() {
876 let series = stage_series(&[probe(500, 12.0, 0.4), probe(1000, 12.5, 0.6)]);
877 let rows = series.as_array().unwrap();
878 assert_eq!(rows.len(), 2);
879 assert_eq!(rows[0]["samples"], 500);
880 assert_eq!(rows[1]["samples"], 1000);
881 assert_eq!(rows[0]["files_probed"], 25);
882 assert!((rows[1]["persist_ms"].as_f64().unwrap() - 0.6).abs() < 1e-9);
883 }
884
885 #[test]
886 fn a_failed_persist_serialises_as_null_not_as_a_number() {
887 // NaN has no JSON spelling. Emitting it as 0.0 would read downstream as
888 // "persistence was free", which is the opposite of what happened.
889 let series = stage_series(&[probe(500, 12.0, f64::NAN)]);
890 assert!(series[0]["persist_ms"].is_null());
891 }
892
893 #[test]
894 fn a_trend_needs_more_than_two_checkpoints() {
895 // Regression guard on a real mistake: an early version compared the
896 // first probe to the last and reported "DSP total -84.7%" on a run
897 // where nothing degraded. The first probe is warmup, so it is dropped,
898 // and with fewer than three checkpoints nothing survives to compare.
899 // These call the printer for absence of panic; the contract they pin is
900 // that `steady` is `probes[1..]` and is only read when len >= 3.
901 report_stage_series(&[]);
902 report_stage_series(&[probe(100, 40.0, 0.05)]);
903 report_stage_series(&[probe(100, 40.0, 0.05), probe(200, 12.0, 0.05)]);
904 report_stage_series(&[
905 probe(100, 40.0, 0.05),
906 probe(200, 12.0, 0.05),
907 probe(300, 12.5, 0.06),
908 ]);
909 }
910
911 #[test]
912 fn stage_probe_returns_nothing_for_an_empty_file_set() {
913 let db = Database::open_in_memory().unwrap();
914 let config = AnalysisConfig::default();
915 assert!(stage_probe(&db, 0, &[], &config).is_none());
916 }
917 }
918