Rename Inner to ReadyToRunQueue for clarity

MajorBreakfast · cramertj · commit 4584e5070990 · 2018-06-22T16:32:02.000-07:00
diff --git a/futures-util/src/stream/futures_unordered/mod.rs b/futures-util/src/stream/futures_unordered/mod.rs
@@ -15,11 +15,11 @@ use futures_core::{Stream, Future, Poll};
 use futures_core::task::{self, AtomicWaker};
 
 mod abort;
-mod inner;
+mod ready_to_run_queue;
 mod iter_mut;
 mod node;
 
-use self::inner::{Inner, Dequeue};
+use self::ready_to_run_queue::{ReadyToRunQueue, Dequeue};
 use self::iter_mut::IterMut;
 use self::node::Node;
 
@@ -40,7 +40,7 @@ use self::node::Node;
 /// an empty set with the `FuturesUnordered::new` constructor.
 #[must_use = "streams do nothing unless polled"]
 pub struct FuturesUnordered<F> {
-    inner: Arc<Inner<F>>,
+    ready_to_run_queue: Arc<ReadyToRunQueue<F>>,
     len: usize,
     head_all: *const Node<F>,
 }
@@ -85,22 +85,22 @@ impl<T> FuturesUnordered<T>
             future: UnsafeCell::new(None),
             next_all: UnsafeCell::new(ptr::null()),
             prev_all: UnsafeCell::new(ptr::null()),
-            next_readiness: AtomicPtr::new(ptr::null_mut()),
+            next_ready_to_run: AtomicPtr::new(ptr::null_mut()),
             queued: AtomicBool::new(true),
-            queue: Weak::new(),
+            ready_to_run_queue: Weak::new(),
         });
         let stub_ptr = &*stub as *const Node<T>;
-        let inner = Arc::new(Inner {
+        let ready_to_run_queue = Arc::new(ReadyToRunQueue {
             parent: AtomicWaker::new(),
-            head_readiness: AtomicPtr::new(stub_ptr as *mut _),
-            tail_readiness: UnsafeCell::new(stub_ptr),
+            head: AtomicPtr::new(stub_ptr as *mut _),
+            tail: UnsafeCell::new(stub_ptr),
             stub: stub,
         });
 
         FuturesUnordered {
             len: 0,
             head_all: ptr::null_mut(),
-            inner: inner,
+            ready_to_run_queue: ready_to_run_queue,
         }
     }
 }
@@ -129,9 +129,9 @@ impl<T> FuturesUnordered<T> {
             future: UnsafeCell::new(Some(future)),
             next_all: UnsafeCell::new(ptr::null_mut()),
             prev_all: UnsafeCell::new(ptr::null_mut()),
-            next_readiness: AtomicPtr::new(ptr::null_mut()),
+            next_ready_to_run: AtomicPtr::new(ptr::null_mut()),
             queued: AtomicBool::new(true),
-            queue: Arc::downgrade(&self.inner),
+            ready_to_run_queue: Arc::downgrade(&self.ready_to_run_queue),
         });
 
         // Right now our node has a strong reference count of 1. We transfer
@@ -143,7 +143,7 @@ impl<T> FuturesUnordered<T> {
         // e.g. getting its unpark notifications going to us tracking which
         // futures are ready. To do that we unconditionally enqueue it for
         // polling here.
-        self.inner.enqueue(ptr);
+        self.ready_to_run_queue.enqueue(ptr);
     }
 
     /// Returns an iterator that allows modifying each future in the set.
@@ -232,12 +232,12 @@ impl<T> Stream for FuturesUnordered<T>
         -> Poll<Option<Self::Item>>
     {
         // Ensure `parent` is correctly set.
-        self.inner.parent.register(cx.waker());
+        self.ready_to_run_queue.parent.register(cx.waker());
 
         loop {
             // Safety: &mut self guarantees the mutual exclusion `dequeue`
             // expects
-            let node = match unsafe { self.inner.dequeue() } {
+            let node = match unsafe { self.ready_to_run_queue.dequeue() } {
                 Dequeue::Empty => {
                     if self.is_empty() {
                         return Poll::Ready(None);
@@ -255,7 +255,7 @@ impl<T> Stream for FuturesUnordered<T>
                 Dequeue::Data(node) => node,
             };
 
-            debug_assert!(node != self.inner.stub());
+            debug_assert!(node != self.ready_to_run_queue.stub());
 
             // Safety:
             // - Node is a valid pointer.
@@ -376,7 +376,7 @@ impl<T> Drop for FuturesUnordered<T> {
         // mpsc queue. None of those nodes, however, have futures associated
         // with them so they're safe to destroy on any thread. At this point
         // the `FuturesUnordered` struct, the owner of the one strong reference
-        // to `Inner<T>` will drop the strong reference. At that point
+        // to `MPSCQueue<T>` will drop the strong reference. At that point
         // whichever thread releases the strong refcount last (be it this
         // thread or some other thread as part of an `upgrade`) will clear out
         // the mpsc queue and free all remaining nodes.
diff --git a/futures-util/src/stream/futures_unordered/node.rs b/futures-util/src/stream/futures_unordered/node.rs
@@ -8,7 +8,7 @@ use std::sync::atomic::Ordering::SeqCst;
 
 use futures_core::task::{UnsafeWake, Waker, LocalWaker};
 
-use super::Inner;
+use super::ReadyToRunQueue;
 use super::abort::abort;
 
 pub(super) struct Node<T> {
@@ -22,18 +22,18 @@ pub(super) struct Node<T> {
     pub(super) prev_all: UnsafeCell<*const Node<T>>,
 
     // Next pointer in readiness queue
-    pub(super) next_readiness: AtomicPtr<Node<T>>,
+    pub(super) next_ready_to_run: AtomicPtr<Node<T>>,
 
     // Queue that we'll be enqueued to when notified
-    pub(super) queue: Weak<Inner<T>>,
+    pub(super) ready_to_run_queue: Weak<ReadyToRunQueue<T>>,
 
     // Whether or not this node is currently in the mpsc queue.
     pub(super) queued: AtomicBool,
 }
 
 impl<T> Node<T> {
     pub(super) fn wake(self: &Arc<Node<T>>) {
-        let inner = match self.queue.upgrade() {
+        let inner = match self.ready_to_run_queue.upgrade() {
             Some(inner) => inner,
             None => return,
         };
diff --git a/futures-util/src/stream/futures_unordered/ready_to_run_queue.rs b/futures-util/src/stream/futures_unordered/ready_to_run_queue.rs
@@ -15,29 +15,29 @@ pub(super) enum Dequeue<T> {
     Inconsistent,
 }
 
-pub(super) struct Inner<T> {
+pub(super) struct ReadyToRunQueue<T> {
     // The task using `FuturesUnordered`.
     pub(super) parent: AtomicWaker,
 
     // Head/tail of the readiness queue
-    pub(super) head_readiness: AtomicPtr<Node<T>>,
-    pub(super) tail_readiness: UnsafeCell<*const Node<T>>,
+    pub(super) head: AtomicPtr<Node<T>>,
+    pub(super) tail: UnsafeCell<*const Node<T>>,
     pub(super) stub: Arc<Node<T>>,
 }
 
-impl<T> Inner<T> {
+impl<T> ReadyToRunQueue<T> {
     /// The enqueue function from the 1024cores intrusive MPSC queue algorithm.
     pub(super) fn enqueue(&self, node: *const Node<T>) {
         unsafe {
             debug_assert!((*node).queued.load(Relaxed));
 
             // This action does not require any coordination
-            (*node).next_readiness.store(ptr::null_mut(), Relaxed);
+            (*node).next_ready_to_run.store(ptr::null_mut(), Relaxed);
 
             // Note that these atomic orderings come from 1024cores
             let node = node as *mut _;
-            let prev = self.head_readiness.swap(node, AcqRel);
-            (*prev).next_readiness.store(node, Release);
+            let prev = self.head.swap(node, AcqRel);
+            (*prev).next_ready_to_run.store(node, Release);
         }
     }
 
@@ -46,35 +46,35 @@ impl<T> Inner<T> {
     /// Note that this is unsafe as it required mutual exclusion (only one
     /// thread can call this) to be guaranteed elsewhere.
     pub(super) unsafe fn dequeue(&self) -> Dequeue<T> {
-        let mut tail = *self.tail_readiness.get();
-        let mut next = (*tail).next_readiness.load(Acquire);
+        let mut tail = *self.tail.get();
+        let mut next = (*tail).next_ready_to_run.load(Acquire);
 
         if tail == self.stub() {
             if next.is_null() {
                 return Dequeue::Empty;
             }
 
-            *self.tail_readiness.get() = next;
+            *self.tail.get() = next;
             tail = next;
-            next = (*next).next_readiness.load(Acquire);
+            next = (*next).next_ready_to_run.load(Acquire);
         }
 
         if !next.is_null() {
-            *self.tail_readiness.get() = next;
+            *self.tail.get() = next;
             debug_assert!(tail != self.stub());
             return Dequeue::Data(tail);
         }
 
-        if self.head_readiness.load(Acquire) as *const _ != tail {
+        if self.head.load(Acquire) as *const _ != tail {
             return Dequeue::Inconsistent;
         }
 
         self.enqueue(self.stub());
 
-        next = (*tail).next_readiness.load(Acquire);
+        next = (*tail).next_ready_to_run.load(Acquire);
 
         if !next.is_null() {
-            *self.tail_readiness.get() = next;
+            *self.tail.get() = next;
             return Dequeue::Data(tail);
         }
 
@@ -86,7 +86,7 @@ impl<T> Inner<T> {
     }
 }
 
-impl<T> Drop for Inner<T> {
+impl<T> Drop for ReadyToRunQueue<T> {
     fn drop(&mut self) {
         // Once we're in the destructor for `Inner<T>` we need to clear out the
         // mpsc queue of nodes if there's anything left in there.
