More refining

Signed-off-by: Moritz Hoffmann <[email protected]>

container/src/lib.rs

@@ -93,7 +93,7 @@ pub trait PushInto<T> {
 /// needs to produce all outputs, even partial ones. Caller should repeatedly call the functions
 /// to drain pending or finished data.
 ///
-/// The caller is responsible to fully consume the containers returned by [`Self::extract`] and
+/// The caller should consume the containers returned by [`Self::extract`] and
 /// [`Self::finish`]. Implementations can recycle buffers, but should ensure that they clear
 /// any remaining elements.
 ///
@@ -109,9 +109,11 @@ pub trait ContainerBuilder: Default + 'static {
     /// be called repeatedly, for example while the caller can send data.
     ///
     /// Returns a `Some` if there is data ready to be shipped, and `None` otherwise.
+    #[must_use]
     fn extract(&mut self) -> Option<&mut Self::Container>;
     /// Extract assembled containers and any unfinished data. Should
     /// be called repeatedly until it returns `None`.
+    #[must_use]
     fn finish(&mut self) -> Option<&mut Self::Container>;
 }
 
@@ -132,6 +134,7 @@ pub struct CapacityContainerBuilder<C>{
 }
 
 impl<T, C: SizableContainer + PushInto<T>> PushInto<T> for CapacityContainerBuilder<C> {
+    #[inline]
     fn push_into(&mut self, item: T) {
         if self.current.capacity() == 0 {
             self.current = self.empty.take().unwrap_or_default();

timely/src/dataflow/operators/core/input.rs

@@ -360,7 +360,7 @@ impl<T: Timestamp, CB: ContainerBuilder> Handle<T, CB> {
 
     /// Extract all ready contents from the builder and distribute to downstream operators.
     #[inline]
-    fn extract(&mut self) {
+    fn extract_and_send(&mut self) {
         while let Some(container) = self.builder.extract() {
             Self::send_container(container, &mut self.buffer, &mut self.pushers, &self.now_at);
         }
@@ -378,6 +378,7 @@ impl<T: Timestamp, CB: ContainerBuilder> Handle<T, CB> {
     /// Does not take `self` because `flush` and `extract` borrow `self` mutably.
     /// Clears the container.
     // TODO: Find a better name for this function.
+    #[inline]
     fn send_container(
         container: &mut CB::Container,
         buffer: &mut CB::Container,
@@ -396,7 +397,7 @@ impl<T: Timestamp, CB: ContainerBuilder> Handle<T, CB> {
         container.clear();
     }
 
-    /// closes the current epoch, flushing if needed, shutting if needed, and updating the frontier.
+    /// Closes the current epoch, flushing if needed, shutting if needed, and updating the frontier.
     // TODO: Find a better name for this function.
     fn close_epoch(&mut self) {
         self.flush();
@@ -490,7 +491,7 @@ where
     #[inline]
     fn push_into(&mut self, item: D) {
         self.builder.push_into(item);
-        self.extract();
+        self.extract_and_send();
     }
 }
 

timely/src/dataflow/operators/to_stream.rs

@@ -27,6 +27,6 @@ pub trait ToStream<D: Data> {
 
 impl<I: IntoIterator+'static> ToStream<I::Item> for I where I::Item: Data {
     fn to_stream<S: Scope>(self, scope: &mut S) -> Stream<S, I::Item> {
-        ToStreamCore::<_>::to_stream(self, scope)
+        ToStreamCore::to_stream(self, scope)
     }
 }