mirror/async-std
2
0
Fork 1
mirror of https://github.com/async-rs/async-std.git synced 2025-04-24 09:16:46 +00:00
Code Issues Releases Wiki Activity

Spawn several threads when we fail to enqueue work in the blocki… (#181)

Browse source 
* Rebase onto master

* Switch to unbounded channels
This commit is contained in:
Tyler Neely 2019-11-01 21:53:13 +01:00 committed by Stjepan Glavina
parent 1a51ca424a
commit 5adc608791
1 changed files with 25 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
src/task/spawn_blocking.rs
View file

@ -2,7 +2,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
use std::thread; use std::thread;
use std::time::Duration; use std::time::Duration;
use crossbeam_channel::{bounded, Receiver, Sender}; use crossbeam_channel::{unbounded, Receiver, Sender};
use once_cell::sync::Lazy; use once_cell::sync::Lazy;
use crate::task::{JoinHandle, Task}; use crate::task::{JoinHandle, Task};
@ -79,7 +79,7 @@ static POOL: Lazy<Pool> = Lazy::new(|| {
// before being acted on by a core. This helps keep // before being acted on by a core. This helps keep
// latency snappy in the overall async system by // latency snappy in the overall async system by
// reducing bufferbloat. // reducing bufferbloat.
let (sender, receiver) = bounded(0); let (sender, receiver) = unbounded();
Pool { sender, receiver } Pool { sender, receiver }
}); });
@ -95,27 +95,31 @@ fn maybe_create_another_blocking_thread() {
return; return;
} }
// We want to avoid having all threads terminate at let n_to_spawn = std::cmp::min(2 + (workers / 10), 10);
// exactly the same time, causing thundering herd
// effects. We want to stagger their destruction over
// 10 seconds or so to make the costs fade into
// background noise.
//
// Generate a simple random number of milliseconds
let rand_sleep_ms = u64::from(random(10_000));
thread::Builder::new() for _ in 0..n_to_spawn {
.name("async-std/blocking".to_string()) // We want to avoid having all threads terminate at
.spawn(move || { // exactly the same time, causing thundering herd
let wait_limit = Duration::from_millis(1000 + rand_sleep_ms); // effects. We want to stagger their destruction over
// 10 seconds or so to make the costs fade into
// background noise.
//
// Generate a simple random number of milliseconds
let rand_sleep_ms = u64::from(random(10_000));
DYNAMIC_THREAD_COUNT.fetch_add(1, Ordering::Relaxed); thread::Builder::new()
while let Ok(task) = POOL.receiver.recv_timeout(wait_limit) { .name("async-std/blocking".to_string())
abort_on_panic(|| task.run()); .spawn(move || {
} let wait_limit = Duration::from_millis(1000 + rand_sleep_ms);
DYNAMIC_THREAD_COUNT.fetch_sub(1, Ordering::Relaxed);
}) DYNAMIC_THREAD_COUNT.fetch_add(1, Ordering::Relaxed);
.expect("cannot start a dynamic thread driving blocking tasks"); while let Ok(task) = POOL.receiver.recv_timeout(wait_limit) {
abort_on_panic(|| task.run());
}
DYNAMIC_THREAD_COUNT.fetch_sub(1, Ordering::Relaxed);
})
.expect("cannot start a dynamic thread driving blocking tasks");
}
} }
// Enqueues work, attempting to send to the threadpool in a // Enqueues work, attempting to send to the threadpool in a