Move opinions about encoding from communication to timely

Author: frankmcsherry

communication/Cargo.toml

@@ -18,7 +18,6 @@ default = ["getopts"]
 
 [dependencies]
 getopts = { version = "0.2.14", optional = true }
-bincode = { version = "1.0" }
 byteorder = "1.5"
 serde = { version = "1.0", features = ["derive"] }
 timely_bytes = { path = "../bytes", version = "0.12" }

communication/examples/comm_hello.rs

@@ -1,7 +1,26 @@
 extern crate timely_communication;
 
-use std::ops::Deref;
-use timely_communication::{Message, Allocate};
+use timely_communication::{Allocate, Bytesable};
+
+/// A wrapper that indicates `bincode` as the serialization/deserialization strategy.
+pub struct Message {
+    /// Text contents.
+    pub payload: String,
+}
+
+impl Bytesable for Message {
+    fn from_bytes(bytes: timely_bytes::arc::Bytes) -> Self {
+        Message { payload: std::str::from_utf8(&bytes[..]).unwrap().to_string() }
+    }
+
+    fn length_in_bytes(&self) -> usize {
+        self.payload.len()
+    }
+
+    fn into_bytes<W: ::std::io::Write>(&self, writer: &mut W) {
+        writer.write_all(self.payload.as_bytes()).unwrap();
+    }
+}
 
 fn main() {
 
@@ -16,7 +35,7 @@ fn main() {
 
         // send typed data along each channel
         for i in 0 .. allocator.peers() {
-            senders[i].send(Message::from_typed(format!("hello, {}", i)));
+            senders[i].send(Message { payload: format!("hello, {}", i)});
             senders[i].done();
         }
 
@@ -28,7 +47,7 @@ fn main() {
             allocator.receive();
 
             if let Some(message) = receiver.recv() {
-                println!("worker {}: received: <{}>", allocator.index(), message.deref());
+                println!("worker {}: received: <{}>", allocator.index(), message.payload);
                 received += 1;
             }
 

communication/src/allocator/mod.rs

@@ -17,7 +17,7 @@ pub mod counters;
 
 pub mod zero_copy;
 
-use crate::{Push, Pull};
+use crate::{Bytesable, Push, Pull};
 
 /// A proto-allocator, which implements `Send` and can be completed with `build`.
 ///
@@ -33,11 +33,10 @@ pub trait AllocateBuilder : Send {
 }
 
 use std::any::Any;
-use crate::message::Bytesable;
 
 /// A type that can be sent along an allocated channel.
-pub trait Exchangeable : Send+Sync+Any+Bytesable+'static { }
-impl<T: Send+Sync+Any+Bytesable+'static> Exchangeable for T { }
+pub trait Exchangeable : Send+Any+Bytesable+'static { }
+impl<T: Send+Any+Bytesable+'static> Exchangeable for T { }
 
 /// A type capable of allocating channels.
 ///

communication/src/allocator/process.rs

@@ -110,7 +110,7 @@ impl Process {
 impl Allocate for Process {
     fn index(&self) -> usize { self.index }
     fn peers(&self) -> usize { self.peers }
-    fn allocate<T: Any+Send+Sync+'static>(&mut self, identifier: usize) -> (Vec<Box<dyn Push<T>>>, Box<dyn Pull<T>>) {
+    fn allocate<T: Any+Send+'static>(&mut self, identifier: usize) -> (Vec<Box<dyn Push<T>>>, Box<dyn Pull<T>>) {
 
         // this is race-y global initialisation of all channels for all workers, performed by the
         // first worker that enters this critical section

communication/src/allocator/zero_copy/push_pull.rs

@@ -8,8 +8,7 @@ use bytes::arc::Bytes;
 
 use crate::allocator::canary::Canary;
 use crate::networking::MessageHeader;
-use crate::message::Bytesable;
-use crate::{Push, Pull};
+use crate::{Bytesable, Push, Pull};
 
 use super::bytes_exchange::{BytesPush, SendEndpoint};
 

communication/src/initialize.rs

@@ -182,47 +182,71 @@ impl Config {
 ///
 /// # Examples
 /// ```
-/// use timely_communication::Allocate;
-///
-/// // configure for two threads, just one process.
-/// let config = timely_communication::Config::Process(2);
-///
-/// // initializes communication, spawns workers
-/// let guards = timely_communication::initialize(config, |mut allocator| {
-///     println!("worker {} started", allocator.index());
-///
-///     // allocates a pair of senders list and one receiver.
-///     let (mut senders, mut receiver) = allocator.allocate(0);
-///
-///     // send typed data along each channel
-///     use timely_communication::Message;
-///     senders[0].send(Message::from_typed(format!("hello, {}", 0)));
-///     senders[1].send(Message::from_typed(format!("hello, {}", 1)));
-///
-///     // no support for termination notification,
-///     // we have to count down ourselves.
-///     let mut expecting = 2;
-///     while expecting > 0 {
-///         allocator.receive();
-///         if let Some(message) = receiver.recv() {
-///             use std::ops::Deref;
-///             println!("worker {}: received: <{}>", allocator.index(), message.deref());
-///             expecting -= 1;
-///         }
-///         allocator.release();
+/// use timely_communication::{Allocate, Bytesable};
+/// 
+/// /// A wrapper that indicates `bincode` as the serialization/deserialization strategy.
+/// pub struct Message {
+///     /// Text contents.
+///     pub payload: String,
+/// }
+/// 
+/// impl Bytesable for Message {
+///     fn from_bytes(bytes: timely_bytes::arc::Bytes) -> Self {
+///         Message { payload: std::str::from_utf8(&bytes[..]).unwrap().to_string() }
 ///     }
-///
-///     // optionally, return something
-///     allocator.index()
-/// });
-///
-/// // computation runs until guards are joined or dropped.
-/// if let Ok(guards) = guards {
-///     for guard in guards.join() {
-///         println!("result: {:?}", guard);
+/// 
+///     fn length_in_bytes(&self) -> usize {
+///         self.payload.len()
+///     }
+/// 
+///     fn into_bytes<W: ::std::io::Write>(&self, writer: &mut W) {
+///         writer.write_all(self.payload.as_bytes()).unwrap();
+///     }
+/// }
+/// 
+/// fn main() {
+/// 
+///     // extract the configuration from user-supplied arguments, initialize the computation.
+///     let config = timely_communication::Config::from_args(std::env::args()).unwrap();
+///     let guards = timely_communication::initialize(config, |mut allocator| {
+/// 
+///         println!("worker {} of {} started", allocator.index(), allocator.peers());
+/// 
+///         // allocates a pair of senders list and one receiver.
+///         let (mut senders, mut receiver) = allocator.allocate(0);
+/// 
+///         // send typed data along each channel
+///         for i in 0 .. allocator.peers() {
+///             senders[i].send(Message { payload: format!("hello, {}", i)});
+///             senders[i].done();
+///         }
+/// 
+///         // no support for termination notification,
+///         // we have to count down ourselves.
+///         let mut received = 0;
+///         while received < allocator.peers() {
+/// 
+///             allocator.receive();
+/// 
+///             if let Some(message) = receiver.recv() {
+///                 println!("worker {}: received: <{}>", allocator.index(), message.payload);
+///                 received += 1;
+///             }
+/// 
+///             allocator.release();
+///         }
+/// 
+///         allocator.index()
+///     });
+/// 
+///     // computation runs until guards are joined or dropped.
+///     if let Ok(guards) = guards {
+///         for guard in guards.join() {
+///             println!("result: {:?}", guard);
+///         }
 ///     }
+///     else { println!("error in computation"); }
 /// }
-/// else { println!("error in computation"); }
 /// ```
 ///
 /// The should produce output like:
@@ -254,47 +278,71 @@ pub fn initialize<T:Send+'static, F: Fn(Generic)->T+Send+Sync+'static>(
 ///
 /// # Examples
 /// ```
-/// use timely_communication::Allocate;
-///
-/// // configure for two threads, just one process.
-/// let builders = timely_communication::allocator::process::Process::new_vector(2);
-///
-/// // initializes communication, spawns workers
-/// let guards = timely_communication::initialize_from(builders, Box::new(()), |mut allocator| {
-///     println!("worker {} started", allocator.index());
-///
-///     // allocates a pair of senders list and one receiver.
-///     let (mut senders, mut receiver) = allocator.allocate(0);
-///
-///     // send typed data along each channel
-///     use timely_communication::Message;
-///     senders[0].send(Message::from_typed(format!("hello, {}", 0)));
-///     senders[1].send(Message::from_typed(format!("hello, {}", 1)));
-///
-///     // no support for termination notification,
-///     // we have to count down ourselves.
-///     let mut expecting = 2;
-///     while expecting > 0 {
-///         allocator.receive();
-///         if let Some(message) = receiver.recv() {
-///             use std::ops::Deref;
-///             println!("worker {}: received: <{}>", allocator.index(), message.deref());
-///             expecting -= 1;
-///         }
-///         allocator.release();
+/// use timely_communication::{Allocate, Bytesable};
+/// 
+/// /// A wrapper that indicates `bincode` as the serialization/deserialization strategy.
+/// pub struct Message {
+///     /// Text contents.
+///     pub payload: String,
+/// }
+/// 
+/// impl Bytesable for Message {
+///     fn from_bytes(bytes: timely_bytes::arc::Bytes) -> Self {
+///         Message { payload: std::str::from_utf8(&bytes[..]).unwrap().to_string() }
 ///     }
-///
-///     // optionally, return something
-///     allocator.index()
-/// });
-///
-/// // computation runs until guards are joined or dropped.
-/// if let Ok(guards) = guards {
-///     for guard in guards.join() {
-///         println!("result: {:?}", guard);
+/// 
+///     fn length_in_bytes(&self) -> usize {
+///         self.payload.len()
+///     }
+/// 
+///     fn into_bytes<W: ::std::io::Write>(&self, writer: &mut W) {
+///         writer.write_all(self.payload.as_bytes()).unwrap();
+///     }
+/// }
+/// 
+/// fn main() {
+/// 
+///     // extract the configuration from user-supplied arguments, initialize the computation.
+///     let config = timely_communication::Config::from_args(std::env::args()).unwrap();
+///     let guards = timely_communication::initialize(config, |mut allocator| {
+/// 
+///         println!("worker {} of {} started", allocator.index(), allocator.peers());
+/// 
+///         // allocates a pair of senders list and one receiver.
+///         let (mut senders, mut receiver) = allocator.allocate(0);
+/// 
+///         // send typed data along each channel
+///         for i in 0 .. allocator.peers() {
+///             senders[i].send(Message { payload: format!("hello, {}", i)});
+///             senders[i].done();
+///         }
+/// 
+///         // no support for termination notification,
+///         // we have to count down ourselves.
+///         let mut received = 0;
+///         while received < allocator.peers() {
+/// 
+///             allocator.receive();
+/// 
+///             if let Some(message) = receiver.recv() {
+///                 println!("worker {}: received: <{}>", allocator.index(), message.payload);
+///                 received += 1;
+///             }
+/// 
+///             allocator.release();
+///         }
+/// 
+///         allocator.index()
+///     });
+/// 
+///     // computation runs until guards are joined or dropped.
+///     if let Ok(guards) = guards {
+///         for guard in guards.join() {
+///             println!("result: {:?}", guard);
+///         }
 ///     }
+///     else { println!("error in computation"); }
 /// }
-/// else { println!("error in computation"); }
 /// ```
 pub fn initialize_from<A, T, F>(
     builders: Vec<A>,