Introduce and use Bytes

Author: frankmcsherry

bytes/src/lib.rs

@@ -55,7 +55,7 @@ pub mod arc {
         sequestered: Arc<dyn Any>,
     }
 
-    // Synchronization happens through `self.sequestered`, which mean to ensure that even
+    // Synchronization happens through `self.sequestered`, which means to ensure that even
     // across multiple threads each region of the slice is uniquely "owned", if not in the
     // traditional Rust sense.
     unsafe impl Send for BytesMut { }
@@ -90,7 +90,7 @@ pub mod arc {
         ///
         /// This method first tests `index` against `self.len`, which should ensure that both
         /// the returned `BytesMut` contains valid memory, and that `self` can no longer access it.
-        pub fn extract_to(&mut self, index: usize) -> BytesMut {
+        pub fn extract_to(&mut self, index: usize) -> Bytes {
 
             assert!(index <= self.len);
 
@@ -103,7 +103,7 @@ pub mod arc {
             self.ptr = self.ptr.wrapping_add(index);
             self.len -= index;
 
-            result
+            result.freeze()
         }
 
         /// Regenerates the BytesMut if it is uniquely held.
@@ -142,6 +142,77 @@ pub mod arc {
             }
         }
 
+        /// Converts a writeable byte slice to an shareable byte slice.
+        pub fn freeze(self) -> Bytes {
+            Bytes {
+                ptr: self.ptr,
+                len: self.len,
+                sequestered: self.sequestered,
+            }
+        }
+    }
+
+    impl Deref for BytesMut {
+        type Target = [u8];
+        fn deref(&self) -> &[u8] {
+            unsafe { ::std::slice::from_raw_parts(self.ptr, self.len) }
+        }
+    }
+
+    impl DerefMut for BytesMut {
+        fn deref_mut(&mut self) -> &mut [u8] {
+            unsafe { ::std::slice::from_raw_parts_mut(self.ptr, self.len) }
+        }
+    }
+
+
+    /// A thread-safe shared byte buffer backed by a shared allocation.
+    ///
+    /// An instance of this type contends that `ptr` is valid for `len` bytes,
+    /// and that no other mutable reference to these bytes exists, other than
+    /// through the type currently held in `sequestered`.
+    #[derive(Clone)]
+    pub struct Bytes {
+        /// Pointer to the start of this slice (not the allocation).
+        ptr: *const u8,
+        /// Length of this slice.
+        len: usize,
+        /// Shared access to underlying resources.
+        ///
+        /// 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 stronger statement about this.
+        sequestered: Arc<dyn Any>,
+    }
+
+    // Synchronization happens through `self.sequestered`, which means to ensure that even
+    // across multiple threads the referenced range of bytes remain valid.
+    unsafe impl Send for Bytes { }
+
+    impl Bytes {
+
+        /// Extracts [0, index) into a new `BytesMut` which is returned, updating `self`.
+        ///
+        /// # Safety
+        ///
+        /// This method first tests `index` against `self.len`, which should ensure that both
+        /// the returned `BytesMut` contains valid memory, and that `self` can no longer access it.
+        pub fn extract_to(&mut self, index: usize) -> Bytes {
+
+            assert!(index <= self.len);
+
+            let result = Bytes {
+                ptr: self.ptr,
+                len: index,
+                sequestered: self.sequestered.clone(),
+            };
+
+            self.ptr = self.ptr.wrapping_add(index);
+            self.len -= index;
+
+            result
+        }
+
         /// Attempts to merge adjacent slices from the same allocation.
         ///
         /// If the merge succeeds then `other.len` is added to `self` and the result is `Ok(())`.
@@ -164,7 +235,7 @@ pub mod arc {
         /// shared2.try_merge(shared1).ok().expect("Failed to merge 23 and 1");
         /// shared4.try_merge(shared2).ok().expect("Failed to merge 4 and 231");
         /// ```
-        pub fn try_merge(&mut self, other: BytesMut) -> Result<(), BytesMut> {
+        pub fn try_merge(&mut self, other: Bytes) -> Result<(), Bytes> {
             if Arc::ptr_eq(&self.sequestered, &other.sequestered) && ::std::ptr::eq(self.ptr.wrapping_add(self.len), other.ptr) {
                 self.len += other.len;
                 Ok(())
@@ -175,16 +246,10 @@ pub mod arc {
         }
     }
 
-    impl Deref for BytesMut {
+    impl Deref for Bytes {
         type Target = [u8];
         fn deref(&self) -> &[u8] {
             unsafe { ::std::slice::from_raw_parts(self.ptr, self.len) }
         }
     }
-
-    impl DerefMut for BytesMut {
-        fn deref_mut(&mut self) -> &mut [u8] {
-            unsafe { ::std::slice::from_raw_parts_mut(self.ptr, self.len) }
-        }
-    }
 }

communication/examples/comm_hello.rs

@@ -7,7 +7,7 @@ pub struct Message {
 }
 
 impl Bytesable for Message {
-    fn from_bytes(bytes: timely_bytes::arc::BytesMut) -> Self {
+    fn from_bytes(bytes: timely_bytes::arc::Bytes) -> Self {
         Message { payload: std::str::from_utf8(&bytes[..]).unwrap().to_string() }
     }
 

communication/src/allocator/zero_copy/allocator.rs

@@ -4,7 +4,7 @@ use std::cell::RefCell;
 use std::collections::{VecDeque, HashMap, hash_map::Entry};
 use crossbeam_channel::{Sender, Receiver};
 
-use timely_bytes::arc::BytesMut;
+use timely_bytes::arc::Bytes;
 
 use crate::networking::MessageHeader;
 
@@ -121,15 +121,15 @@ pub struct TcpAllocator<A: Allocate> {
     index:      usize,                              // number out of peers
     peers:      usize,                              // number of peer allocators (for typed channel allocation).
 
-    staged:     Vec<BytesMut>,                      // staging area for incoming Bytes
+    staged:     Vec<Bytes>,                         // staging area for incoming Bytes
     canaries:   Rc<RefCell<Vec<usize>>>,
 
     channel_id_bound: Option<usize>,
 
     // sending, receiving, and responding to binary buffers.
     sends:      Vec<Rc<RefCell<SendEndpoint<MergeQueue>>>>,     // sends[x] -> goes to process x.
     recvs:      Vec<MergeQueue>,                                // recvs[x] <- from process x.
-    to_local:   HashMap<usize, Rc<RefCell<VecDeque<BytesMut>>>>,// to worker-local typed pullers.
+    to_local:   HashMap<usize, Rc<RefCell<VecDeque<Bytes>>>>,   // to worker-local typed pullers.
 }
 
 impl<A: Allocate> Allocate for TcpAllocator<A> {
@@ -220,7 +220,7 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
             // No splitting occurs across allocations.
             while bytes.len() > 0 {
 
-                if let Some(header) = MessageHeader::try_read(&mut bytes[..]) {
+                if let Some(header) = MessageHeader::try_read(&bytes[..]) {
 
                     // Get the header and payload, ditch the header.
                     let mut peel = bytes.extract_to(header.required_bytes());

communication/src/allocator/zero_copy/allocator_process.rs

@@ -5,7 +5,7 @@ use std::cell::RefCell;
 use std::collections::{VecDeque, HashMap, hash_map::Entry};
 use crossbeam_channel::{Sender, Receiver};
 
-use timely_bytes::arc::BytesMut;
+use timely_bytes::arc::Bytes;
 
 use crate::networking::MessageHeader;
 
@@ -110,10 +110,10 @@ pub struct ProcessAllocator {
     channel_id_bound: Option<usize>,
 
     // sending, receiving, and responding to binary buffers.
-    staged:     Vec<BytesMut>,
+    staged:     Vec<Bytes>,
     sends:      Vec<Rc<RefCell<SendEndpoint<MergeQueue>>>>, // sends[x] -> goes to thread x.
     recvs:      Vec<MergeQueue>,                            // recvs[x] <- from thread x.
-    to_local:   HashMap<usize, Rc<RefCell<VecDeque<BytesMut>>>>,          // to worker-local typed pullers.
+    to_local:   HashMap<usize, Rc<RefCell<VecDeque<Bytes>>>>,          // to worker-local typed pullers.
 }
 
 impl Allocate for ProcessAllocator {
@@ -188,7 +188,7 @@ impl Allocate for ProcessAllocator {
             // No splitting occurs across allocations.
             while bytes.len() > 0 {
 
-                if let Some(header) = MessageHeader::try_read(&mut bytes[..]) {
+                if let Some(header) = MessageHeader::try_read(&bytes[..]) {
 
                     // Get the header and payload, ditch the header.
                     let mut peel = bytes.extract_to(header.required_bytes());

communication/src/allocator/zero_copy/bytes_exchange.rs

@@ -3,22 +3,22 @@
 use std::sync::{Arc, Mutex};
 use std::collections::VecDeque;
 
-use timely_bytes::arc::BytesMut;
+use timely_bytes::arc::Bytes;
 use super::bytes_slab::BytesSlab;
 
 /// A target for `Bytes`.
 pub trait BytesPush {
     // /// Pushes bytes at the instance.
     // fn push(&mut self, bytes: Bytes);
     /// Pushes many bytes at the instance.
-    fn extend<I: IntoIterator<Item=BytesMut>>(&mut self, iter: I);
+    fn extend<I: IntoIterator<Item=Bytes>>(&mut self, iter: I);
 }
 /// A source for `Bytes`.
 pub trait BytesPull {
     // /// Pulls bytes from the instance.
     // fn pull(&mut self) -> Option<Bytes>;
     /// Drains many bytes from the instance.
-    fn drain_into(&mut self, vec: &mut Vec<BytesMut>);
+    fn drain_into(&mut self, vec: &mut Vec<Bytes>);
 }
 
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -28,7 +28,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
 /// TODO: explain "extend"
 #[derive(Clone)]
 pub struct MergeQueue {
-    queue: Arc<Mutex<VecDeque<BytesMut>>>, // queue of bytes.
+    queue: Arc<Mutex<VecDeque<Bytes>>>, // queue of bytes.
     buzzer: crate::buzzer::Buzzer,  // awakens receiver thread.
     panic: Arc<AtomicBool>,
 }
@@ -50,7 +50,7 @@ impl MergeQueue {
 }
 
 impl BytesPush for MergeQueue {
-    fn extend<I: IntoIterator<Item=BytesMut>>(&mut self, iterator: I) {
+    fn extend<I: IntoIterator<Item=Bytes>>(&mut self, iterator: I) {
 
         if self.panic.load(Ordering::SeqCst) { panic!("MergeQueue poisoned."); }
 
@@ -92,7 +92,7 @@ impl BytesPush for MergeQueue {
 }
 
 impl BytesPull for MergeQueue {
-    fn drain_into(&mut self, vec: &mut Vec<BytesMut>) {
+    fn drain_into(&mut self, vec: &mut Vec<Bytes>) {
         if self.panic.load(Ordering::SeqCst) { panic!("MergeQueue poisoned."); }
 
         // try to acquire lock without going to sleep (Rust's lock() might yield)

communication/src/allocator/zero_copy/bytes_slab.rs

@@ -1,6 +1,6 @@
 //! A large binary allocation for writing and sharing.
 
-use timely_bytes::arc::BytesMut;
+use timely_bytes::arc::{Bytes, BytesMut};
 
 /// A large binary allocation for writing and sharing.
 ///
@@ -39,7 +39,7 @@ impl BytesSlab {
         self.valid += bytes;
     }
     /// Extracts the first `bytes` valid bytes.
-    pub fn extract(&mut self, bytes: usize) -> BytesMut {
+    pub fn extract(&mut self, bytes: usize) -> Bytes {
         debug_assert!(bytes <= self.valid);
         self.valid -= bytes;
         self.buffer.extract_to(bytes)

communication/src/allocator/zero_copy/push_pull.rs

@@ -4,7 +4,7 @@ use std::rc::Rc;
 use std::cell::RefCell;
 use std::collections::VecDeque;
 
-use timely_bytes::arc::BytesMut;
+use timely_bytes::arc::Bytes;
 
 use crate::allocator::canary::Canary;
 use crate::networking::MessageHeader;
@@ -67,12 +67,12 @@ impl<T: Bytesable, P: BytesPush> Push<T> for Pusher<T, P> {
 pub struct Puller<T> {
     _canary: Canary,
     current: Option<T>,
-    receiver: Rc<RefCell<VecDeque<BytesMut>>>,    // source of serialized buffers
+    receiver: Rc<RefCell<VecDeque<Bytes>>>,    // source of serialized buffers
 }
 
 impl<T: Bytesable> Puller<T> {
     /// Creates a new `Puller` instance from a shared queue.
-    pub fn new(receiver: Rc<RefCell<VecDeque<BytesMut>>>, _canary: Canary) -> Puller<T> {
+    pub fn new(receiver: Rc<RefCell<VecDeque<Bytes>>>, _canary: Canary) -> Puller<T> {
         Puller {
             _canary,
             current: None,
@@ -104,12 +104,12 @@ pub struct PullerInner<T> {
     inner: Box<dyn Pull<T>>,               // inner pullable (e.g. intra-process typed queue)
     _canary: Canary,
     current: Option<T>,
-    receiver: Rc<RefCell<VecDeque<BytesMut>>>,     // source of serialized buffers
+    receiver: Rc<RefCell<VecDeque<Bytes>>>,     // source of serialized buffers
 }
 
 impl<T: Bytesable> PullerInner<T> {
     /// Creates a new `PullerInner` instance from a shared queue.
-    pub fn new(inner: Box<dyn Pull<T>>, receiver: Rc<RefCell<VecDeque<BytesMut>>>, _canary: Canary) -> Self {
+    pub fn new(inner: Box<dyn Pull<T>>, receiver: Rc<RefCell<VecDeque<Bytes>>>, _canary: Canary) -> Self {
         PullerInner {
             inner,
             _canary,

communication/src/allocator/zero_copy/tcp.rs

@@ -174,12 +174,12 @@ pub fn send_loop<S: Stream>(
         }
         else {
             // TODO: Could do scatter/gather write here.
-            for mut bytes in stash.drain(..) {
+            for bytes in stash.drain(..) {
 
                 // Record message sends.
                 logger.as_mut().map(|logger| {
                     let mut offset = 0;
-                    while let Some(header) = MessageHeader::try_read(&mut bytes[offset..]) {
+                    while let Some(header) = MessageHeader::try_read(&bytes[offset..]) {
                         logger.log(MessageEvent { is_send: true, header, });
                         offset += header.required_bytes();
                     }

communication/src/initialize.rs

@@ -189,7 +189,7 @@ impl Config {
 /// }
 ///
 /// impl Bytesable for Message {
-///     fn from_bytes(bytes: timely_bytes::arc::BytesMut) -> Self {
+///     fn from_bytes(bytes: timely_bytes::arc::Bytes) -> Self {
 ///         Message { payload: std::str::from_utf8(&bytes[..]).unwrap().to_string() }
 ///     }
 ///
@@ -285,7 +285,7 @@ pub fn initialize<T:Send+'static, F: Fn(Generic)->T+Send+Sync+'static>(
 /// }
 ///
 /// impl Bytesable for Message {
-///     fn from_bytes(bytes: timely_bytes::arc::BytesMut) -> Self {
+///     fn from_bytes(bytes: timely_bytes::arc::Bytes) -> Self {
 ///         Message { payload: std::str::from_utf8(&bytes[..]).unwrap().to_string() }
 ///     }
 ///

communication/src/lib.rs

@@ -24,7 +24,7 @@
 //! }
 //! 
 //! impl Bytesable for Message {
-//!     fn from_bytes(bytes: timely_bytes::arc::BytesMut) -> Self {
+//!     fn from_bytes(bytes: timely_bytes::arc::Bytes) -> Self {
 //!         Message { payload: std::str::from_utf8(&bytes[..]).unwrap().to_string() }
 //!     }
 //! 
@@ -107,12 +107,12 @@ pub use allocator::Generic as Allocator;
 pub use allocator::{Allocate, Exchangeable};
 pub use initialize::{initialize, initialize_from, Config, WorkerGuards};
 
-use timely_bytes::arc::BytesMut;
+use timely_bytes::arc::Bytes;
 
-/// A type that can be serialized and deserialized through `BytesMut`.
+/// A type that can be serialized and deserialized through `Bytes`.
 pub trait Bytesable {
     /// Wrap bytes as `Self`.
-    fn from_bytes(bytes: BytesMut) -> Self;
+    fn from_bytes(bytes: Bytes) -> Self;
 
     /// The number of bytes required to serialize the data.
     fn length_in_bytes(&self) -> usize;