tokio-rs/tokio
1
0
Fork 0
mirror of https://github.com/tokio-rs/tokio.git synced 2026-08-21 00:00:10 +02:00
Code Issues Packages Projects Releases Wiki Activity

runtime: recover when OS fails to spawn a new thread (#4485)

Browse Source
This commit is contained in:
Antonin Amand
2022-02-25 12:28:56 +01:00
committed by GitHub
parent 8e0e56fdf2
commit 70c10bae60
1 changed files with 30 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tokio/src/runtime/blocking/pool.rs
+30 -14
View File
@@ -240,17 +240,29 @@ impl Spawner {
if shared.num_th == self.inner.thread_cap {
// At max number of threads
} else {
shared.num_th += 1;
assert!(shared.shutdown_tx.is_some());
let shutdown_tx = shared.shutdown_tx.clone();
if let Some(shutdown_tx) = shutdown_tx {
let id = shared.worker_thread_index;
shared.worker_thread_index += 1;
let handle = self.spawn_thread(shutdown_tx, rt, id);
shared.worker_threads.insert(id, handle);
match self.spawn_thread(shutdown_tx, rt, id) {
Ok(handle) => {
shared.num_th += 1;
shared.worker_thread_index += 1;
shared.worker_threads.insert(id, handle);
}
Err(ref e) if is_temporary_os_thread_error(e) && shared.num_th > 0 => {
// OS temporarily failed to spawn a new thread.
// The task will be picked up eventually by a currently
// busy thread.
}
Err(e) => {
// The OS refused to spawn the thread and there is no thread
// to pick up the task that has just been pushed to the queue.
panic!("OS can't spawn worker thread: {}", e)
}
}
}
}
} else {
@@ -272,7 +284,7 @@ impl Spawner {
shutdown_tx: shutdown::Sender,
rt: &Handle,
id: usize,
) -> thread::JoinHandle<()> {
) -> std::io::Result<thread::JoinHandle<()>> {
let mut builder = thread::Builder::new().name((self.inner.thread_name)());
if let Some(stack_size) = self.inner.stack_size {
@@ -281,17 +293,21 @@ impl Spawner {
let rt = rt.clone();
builder
.spawn(move || {
// Only the reference should be moved into the closure
let _enter = crate::runtime::context::enter(rt.clone());
rt.blocking_spawner.inner.run(id);
drop(shutdown_tx);
})
.expect("OS can't spawn a new worker thread")
builder.spawn(move || {
// Only the reference should be moved into the closure
let _enter = crate::runtime::context::enter(rt.clone());
rt.blocking_spawner.inner.run(id);
drop(shutdown_tx);
})
}
}
// Tells whether the error when spawning a thread is temporary.
#[inline]
fn is_temporary_os_thread_error(error: &std::io::Error) -> bool {
matches!(error.kind(), std::io::ErrorKind::WouldBlock)
}
impl Inner {
fn run(&self, worker_thread_id: usize) {
if let Some(f) = &self.after_start {