Dataplane mockup

Cargo.toml

@@ -18,9 +18,11 @@ license = "MIT"
 [dependencies]
 abomonation = "0.5"
 abomonation_derive = "0.3"
-timely_communication = "0.5"
+#timely_communication = "0.5"
+timely_communication = { path = "./communication"}
 byteorder="1"
 time="0.1.34"
+bytes = { path = "./bytes" }
 
 [dev-dependencies]
 timely_sort="0.1.6"

bytes/src/lib.rs

@@ -25,7 +25,7 @@
 //! drop(shared2);
 //! drop(shared3);
 //!
-//! if let Ok(bytes) = shared4.try_recover() {
+//! if let Ok(bytes) = shared4.try_recover::<Vec<u8>>() {
 //!     assert_eq!(bytes[200..1024].to_vec(), [1u8;824].to_vec());
 //!     assert_eq!(bytes[60..100].to_vec(),   [2u8;40].to_vec());
 //!     assert_eq!(bytes[100..200].to_vec(),  [3u8;100].to_vec());
@@ -42,9 +42,10 @@ pub mod rc {
 
     use std::ops::{Deref, DerefMut};
     use std::rc::Rc;
+    use std::any::Any;
 
     /// A thread-local byte buffer backed by a shared allocation.
-    pub struct Bytes<B: DerefMut<Target=[u8]>> {
+    pub struct Bytes {
         /// Pointer to the start of this slice (not the allocation).
         ptr: *mut u8,
         /// Length of this slice.
@@ -54,20 +55,24 @@ pub mod rc {
         /// Importantly, this is unavailable for as long as the struct exists, which may
         /// prevent shared access to ptr[0 .. len]. I'm not sure I understand Rust's rules
         /// enough to make a strong statement about this.
-        sequestered: Rc<B>,
+        sequestered: Rc<Box<Any>>,
     }
 
-    impl<B: DerefMut<Target=[u8]>> Bytes<B> {
+    impl Bytes {
 
         /// Create a new instance from a byte allocation.
-        pub fn from(bytes: B) -> Bytes<B> {
+        pub fn from<B>(bytes: B) -> Bytes where B: DerefMut<Target=[u8]>+'static {
 
-            let mut rc = Rc::new(bytes);
+            let mut boxed = Box::new(bytes) as Box<Any>;
+
+            let ptr = boxed.downcast_mut::<B>().unwrap().as_mut_ptr();
+            let len = boxed.downcast_ref::<B>().unwrap().len();
+            let sequestered = Rc::new(boxed);
 
             Bytes {
-                ptr: Rc::get_mut(&mut rc).unwrap().as_mut_ptr(),
-                len: rc.len(),
-                sequestered: rc,
+                ptr,
+                len,
+                sequestered,
             }
         }
 
@@ -77,7 +82,7 @@ pub mod rc {
         ///
         /// This method uses an `unsafe` region to advance the pointer by `index`. It first
         /// tests `index` against `self.len`, which should ensure that the offset is in-bounds.
-        pub fn extract_to(&mut self, index: usize) -> Bytes<B> {
+        pub fn extract_to(&mut self, index: usize) -> Bytes {
 
             assert!(index <= self.len);
 
@@ -97,9 +102,9 @@ pub mod rc {
         ///
         /// This method either results in the underlying storage if it is uniquely held, or the
         /// input `Bytes` if it is not uniquely held.
-        pub fn try_recover(self) -> Result<B, Bytes<B>> {
+        pub fn try_recover<B>(self) -> Result<B, Bytes> where B: DerefMut<Target=[u8]>+'static {
             match Rc::try_unwrap(self.sequestered) {
-                Ok(bytes) => Ok(bytes),
+                Ok(bytes) => Ok(*bytes.downcast::<B>().unwrap()),
                 Err(rc) => Err(Bytes {
                     ptr: self.ptr,
                     len: self.len,
@@ -109,14 +114,14 @@ pub mod rc {
         }
     }
 
-    impl<B: DerefMut<Target=[u8]>> Deref for Bytes<B> {
+    impl Deref for Bytes {
         type Target = [u8];
         fn deref(&self) -> &[u8] {
             unsafe { ::std::slice::from_raw_parts(self.ptr, self.len) }
         }
     }
 
-    impl<B: DerefMut<Target=[u8]>> DerefMut for Bytes<B> {
+    impl DerefMut for Bytes {
         fn deref_mut(&mut self) -> &mut [u8] {
             unsafe { ::std::slice::from_raw_parts_mut(self.ptr, self.len) }
         }
@@ -128,9 +133,10 @@ pub mod arc {
 
     use std::ops::{Deref, DerefMut};
     use std::sync::Arc;
+    use std::any::Any;
 
     /// A thread-safe byte buffer backed by a shared allocation.
-    pub struct Bytes<B: DerefMut<Target=[u8]>> {
+    pub struct Bytes {
         /// Pointer to the start of this slice (not the allocation).
         ptr: *mut u8,
         /// Length of this slice.
@@ -140,20 +146,26 @@ pub mod arc {
         /// Importantly, this is unavailable for as long as the struct exists, which may
         /// prevent shared access to ptr[0 .. len]. I'm not sure I understand Rust's rules
         /// enough to make a strong statement about this.
-        sequestered: Arc<B>,
+        sequestered: Arc<Box<Any>>,
     }
 
-    impl<B: DerefMut<Target=[u8]>> Bytes<B> {
+    unsafe impl Send for Bytes { }
+
+    impl Bytes {
 
         /// Create a new instance from a byte allocation.
-        pub fn from(bytes: B) -> Bytes<B> {
+        pub fn from<B>(bytes: B) -> Bytes where B : DerefMut<Target=[u8]>+'static {
+
+            let mut boxed = Box::new(bytes) as Box<Any>;
 
-            let mut arc = Arc::new(bytes);
+            let ptr = boxed.downcast_mut::<B>().unwrap().as_mut_ptr();
+            let len = boxed.downcast_ref::<B>().unwrap().len();
+            let sequestered = Arc::new(boxed);
 
             Bytes {
-                ptr: Arc::get_mut(&mut arc).unwrap().as_mut_ptr(),
-                len: arc.len(),
-                sequestered: arc,
+                ptr,
+                len,
+                sequestered,
             }
         }
 
@@ -163,7 +175,7 @@ pub mod arc {
         ///
         /// This method uses an `unsafe` region to advance the pointer by `index`. It first
         /// tests `index` against `self.len`, which should ensure that the offset is in-bounds.
-        pub fn extract_to(&mut self, index: usize) -> Bytes<B> {
+        pub fn extract_to(&mut self, index: usize) -> Bytes {
 
             assert!(index <= self.len);
 
@@ -183,9 +195,10 @@ pub mod arc {
         ///
         /// This method either results in the underlying storage if it is uniquely held, or the
         /// input `Bytes` if it is not uniquely held.
-        pub fn try_recover(self) -> Result<B, Bytes<B>> {
+        pub fn try_recover<B>(self) -> Result<B, Bytes> where B: DerefMut<Target=[u8]>+'static {
+            // println!("Trying recovery; strong count: {:?}", Arc::strong_count(&self.sequestered));
             match Arc::try_unwrap(self.sequestered) {
-                Ok(bytes) => Ok(bytes),
+                Ok(bytes) => Ok(*bytes.downcast::<B>().unwrap()),
                 Err(arc) => Err(Bytes {
                     ptr: self.ptr,
                     len: self.len,
@@ -195,14 +208,14 @@ pub mod arc {
         }
     }
 
-    impl<B: DerefMut<Target=[u8]>> Deref for Bytes<B> {
+    impl Deref for Bytes {
         type Target = [u8];
         fn deref(&self) -> &[u8] {
             unsafe { ::std::slice::from_raw_parts(self.ptr, self.len) }
         }
     }
 
-    impl<B: DerefMut<Target=[u8]>> DerefMut for Bytes<B> {
+    impl DerefMut for Bytes {
         fn deref_mut(&mut self) -> &mut [u8] {
             unsafe { ::std::slice::from_raw_parts_mut(self.ptr, self.len) }
         }

communication/Cargo.toml

@@ -22,6 +22,7 @@ getopts={version="0.2.14", optional=true}
 time="0.1.34"
 abomonation = "0.5"
 abomonation_derive = "0.3"
+bytes = { path = "../bytes" }
 
 [[examples]]
 features = "arg_parse"

communication/examples/hello.rs

@@ -22,10 +22,15 @@ fn main() {
         // we have to count down ourselves.
         let mut received = 0;
         while received < allocator.peers() {
+
+            allocator.pre_work();
+
             if let Some(message) = receiver.recv() {
                 println!("worker {}: received: <{}>", allocator.index(), message);
                 received += 1;
             }
+
+            allocator.post_work();
         }
 
         allocator.index()

communication/src/allocator/binary.rs

@@ -139,13 +139,15 @@ impl<T:Data> Pull<T> for Puller<T> {
         let log_sender = &self.log_sender;
         if inner.is_some() { inner }
         else {
-            self.current = self.receiver.try_recv().ok().map(|mut bytes| {
+            self.current = self.receiver.try_recv().ok().map(|bytes| {
                 log_sender.when_enabled(|l| l.log(
                     ::logging::CommsEvent::Serialization(::logging::SerializationEvent {
                         seq_no: None,
                         is_start: true,
                     })));
-                let result = <T as Serialize>::from_bytes(&mut bytes);
+
+                let bytes = ::bytes::arc::Bytes::from(bytes);
+                let result = <T as Serialize>::from_bytes(bytes);
                 log_sender.when_enabled(|l| l.log(
                     ::logging::CommsEvent::Serialization(::logging::SerializationEvent {
                         seq_no: None,

communication/src/allocator/generic.rs

@@ -4,6 +4,7 @@
 //! for example closures whose type arguments must be specified.
 
 use allocator::{Allocate, Thread, Process, Binary};
+use allocator::process_binary::{ProcessBinary, ProcessBinaryBuilder};
 use {Push, Pull, Data};
 
 /// Enumerates known implementors of `Allocate`.
@@ -12,31 +13,52 @@ pub enum Generic {
     Thread(Thread),
     Process(Process),
     Binary(Binary),
+    ProcessBinary(ProcessBinary<::allocator::process_binary::vec::VecBytesExchange>),
 }
 
 impl Generic {
     /// The index of the worker out of `(0..self.peers())`.
     pub fn index(&self) -> usize {
-        match *self {
-            Generic::Thread(ref t) => t.index(),
-            Generic::Process(ref p) => p.index(),
-            Generic::Binary(ref b) => b.index(),
+        match self {
+            &Generic::Thread(ref t) => t.index(),
+            &Generic::Process(ref p) => p.index(),
+            &Generic::Binary(ref b) => b.index(),
+            &Generic::ProcessBinary(ref pb) => pb.index(),
         }
     }
     /// The number of workers.
     pub fn peers(&self) -> usize {
-        match *self {
-            Generic::Thread(ref t) => t.peers(),
-            Generic::Process(ref p) => p.peers(),
-            Generic::Binary(ref b) => b.peers(),
+        match self {
+            &Generic::Thread(ref t) => t.peers(),
+            &Generic::Process(ref p) => p.peers(),
+            &Generic::Binary(ref b) => b.peers(),
+            &Generic::ProcessBinary(ref pb) => pb.peers(),
         }
     }
     /// Constructs several send endpoints and one receive endpoint.
     pub fn allocate<T: Data>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>, Option<usize>) {
-        match *self {
-            Generic::Thread(ref mut t) => t.allocate(),
-            Generic::Process(ref mut p) => p.allocate(),
-            Generic::Binary(ref mut b) => b.allocate(),
+        match self {
+            &mut Generic::Thread(ref mut t) => t.allocate(),
+            &mut Generic::Process(ref mut p) => p.allocate(),
+            &mut Generic::Binary(ref mut b) => b.allocate(),
+            &mut Generic::ProcessBinary(ref mut pb) => pb.allocate(),
+        }
+    }
+
+    pub fn pre_work(&mut self) {
+        match self {
+            &mut Generic::Thread(ref mut t) => t.pre_work(),
+            &mut Generic::Process(ref mut p) => p.pre_work(),
+            &mut Generic::Binary(ref mut b) => b.pre_work(),
+            &mut Generic::ProcessBinary(ref mut pb) => pb.pre_work(),
+        }
+    }
+    pub fn post_work(&mut self) {
+        match self {
+            &mut Generic::Thread(ref mut t) => t.post_work(),
+            &mut Generic::Process(ref mut p) => p.post_work(),
+            &mut Generic::Binary(ref mut b) => b.post_work(),
+            &mut Generic::ProcessBinary(ref mut pb) => pb.post_work(),
         }
     }
 }
@@ -47,4 +69,31 @@ impl Allocate for Generic {
     fn allocate<T: Data>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>, Option<usize>) {
         self.allocate()
     }
+
+    fn pre_work(&mut self) { self.pre_work(); }
+    fn post_work(&mut self) { self.post_work(); }
+}
+
+
+/// Enumerations of constructable implementors of `Allocate`.
+///
+/// The builder variants are meant to be `Send`, so that they can be moved across threads,
+/// whereas the allocator they construct may not. As an example, the `ProcessBinary` type
+/// contains `Rc` wrapped state, and so cannot itself be moved across threads.
+pub enum GenericBuilder {
+    Thread(Thread),
+    Process(Process),
+    Binary(Binary),
+    ProcessBinary(ProcessBinaryBuilder<::allocator::process_binary::vec::VecBytesExchange>),
+}
+
+impl GenericBuilder {
+    pub fn build(self) -> Generic {
+        match self {
+            GenericBuilder::Thread(t) => Generic::Thread(t),
+            GenericBuilder::Process(p) => Generic::Process(p),
+            GenericBuilder::Binary(b) => Generic::Binary(b),
+            GenericBuilder::ProcessBinary(pb) => Generic::ProcessBinary(pb.build()),
+        }
+    }
 }

communication/src/allocator/mod.rs

@@ -3,12 +3,13 @@
 pub use self::thread::Thread;
 pub use self::process::Process;
 pub use self::binary::Binary;
-pub use self::generic::Generic;
+pub use self::generic::{Generic, GenericBuilder};
 
 pub mod thread;
 pub mod process;
 pub mod binary;
 pub mod generic;
+pub mod process_binary;
 
 use {Data, Push, Pull};
 
@@ -22,4 +23,7 @@ pub trait Allocate {
     fn peers(&self) -> usize;
     /// Constructs several send endpoints and one receive endpoint.
     fn allocate<T: Data>(&mut self) -> (Vec<Box<Push<T>>>, Box<Pull<T>>, Option<usize>);
+
+    fn pre_work(&mut self) { }
+    fn post_work(&mut self) { }
 }