exchange: remove one layer of boxing (#456)

* exchange: remove one layer of boxing

We were previously forced to use a boxed trait object because we were
unable to name the type of the constructed closure.

This patch fixes it by introducing a custom `ParallelizationHasher`
trait that the `ParallelizationContractCore` expects with a blanket
implementation for all closures taking two arguments. This is to
maintain backwards compatibility with any code that passes a closure
directly to `ExchangeCore::new`.

Then a wrapper type `DataHasher<F>` is introduced that allows us to both
specialize the `ParallelizationHasher` implementation for single
argument closures and at the same time name the type which is what we
need to remove one layer of boxed trait objects.

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

* exchange: disallow exchanging based on time

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

Author: petrosagg

timely/src/dataflow/channels/pact.rs

@@ -73,14 +73,13 @@ impl<T: Timestamp, C, D: Data+Clone, F: FnMut(&D)->u64+'static> ParallelizationC
 where
     C: Data + Container + PushPartitioned<Item=D>,
 {
-    // TODO: The closure in the type prevents us from naming it.
-    //       Could specialize `ExchangePusher` to a time-free version.
-    type Pusher = Box<dyn Push<BundleCore<T, C>>>;
-    type Puller = Box<dyn Pull<BundleCore<T, C>>>;
-    fn connect<A: AsWorker>(mut self, allocator: &mut A, identifier: usize, address: &[usize], logging: Option<Logger>) -> (Self::Pusher, Self::Puller) {
+    type Pusher = ExchangePusher<T, C, D, LogPusher<T, C, Box<dyn Push<BundleCore<T, C>>>>, F>;
+    type Puller = LogPuller<T, C, Box<dyn Pull<BundleCore<T, C>>>>;
+
+    fn connect<A: AsWorker>(self, allocator: &mut A, identifier: usize, address: &[usize], logging: Option<Logger>) -> (Self::Pusher, Self::Puller) {
         let (senders, receiver) = allocator.allocate::<Message<T, C>>(identifier, address);
         let senders = senders.into_iter().enumerate().map(|(i,x)| LogPusher::new(x, allocator.index(), i, identifier, logging.clone())).collect::<Vec<_>>();
-        (Box::new(ExchangePusher::new(senders, move |_, d| (self.hash_func)(d))), Box::new(LogPuller::new(receiver, allocator.index(), identifier, logging.clone())))
+        (ExchangePusher::new(senders, self.hash_func), LogPuller::new(receiver, allocator.index(), identifier, logging.clone()))
     }
 }
 

timely/src/dataflow/channels/pushers/exchange.rs

@@ -7,15 +7,15 @@ use crate::dataflow::channels::{BundleCore, Message};
 
 // TODO : Software write combining
 /// Distributes records among target pushees according to a distribution function.
-pub struct Exchange<T, C: Container, D, P: Push<BundleCore<T, C>>, H: FnMut(&T, &D) -> u64> {
+pub struct Exchange<T, C: Container, D, P: Push<BundleCore<T, C>>, H: FnMut(&D) -> u64> {
     pushers: Vec<P>,
     buffers: Vec<C>,
     current: Option<T>,
     hash_func: H,
     phantom: std::marker::PhantomData<D>,
 }
 
-impl<T: Clone, C: Container, D: Data, P: Push<BundleCore<T, C>>, H: FnMut(&T, &D)->u64>  Exchange<T, C, D, P, H> {
+impl<T: Clone, C: Container, D: Data, P: Push<BundleCore<T, C>>, H: FnMut(&D) -> u64>  Exchange<T, C, D, P, H> {
     /// Allocates a new `Exchange` from a supplied set of pushers and a distribution function.
     pub fn new(pushers: Vec<P>, key: H) -> Exchange<T, C, D, P, H> {
         let mut buffers = vec![];
@@ -40,7 +40,7 @@ impl<T: Clone, C: Container, D: Data, P: Push<BundleCore<T, C>>, H: FnMut(&T, &D
     }
 }
 
-impl<T: Eq+Data, C: Container, D: Data, P: Push<BundleCore<T, C>>, H: FnMut(&T, &D)->u64> Push<BundleCore<T, C>> for Exchange<T, C, D, P, H>
+impl<T: Eq+Data, C: Container, D: Data, P: Push<BundleCore<T, C>>, H: FnMut(&D) -> u64> Push<BundleCore<T, C>> for Exchange<T, C, D, P, H>
 where
     C: PushPartitioned<Item=D>
 {
@@ -72,7 +72,7 @@ where
                 let pushers = &mut self.pushers;
                 data.push_partitioned(
                     &mut self.buffers,
-                    move |datum| ((hash_func)(time, datum) & mask) as usize,
+                    move |datum| ((hash_func)(datum) & mask) as usize,
                     |index, buffer| {
                             Message::push_at(buffer, time.clone(), &mut pushers[index]);
                     }
@@ -84,7 +84,7 @@ where
                 let pushers = &mut self.pushers;
                 data.push_partitioned(
                     &mut self.buffers,
-                    move |datum| ((hash_func)(time, datum) % num_pushers) as usize,
+                    move |datum| ((hash_func)(datum) % num_pushers) as usize,
                     |index, buffer| {
                         Message::push_at(buffer, time.clone(), &mut pushers[index]);
                     }