Merge remote-tracking branch 'upstream/master' into calculate-chance

Signed-off-by: Yilin Chen <sticnarf@gmail.com>

Author: sticnarf

Cargo.toml

@@ -45,3 +45,4 @@ harness = false
 
 [profile.bench]
 codegen-units = 1
+debug = true

benches/chained_spawn.rs

@@ -39,9 +39,7 @@ mod yatp_future {
     use yatp::task::future::TaskCell;
     use yatp::Remote;
 
-    pub fn chained_spawn(b: &mut Bencher<'_>, iter_count: usize) {
-        let pool = yatp::Builder::new("chained_spawn").build_future_pool();
-
+    fn chained_spawn(b: &mut Bencher<'_>, pool: yatp::ThreadPool<TaskCell>, iter_count: usize) {
         fn iter(remote: Remote<TaskCell>, done_tx: mpsc::SyncSender<()>, n: usize) {
             if n == 0 {
                 done_tx.send(()).unwrap();
@@ -65,6 +63,16 @@ mod yatp_future {
             done_rx.recv().unwrap();
         });
     }
+
+    pub fn chained_spawn_single_level(b: &mut Bencher<'_>, iter_count: usize) {
+        let pool = yatp::Builder::new("chained_spawn").build_future_pool();
+        chained_spawn(b, pool, iter_count)
+    }
+
+    pub fn chained_spawn_multilevel(b: &mut Bencher<'_>, iter_count: usize) {
+        let pool = yatp::Builder::new("chained_spawn").build_multilevel_future_pool();
+        chained_spawn(b, pool, iter_count)
+    }
 }
 
 mod tokio {
@@ -134,13 +142,18 @@ mod async_std {
 
 pub fn chained_spawn(b: &mut Criterion) {
     let mut group = b.benchmark_group("chained_spawn");
-    for i in &[100, 400, 700, 1000] {
+    for i in &[256, 512, 1024] {
         group.bench_with_input(BenchmarkId::new("yatp::future", i), i, |b, i| {
-            yatp_future::chained_spawn(b, *i)
+            yatp_future::chained_spawn_single_level(b, *i)
         });
         group.bench_with_input(BenchmarkId::new("yatp::callback", i), i, |b, i| {
             yatp_callback::chained_spawn(b, *i)
         });
+        group.bench_with_input(
+            BenchmarkId::new("yatp::future::multilevel", i),
+            i,
+            |b, i| yatp_future::chained_spawn_multilevel(b, *i),
+        );
         group.bench_with_input(BenchmarkId::new("tokio", i), i, |b, i| {
             tokio::chained_spawn(b, *i)
         });

benches/ping_pong.rs

@@ -53,10 +53,9 @@ mod yatp_future {
     use std::sync::atomic::*;
     use std::sync::*;
     use tokio::sync::oneshot;
+    use yatp::task::future::TaskCell;
 
-    pub fn ping_pong(b: &mut Bencher<'_>, ping_count: usize) {
-        let pool = yatp::Builder::new("ping_pong").build_future_pool();
-
+    fn ping_pong(b: &mut Bencher<'_>, pool: yatp::ThreadPool<TaskCell>, ping_count: usize) {
         let (done_tx, done_rx) = mpsc::sync_channel(1000);
         let rem = Arc::new(AtomicUsize::new(0));
 
@@ -96,6 +95,16 @@ mod yatp_future {
             done_rx.recv().unwrap();
         });
     }
+
+    pub fn ping_pong_single_level(b: &mut Bencher<'_>, ping_count: usize) {
+        let pool = yatp::Builder::new("ping_pong").build_future_pool();
+        ping_pong(b, pool, ping_count)
+    }
+
+    pub fn ping_pong_multilevel(b: &mut Bencher<'_>, ping_count: usize) {
+        let pool = yatp::Builder::new("ping_pong").build_multilevel_future_pool();
+        ping_pong(b, pool, ping_count)
+    }
 }
 
 mod tokio {
@@ -199,13 +208,18 @@ mod async_std {
 
 pub fn ping_pong(b: &mut Criterion) {
     let mut group = b.benchmark_group("ping_pong");
-    for i in &[100, 400, 700, 1000] {
+    for i in &[256, 512, 1024] {
         group.bench_with_input(BenchmarkId::new("yatp::future", i), i, |b, i| {
-            yatp_future::ping_pong(b, *i)
+            yatp_future::ping_pong_single_level(b, *i)
         });
         group.bench_with_input(BenchmarkId::new("yatp::callback", i), i, |b, i| {
             yatp_callback::ping_pong(b, *i)
         });
+        group.bench_with_input(
+            BenchmarkId::new("yatp::future::multilevel", i),
+            i,
+            |b, i| yatp_future::ping_pong_multilevel(b, *i),
+        );
         group.bench_with_input(BenchmarkId::new("tokio", i), i, |b, i| {
             tokio::ping_pong(b, *i)
         });

benches/spawn_many.rs

@@ -35,11 +35,11 @@ mod yatp_future {
     use criterion::*;
     use std::sync::atomic::*;
     use std::sync::*;
+    use yatp::task::future::TaskCell;
 
-    pub fn spawn_many(b: &mut Bencher<'_>, spawn_count: usize) {
+    fn spawn_many(b: &mut Bencher<'_>, pool: yatp::ThreadPool<TaskCell>, spawn_count: usize) {
         let (tx, rx) = mpsc::sync_channel(1000);
         let rem = Arc::new(AtomicUsize::new(0));
-        let pool = yatp::Builder::new("spawn_many").build_future_pool();
 
         b.iter(|| {
             rem.store(spawn_count, Ordering::Relaxed);
@@ -58,6 +58,16 @@ mod yatp_future {
             let _ = rx.recv().unwrap();
         });
     }
+
+    pub fn spawn_many_single_level(b: &mut Bencher<'_>, spawn_count: usize) {
+        let pool = yatp::Builder::new("spawn_many").build_future_pool();
+        spawn_many(b, pool, spawn_count)
+    }
+
+    pub fn spawn_many_multilevel(b: &mut Bencher<'_>, spawn_count: usize) {
+        let pool = yatp::Builder::new("spawn_many").build_multilevel_future_pool();
+        spawn_many(b, pool, spawn_count)
+    }
 }
 
 mod threadpool {
@@ -153,13 +163,18 @@ mod async_std {
 
 pub fn spawn_many(b: &mut Criterion) {
     let mut group = b.benchmark_group("spawn_many");
-    for i in &[1000, 4000, 7000, 10000] {
+    for i in &[1024, 4096, 8192, 16384] {
         group.bench_with_input(BenchmarkId::new("yatp::future", i), i, |b, i| {
-            yatp_future::spawn_many(b, *i)
+            yatp_future::spawn_many_single_level(b, *i)
         });
         group.bench_with_input(BenchmarkId::new("yatp::callback", i), i, |b, i| {
             yatp_callback::spawn_many(b, *i)
         });
+        group.bench_with_input(
+            BenchmarkId::new("yatp::future::multilevel", i),
+            i,
+            |b, i| yatp_future::spawn_many_multilevel(b, *i),
+        );
         group.bench_with_input(BenchmarkId::new("threadpool", i), i, |b, i| {
             threadpool::spawn_many(b, *i)
         });

benches/yield_many.rs

@@ -62,11 +62,11 @@ mod yatp_callback {
 mod yatp_future {
     use criterion::*;
     use std::sync::mpsc;
+    use yatp::task::future::TaskCell;
 
-    pub fn yield_many(b: &mut Bencher<'_>, yield_count: usize) {
+    fn yield_many(b: &mut Bencher<'_>, pool: yatp::ThreadPool<TaskCell>, yield_count: usize) {
         let tasks = super::TASKS_PER_CPU * num_cpus::get();
         let (tx, rx) = mpsc::sync_channel(tasks);
-        let pool = yatp::Builder::new("yield_many").build_future_pool();
 
         b.iter(move || {
             for _ in 0..tasks {
@@ -84,6 +84,16 @@ mod yatp_future {
             }
         });
     }
+
+    pub fn yield_many_single_level(b: &mut Bencher<'_>, yield_count: usize) {
+        let pool = yatp::Builder::new("yield_many").build_future_pool();
+        yield_many(b, pool, yield_count)
+    }
+
+    pub fn yield_many_multilevel(b: &mut Bencher<'_>, yield_count: usize) {
+        let pool = yatp::Builder::new("yield_many").build_multilevel_future_pool();
+        yield_many(b, pool, yield_count)
+    }
 }
 
 mod tokio {
@@ -146,13 +156,18 @@ mod async_std {
 
 pub fn yield_many(b: &mut Criterion) {
     let mut group = b.benchmark_group("yield_many");
-    for i in &[100, 400, 700, 1000] {
+    for i in &[256, 512, 1024] {
         group.bench_with_input(BenchmarkId::new("yatp::future", i), i, |b, i| {
-            yatp_future::yield_many(b, *i)
+            yatp_future::yield_many_single_level(b, *i)
         });
         group.bench_with_input(BenchmarkId::new("yatp::callback", i), i, |b, i| {
             yatp_callback::yield_many(b, *i)
         });
+        group.bench_with_input(
+            BenchmarkId::new("yatp::future::multilevel", i),
+            i,
+            |b, i| yatp_future::yield_many_multilevel(b, *i),
+        );
         group.bench_with_input(BenchmarkId::new("tokio", i), i, |b, i| {
             tokio::yield_many(b, *i)
         });

src/pool.rs

@@ -17,7 +17,7 @@ pub use self::spawn::{build_spawn, Local, Remote};
 use crate::queue::{TaskCell, WithExtras};
 use std::mem;
 use std::sync::Mutex;
-use std::thread::JoinHandle;
+use std::thread::{self, JoinHandle};
 
 /// A generic thread pool.
 pub struct ThreadPool<T: TaskCell + Send> {
@@ -39,8 +39,11 @@ impl<T: TaskCell + Send> ThreadPool<T> {
     pub fn shutdown(&self) {
         self.remote.stop();
         let mut threads = mem::replace(&mut *self.threads.lock().unwrap(), Vec::new());
+        let curr_id = thread::current().id();
         for j in threads.drain(..) {
-            j.join().unwrap();
+            if curr_id != j.thread().id() {
+                j.join().unwrap();
+            }
         }
     }
 

src/pool/builder.rs

@@ -3,7 +3,7 @@
 use crate::pool::spawn::QueueCore;
 use crate::pool::worker::WorkerThread;
 use crate::pool::{CloneRunnerBuilder, Local, Remote, Runner, RunnerBuilder, ThreadPool};
-use crate::queue::{self, LocalQueue, QueueType, TaskCell};
+use crate::queue::{self, multilevel, LocalQueue, QueueType, TaskCell};
 use crate::task::{callback, future};
 use std::sync::{Arc, Mutex};
 use std::thread;
@@ -233,6 +233,16 @@ impl Builder {
         self.build_with_queue_and_runner(QueueType::SingleLevel, fb)
     }
 
+    /// Spawns a multilevel future pool.
+    ///
+    /// It setups the pool with multi level queue.
+    pub fn build_multilevel_future_pool(&self) -> ThreadPool<future::TaskCell> {
+        let fb = CloneRunnerBuilder(future::Runner::default());
+        let queue_builder = multilevel::Builder::new(multilevel::Config::default());
+        let runner_builder = queue_builder.runner_builder(fb);
+        self.build_with_queue_and_runner(QueueType::Multilevel(queue_builder), runner_builder)
+    }
+
     /// Spawns the thread pool immediately.
     ///
     /// `queue_builder` is a closure that creates a task queue. It accepts the

src/pool/tests.rs

@@ -96,3 +96,17 @@ fn test_remote() {
     let res = rx.recv_timeout(Duration::from_millis(500));
     assert_eq!(res, Err(mpsc::RecvTimeoutError::Timeout));
 }
+
+#[test]
+fn test_shutdown_in_pool() {
+    let pool = Builder::new("test_shutdown_in_pool")
+        .max_thread_count(4)
+        .build_callback_pool();
+    let remote = pool.remote().clone();
+    let (tx, rx) = mpsc::channel();
+    remote.spawn(move |_: &mut Handle<'_>| {
+        pool.shutdown();
+        tx.send(()).unwrap();
+    });
+    rx.recv_timeout(Duration::from_secs(1)).unwrap();
+}

src/queue/multilevel.rs

@@ -17,7 +17,7 @@ use prometheus::local::LocalIntCounter;
 use prometheus::{Gauge, IntCounter};
 use rand::prelude::*;
 use std::cell::Cell;
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst};
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering::*};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 use std::{f64, fmt, iter};
@@ -291,12 +291,12 @@ impl LevelManager {
     {
         let extras = task_cell.mut_extras();
         let task_id = extras.task_id;
-        let running_time = extras
-            .running_time
-            .get_or_insert_with(|| self.task_elapsed_map.get_elapsed(task_id));
         let current_level = match extras.fixed_level {
             Some(level) => level,
             None => {
+                let running_time = extras
+                    .running_time
+                    .get_or_insert_with(|| self.task_elapsed_map.get_elapsed(task_id));
                 let running_time = running_time.as_duration();
                 self.level_time_threshold
                     .iter()
@@ -363,15 +363,15 @@ fn calculate_level0_chance(
     }
 }
 
-pub(crate) struct ElapsedTime(IntCounter);
+pub(crate) struct ElapsedTime(AtomicU64);
 
 impl ElapsedTime {
     fn as_duration(&self) -> Duration {
-        Duration::from_micros(self.0.get() as u64)
+        Duration::from_micros(self.0.load(Relaxed) as u64)
     }
 
     fn inc_by(&self, t: Duration) {
-        self.0.inc_by(t.as_micros() as i64);
+        self.0.fetch_add(t.as_micros() as u64, Relaxed);
     }
 
     #[cfg(test)]
@@ -384,7 +384,7 @@ impl ElapsedTime {
 
 impl Default for ElapsedTime {
     fn default() -> ElapsedTime {
-        ElapsedTime(IntCounter::new("_", "_").unwrap())
+        ElapsedTime(AtomicU64::new(0))
     }
 }