Simplify communication Message type (#596)

* Remove MessageContents::Arc

* Remove MessageContents

* Remove RefOrMut::Ref

* Remove RefOrMut

* Remove accessory methods

* mdbook fix-up

Author: frankmcsherry

communication/src/message.rs

@@ -1,184 +1,42 @@
 //! Types wrapping typed data.
 
-use std::sync::Arc;
 use bytes::arc::Bytes;
 use crate::Data;
 
-/// Either an immutable or mutable reference.
-pub enum RefOrMut<'a, T> where T: 'a {
-    /// An immutable reference.
-    Ref(&'a T),
-    /// A mutable reference.
-    Mut(&'a mut T),
-}
-
-impl<'a, T: 'a> ::std::ops::Deref for RefOrMut<'a, T> {
-    type Target = T;
-    fn deref(&self) -> &Self::Target {
-        match self {
-            RefOrMut::Ref(reference) => reference,
-            RefOrMut::Mut(reference) => reference,
-        }
-    }
-}
-
-impl<'a, T: 'a> ::std::borrow::Borrow<T> for RefOrMut<'a, T> {
-    fn borrow(&self) -> &T {
-        match self {
-            RefOrMut::Ref(reference) => reference,
-            RefOrMut::Mut(reference) => reference,
-        }
-    }
-}
-
-impl<'a, T: Clone+'a> RefOrMut<'a, T> {
-    /// Extracts the contents of `self`, either by cloning or swapping.
-    ///
-    /// This consumes `self` because its contents are now in an unknown state.
-    pub fn swap<'b>(self, element: &'b mut T) {
-        match self {
-            RefOrMut::Ref(reference) => element.clone_from(reference),
-            RefOrMut::Mut(reference) => ::std::mem::swap(reference, element),
-        };
-    }
-    /// Extracts the contents of `self`, either by cloning or swapping.
-    ///
-    /// This consumes `self` because its contents are now in an unknown state.
-    pub fn replace(self, mut element: T) -> T {
-        self.swap(&mut element);
-        element
-    }
-
-    /// Extracts the contents of `self`, either by cloning, or swapping and leaving a default
-    /// element in place.
-    ///
-    /// This consumes `self` because its contents are now in an unknown state.
-    pub fn take(self) -> T where T: Default {
-        let mut element = Default::default();
-        self.swap(&mut element);
-        element
-    }
-}
-
-/// A wrapped message which may be either typed or binary data.
+/// A wrapped message which supports serialization and deserialization.
 pub struct Message<T> {
-    payload: MessageContents<T>,
-}
-
-/// Possible returned representations from a channel.
-enum MessageContents<T> {
-    /// Rust typed instance. Available for ownership.
-    Owned(T),
-    /// Atomic reference counted. Only available as a reference.
-    Arc(Arc<T>),
+    /// Message contents.
+    pub payload: T,
 }
 
 impl<T> Message<T> {
     /// Wrap a typed item as a message.
     pub fn from_typed(typed: T) -> Self {
-        Message { payload: MessageContents::Owned(typed) }
-    }
-    /// Wrap a shared typed item as a message.
-    pub fn from_arc(typed: Arc<T>) -> Self {
-        Message { payload: MessageContents::Arc(typed) }
-    }
-    /// Destructures and returns any typed data.
-    pub fn if_typed(self) -> Option<T> {
-        match self.payload {
-            MessageContents::Owned(typed) => Some(typed),
-            MessageContents::Arc(_) => None,
-        }
-    }
-    /// Returns a mutable reference, if typed.
-    pub fn if_mut(&mut self) -> Option<&mut T> {
-        match &mut self.payload {
-            MessageContents::Owned(typed) => Some(typed),
-            MessageContents::Arc(_) => None,
-        }
-    }
-    /// Returns an immutable or mutable typed reference.
-    ///
-    /// This method returns a mutable reference if the underlying data are typed Rust
-    /// instances, which admit mutation, and it returns an immutable reference if the
-    /// data are serialized binary data.
-    pub fn as_ref_or_mut(&mut self) -> RefOrMut<T> {
-        match &mut self.payload {
-            MessageContents::Owned(typed) => { RefOrMut::Mut(typed) },
-            MessageContents::Arc(typed) => { RefOrMut::Ref(typed) },
-        }
+        Message { payload: typed }
     }
 }
 
 impl<T: Data> Message<T> {
     /// Wrap bytes as a message.
     pub fn from_bytes(bytes: Bytes) -> Self {
         let typed = ::bincode::deserialize(&bytes[..]).expect("bincode::deserialize() failed");
-        Message { payload: MessageContents::Owned(typed) }
+        Message { payload: typed }
     }
 
     /// The number of bytes required to serialize the data.
     pub fn length_in_bytes(&self) -> usize {
-        match &self.payload {
-            MessageContents::Owned(typed) => {
-                ::bincode::serialized_size(&typed).expect("bincode::serialized_size() failed") as usize
-            },
-            MessageContents::Arc(typed) => {
-                ::bincode::serialized_size(&**typed).expect("bincode::serialized_size() failed") as usize
-            },
-        }
+        ::bincode::serialized_size(&self.payload).expect("bincode::serialized_size() failed") as usize
     }
 
     /// Writes the binary representation into `writer`.
     pub fn into_bytes<W: ::std::io::Write>(&self, writer: &mut W) {
-        match &self.payload {
-            MessageContents::Owned(typed) => {
-                ::bincode::serialize_into(writer, &typed).expect("bincode::serialize_into() failed");
-            },
-            MessageContents::Arc(typed) => {
-                ::bincode::serialize_into(writer, &**typed).expect("bincode::serialize_into() failed");
-            },
-        }
+        ::bincode::serialize_into(writer, &self.payload).expect("bincode::serialize_into() failed");
     }
 }
 
 impl<T> ::std::ops::Deref for Message<T> {
     type Target = T;
     fn deref(&self) -> &Self::Target {
-        // TODO: In principle we have already decoded, but let's go again
-        match &self.payload {
-            MessageContents::Owned(typed) => { typed },
-            MessageContents::Arc(typed) => { typed },
-        }
+        &self.payload
     }
 }
-
-impl<T: Clone> Message<T> {
-    /// Produces a typed instance of the wrapped element.
-    pub fn into_typed(self) -> T {
-        match self.payload {
-            MessageContents::Owned(instance) => instance,
-            // TODO: Could attempt `Arc::try_unwrap()` here.
-            MessageContents::Arc(instance) => (*instance).clone(),
-        }
-    }
-    /// Ensures the message is typed data and returns a mutable reference to it.
-    pub fn as_mut(&mut self) -> &mut T {
-
-        let cloned: Option<T> = match &self.payload {
-            MessageContents::Owned(_) => None,
-            // TODO: Could attempt `Arc::try_unwrap()` here.
-            MessageContents::Arc(typed) => Some((**typed).clone()),
-        };
-
-        if let Some(cloned) = cloned {
-            self.payload = MessageContents::Owned(cloned);
-        }
-
-        if let MessageContents::Owned(typed) = &mut self.payload {
-            typed
-        }
-        else {
-            unreachable!()
-        }
-    }
-}

mdbook/src/chapter_2/chapter_2_4.md

@@ -19,12 +19,10 @@ fn main() {
             .to_stream(scope)
             .unary(Pipeline, "increment", |capability, info| {
 
-                let mut vector = Vec::new();
                 move |input, output| {
                     while let Some((time, data)) = input.next() {
-                        data.swap(&mut vector);
                         let mut session = output.session(&time);
-                        for datum in vector.drain(..) {
+                        for datum in data.drain(..) {
                             session.give(datum + 1);
                         }
                     }
@@ -136,13 +134,11 @@ fn main() {
             .unary(Pipeline, "increment", |capability, info| {
 
                 let mut maximum = 0;    // define this here; use in the closure
-                let mut vector = Vec::new();
 
                 move |input, output| {
                     while let Some((time, data)) = input.next() {
-                        data.swap(&mut vector);
                         let mut session = output.session(&time);
-                        for datum in vector.drain(..) {
+                        for datum in data.drain(..) {
                             if datum > maximum {
                                 session.give(datum + 1);
                                 maximum = datum;
@@ -195,13 +191,13 @@ fn main() {
                 while let Some((time, data)) = input1.next() {
                     stash.entry(time.time().clone())
                          .or_insert(Vec::new())
-                         .push(data.replace(Vec::new()));
+                         .push(std::mem::take(data));
                     notificator.notify_at(time.retain());
                 }
                 while let Some((time, data)) = input2.next() {
                     stash.entry(time.time().clone())
                          .or_insert(Vec::new())
-                         .push(data.replace(Vec::new()));
+                         .push(std::mem::take(data));
                     notificator.notify_at(time.retain());
                 }
 
@@ -246,12 +242,12 @@ fn main() {
                 while let Some((time, data)) = input1.next() {
                     stash.entry(time.retain())
                          .or_insert(Vec::new())
-                         .push(data.replace(Vec::new()));
+                         .push(std::mem::take(data));
                 }
                 while let Some((time, data)) = input2.next() {
                     stash.entry(time.retain())
                          .or_insert(Vec::new())
-                         .push(data.replace(Vec::new()));
+                         .push(std::mem::take(data));
                 }
 
                 // consider sending everything in `stash`.

mdbook/src/chapter_2/chapter_2_5.md

@@ -212,7 +212,7 @@ As before, I'm just going to show you the new code, which now lives just after `
                         while let Some((time, data)) = input.next() {
                             queues.entry(time.retain())
                                   .or_insert(Vec::new())
-                                  .extend(data.replace(Vec::new()));
+                                  .extend(std::mem::take(data));
                         }
 
                         // enable each stashed time if ready.
@@ -297,7 +297,7 @@ Inside the closure, we do two things: (i) read inputs and (ii) update counts and
         while let Some((time, data)) = input.next() {
             queues.entry(time.retain())
                   .or_insert(Vec::new())
-                  .extend(data.replace(Vec::new()));
+                  .extend(std::mem::take(data));
         }
 ```
 

mdbook/src/chapter_4/chapter_4_3.md

@@ -78,16 +78,13 @@ fn main() {
             // Buffer records until all prior timestamps have completed.
             .binary_frontier(&cycle, Pipeline, Pipeline, "Buffer", move |capability, info| {
 
-                let mut vector = Vec::new();
-
                 move |input1, input2, output| {
 
                     // Stash received data.
                     input1.for_each(|time, data| {
-                        data.swap(&mut vector);
                         stash.entry(time.retain())
                              .or_insert(Vec::new())
-                             .extend(vector.drain(..));
+                             .extend(data.drain(..));
                     });
 
                     // Consider sending stashed data.

timely/examples/bfs.rs

@@ -58,7 +58,7 @@ fn main() {
                     // receive edges, start to sort them
                     input1.for_each(|time, data| {
                         notify.notify_at(time.retain());
-                        edge_list.push(data.replace(Vec::new()));
+                        edge_list.push(std::mem::take(data));
                     });
 
                     // receive (node, worker) pairs, note any new ones.
@@ -68,7 +68,7 @@ fn main() {
                                       notify.notify_at(time.retain());
                                       Vec::new()
                                   })
-                                  .push(data.replace(Vec::new()));
+                                  .push(std::mem::take(data));
                     });
 
                     notify.for_each(|time, _num, _notify| {

timely/examples/flatcontainer.rs

@@ -52,7 +52,7 @@ fn main() {
                                 queues
                                     .entry(time.retain())
                                     .or_insert(Vec::new())
-                                    .push(data.take());
+                                    .push(std::mem::take(data));
                             }
 
                             for (key, val) in queues.iter_mut() {

timely/examples/hashjoin.rs

@@ -40,16 +40,12 @@ fn main() {
                     let mut map1 = HashMap::<u64, Vec<u64>>::new();
                     let mut map2 = HashMap::<u64, Vec<u64>>::new();
 
-                    let mut vector1 = Vec::new();
-                    let mut vector2 = Vec::new();
-
                     move |input1, input2, output| {
 
                         // Drain first input, check second map, update first map.
                         input1.for_each(|time, data| {
-                            data.swap(&mut vector1);
                             let mut session = output.session(&time);
-                            for (key, val1) in vector1.drain(..) {
+                            for (key, val1) in data.drain(..) {
                                 if let Some(values) = map2.get(&key) {
                                     for val2 in values.iter() {
                                         session.give((val1.clone(), val2.clone()));
@@ -62,9 +58,8 @@ fn main() {
 
                         // Drain second input, check first map, update second map.
                         input2.for_each(|time, data| {
-                            data.swap(&mut vector2);
                             let mut session = output.session(&time);
-                            for (key, val2) in vector2.drain(..) {
+                            for (key, val2) in data.drain(..) {
                                 if let Some(values) = map1.get(&key) {
                                     for val1 in values.iter() {
                                         session.give((val1.clone(), val2.clone()));

timely/examples/pagerank.rs

@@ -42,23 +42,18 @@ fn main() {
                     let mut diffs = Vec::new(); // for received but un-acted upon deltas.
                     let mut delta = Vec::new();
 
-                    let mut edge_vec = Vec::new();
-                    let mut rank_vec = Vec::new();
-
                     let timer = ::std::time::Instant::now();
 
                     move |input1, input2, output| {
 
                         // hold on to edge changes until it is time.
                         input1.for_each(|time, data| {
-                            data.swap(&mut edge_vec);
-                            edge_stash.entry(time.retain()).or_insert(Vec::new()).extend(edge_vec.drain(..));
+                            edge_stash.entry(time.retain()).or_insert(Vec::new()).extend(data.drain(..));
                         });
 
                         // hold on to rank changes until it is time.
                         input2.for_each(|time, data| {
-                            data.swap(&mut rank_vec);
-                            rank_stash.entry(time.retain()).or_insert(Vec::new()).extend(rank_vec.drain(..));
+                            rank_stash.entry(time.retain()).or_insert(Vec::new()).extend(data.drain(..));
                         });
 
                         let frontiers = &[input1.frontier(), input2.frontier()];

timely/examples/wordcount.rs

@@ -35,7 +35,7 @@ fn main() {
                         while let Some((time, data)) = input.next() {
                             queues.entry(time.retain())
                                   .or_insert(Vec::new())
-                                  .push(data.replace(Vec::new()));
+                                  .push(std::mem::take(data));
                         }
 
                         for (key, val) in queues.iter_mut() {

timely/src/dataflow/channels/mod.rs

@@ -53,10 +53,8 @@ impl<T, C: Container> Message<T, C> {
         pusher.push(&mut bundle);
 
         if let Some(message) = bundle {
-            if let Some(message) = message.if_typed() {
-                *buffer = message.data;
-                buffer.clear();
-            }
+            *buffer = message.payload.data;
+            buffer.clear();
         }
     }
 }