diff --git a/soulseek-rs-lib/src/utils/thread_pool.rs b/soulseek-rs-lib/src/utils/thread_pool.rs index eafd0a1..371c5bc 100644 --- a/soulseek-rs-lib/src/utils/thread_pool.rs +++ b/soulseek-rs-lib/src/utils/thread_pool.rs @@ -39,7 +39,7 @@ impl ThreadPool { { let job = Box::new(f); if let Some(ref sender) = self.sender { - sender.send(Message::NewJob(job)).unwrap(); + let _ = sender.send(Message::NewJob(job)); } } } @@ -48,18 +48,47 @@ impl Drop for ThreadPool { fn drop(&mut self) { if let Some(sender) = self.sender.take() { for _ in &self.workers { - sender.send(Message::Terminate).unwrap(); + let _ = sender.send(Message::Terminate); } } + // Ignore join errors: a worker whose job panicked returns Err here, but + // Drop must never itself panic (that would abort the process). for worker in &mut self.workers { if let Some(thread) = worker.thread.take() { - thread.join().unwrap(); + let _ = thread.join(); } } } } +#[cfg(test)] +mod tests { + use super::ThreadPool; + use std::sync::mpsc; + use std::time::Duration; + + // A single panicking job must not permanently kill its worker: subsequent + // jobs still need to run. Otherwise one malformed network message could + // leak a pool worker until the whole pool is exhausted (a DoS). + #[test] + fn pool_survives_a_panicking_job() { + let pool = ThreadPool::new(1); + pool.execute(|| panic!("boom")); + + let (tx, rx) = mpsc::channel(); + pool.execute(move || { + let _ = tx.send(42); + }); + + assert_eq!( + rx.recv_timeout(Duration::from_secs(5)).ok(), + Some(42), + "worker died after a panicking job" + ); + } +} + struct Worker { thread: Option>, } @@ -73,7 +102,15 @@ impl Worker { Err(_) => break, }; match message { - Ok(Message::NewJob(job)) => job(), + // Contain a panicking job so it kills only that job, not the + // worker: the pool has a fixed number of workers and a lost + // one is never replaced. The lock is already released here, + // so a panic cannot poison the shared receiver. + Ok(Message::NewJob(job)) => { + let _ = std::panic::catch_unwind( + std::panic::AssertUnwindSafe(job), + ); + } Ok(Message::Terminate) | Err(_) => break, } }