mirror/async-std
2
0
Fork 1
mirror of https://github.com/async-rs/async-std.git synced 2025-02-28 23:29:41 +00:00
Code Issues Releases Wiki Activity

Merge pull request #510 from stjepang/fix-spawn-blocking

Browse source 
Improve thread startup/shutdown algorithm in spawn_blocking
This commit is contained in:
Florian Gilcher 2019-11-12 00:39:10 +01:00 committed by GitHub
commit 54371c21c1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 19 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
src/task/spawn_blocking.rs
View file

@ -6,7 +6,7 @@ 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};
use crate::utils::{abort_on_panic, random}; use crate::utils::abort_on_panic;
/// Spawns a blocking task. /// Spawns a blocking task.
/// ///
@ -68,16 +68,13 @@ static POOL: Lazy<Pool> = Lazy::new(|| {
fn start_thread() { fn start_thread() {
SLEEPING.fetch_add(1, Ordering::SeqCst); SLEEPING.fetch_add(1, Ordering::SeqCst);
let timeout = Duration::from_secs(1);
// Generate a random duration of time between 1 second and 10 seconds. If the thread doesn't
// receive the next task in this duration of time, it will stop running.
let timeout = Duration::from_millis(1000 + u64::from(random(9_000)));
thread::Builder::new() thread::Builder::new()
.name("async-std/blocking".to_string()) .name("async-std/blocking".to_string())
.spawn(move || { .spawn(move || {
loop { loop {
let task = match POOL.receiver.recv_timeout(timeout) { let mut task = match POOL.receiver.recv_timeout(timeout) {
Ok(task) => task, Ok(task) => task,
Err(_) => { Err(_) => {
// Check whether this is the last sleeping thread. // Check whether this is the last sleeping thread.
@ -100,9 +97,23 @@ fn start_thread() {
start_thread(); start_thread();
} }
loop {
// Run the task. // Run the task.
abort_on_panic(|| task.run()); abort_on_panic(|| task.run());
// Try taking another task if there are any available.
task = match POOL.receiver.try_recv() {
Ok(task) => task,
Err(_) => break,
};
}
// If there is at least one sleeping thread, stop this thread instead of putting it
// to sleep.
if SLEEPING.load(Ordering::SeqCst) > 0 {
return;
}
SLEEPING.fetch_add(1, Ordering::SeqCst); SLEEPING.fetch_add(1, Ordering::SeqCst);
} }
}) })