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

@@ -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

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 }
     }
 }
 

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<Pool>,
+    pub pool: Arc<Pool>,
 }
 
 /// 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);
             }
         }
     }

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<Task>, inner: &Arc<Pool>) {
-        debug_assert_eq!(*self, **inner);
+    pub fn submit(&self, task: Arc<Task>, pool: &Arc<Pool>) {
+        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<Task>, inner: &Arc<Pool>) {
-        debug_assert_eq!(*self, **inner);
+    pub fn submit_external(&self, task: Arc<Task>, pool: &Arc<Pool>) {
+        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<Task>, inner: &Arc<Pool>) {
-        debug_assert_eq!(*self, **inner);
+    pub fn submit_to_random(&self, task: Arc<Task>, pool: &Arc<Pool>) {
+        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<Task>,
                           state: worker::State,
-                          inner: &Arc<Pool>)
+                          pool: &Arc<Pool>)
     {
-        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<Pool>) {
-        debug_assert_eq!(*self, **inner);
-        self.blocking.notify_task(&inner);
+    pub fn notify_blocking_task(&self, pool: &Arc<Pool>) {
+        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<Pool>) {
-        debug_assert_eq!(*self, **inner);
+    pub fn spawn_thread(&self, id: WorkerId, pool: &Arc<Pool>) {
+        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<Pool>) {
-        debug_assert_eq!(*self, **inner);
+    pub fn signal_work(&self, pool: &Arc<Pool>) {
+        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.

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<Pool>,
+    pub(crate) pool: Arc<Pool>,
 }
 
 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<Item = (), Error = ()> + 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 }
     }
 }

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);
         }
 

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<Sender>,
+    pub(crate) sender: Option<Sender>,
 }
 
 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<T, E> Future for SpawnHandle<T, E> {
     fn poll(&mut self) -> Poll<T, E> {
         self.0.poll()
     }
-}
+}

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<Pool>,
+    pub(crate) pool: Arc<Pool>,
 
     // 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<Pool>) -> Worker {
+    pub(crate) fn new(id: WorkerId, backup_id: BackupId, pool: Arc<Pool>) -> 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]
     }
 }
 

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;