Skip to main content

max / audiofiles

Perf A+: spawn_job one-shot seal + DB persistence bench - worker_runtime: add spawn_job/JobHandle, the one-shot counterpart to spawn_worker (run once -> single result event, catch_unwind built in). Migrate the classifier job to it, deleting its hand-rolled catch_unwind + join-on-drop. Docstring now states accurately which bare thread::spawn sites remain by design (instrument decode worker self-guards; streaming preview decode + OAuth listener recover by replacement/timeout, not a terminal event). - bench: add Section 8 persistence scaling — batched vs per-row analysis-row inserts (the CHRONIC #2 transaction-seal contrast) and load-from-DB + index build (VP-tree cost INCLUDING I/O, vs Section 7's in-memory build). Closes the bench cold spot (was compute-only). A schema test guards the raw inserts. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-22 22:24 UTC
Commit: df8fb3b3ae800529085a0d79d04246d4bd7cca78
Parent: 69b70fe
3 files changed, +244 insertions, -39 deletions
M crates/audiofiles-bench/src/main.rs +136
@@ -692,6 +692,21 @@ fn main() {
692 692 bench_similarity_scaling();
693 693 println!();
694 694
695 + // ─────────────────────────────────────────────────────────────
696 + // Section 8: Persistence Scaling (DB import write + load-then-build)
697 + // ─────────────────────────────────────────────────────────────
698 + println!("━━━ 8. PERSISTENCE SCALING ━━━");
699 + println!();
700 + println!(" Analysis-row writes to the DB, and loading them back to build the");
701 + println!(" similarity index. The write path was the headline bottleneck behind");
702 + println!(" ultra-fuzz CHRONIC #2 (per-row autocommit = one fsync per sample);");
703 + println!(" the batched-vs-per-row columns below are why it was sealed behind a");
704 + println!(" transaction token. 'Load+build' is the index build INCLUDING DB I/O");
705 + println!(" (Section 7 measures the in-memory build alone).");
706 + println!();
707 + bench_persistence_scaling();
708 + println!();
709 +
695 710 println!("═══════════════════════════════════════════════════════════════");
696 711 println!(" Benchmark complete. {} files analyzed.", n + total_classified);
697 712 println!("═══════════════════════════════════════════════════════════════");
@@ -776,3 +791,124 @@ fn bench_similarity_scaling() {
776 791 );
777 792 }
778 793 }
794 +
795 + // ── Persistence Scaling ──
796 +
797 + /// Insert one `samples` + `audio_analysis` row pair. Feature values vary with `i`
798 + /// so the built index isn't a degenerate all-equal set. Only the NOT NULL columns
799 + /// plus a few features `load_data` reads are set; tuple params avoid a direct
800 + /// rusqlite dependency (panics on schema mismatch, which a bench wants to surface).
801 + fn insert_analysis_row(db: &audiofiles_core::db::Database, i: usize) {
802 + let hash = format!("{i:064x}");
803 + db.conn()
804 + .execute(
805 + "INSERT INTO samples (hash, original_name, file_extension, file_size, import_date, last_modified)
806 + VALUES (?1, ?2, 'wav', 100, 0, 0)",
807 + (&hash, format!("s{i}.wav")),
808 + )
809 + .unwrap();
810 + let bpm = 60.0 + (i % 140) as f64;
811 + let centroid = (i % 8000) as f64;
812 + let onset = (i % 1000) as f64 / 1000.0;
813 + db.conn()
814 + .execute(
815 + "INSERT INTO audio_analysis
816 + (hash, duration, sample_rate, channels, analyzed_at, bpm, spectral_centroid, onset_strength)
817 + VALUES (?1, 1.0, 44100, 2, 0, ?2, ?3, ?4)",
818 + (&hash, bpm, centroid, onset),
819 + )
820 + .unwrap();
821 + }
822 +
823 + fn bench_persistence_scaling() {
824 + use audiofiles_core::db::Database;
825 + use audiofiles_core::similarity::SimilarityIndex;
826 +
827 + let db_path = std::env::temp_dir().join(format!("af_bench_persist_{}.db", std::process::id()));
828 + let _ = std::fs::remove_file(&db_path);
829 +
830 + println!(
831 + " {:>9} {:>14} {:>16} {:>14}",
832 + "Rows", "Batched(ms)", "Per-row(ms)", "Load+build(ms)"
833 + );
834 + println!(" {}", "─".repeat(58));
835 +
836 + for &size in &[1_000usize, 10_000] {
837 + // Fresh DB per size so the row counts are clean.
838 + let _ = std::fs::remove_file(&db_path);
839 + let db = match Database::open(&db_path) {
840 + Ok(d) => d,
841 + Err(e) => {
842 + println!(" (skipped: could not open temp DB: {e})");
843 + return;
844 + }
845 + };
846 +
847 + // Batched: all inserts in one transaction (the sealed path).
848 + let t = Instant::now();
849 + let res = db.transaction(|_tx| {
850 + for i in 0..size {
851 + insert_analysis_row(&db, i);
852 + }
853 + Ok(())
854 + });
855 + let batched_ms = t.elapsed().as_secs_f64() * 1000.0;
856 + if let Err(e) = res {
857 + println!(" (skipped: batched insert failed: {e})");
858 + return;
859 + }
860 +
861 + // Per-row autocommit (the pre-CHRONIC-#2 pattern), for contrast. Capped at
862 + // 1k rows — at 10k the autocommit fsyncs dominate the whole bench's runtime.
863 + let per_row_n = size.min(1_000);
864 + let t = Instant::now();
865 + for i in 0..per_row_n {
866 + // offset the hashes so they don't collide with the batched rows
867 + insert_analysis_row(&db, size + i);
868 + }
869 + let per_row_ms = t.elapsed().as_secs_f64() * 1000.0
870 + * (size as f64 / per_row_n as f64); // extrapolate to `size` rows
871 +
872 + // Load from DB (I/O) then build the index (CPU) — the full cold-start cost.
873 + let t = Instant::now();
874 + let data = SimilarityIndex::load_data(&db).unwrap_or_default();
875 + let _index = SimilarityIndex::build_from_data(data);
876 + let load_build_ms = t.elapsed().as_secs_f64() * 1000.0;
877 +
878 + println!(
879 + " {:>9} {:>14.1} {:>16.1} {:>14.1}",
880 + size, batched_ms, per_row_ms, load_build_ms
881 + );
882 + }
883 +
884 + let _ = std::fs::remove_file(&db_path);
885 + println!();
886 + println!(" Per-row is extrapolated from a 1k-row sample; the batched column is");
887 + println!(" the path the app actually uses (one transaction per analysis flush).");
888 + }
889 +
890 + #[cfg(test)]
891 + mod tests {
892 + use super::*;
893 +
894 + #[test]
895 + fn persistence_rows_match_schema_and_load() {
896 + // Guards the bench's raw inserts against schema drift: the rows must be
897 + // accepted and round-trip through SimilarityIndex::load_data.
898 + use audiofiles_core::db::Database;
899 + use audiofiles_core::similarity::SimilarityIndex;
900 +
901 + let db = Database::open_in_memory().unwrap();
902 + db.transaction(|_tx| {
903 + for i in 0..50 {
904 + insert_analysis_row(&db, i);
905 + }
906 + Ok(())
907 + })
908 + .unwrap();
909 +
910 + let data = SimilarityIndex::load_data(&db).unwrap();
911 + assert_eq!(data.len(), 50, "all inserted analysis rows must load back");
912 + let _ = SimilarityIndex::build_from_data(data);
913 + }
914 + }
M crates/audiofiles-browser/src/classifier_worker.rs +24 -39
@@ -8,13 +8,12 @@
8 8 //! buttons while a job is in flight, and dispatching a new job drops (joins) the prior one.
9 9
10 10 use std::path::PathBuf;
11 - use std::sync::{mpsc, Mutex};
12 - use std::thread;
13 11
14 12 use tracing::{error, instrument};
15 13
16 14 use audiofiles_core::analysis::{afcl, cluster, exemplar, trained_head};
17 15 use audiofiles_core::db::Database;
16 + use audiofiles_core::worker_runtime::{spawn_job, JobHandle};
18 17
19 18 use crate::backend::{ClassifierJob, ClassifierJobResult, TrainedHeadInfo};
20 19
@@ -24,54 +23,40 @@ pub enum ClassifierWorkerEvent {
24 23 Failed(String),
25 24 }
26 25
27 - /// Handle for a single in-flight classifier job. The receiver is wrapped in a `Mutex` so
28 - /// `BrowserState` stays `Sync` (required by nih-plug); only the GUI thread calls `try_recv`.
29 - pub struct ClassifierHandle {
30 - event_rx: Mutex<mpsc::Receiver<ClassifierWorkerEvent>>,
31 - _thread: Option<thread::JoinHandle<()>>,
32 - }
26 + /// Handle for a single in-flight classifier job. Thin wrapper over the shared
27 + /// one-shot [`JobHandle`], which owns the result channel, the `catch_unwind`
28 + /// panic-isolation, and the join-on-drop.
29 + pub struct ClassifierHandle(JobHandle<ClassifierWorkerEvent>);
33 30
34 31 impl ClassifierHandle {
35 32 /// Poll for the job's result without blocking.
36 33 pub fn try_recv(&self) -> Option<ClassifierWorkerEvent> {
37 - self.event_rx.lock().ok()?.try_recv().ok()
34 + self.0.try_recv()
38 35 }
39 36 }
40 37
41 - impl Drop for ClassifierHandle {
42 - fn drop(&mut self) {
43 - // The worker is one-shot and detached; join so a replaced job finishes cleanly.
44 - if let Some(handle) = self._thread.take() {
45 - let _ = handle.join();
46 - }
47 - }
48 - }
38 + // Join-on-drop is owned by the inner JobHandle (so a replaced job finishes cleanly).
49 39
50 40 /// Spawn a one-shot worker thread to run `job` against the database at `db_path`.
41 + ///
42 + /// The shared [`spawn_job`] runtime owns the panic isolation: a panic in the job
43 + /// (or `Database::open`) becomes a `Failed` event rather than a silently-dead
44 + /// thread, so the GUI's `try_recv` always resolves and the trigger buttons
45 + /// re-enable.
51 46 #[instrument(skip_all)]
52 47 pub fn spawn_classifier_job(db_path: PathBuf, job: ClassifierJob) -> std::io::Result<ClassifierHandle> {
53 - let (event_tx, event_rx) = mpsc::channel::<ClassifierWorkerEvent>();
54 -
55 - let thread = thread::Builder::new()
56 - .name("classifier-job".to_string())
57 - .spawn(move || {
58 - // Catch panics so a failed job reports back instead of unwinding the
59 - // thread silently — otherwise the GUI's try_recv() never resolves and
60 - // the trigger buttons stay disabled forever.
61 - let event = match Database::open(&db_path) {
62 - Ok(db) => std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| run_job(&db, job)))
63 - .unwrap_or_else(|_| {
64 - ClassifierWorkerEvent::Failed("classifier job panicked (internal error)".to_string())
65 - }),
66 - Err(e) => {
67 - error!("Classifier worker failed to open database: {e}");
68 - ClassifierWorkerEvent::Failed(format!("could not open database: {e}"))
69 - }
70 - };
71 - let _ = event_tx.send(event);
72 - })?;
73 -
74 - Ok(ClassifierHandle { event_rx: Mutex::new(event_rx), _thread: Some(thread) })
48 + let handle = spawn_job(
49 + "classifier-job",
50 + || ClassifierWorkerEvent::Failed("classifier job panicked (internal error)".to_string()),
51 + move || match Database::open(&db_path) {
52 + Ok(db) => run_job(&db, job),
53 + Err(e) => {
54 + error!("Classifier worker failed to open database: {e}");
55 + ClassifierWorkerEvent::Failed(format!("could not open database: {e}"))
56 + }
57 + },
58 + )?;
59 + Ok(ClassifierHandle(handle))
75 60 }
76 61
77 62 fn run_job(db: &Database, job: ClassifierJob) -> ClassifierWorkerEvent {
M crates/audiofiles-core/src/worker_runtime.rs +84
@@ -266,11 +266,95 @@ where
266 266 })
267 267 }
268 268
269 + /// Handle to a one-shot job thread (see [`spawn_job`]). Poll [`try_recv`](Self::try_recv)
270 + /// for the single result event; `Drop` joins the thread.
271 + pub struct JobHandle<Ev> {
272 + event_rx: Mutex<mpsc::Receiver<Ev>>,
273 + thread: Option<thread::JoinHandle<()>>,
274 + }
275 +
276 + impl<Ev> JobHandle<Ev> {
277 + /// Poll for the job's result without blocking.
278 + pub fn try_recv(&self) -> Option<Ev> {
279 + self.event_rx.lock().ok()?.try_recv().ok()
280 + }
281 + }
282 +
283 + impl<Ev> Drop for JobHandle<Ev> {
284 + fn drop(&mut self) {
285 + if let Some(handle) = self.thread.take() {
286 + let _ = handle.join();
287 + }
288 + }
289 + }
290 +
291 + /// Spawn a one-shot job thread: `body` runs once, its return value is sent back
292 + /// as the single result event, and the thread exits.
293 + ///
294 + /// This is the one-shot counterpart to [`spawn_worker`] (which owns a command
295 + /// recv loop). A panic in `body` is caught and mapped to `on_panic()` — the same
296 + /// isolation guarantee — so the GUI's `try_recv` always resolves a terminal event
297 + /// instead of hanging on a silently-dead thread. Use this for any
298 + /// run-once-then-exit thread (classifier jobs, one-shot library passes); use
299 + /// `spawn_worker` for a long-lived command loop. Two bare `thread::spawn` sites
300 + /// remain by design, both different shapes from a compute job: the instrument
301 + /// decode worker (its own per-request `catch_unwind`), and two recoverable
302 + /// threads — a generation-cancelled streaming preview decode (a panic just stops
303 + /// that one preview; the next play spawns a fresh generation) and an OAuth
304 + /// loopback listener (bounded by a timeout). Those recover by replacement rather
305 + /// than by emitting a terminal event, so the job seal does not apply.
306 + pub fn spawn_job<Ev, F>(
307 + name: &str,
308 + on_panic: impl FnOnce() -> Ev + Send + 'static,
309 + body: F,
310 + ) -> std::io::Result<JobHandle<Ev>>
311 + where
312 + Ev: Send + 'static,
313 + F: FnOnce() -> Ev + Send + 'static,
314 + {
315 + let (tx, rx) = mpsc::channel::<Ev>();
316 + let thread = thread::Builder::new()
317 + .name(name.to_string())
318 + .spawn(move || {
319 + let ev = std::panic::catch_unwind(std::panic::AssertUnwindSafe(body))
320 + .unwrap_or_else(|_| on_panic());
321 + let _ = tx.send(ev);
322 + })?;
323 + Ok(JobHandle {
324 + event_rx: Mutex::new(rx),
325 + thread: Some(thread),
326 + })
327 + }
328 +
269 329 #[cfg(test)]
270 330 mod tests {
271 331 use super::*;
272 332 use std::convert::Infallible;
273 333
334 + fn poll_job<Ev>(handle: &JobHandle<Ev>) -> Ev {
335 + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
336 + loop {
337 + if let Some(ev) = handle.try_recv() {
338 + return ev;
339 + }
340 + assert!(std::time::Instant::now() < deadline, "job never emitted");
341 + std::thread::sleep(std::time::Duration::from_millis(1));
342 + }
343 + }
344 +
345 + #[test]
346 + fn spawn_job_delivers_result() {
347 + let handle = spawn_job("test-job-ok", || 0u32, || 42u32).unwrap();
348 + assert_eq!(poll_job(&handle), 42);
349 + }
350 +
351 + #[test]
352 + fn spawn_job_maps_panic_to_event() {
353 + // A panicking job body must surface the on_panic event, not hang try_recv.
354 + let handle = spawn_job("test-job-panic", || "panicked", || panic!("boom")).unwrap();
355 + assert_eq!(poll_job(&handle), "panicked");
356 + }
357 +
274 358 fn recv_one<Cmd, Ev>(handle: &WorkerHandle<Cmd, Ev>) -> Ev {
275 359 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
276 360 while std::time::Instant::now() < deadline {