Skip to main content

max / audiofiles

Perf: bound the worker event channel with cancel-aware backpressure The per-worker event channel was unbounded — the analysis batch fans out over rayon and emits a per-sample event, which on a large import outruns a slow GUI drain and grows the queue without limit (the every-8th-sample throttle only reduced the rate). Switch events to a bounded sync_channel(4096); `emit` retries on a full channel (backpressure) but drops the event and returns once the cancel flag is set, so a worker can never hang the handle's join on Drop/quit. Commands stay unbounded (low-volume; blocking the GUI on a command send would be worse). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-22 21:51 UTC
Commit: 581abadd3374e0e96c20866f6aabbc6369ec2e6e
Parent: 4d1ea2f
1 file changed, +58 insertions, -11 deletions
M crates/audiofiles-core/src/worker_runtime.rs +58 -11
@@ -32,32 +32,74 @@
32 32 use std::sync::atomic::{AtomicBool, Ordering};
33 33 use std::sync::{mpsc, Arc, Mutex};
34 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;
35 44
36 45 /// Sink handed to a worker step so it can emit zero or more events per command.
37 46 ///
38 - /// Cloneable and `Sync` (it wraps an `mpsc::Sender`), so a step may share it
39 - /// across rayon worker threads (e.g. the analysis batch fan-out).
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.
40 50 pub struct EventSink<Ev> {
41 - tx: mpsc::Sender<Ev>,
51 + tx: mpsc::SyncSender<Ev>,
52 + cancel: Arc<AtomicBool>,
42 53 }
43 54
44 55 impl<Ev> Clone for EventSink<Ev> {
45 56 fn clone(&self) -> Self {
46 - Self { tx: self.tx.clone() }
57 + Self {
58 + tx: self.tx.clone(),
59 + cancel: Arc::clone(&self.cancel),
60 + }
47 61 }
48 62 }
49 63
50 64 impl<Ev> EventSink<Ev> {
51 - /// Emit one event back to the handle. Silently dropped if the handle is gone.
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).
52 73 pub fn emit(&self, ev: Ev) {
53 - let _ = self.tx.send(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 + }
54 88 }
55 89
56 90 /// Create a sink paired with a receiver. For tests that exercise functions
57 - /// taking an `&EventSink` directly, outside a spawned worker.
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.
58 94 pub fn channel() -> (Self, mpsc::Receiver<Ev>) {
59 - let (tx, rx) = mpsc::channel();
60 - (Self { tx }, rx)
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 + )
61 103 }
62 104 }
63 105
@@ -179,8 +221,10 @@ where
179 221 State: 'static,
180 222 InitErr: 'static,
181 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.
182 226 let (cmd_tx, cmd_rx) = mpsc::channel::<Cmd>();
183 - let (event_tx, event_rx) = mpsc::channel::<Ev>();
227 + let (event_tx, event_rx) = mpsc::sync_channel::<Ev>(EVENT_CHANNEL_CAP);
184 228 let cancel = Arc::new(AtomicBool::new(false));
185 229 let cancel_worker = Arc::clone(&cancel);
186 230
@@ -188,7 +232,10 @@ where
188 232 .name(name.to_string())
189 233 .spawn(move || {
190 234 let ctx = WorkerCtx {
191 - sink: EventSink { tx: event_tx },
235 + sink: EventSink {
236 + tx: event_tx,
237 + cancel: Arc::clone(&cancel_worker),
238 + },
192 239 cancel: cancel_worker,
193 240 };
194 241 let mut state = match init() {