From 9c037044c4b837a7803be34c6d1a5e23d16011de Mon Sep 17 00:00:00 2001 From: Stjepan Glavina Date: Tue, 20 Nov 2018 20:05:14 +0100 Subject: [PATCH] threadpool: rename inner to something more descriptive (#768) `inner` is a fitting name for variables of type named `Inner`, but in other cases I find them confusing - sometimes `inner` refers to a `Pool`, sometimes to a `Sender`. I renamed a bunch of variables named `inner` to be more descriptive. This PR is the first step in an effort of splitting https://github.com/tokio-rs/tokio/pull/722#issuecomment-439552671 into multiple PRs. --- src/runtime/mod.rs | 2 +- tokio-threadpool/src/builder.rs | 8 +-- tokio-threadpool/src/notifier.rs | 4 +- tokio-threadpool/src/pool/mod.rs | 82 ++++++++++++++--------------- tokio-threadpool/src/sender.rs | 14 ++--- tokio-threadpool/src/shutdown.rs | 10 ++-- tokio-threadpool/src/thread_pool.rs | 30 +++++------ tokio-threadpool/src/worker/mod.rs | 50 +++++++++--------- tokio-threadpool/tests/blocking.rs | 1 - 9 files changed, 100 insertions(+), 101 deletions(-) diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 8dbff44d1..2ea0d5483 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -153,7 +153,7 @@ pub struct Runtime { #[derive(Debug)] struct Inner { - /// A handle to the reactor in the background thread. + /// A handle to the reactor in the background thread. reactor_handle: Handle, // TODO: This should go away in 0.2 diff --git a/tokio-threadpool/src/builder.rs b/tokio-threadpool/src/builder.rs index a02d0e1d3..ca701f2c1 100644 --- a/tokio-threadpool/src/builder.rs +++ b/tokio-threadpool/src/builder.rs @@ -409,18 +409,18 @@ impl Builder { } // Create the pool - let inner = Arc::new( + let pool = Arc::new( Pool::new( workers.into_boxed_slice(), self.max_blocking, self.config.clone())); // Wrap with `Sender` - let inner = Some(Sender { - inner + let sender = Some(Sender { + pool }); - ThreadPool { inner } + ThreadPool { sender } } } diff --git a/tokio-threadpool/src/notifier.rs b/tokio-threadpool/src/notifier.rs index 3c4dfa698..fe8e8f4ad 100644 --- a/tokio-threadpool/src/notifier.rs +++ b/tokio-threadpool/src/notifier.rs @@ -13,7 +13,7 @@ use futures::executor::Notify; /// to poll the future again. #[derive(Debug)] pub(crate) struct Notifier { - pub inner: Arc, + pub pool: Arc, } /// A guard that ensures that the inner value gets forgotten. @@ -38,7 +38,7 @@ impl Notify for Notifier { // Bump the ref count let task = task.clone(); - let _ = self.inner.submit(task, &self.inner); + let _ = self.pool.submit(task, &self.pool); } } } diff --git a/tokio-threadpool/src/pool/mod.rs b/tokio-threadpool/src/pool/mod.rs index b4e8e388f..c6dce5473 100644 --- a/tokio-threadpool/src/pool/mod.rs +++ b/tokio-threadpool/src/pool/mod.rs @@ -270,8 +270,8 @@ impl Pool { /// /// Called from either inside or outside of the scheduler. If currently on /// the scheduler, then a fast path is taken. - pub fn submit(&self, task: Arc, inner: &Arc) { - debug_assert_eq!(*self, **inner); + pub fn submit(&self, task: Arc, pool: &Arc) { + debug_assert_eq!(*self, **pool); Worker::with_current(|worker| { if let Some(worker) = worker { @@ -283,18 +283,18 @@ impl Pool { // The second check handles the case where the current thread is // part of a different threadpool than the one being submitted // to. - if !worker.is_blocking() && *self == *worker.inner { + if !worker.is_blocking() && *self == *worker.pool { let idx = worker.id.0; trace!(" -> submit internal; idx={}", idx); - worker.inner.workers[idx].submit_internal(task); - worker.inner.signal_work(inner); + worker.pool.workers[idx].submit_internal(task); + worker.pool.signal_work(pool); return; } } - self.submit_external(task, inner); + self.submit_external(task, pool); }); } @@ -302,8 +302,8 @@ impl Pool { /// /// Called from outside of the scheduler, this function is how new tasks /// enter the system. - pub fn submit_external(&self, task: Arc, inner: &Arc) { - debug_assert_eq!(*self, **inner); + pub fn submit_external(&self, task: Arc, pool: &Arc) { + debug_assert_eq!(*self, **pool); use worker::Lifecycle::Notified; @@ -311,21 +311,21 @@ impl Pool { // sleeping tasks get woken up if let Some((idx, worker_state)) = self.sleep_stack.pop(&self.workers, Notified, false) { trace!("submit to existing worker; idx={}; state={:?}", idx, worker_state); - self.submit_to_external(idx, task, worker_state, inner); + self.submit_to_external(idx, task, worker_state, pool); return; } // All workers are active, so pick a random worker and submit the // task to it. - self.submit_to_random(task, inner); + self.submit_to_random(task, pool); } /// Submit a task to a random worker /// /// Called from outside of the scheduler, this function is how new tasks /// enter the system. - pub fn submit_to_random(&self, task: Arc, inner: &Arc) { - debug_assert_eq!(*self, **inner); + pub fn submit_to_random(&self, task: Arc, pool: &Arc) { + debug_assert_eq!(*self, **pool); let len = self.workers.len(); let idx = self.rand_usize() % len; @@ -333,21 +333,21 @@ impl Pool { trace!(" -> submitting to random; idx={}", idx); let state = self.workers[idx].load_state(); - self.submit_to_external(idx, task, state, inner); + self.submit_to_external(idx, task, state, pool); } fn submit_to_external(&self, idx: usize, task: Arc, state: worker::State, - inner: &Arc) + pool: &Arc) { - debug_assert_eq!(*self, **inner); + debug_assert_eq!(*self, **pool); let entry = &self.workers[idx]; if !entry.submit_external(task, state) { - self.spawn_thread(WorkerId::new(idx), inner); + self.spawn_thread(WorkerId::new(idx), pool); } } @@ -360,14 +360,14 @@ impl Pool { self.backup_stack.push(&self.backup, backup_id) } - pub fn notify_blocking_task(&self, inner: &Arc) { - debug_assert_eq!(*self, **inner); - self.blocking.notify_task(&inner); + pub fn notify_blocking_task(&self, pool: &Arc) { + debug_assert_eq!(*self, **pool); + self.blocking.notify_task(&pool); } /// Provision a thread to run a worker - pub fn spawn_thread(&self, id: WorkerId, inner: &Arc) { - debug_assert_eq!(*self, **inner); + pub fn spawn_thread(&self, id: WorkerId, pool: &Arc) { + debug_assert_eq!(*self, **pool); let backup_id = match self.backup_stack.pop(&self.backup, false) { Ok(Some(backup_id)) => backup_id, @@ -392,31 +392,31 @@ impl Pool { let mut th = thread::Builder::new(); - if let Some(ref prefix) = inner.config.name_prefix { + if let Some(ref prefix) = pool.config.name_prefix { th = th.name(format!("{}{}", prefix, backup_id.0)); } - if let Some(stack) = inner.config.stack_size { + if let Some(stack) = pool.config.stack_size { th = th.stack_size(stack); } - let inner = inner.clone(); + let pool = pool.clone(); let res = th.spawn(move || { - if let Some(ref f) = inner.config.after_start { + if let Some(ref f) = pool.config.after_start { f(); } let mut worker_id = id; - inner.backup[backup_id.0].start(&worker_id); + pool.backup[backup_id.0].start(&worker_id); loop { // The backup token should be in the running state. - debug_assert!(inner.backup[backup_id.0].is_running()); + debug_assert!(pool.backup[backup_id.0].is_running()); // TODO: Avoid always cloning - let worker = Worker::new(worker_id, backup_id, inner.clone()); + let worker = Worker::new(worker_id, backup_id, pool.clone()); // Run the worker. If the worker transitioned to a "blocking" // state, then `is_blocking` will be true. @@ -425,14 +425,14 @@ impl Pool { break; } - debug_assert!(!inner.backup[backup_id.0].is_pushed()); + debug_assert!(!pool.backup[backup_id.0].is_pushed()); // Push the thread back onto the backup stack. This makes it // available for future handoffs. // // This **must** happen before notifying the task. - let res = inner.backup_stack - .push(&inner.backup, backup_id); + let res = pool.backup_stack + .push(&pool.backup, backup_id); if res.is_err() { // The pool is being shutdown. @@ -441,17 +441,17 @@ impl Pool { // The task switched the current thread to blocking mode. // Now that the blocking task completed, any tasks - inner.notify_blocking_task(&inner); + pool.notify_blocking_task(&pool); - debug_assert!(inner.backup[backup_id.0].is_running()); + debug_assert!(pool.backup[backup_id.0].is_running()); // Wait for a handoff - let handoff = inner.backup[backup_id.0] - .wait_for_handoff(inner.config.keep_alive); + let handoff = pool.backup[backup_id.0] + .wait_for_handoff(pool.config.keep_alive); match handoff { Handoff::Worker(id) => { - debug_assert!(inner.backup[backup_id.0].is_running()); + debug_assert!(pool.backup[backup_id.0].is_running()); worker_id = id; } Handoff::Idle | Handoff::Terminated => { @@ -460,11 +460,11 @@ impl Pool { } } - if let Some(ref f) = inner.config.before_stop { + if let Some(ref f) = pool.config.before_stop { f(); } - inner.thread_stopped(); + pool.thread_stopped(); }); if let Err(e) = res { @@ -474,8 +474,8 @@ impl Pool { /// If there are any other workers currently relaxing, signal them that work /// is available so that they can try to find more work to process. - pub fn signal_work(&self, inner: &Arc) { - debug_assert_eq!(*self, **inner); + pub fn signal_work(&self, pool: &Arc) { + debug_assert_eq!(*self, **pool); use worker::Lifecycle::*; @@ -509,7 +509,7 @@ impl Pool { } Shutdown => { trace!("signal_work -- spawn; idx={}", idx); - self.spawn_thread(WorkerId(idx), inner); + self.spawn_thread(WorkerId(idx), pool); } Running | Notified | Signaled => { // The workers are already active. No need to wake them up. diff --git a/tokio-threadpool/src/sender.rs b/tokio-threadpool/src/sender.rs index 211aca4dd..0ff99d736 100644 --- a/tokio-threadpool/src/sender.rs +++ b/tokio-threadpool/src/sender.rs @@ -23,7 +23,7 @@ use futures::{future, Future}; /// [`ThreadPool::sender`]: struct.ThreadPool.html#method.sender #[derive(Debug)] pub struct Sender { - pub(crate) inner: Arc, + pub(crate) pool: Arc, } impl Sender { @@ -85,7 +85,7 @@ impl Sender { /// Logic to prepare for spawning fn prepare_for_spawn(&self) -> Result<(), SpawnError> { - let mut state: pool::State = self.inner.state.load(Acquire).into(); + let mut state: pool::State = self.pool.state.load(Acquire).into(); // Increment the number of futures spawned on the pool as well as // validate that the pool is still running/ @@ -104,7 +104,7 @@ impl Sender { next.inc_num_futures(); - let actual = self.inner.state.compare_and_swap( + let actual = self.pool.state.compare_and_swap( state.into(), next.into(), AcqRel).into(); if actual == state { @@ -135,7 +135,7 @@ impl tokio_executor::Executor for Sender { impl<'a> tokio_executor::Executor for &'a Sender { fn status(&self) -> Result<(), tokio_executor::SpawnError> { - let state: pool::State = self.inner.state.load(Acquire).into(); + let state: pool::State = self.pool.state.load(Acquire).into(); if state.num_futures() == MAX_FUTURES { // No capacity @@ -161,7 +161,7 @@ impl<'a> tokio_executor::Executor for &'a Sender { // Create a new task for the future let task = Arc::new(Task::new(future)); - self.inner.submit_to_random(task, &self.inner); + self.pool.submit_to_random(task, &self.pool); Ok(()) } @@ -189,7 +189,7 @@ where T: Future + Send + 'static, impl Clone for Sender { #[inline] fn clone(&self) -> Sender { - let inner = self.inner.clone(); - Sender { inner } + let pool = self.pool.clone(); + Sender { pool } } } diff --git a/tokio-threadpool/src/shutdown.rs b/tokio-threadpool/src/shutdown.rs index eb58ff084..957276079 100644 --- a/tokio-threadpool/src/shutdown.rs +++ b/tokio-threadpool/src/shutdown.rs @@ -16,12 +16,12 @@ use futures::{Future, Poll, Async}; /// [`shutdown_now`]: struct.ThreadPool.html#method.shutdown_now #[derive(Debug)] pub struct Shutdown { - pub(crate) inner: Sender, + pub(crate) sender: Sender, } impl Shutdown { - fn inner(&self) -> &Pool { - &*self.inner.inner + fn pool(&self) -> &Pool { + &*self.sender.pool } } @@ -32,9 +32,9 @@ impl Future for Shutdown { fn poll(&mut self) -> Poll<(), ()> { use futures::task; - self.inner().shutdown_task.task.register_task(task::current()); + self.pool().shutdown_task.task.register_task(task::current()); - if !self.inner().is_shutdown() { + if !self.pool().is_shutdown() { return Ok(Async::NotReady); } diff --git a/tokio-threadpool/src/thread_pool.rs b/tokio-threadpool/src/thread_pool.rs index 0bebd183e..c9e772921 100644 --- a/tokio-threadpool/src/thread_pool.rs +++ b/tokio-threadpool/src/thread_pool.rs @@ -15,7 +15,7 @@ use futures::sync::oneshot; /// Create `ThreadPool` instances using `Builder`. #[derive(Debug)] pub struct ThreadPool { - pub(crate) inner: Option, + pub(crate) sender: Option, } impl ThreadPool { @@ -111,12 +111,12 @@ impl ThreadPool { /// The handle is used to spawn futures onto the thread pool. It also /// implements the `Executor` trait. pub fn sender(&self) -> &Sender { - self.inner.as_ref().unwrap() + self.sender.as_ref().unwrap() } /// Return a mutable reference to the sender handle pub fn sender_mut(&mut self) -> &mut Sender { - self.inner.as_mut().unwrap() + self.sender.as_mut().unwrap() } /// Shutdown the pool once it becomes idle. @@ -130,8 +130,8 @@ impl ThreadPool { /// shutdown. The returned future completes once all worker threads have /// completed the shutdown process. pub fn shutdown_on_idle(mut self) -> Shutdown { - self.inner().shutdown(false, false); - Shutdown { inner: self.inner.take().unwrap() } + self.pool().shutdown(false, false); + Shutdown { sender: self.sender.take().unwrap() } } /// Shutdown the pool @@ -143,8 +143,8 @@ impl ThreadPool { /// worker threads are signaled and will shutdown. The returned future /// completes once all worker threads have completed the shutdown process. pub fn shutdown(mut self) -> Shutdown { - self.inner().shutdown(true, false); - Shutdown { inner: self.inner.take().unwrap() } + self.pool().shutdown(true, false); + Shutdown { sender: self.sender.take().unwrap() } } /// Shutdown the pool immediately @@ -156,20 +156,20 @@ impl ThreadPool { /// worker threads are signaled and will shutdown. The returned future /// completes once all worker threads have completed the shutdown process. pub fn shutdown_now(mut self) -> Shutdown { - self.inner().shutdown(true, true); - Shutdown { inner: self.inner.take().unwrap() } + self.pool().shutdown(true, true); + Shutdown { sender: self.sender.take().unwrap() } } - fn inner(&self) -> &Pool { - &*self.inner.as_ref().unwrap().inner + fn pool(&self) -> &Pool { + &*self.sender.as_ref().unwrap().pool } } impl Drop for ThreadPool { fn drop(&mut self) { - if let Some(sender) = self.inner.take() { - sender.inner.shutdown(true, true); - let shutdown = Shutdown { inner: sender }; + if let Some(sender) = self.sender.take() { + sender.pool.shutdown(true, true); + let shutdown = Shutdown { sender }; let _ = shutdown.wait(); } } @@ -191,4 +191,4 @@ impl Future for SpawnHandle { fn poll(&mut self) -> Poll { self.0.poll() } -} \ No newline at end of file +} diff --git a/tokio-threadpool/src/worker/mod.rs b/tokio-threadpool/src/worker/mod.rs index 438c28d76..9999d8639 100644 --- a/tokio-threadpool/src/worker/mod.rs +++ b/tokio-threadpool/src/worker/mod.rs @@ -39,7 +39,7 @@ use std::time::Duration; #[derive(Debug)] pub struct Worker { // Shared scheduler data - pub(crate) inner: Arc, + pub(crate) pool: Arc, // WorkerEntry index pub(crate) id: WorkerId, @@ -86,9 +86,9 @@ pub struct WorkerId(pub(crate) usize); thread_local!(static CURRENT_WORKER: Cell<*const Worker> = Cell::new(0 as *const _)); impl Worker { - pub(crate) fn new(id: WorkerId, backup_id: BackupId, inner: Arc) -> Worker { + pub(crate) fn new(id: WorkerId, backup_id: BackupId, pool: Arc) -> Worker { Worker { - inner, + pool, id, backup_id, current_task: CurrentTask::new(), @@ -111,14 +111,14 @@ impl Worker { CURRENT_WORKER.with(|c| { c.set(self as *const _); - let inner = self.inner.clone(); - let mut sender = Sender { inner }; + let pool = self.pool.clone(); + let mut sender = Sender { pool }; // Enter an execution context let mut enter = tokio_executor::enter().unwrap(); tokio_executor::with_default(&mut sender, &mut enter, |enter| { - if let Some(ref callback) = self.inner.config.around_worker { + if let Some(ref callback) = self.pool.config.around_worker { callback.call(self, enter); } else { self.run(); @@ -165,7 +165,7 @@ impl Worker { // Atomically attempt to acquire blocking capacity, and if none // is available, register the task to be notified once capacity // becomes available. - match self.inner.poll_blocking_capacity(task_ref)? { + match self.pool.poll_blocking_capacity(task_ref)? { Async::Ready(()) => { self.current_task.set_can_block(Allocated); } @@ -192,7 +192,7 @@ impl Worker { // Transitioning to blocking requires handing over the worker state to // another thread so that the work queue can continue to be processed. - self.inner.spawn_thread(self.id.clone(), &self.inner); + self.pool.spawn_thread(self.id.clone(), &self.pool); // Track that the thread has now fully entered the blocking state. self.is_blocking.set(true); @@ -222,7 +222,7 @@ impl Worker { // Get the notifier. let notify = Arc::new(Notifier { - inner: self.inner.clone(), + pool: self.pool.clone(), }); let mut first = true; @@ -286,7 +286,7 @@ impl Worker { // // The returned result is ignored because `Err` represents the pool // shutting down. We are currently aware of this fact. - let _ = self.inner.release_backup(self.backup_id); + let _ = self.pool.release_backup(self.backup_id); self.should_finalize.set(true); } @@ -315,7 +315,7 @@ impl Worker { let mut state: State = self.entry().state.load(Acquire).into(); loop { - let pool_state: pool::State = self.inner.state.load(Acquire).into(); + let pool_state: pool::State = self.pool.state.load(Acquire).into(); if pool_state.is_terminated() { return false; @@ -373,7 +373,7 @@ impl Worker { trace!("Worker::check_run_state; delegate signal"); // This worker is not ready to be signaled, so delegate the signal // to another worker. - self.inner.signal_work(&self.inner); + self.pool.signal_work(&self.pool); } true @@ -404,14 +404,14 @@ impl Worker { debug_assert!(!self.is_blocking.get()); - let len = self.inner.workers.len(); - let mut idx = self.inner.rand_usize() % len; + let len = self.pool.workers.len(); + let mut idx = self.pool.rand_usize() % len; let mut found_work = false; let start = idx; loop { if idx < len { - match self.inner.workers[idx].steal_tasks(self.entry()) { + match self.pool.workers[idx].steal_tasks(self.entry()) { Steal::Data(task) => { trace!("stole task"); @@ -424,7 +424,7 @@ impl Worker { // // TODO: Should this be called here or before // `run_task`? - self.inner.signal_work(&self.inner); + self.pool.signal_work(&self.pool); return true; } @@ -466,19 +466,19 @@ impl Worker { // // We have to call `submit_external` instead of `submit` // here because `self` is still set as the current worker. - self.inner.submit_external(task, &self.inner); + self.pool.submit_external(task, &self.pool); } else { self.entry().push_internal(task); } } Complete => { - let mut state: pool::State = self.inner.state.load(Acquire).into(); + let mut state: pool::State = self.pool.state.load(Acquire).into(); loop { let mut next = state; next.dec_num_futures(); - let actual = self.inner.state.compare_and_swap( + let actual = self.pool.state.compare_and_swap( state.into(), next.into(), AcqRel).into(); if actual == state { @@ -490,7 +490,7 @@ impl Worker { // up any sleeping worker so that they can notice // the shutdown state. if next.is_terminated() { - self.inner.terminate_sleeping_workers(); + self.pool.terminate_sleeping_workers(); } } @@ -528,7 +528,7 @@ impl Worker { // transitioned to blocking in this call, then another task has // to be notified. if self.allocated_at_run && !self.worker.is_blocking.get() { - self.worker.inner.notify_blocking_task(&self.worker.inner); + self.worker.pool.notify_blocking_task(&self.worker.pool); } self.worker.current_task.clear(); @@ -571,7 +571,7 @@ impl Worker { // not be better to only signal when work was found // after waking up? trace!("found work while draining; signal_work"); - self.inner.signal_work(&self.inner); + self.pool.signal_work(&self.pool); } return true; @@ -579,7 +579,7 @@ impl Worker { Inconsistent => { if found_work { trace!("found work while draining; signal_work"); - self.inner.signal_work(&self.inner); + self.pool.signal_work(&self.pool); } return false; @@ -649,7 +649,7 @@ impl Worker { // We obtained permission to push the worker into the // sleeper queue. - if let Err(_) = self.inner.push_sleeper(self.id.0) { + if let Err(_) = self.pool.push_sleeper(self.id.0) { trace!(" sleeping -- push to stack failed; idx={}", self.id.0); // The push failed due to the pool being terminated. // @@ -726,7 +726,7 @@ impl Worker { fn entry(&self) -> &Entry { debug_assert!(!self.is_blocking.get()); - &self.inner.workers[self.id.0] + &self.pool.workers[self.id.0] } } diff --git a/tokio-threadpool/tests/blocking.rs b/tokio-threadpool/tests/blocking.rs index 2d63d0dc8..47a8a9bc9 100644 --- a/tokio-threadpool/tests/blocking.rs +++ b/tokio-threadpool/tests/blocking.rs @@ -1,7 +1,6 @@ extern crate tokio_threadpool; extern crate env_logger; -#[macro_use] extern crate futures; extern crate rand;