fix: contain panicking jobs so a thread-pool worker is never lost
A job that panicked unwound out of the worker's run loop and permanently killed that worker; the pool has a fixed worker count and never respawns, so repeated panics (e.g. from malformed network messages) would exhaust the pool and wedge the whole actor system — a DoS. The worker now runs each job under catch_unwind (the shared receiver lock is already released, so a panic cannot poison it) and survives to process the next job. Drop and execute no longer unwrap their channel/join results, so shutdown can never itself panic.
This commit is contained in:
@@ -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<thread::JoinHandle<()>>,
|
||||
}
|
||||
@@ -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,
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user