frontier: tighter integration with rust traits

timely/src/dataflow/operators/capability.rs

@@ -403,7 +403,7 @@ impl<T: Timestamp> CapabilitySet<T> {
     ///             let mut cap = CapabilitySet::from_elem(default_cap);
     ///             let mut vector = Vec::new();
     ///             move |input, output| {
-    ///                 cap.downgrade(&input.frontier().frontier());
+    ///                 cap.downgrade(&**input.frontier());
     ///                 while let Some((time, data)) = input.next() {
     ///                     data.swap(&mut vector);
     ///                 }

timely/src/dataflow/operators/generic/notificator.rs

@@ -40,8 +40,8 @@ impl<'a, T: Timestamp> Notificator<'a, T> {
     }
 
     /// Reveals the elements in the frontier of the indicated input.
-    pub fn frontier(&self, input: usize) -> AntichainRef<T> {
-        self.frontiers[input].frontier()
+    pub fn frontier(&self, input: usize) -> &AntichainRef<T> {
+        self.frontiers[input]
     }
 
     /// Requests a notification at the time associated with capability `cap`.

timely/src/dataflow/operators/inspect.rs

@@ -124,9 +124,9 @@ impl<G: Scope, C: Container> InspectCore<G, C> for StreamCore<G, C> {
         let mut frontier = crate::progress::Antichain::from_elem(G::Timestamp::minimum());
         let mut vector = Default::default();
         self.unary_frontier(Pipeline, "InspectBatch", move |_,_| move |input, output| {
-            if input.frontier.frontier() != frontier.borrow() {
+            if input.frontier != &frontier {
                 frontier.clear();
-                frontier.extend(input.frontier.frontier().iter().cloned());
+                frontier.extend(input.frontier.iter().cloned());
                 func(Err(frontier.elements()));
             }
             input.for_each(|time, data| {

timely/src/dataflow/operators/probe.rs

@@ -166,8 +166,8 @@ impl<T: Timestamp> Handle<T> {
     /// let frontier = handle.with_frontier(|frontier| frontier.to_vec());
     /// ```
     #[inline]
-    pub fn with_frontier<R, F: FnMut(AntichainRef<T>)->R>(&self, mut function: F) -> R {
-        function(self.frontier.borrow().frontier())
+    pub fn with_frontier<R, F: FnMut(&AntichainRef<T>)->R>(&self, mut function: F) -> R {
+        function(&self.frontier.borrow())
     }
 }