Skip to main content

max / audiofiles

17.9 KB · 457 lines History Blame Raw
1 //! Generic background-worker runtime.
2 //!
3 //! Every recv-loop worker in the app (analysis, edit, forge, import, export,
4 //! search, loose-files) routes its command loop through [`spawn_worker`]. The
5 //! factory owns the command channel's `Receiver` and the `recv` loop, and wraps
6 //! each command's handler in `catch_unwind` so a panic becomes a caller-defined
7 //! event instead of silently killing the thread, which would stop every future
8 //! command of that kind with no UI signal.
9 //!
10 //! Every recv-loop worker routes through here, so the `catch_unwind` lives in one
11 //! place instead of being copy-pasted (and forgotten) per worker, the
12 //! historically-recurring "added a worker, forgot the guard" drift (the
13 //! `loose_files_worker` regression). This is enforced by convention, not by the
14 //! compiler: nothing *prevents* a new worker from hand-rolling its own
15 //! `mpsc::channel()` + `rx.recv()` loop. The rule is: any worker that loops over
16 //! commands MUST be spawned via [`spawn_worker`]; one-shot job threads (which run
17 //! a single task and exit, e.g. the classifier job and the preview decode) carry
18 //! their own `catch_unwind` and are a different shape. A new bare recv loop is a
19 //! review smell, not a build error.
20 //!
21 //! ## Cancellation
22 //!
23 //! Every worker gets a shared cancel flag for free (in [`WorkerCtx`]). It is set
24 //! to `true` automatically when the [`WorkerHandle`] is dropped, so an in-flight
25 //! long operation aborts promptly on shutdown instead of blocking the join.
26 //! Workers that support an explicit Cancel command wrap the handle in a thin
27 //! newtype whose `send` calls [`WorkerHandle::request_cancel`] before enqueuing,
28 //! so the flag is set synchronously by the GUI thread (the queued command itself
29 //! becomes a no-op). Long operations reset the flag at their start via
30 //! [`WorkerCtx::reset_cancel`] so a stale cancel does not abort the next command.
31
32 use std::sync::atomic::{AtomicBool, Ordering};
33 use std::sync::{Arc, Mutex, mpsc};
34 use std::thread;
35 use std::time::Duration;
36
37 /// Bound on the per-worker event channel. The channel is the one place an
38 /// unbounded queue could balloon, the analysis batch fans out over rayon and
39 /// emits a per-sample event, which on a 100k-sample import outruns a slow GUI
40 /// drain. A bounded channel applies backpressure (the producer waits) instead of
41 /// growing without limit; the cap is generous so a normally-draining GUI never
42 /// blocks a worker.
43 const EVENT_CHANNEL_CAP: usize = 4096;
44
45 /// Sink handed to a worker step so it can emit zero or more events per command.
46 ///
47 /// Cloneable and `Sync` (it wraps an `mpsc::SyncSender`), so a step may share it
48 /// across rayon worker threads (e.g. the analysis batch fan-out). [`emit`](Self::emit)
49 /// applies bounded backpressure but never blocks past cancellation.
50 pub struct EventSink<Ev> {
51 tx: mpsc::SyncSender<Ev>,
52 cancel: Arc<AtomicBool>,
53 }
54
55 impl<Ev> Clone for EventSink<Ev> {
56 fn clone(&self) -> Self {
57 Self {
58 tx: self.tx.clone(),
59 cancel: Arc::clone(&self.cancel),
60 }
61 }
62 }
63
64 impl<Ev> EventSink<Ev> {
65 /// Emit one event back to the handle.
66 ///
67 /// The event channel is bounded, so this applies backpressure when the GUI
68 /// hasn't drained yet: it retries rather than dropping the event or growing
69 /// the queue. It must never block *past cancellation*, though, a worker stuck
70 /// here while the handle is being dropped would hang the join on quit, so once
71 /// the cancel flag is set it drops the event and returns. Also returns if the
72 /// receiver is gone (handle dropped).
73 pub fn emit(&self, ev: Ev) {
74 let mut pending = ev;
75 loop {
76 match self.tx.try_send(pending) {
77 Ok(()) => return,
78 Err(mpsc::TrySendError::Full(returned)) => {
79 if self.cancel.load(Ordering::Acquire) {
80 return; // shutting down, drop the event rather than hang
81 }
82 pending = returned;
83 thread::sleep(Duration::from_millis(1));
84 }
85 Err(mpsc::TrySendError::Disconnected(_)) => return,
86 }
87 }
88 }
89
90 /// Create a sink paired with a receiver. For tests that exercise functions
91 /// taking an `&EventSink` directly, outside a spawned worker. The test must
92 /// stay under [`EVENT_CHANNEL_CAP`] unsent events or drain, since the standalone
93 /// sink has no cancel flag to release a full-channel wait.
94 pub fn channel() -> (Self, mpsc::Receiver<Ev>) {
95 let (tx, rx) = mpsc::sync_channel(EVENT_CHANNEL_CAP);
96 (
97 Self {
98 tx,
99 cancel: Arc::new(AtomicBool::new(false)),
100 },
101 rx,
102 )
103 }
104 }
105
106 /// Context passed to each worker step: an [`EventSink`] plus the shared cancel
107 /// flag. Both are `Sync`, so `&WorkerCtx` can be shared into rayon closures.
108 pub struct WorkerCtx<Ev> {
109 sink: EventSink<Ev>,
110 cancel: Arc<AtomicBool>,
111 }
112
113 impl<Ev> WorkerCtx<Ev> {
114 /// Emit one event.
115 pub fn emit(&self, ev: Ev) {
116 self.sink.emit(ev);
117 }
118
119 /// The event sink, for handing to code that emits many events.
120 pub fn sink(&self) -> &EventSink<Ev> {
121 &self.sink
122 }
123
124 /// Whether cancellation has been requested for the current operation.
125 pub fn is_cancelled(&self) -> bool {
126 self.cancel.load(Ordering::Acquire)
127 }
128
129 /// The raw cancel flag, for passing to processing functions that take
130 /// `&AtomicBool` (e.g. `chop_to_temp`, `process_edit`).
131 pub fn cancel_flag(&self) -> &AtomicBool {
132 &self.cancel
133 }
134
135 /// Clear the cancel flag at the start of a fresh operation so a stale cancel
136 /// from a previous command does not abort this one.
137 pub fn reset_cancel(&self) {
138 self.cancel.store(false, Ordering::Release);
139 }
140 }
141
142 /// Handle to a background worker. Sends commands, polls events; `Drop` requests
143 /// cancellation, closes the command channel (ending the worker's recv loop), and
144 /// joins the thread.
145 ///
146 /// Workers expose a thin newtype around this rather than the raw handle, so
147 /// worker-specific `send` semantics (mapping a Cancel command to
148 /// [`request_cancel`](Self::request_cancel)) live at the edge, not in the loop.
149 pub struct WorkerHandle<Cmd, Ev> {
150 cmd_tx: Option<mpsc::Sender<Cmd>>,
151 event_rx: Mutex<mpsc::Receiver<Ev>>,
152 cancel: Arc<AtomicBool>,
153 thread: Option<thread::JoinHandle<()>>,
154 }
155
156 impl<Cmd, Ev> WorkerHandle<Cmd, Ev> {
157 /// Poll for the next event without blocking.
158 pub fn try_recv(&self) -> Option<Ev> {
159 self.event_rx.lock().ok()?.try_recv().ok()
160 }
161
162 /// Enqueue a command for the worker. Returns `false` if the worker is no
163 /// longer alive, its receive loop has ended (e.g. `init` failed, or the
164 /// thread panicked out past the guard), so the command was dropped. Callers
165 /// that set a "busy" flag on dispatch must check this: setting busy for a
166 /// command that never reached a worker means no terminal event will ever
167 /// clear the flag.
168 #[must_use]
169 pub fn send(&self, cmd: Cmd) -> bool {
170 match &self.cmd_tx {
171 Some(tx) => tx.send(cmd).is_ok(),
172 None => false,
173 }
174 }
175
176 /// Request cancellation of the in-flight operation synchronously (on the
177 /// caller's thread). Newtype handles call this from `send` for their Cancel
178 /// command so the flag is set before the running operation next checks it.
179 pub fn request_cancel(&self) {
180 self.cancel.store(true, Ordering::Release);
181 }
182 }
183
184 impl<Cmd, Ev> Drop for WorkerHandle<Cmd, Ev> {
185 fn drop(&mut self) {
186 // Abort any in-flight long operation, then close the channel so the
187 // recv loop ends, then join.
188 self.cancel.store(true, Ordering::Release);
189 self.cmd_tx = None;
190 if let Some(handle) = self.thread.take() {
191 let _ = handle.join();
192 }
193 }
194 }
195
196 /// Spawn a background worker.
197 ///
198 /// - `init` runs once on the worker thread to build per-worker state (open a DB
199 /// or store, etc.). If it returns `Err`, `on_init_err` maps the error to an
200 /// event, that event is emitted, and the thread exits.
201 /// - `step` handles one command, using the [`WorkerCtx`] to emit any number of
202 /// events. Every `step` call is wrapped in `catch_unwind`; a panic is mapped
203 /// to a terminal event via `on_panic(&state)` (so the UI un-wedges for the
204 /// operation that was in flight) and the loop continues (the worker survives).
205 /// `on_panic` receives `&State` so it can carry context a step recorded before
206 /// panicking (e.g. the hash of the edit being processed).
207 ///
208 /// For workers with no persistent state, use `State = ()` and an infallible
209 /// `init` returning `Ok::<(), std::convert::Infallible>(())` with
210 /// `on_init_err = |e| match e {}`.
211 pub fn spawn_worker<Cmd, Ev, State, InitErr>(
212 name: &str,
213 init: impl FnOnce() -> Result<State, InitErr> + Send + 'static,
214 on_init_err: impl FnOnce(InitErr) -> Ev + Send + 'static,
215 on_panic: impl Fn(&State) -> Ev + Send + 'static,
216 mut step: impl FnMut(&mut State, Cmd, &WorkerCtx<Ev>) + Send + 'static,
217 ) -> std::io::Result<WorkerHandle<Cmd, Ev>>
218 where
219 Cmd: Send + 'static,
220 Ev: Send + 'static,
221 State: 'static,
222 InitErr: 'static,
223 {
224 // Commands stay unbounded (low-volume, one per user action, blocking the GUI
225 // on a command send would be worse than a tiny queue). Only events are bounded.
226 let (cmd_tx, cmd_rx) = mpsc::channel::<Cmd>();
227 let (event_tx, event_rx) = mpsc::sync_channel::<Ev>(EVENT_CHANNEL_CAP);
228 let cancel = Arc::new(AtomicBool::new(false));
229 let cancel_worker = Arc::clone(&cancel);
230
231 #[allow(clippy::disallowed_methods)] // sanctioned spawn point (per-step catch_unwind)
232 let thread = thread::Builder::new()
233 .name(name.to_string())
234 .spawn(move || {
235 let ctx = WorkerCtx {
236 sink: EventSink {
237 tx: event_tx,
238 cancel: Arc::clone(&cancel_worker),
239 },
240 cancel: cancel_worker,
241 };
242 let mut state = match init() {
243 Ok(s) => s,
244 Err(e) => {
245 ctx.emit(on_init_err(e));
246 return;
247 }
248 };
249 // Owned, guarded recv loop, the only consumer of `cmd_rx`, which
250 // never leaves this function. A panic in `step` becomes an event,
251 // not a dead thread.
252 while let Ok(cmd) = cmd_rx.recv() {
253 let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
254 step(&mut state, cmd, &ctx);
255 }));
256 if result.is_err() {
257 ctx.emit(on_panic(&state));
258 }
259 }
260 })?;
261
262 Ok(WorkerHandle {
263 cmd_tx: Some(cmd_tx),
264 event_rx: Mutex::new(event_rx),
265 cancel,
266 thread: Some(thread),
267 })
268 }
269
270 /// Handle to a one-shot job thread (see [`spawn_job`]). Poll [`try_recv`](Self::try_recv)
271 /// for the single result event; `Drop` joins the thread.
272 pub struct JobHandle<Ev> {
273 event_rx: Mutex<mpsc::Receiver<Ev>>,
274 thread: Option<thread::JoinHandle<()>>,
275 }
276
277 impl<Ev> JobHandle<Ev> {
278 /// Poll for the job's result without blocking.
279 pub fn try_recv(&self) -> Option<Ev> {
280 self.event_rx.lock().ok()?.try_recv().ok()
281 }
282 }
283
284 impl<Ev> Drop for JobHandle<Ev> {
285 fn drop(&mut self) {
286 if let Some(handle) = self.thread.take() {
287 let _ = handle.join();
288 }
289 }
290 }
291
292 /// Spawn a one-shot job thread: `body` runs once, its return value is sent back
293 /// as the single result event, and the thread exits.
294 ///
295 /// This is the one-shot counterpart to [`spawn_worker`] (which owns a command
296 /// recv loop). A panic in `body` is caught and mapped to `on_panic()`, the same
297 /// isolation guarantee, so the GUI's `try_recv` always resolves a terminal event
298 /// instead of hanging on a silently-dead thread. Use this for any
299 /// run-once-then-exit thread (classifier jobs, one-shot library passes); use
300 /// `spawn_worker` for a long-lived command loop, and [`spawn_detached`] for a
301 /// thread that restores its own state via Drop and reports no event. Bare
302 /// `thread::spawn` / `Builder::spawn` is denied by clippy (`disallowed-methods`):
303 /// every worker thread must route through one of these three panic-isolated
304 /// seals, so a new recv/stream loop cannot reintroduce the silent-stuck-state
305 /// failure that recurred across three ultra-fuzz runs.
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 #[allow(clippy::disallowed_methods)] // sanctioned spawn point (panic-isolated)
317 let thread = thread::Builder::new()
318 .name(name.to_string())
319 .spawn(move || {
320 let ev = std::panic::catch_unwind(std::panic::AssertUnwindSafe(body))
321 .unwrap_or_else(|_| on_panic());
322 let _ = tx.send(ev);
323 })?;
324 Ok(JobHandle {
325 event_rx: Mutex::new(rx),
326 thread: Some(thread),
327 })
328 }
329
330 /// Spawn a detached thread whose `body` restores its own invariants on exit
331 /// (typically via an RAII guard whose `Drop` resets shared state) and does NOT
332 /// report a terminal event.
333 ///
334 /// `body` runs inside `catch_unwind`, so a panic unwinds it, running its
335 /// destructors, which is where the state restoration happens, and is then
336 /// logged instead of silently killing the thread with state left wedged. This is
337 /// the sanctioned spawn for the recoverable detached threads that recover by
338 /// replacement rather than by emitting an event (the streaming preview decode,
339 /// whose `StreamingLease::drop` clears the `streaming` flag on every exit path
340 /// including panic). `spawn_job`/`spawn_worker` cover everything that reports a
341 /// result. A bare `thread::spawn` is denied by clippy precisely so this seal
342 /// (Drop + catch_unwind) can't be bypassed by a new recv/stream loop, the exact
343 /// regression that recurred across three ultra-fuzz runs.
344 pub fn spawn_detached(name: &str, body: impl FnOnce() + Send + 'static) -> std::io::Result<()> {
345 let label = name.to_string();
346 #[allow(clippy::disallowed_methods)] // sanctioned spawn point (panic-isolated)
347 thread::Builder::new().name(label.clone()).spawn(move || {
348 if std::panic::catch_unwind(std::panic::AssertUnwindSafe(body)).is_err() {
349 tracing::error!(thread = %label, "detached thread panicked; state restored via Drop");
350 }
351 })?;
352 Ok(())
353 }
354
355 #[cfg(test)]
356 mod tests {
357 use super::*;
358 use std::convert::Infallible;
359
360 fn poll_job<Ev>(handle: &JobHandle<Ev>) -> Ev {
361 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
362 loop {
363 if let Some(ev) = handle.try_recv() {
364 return ev;
365 }
366 assert!(std::time::Instant::now() < deadline, "job never emitted");
367 std::thread::sleep(std::time::Duration::from_millis(1));
368 }
369 }
370
371 #[test]
372 fn spawn_job_delivers_result() {
373 let handle = spawn_job("test-job-ok", || 0u32, || 42u32).unwrap();
374 assert_eq!(poll_job(&handle), 42);
375 }
376
377 #[test]
378 fn spawn_job_maps_panic_to_event() {
379 // A panicking job body must surface the on_panic event, not hang try_recv.
380 let handle = spawn_job("test-job-panic", || "panicked", || panic!("boom")).unwrap();
381 assert_eq!(poll_job(&handle), "panicked");
382 }
383
384 fn recv_one<Cmd, Ev>(handle: &WorkerHandle<Cmd, Ev>) -> Ev {
385 let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
386 while std::time::Instant::now() < deadline {
387 if let Some(ev) = handle.try_recv() {
388 return ev;
389 }
390 std::thread::sleep(std::time::Duration::from_millis(1));
391 }
392 panic!("worker produced no event");
393 }
394
395 #[test]
396 fn step_panic_becomes_event_and_worker_survives() {
397 // A command that panics must not kill the worker: the next command still
398 // gets processed. This is the invariant the loose_files regression broke.
399 let handle = spawn_worker::<u32, &'static str, (), Infallible>(
400 "test-panic",
401 || Ok(()),
402 |e| match e {},
403 |_state| "panicked",
404 |_state, cmd: u32, ctx| {
405 assert!(cmd != 0, "boom");
406 ctx.emit("ok");
407 },
408 )
409 .unwrap();
410
411 assert!(handle.send(0)); // panics (delivered, then caught)
412 assert_eq!(recv_one(&handle), "panicked");
413 assert!(handle.send(1)); // worker must still be alive
414 assert_eq!(recv_one(&handle), "ok");
415 drop(handle);
416 }
417
418 #[test]
419 fn init_failure_emits_event_and_exits() {
420 let handle = spawn_worker::<u32, &'static str, (), &'static str>(
421 "test-init-fail",
422 || Err("no db"),
423 |e| e,
424 |_state| "panicked",
425 |_state, _cmd, _ctx| {},
426 )
427 .unwrap();
428 assert_eq!(recv_one(&handle), "no db");
429 drop(handle);
430 }
431
432 #[test]
433 fn drop_sets_cancel_flag() {
434 // Drop must request cancellation so an in-flight op aborts promptly.
435 let observed = Arc::new(AtomicBool::new(false));
436 let observed_clone = Arc::clone(&observed);
437 let handle = spawn_worker::<u32, (), (), Infallible>(
438 "test-cancel",
439 || Ok(()),
440 |e| match e {},
441 |_state| (),
442 move |_state, _cmd: u32, ctx| {
443 // Spin until cancelled, then record that we saw it.
444 while !ctx.is_cancelled() {
445 std::thread::sleep(std::time::Duration::from_millis(1));
446 }
447 observed_clone.store(true, Ordering::Release);
448 },
449 )
450 .unwrap();
451 let _ = handle.send(1);
452 std::thread::sleep(std::time::Duration::from_millis(10));
453 drop(handle); // sets cancel, the step exits, join completes
454 assert!(observed.load(Ordering::Acquire));
455 }
456 }
457