frontier: tighter integration with rust traits

The `frontier` module implements a common Rust pattern where there is
one type representing a mutable version of something and a separate type
representing its immutable counterpart. Examples in std are `String` and
`str`, `PathBuf` and `Path`, `OsString` and `OsStr`, and other.

In all the examples above the mutable version holds onto a heap
allocated buffer and the immutable version is just a
`#[repr(transparent)]` wrapper over an raw unsized slice of the data
type.

This patch changes `AntichainRef<T>` to simply be a transparent wrapper
over `[T]`. This change enables `Deref` implementations targeting the
immutable antichain from both `MutableAntichain<T>` and `Antichain<T>`.

This has a bunch of ergonomic improvements for users (due to auto-deref)
and also allows having a single implementation of all methods that only
need immutable access to the antichain.

Specifically, the following methods have been deduplicated and are now
implemented in only once place:

* `AntichainRef::less_than`
* `AntichainRef::less_equal`
* `AntichainRef::dominates`
* `AntichainRef::elements`
* `<AntichainRef as PartialEq>::eq`
* `<AntichainRef as PartialOrd>::partial_cmp`
* `<AntichainRef as Hash>::hash`

Finally, this change also enables replacing the inherent
`Antichain::borrow` and `Antichain::to_owned` methods by implementing
the `std::borrow::Borrow` and `std::borrow::ToOwned` traits
respectively.

Signed-off-by: Petros Angelatos <[email protected]>

Author: petrosagg

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