More work on #757

bytes/src/lib.rs

@@ -107,7 +107,8 @@ pub mod arc {
         ///
         /// If uniquely held, this method recovers the initial pointer and length
         /// of the sequestered allocation and re-initializes the BytesMut. The return
-        /// value indicates whether this occurred.
+        /// value indicates whether this occurred. A `None` value indicates that the
+        /// downcast to `B` failed and the type is not correct.
         ///
         /// # Examples
         ///
@@ -123,19 +124,19 @@ pub mod arc {
         /// drop(shared3);
         /// drop(shared2);
         /// drop(shared4);
-        /// assert!(shared1.try_regenerate::<Vec<u8>>());
+        /// assert_eq!(shared1.try_regenerate::<Vec<u8>>(), Some(true));
         /// assert!(shared1.len() == 1024);
         /// ```
-        pub fn try_regenerate<B>(&mut self) -> bool where B: DerefMut<Target=[u8]>+'static {
+        pub fn try_regenerate<B>(&mut self) -> Option<bool> where B: DerefMut<Target=[u8]>+'static {
             // Only possible if this is the only reference to the sequestered allocation.
             if let Some(boxed) = Arc::get_mut(&mut self.sequestered) {
-                let downcast = boxed.downcast_mut::<B>().expect("Downcast failed");
+                let downcast = boxed.downcast_mut::<B>()?;
                 self.ptr = downcast.as_mut_ptr();
                 self.len = downcast.len();
-                true
+                Some(true)
             }
             else {
-                false
+                Some(false)
             }
         }
 

communication/src/allocator/zero_copy/bytes_slab.rs

@@ -80,7 +80,7 @@ impl BytesSlab {
             if self.stash.is_empty() {
                 for shared in self.in_progress.iter_mut() {
                     if let Some(mut bytes) = shared.take() {
-                        if bytes.try_regenerate::<BoxDerefMut>() {
+                        if bytes.try_regenerate::<BoxDerefMut>() == Some(true) {
                             // NOTE: Test should be redundant, but better safe...
                             if bytes.len() == (1 << self.shift) {
                                 self.stash.push(bytes);

timely/src/dataflow/operators/vec/partition.rs

@@ -5,7 +5,7 @@ use crate::dataflow::operators::core::Partition as PartitionCore;
 use crate::dataflow::{Scope, StreamVec};
 
 /// Partition a stream of records into multiple streams.
-pub trait Partition<G: Scope, D: 'static, D2: 'static, F: Fn(D) -> (u64, D2)> {
+pub trait Partition<G: Scope, D: 'static> {
     /// Produces `parts` output streams, containing records produced and assigned by `route`.
     ///
     /// # Examples
@@ -21,11 +21,11 @@ pub trait Partition<G: Scope, D: 'static, D2: 'static, F: Fn(D) -> (u64, D2)> {
     ///     streams.pop().unwrap().inspect(|x| println!("seen 0: {:?}", x));
     /// });
     /// ```
-    fn partition(self, parts: u64, route: F) -> Vec<StreamVec<G, D2>>;
+    fn partition<D2: 'static, F: Fn(D) -> (u64, D2)+'static>(self, parts: u64, route: F) -> Vec<StreamVec<G, D2>>;
 }
 
-impl<G: Scope, D: 'static, D2: 'static, F: Fn(D)->(u64, D2)+'static> Partition<G, D, D2, F> for StreamVec<G, D> {
-    fn partition(self, parts: u64, route: F) -> Vec<StreamVec<G, D2>> {
+impl<G: Scope, D: 'static> Partition<G, D> for StreamVec<G, D> {
+    fn partition<D2: 'static, F: Fn(D)->(u64, D2)+'static>(self, parts: u64, route: F) -> Vec<StreamVec<G, D2>> {
         PartitionCore::partition::<CapacityContainerBuilder<_>, _, _>(self, parts, route)
     }
 }

timely/src/progress/broadcast.rs

@@ -144,9 +144,7 @@ impl<T:Timestamp+Send> Progcaster<T> {
             });
 
             // We clone rather than drain to avoid deserialization.
-            for &(ref update, delta) in recv_changes.iter() {
-                changes.update(update.clone(), delta);
-            }
+            changes.extend(recv_changes.iter().map(|(u,d)| (u.clone(), *d)));
         }
 
     }