spawn through local queue if possible

benches/chained_spawn.rs

@@ -37,17 +37,17 @@ mod yatp_future {
     use criterion::*;
     use std::sync::mpsc;
     use yatp::task::future::TaskCell;
-    use yatp::Remote;
+    use yatp::Handle;
 
     pub fn chained_spawn(b: &mut Bencher<'_>, iter_count: usize) {
         let pool = yatp::Builder::new("chained_spawn").build_future_pool();
 
-        fn iter(remote: Remote<TaskCell>, done_tx: mpsc::SyncSender<()>, n: usize) {
+        fn iter(handle: Handle<TaskCell>, done_tx: mpsc::SyncSender<()>, n: usize) {
             if n == 0 {
                 done_tx.send(()).unwrap();
             } else {
-                let s2 = remote.clone();
-                remote.spawn(async move {
+                let s2 = handle.clone();
+                handle.spawn(async move {
                     iter(s2, done_tx, n - 1);
                 });
             }
@@ -57,9 +57,9 @@ mod yatp_future {
 
         b.iter(move || {
             let done_tx = done_tx.clone();
-            let remote = pool.remote().clone();
+            let handle = pool.handle().clone();
             pool.spawn(async move {
-                iter(remote, done_tx, iter_count);
+                iter(handle, done_tx, iter_count);
             });
 
             done_rx.recv().unwrap();

benches/ping_pong.rs

@@ -65,20 +65,20 @@ mod yatp_future {
             let rem = rem.clone();
             rem.store(ping_count, Ordering::Relaxed);
 
-            let remote = pool.remote().clone();
+            let handle = pool.handle().clone();
 
             pool.spawn(async move {
                 for _ in 0..ping_count {
                     let rem = rem.clone();
                     let done_tx = done_tx.clone();
 
-                    let remote2 = remote.clone();
+                    let handle2 = handle.clone();
 
-                    remote.spawn(async move {
+                    handle.spawn(async move {
                         let (tx1, rx1) = oneshot::channel();
                         let (tx2, rx2) = oneshot::channel();
 
-                        remote2.spawn(async move {
+                        handle2.spawn(async move {
                             rx1.await.unwrap();
                             tx2.send(()).unwrap();
                         });

src/lib.rs

@@ -9,4 +9,4 @@ pub mod pool;
 pub mod queue;
 pub mod task;
 
-pub use self::pool::{Builder, Remote, ThreadPool};
+pub use self::pool::{Builder, Handle, ThreadPool};

src/pool.rs

@@ -12,45 +12,45 @@ mod worker;
 
 pub use self::builder::{Builder, SchedConfig};
 pub use self::runner::{CloneRunnerBuilder, Runner, RunnerBuilder};
-pub use self::spawn::{build_spawn, Local, Remote};
+pub use self::spawn::{build_spawn, Handle, Local};
 
 use crate::queue::{TaskCell, WithExtras};
 use std::mem;
 use std::sync::Mutex;
 use std::thread::JoinHandle;
 
 /// A generic thread pool.
-pub struct ThreadPool<T: TaskCell + Send> {
-    remote: Remote<T>,
+pub struct ThreadPool<T: TaskCell + Send + 'static> {
+    handle: Handle<T>,
     threads: Mutex<Vec<JoinHandle<()>>>,
 }
 
-impl<T: TaskCell + Send> ThreadPool<T> {
+impl<T: TaskCell + Send + 'static> ThreadPool<T> {
     /// Spawns the task into the thread pool.
     ///
     /// If the pool is shutdown, it becomes no-op.
     pub fn spawn(&self, t: impl WithExtras<T>) {
-        self.remote.spawn(t);
+        self.handle.spawn(t);
     }
 
     /// Shutdowns the pool.
     ///
     /// Closes the queue and wait for all threads to exit.
     pub fn shutdown(&self) {
-        self.remote.stop();
+        self.handle.stop();
         let mut threads = mem::replace(&mut *self.threads.lock().unwrap(), Vec::new());
         for j in threads.drain(..) {
             j.join().unwrap();
         }
     }
 
-    /// Get a remote queue for spawning tasks without owning the thread pool.
-    pub fn remote(&self) -> &Remote<T> {
-        &self.remote
+    /// Get a handle for spawning tasks without owning the thread pool.
+    pub fn handle(&self) -> &Handle<T> {
+        &self.handle
     }
 }
 
-impl<T: TaskCell + Send> Drop for ThreadPool<T> {
+impl<T: TaskCell + Send + 'static> Drop for ThreadPool<T> {
     /// Will shutdown the thread pool if it has not.
     fn drop(&mut self) {
         self.shutdown();

src/pool/builder.rs

@@ -2,8 +2,8 @@
 
 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::pool::{CloneRunnerBuilder, Handle, Local, Runner, RunnerBuilder, ThreadPool};
+use crate::queue::{self, LocalQueueBuilder, QueueType, TaskCell};
 use crate::task::{callback, future};
 use std::sync::{Arc, Mutex};
 use std::thread;
@@ -49,10 +49,10 @@ impl Default for SchedConfig {
 pub struct LazyBuilder<T> {
     builder: Builder,
     core: Arc<QueueCore<T>>,
-    local_queues: Vec<LocalQueue<T>>,
+    local_queue_builders: Vec<LocalQueueBuilder<T>>,
 }
 
-impl<T> LazyBuilder<T> {
+impl<T: Send + 'static> LazyBuilder<T> {
     /// Sets the name prefix of threads. The thread name will follow the
     /// format "prefix-index".
     pub fn name(mut self, name_prefix: impl Into<String>) -> LazyBuilder<T> {
@@ -61,7 +61,7 @@ impl<T> LazyBuilder<T> {
     }
 }
 
-impl<T> LazyBuilder<T>
+impl<T: Send + 'static> LazyBuilder<T>
 where
     T: TaskCell + Send + 'static,
 {
@@ -76,25 +76,26 @@ where
         F::Runner: Runner<TaskCell = T> + Send + 'static,
     {
         let mut threads = Vec::with_capacity(self.builder.sched_config.max_thread_count);
-        for (i, local_queue) in self.local_queues.into_iter().enumerate() {
+        for (i, queue_builder) in self.local_queue_builders.into_iter().enumerate() {
             let runner = factory.build();
             let name = format!("{}-{}", self.builder.name_prefix, i);
             let mut builder = thread::Builder::new().name(name);
             if let Some(size) = self.builder.stack_size {
                 builder = builder.stack_size(size)
             }
-            let local = Local::new(i + 1, local_queue, self.core.clone());
-            let thd = WorkerThread::new(local, runner);
+            let core = self.core.clone();
             threads.push(
                 builder
                     .spawn(move || {
+                        let local = Local::new(i + 1, queue_builder(), core);
+                        let thd = WorkerThread::new(local, runner);
                         thd.run();
                     })
                     .unwrap(),
             );
         }
         ThreadPool {
-            remote: Remote::new(self.core.clone()),
+            handle: Handle::new(self.core.clone()),
             threads: Mutex::new(threads),
         }
     }
@@ -183,9 +184,9 @@ impl Builder {
     /// In some cases, especially building up a large application, a task
     /// scheduler is required before spawning new threads. You can use this
     /// to separate the construction and starting.
-    pub fn freeze<T>(&self) -> (Remote<T>, LazyBuilder<T>)
+    pub fn freeze<T>(&self) -> (Handle<T>, LazyBuilder<T>)
     where
-        T: TaskCell + Send,
+        T: TaskCell + Send + 'static,
     {
         self.freeze_with_queue(QueueType::SingleLevel)
     }
@@ -199,20 +200,21 @@ impl Builder {
     /// In some cases, especially building up a large application, a task
     /// scheduler is required before spawning new threads. You can use this
     /// to separate the construction and starting.
-    pub fn freeze_with_queue<T>(&self, queue_type: QueueType) -> (Remote<T>, LazyBuilder<T>)
+    pub fn freeze_with_queue<T>(&self, queue_type: QueueType) -> (Handle<T>, LazyBuilder<T>)
     where
-        T: TaskCell + Send,
+        T: TaskCell + Send + 'static,
     {
         assert!(self.sched_config.min_thread_count <= self.sched_config.max_thread_count);
-        let (injector, local_queues) = queue::build(queue_type, self.sched_config.max_thread_count);
+        let (injector, local_queue_builders) =
+            queue::build(queue_type, self.sched_config.max_thread_count);
         let core = Arc::new(QueueCore::new(injector, self.sched_config.clone()));
 
         (
-            Remote::new(core.clone()),
+            Handle::new(core.clone()),
             LazyBuilder {
                 builder: self.clone(),
                 core,
-                local_queues,
+                local_queue_builders,
             },
         )
     }