Rename Bytes to BytesMut

Author: frankmcsherry

bytes/src/lib.rs

@@ -3,10 +3,10 @@
 //! # Examples
 //!
 //! ```
-//! use timely_bytes::arc::Bytes;
+//! use timely_bytes::arc::BytesMut;
 //!
 //! let bytes = vec![0u8; 1024];
-//! let mut shared1 = Bytes::from(bytes);
+//! let mut shared1 = BytesMut::from(bytes);
 //! let mut shared2 = shared1.extract_to(100);
 //! let mut shared3 = shared1.extract_to(100);
 //! let mut shared4 = shared2.extract_to(60);
@@ -38,7 +38,11 @@ pub mod arc {
     use std::any::Any;
 
     /// A thread-safe byte buffer backed by a shared allocation.
-    pub struct Bytes {
+    ///
+    /// An instance of this type contends that `ptr` is valid for `len` bytes,
+    /// and that no other reference to these bytes exists, other than through
+    /// the type currently held in `sequestered`.
+    pub struct BytesMut {
         /// Pointer to the start of this slice (not the allocation).
         ptr: *mut u8,
         /// Length of this slice.
@@ -54,12 +58,12 @@ pub mod arc {
     // Synchronization happens through `self.sequestered`, which mean 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 Bytes { }
+    unsafe impl Send for BytesMut { }
 
-    impl Bytes {
+    impl BytesMut {
 
         /// Create a new instance from a byte allocation.
-        pub fn from<B>(bytes: B) -> Bytes where B : DerefMut<Target=[u8]>+'static {
+        pub fn from<B>(bytes: B) -> BytesMut where B : DerefMut<Target=[u8]>+'static {
 
             // Sequester allocation behind an `Arc`, which *should* keep the address
             // stable for the lifetime of `sequestered`. The `Arc` also serves as our
@@ -73,24 +77,24 @@ pub mod arc {
                 .map(|a| (a.as_mut_ptr(), a.len()))
                 .unwrap();
 
-            Bytes {
+            BytesMut {
                 ptr,
                 len,
                 sequestered,
             }
         }
 
-        /// Extracts [0, index) into a new `Bytes` which is returned, updating `self`.
+        /// 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 `Bytes` contains valid memory, and that `self` can no longer access it.
-        pub fn extract_to(&mut self, index: usize) -> Bytes {
+        /// the returned `BytesMut` contains valid memory, and that `self` can no longer access it.
+        pub fn extract_to(&mut self, index: usize) -> BytesMut {
 
             assert!(index <= self.len);
 
-            let result = Bytes {
+            let result = BytesMut {
                 ptr: self.ptr,
                 len: index,
                 sequestered: self.sequestered.clone(),
@@ -102,19 +106,19 @@ pub mod arc {
             result
         }
 
-        /// Regenerates the Bytes if it is uniquely held.
+        /// Regenerates the BytesMut if it is uniquely held.
         ///
         /// If uniquely held, this method recovers the initial pointer and length
-        /// of the sequestered allocation and re-initializes the Bytes. The return
+        /// of the sequestered allocation and re-initializes the BytesMut. The return
         /// value indicates whether this occurred.
         ///
         /// # Examples
         ///
         /// ```
-        /// use timely_bytes::arc::Bytes;
+        /// use timely_bytes::arc::BytesMut;
         ///
         /// let bytes = vec![0u8; 1024];
-        /// let mut shared1 = Bytes::from(bytes);
+        /// let mut shared1 = BytesMut::from(bytes);
         /// let mut shared2 = shared1.extract_to(100);
         /// let mut shared3 = shared1.extract_to(100);
         /// let mut shared4 = shared2.extract_to(60);
@@ -147,10 +151,10 @@ pub mod arc {
         /// # Examples
         ///
         /// ```
-        /// use timely_bytes::arc::Bytes;
+        /// use timely_bytes::arc::BytesMut;
         ///
         /// let bytes = vec![0u8; 1024];
-        /// let mut shared1 = Bytes::from(bytes);
+        /// let mut shared1 = BytesMut::from(bytes);
         /// let mut shared2 = shared1.extract_to(100);
         /// let mut shared3 = shared1.extract_to(100);
         /// let mut shared4 = shared2.extract_to(60);
@@ -160,7 +164,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: Bytes) -> Result<(), Bytes> {
+        pub fn try_merge(&mut self, other: BytesMut) -> Result<(), BytesMut> {
             if Arc::ptr_eq(&self.sequestered, &other.sequestered) && ::std::ptr::eq(self.ptr.wrapping_add(self.len), other.ptr) {
                 self.len += other.len;
                 Ok(())
@@ -171,14 +175,14 @@ pub mod arc {
         }
     }
 
-    impl Deref for Bytes {
+    impl Deref for BytesMut {
         type Target = [u8];
         fn deref(&self) -> &[u8] {
             unsafe { ::std::slice::from_raw_parts(self.ptr, self.len) }
         }
     }
 
-    impl DerefMut for Bytes {
+    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::Bytes) -> Self {
+    fn from_bytes(bytes: timely_bytes::arc::BytesMut) -> 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::Bytes;
+use timely_bytes::arc::BytesMut;
 
 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<Bytes>,                         // staging area for incoming Bytes
+    staged:     Vec<BytesMut>,                      // 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<Bytes>>>>,   // to worker-local typed pullers.
+    to_local:   HashMap<usize, Rc<RefCell<VecDeque<BytesMut>>>>,// to worker-local typed pullers.
 }
 
 impl<A: Allocate> Allocate for TcpAllocator<A> {

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::Bytes;
+use timely_bytes::arc::BytesMut;
 
 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<Bytes>,
+    staged:     Vec<BytesMut>,
     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<Bytes>>>>,          // to worker-local typed pullers.
+    to_local:   HashMap<usize, Rc<RefCell<VecDeque<BytesMut>>>>,          // to worker-local typed pullers.
 }
 
 impl Allocate for ProcessAllocator {

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::Bytes;
+use timely_bytes::arc::BytesMut;
 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=Bytes>>(&mut self, iter: I);
+    fn extend<I: IntoIterator<Item=BytesMut>>(&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<Bytes>);
+    fn drain_into(&mut self, vec: &mut Vec<BytesMut>);
 }
 
 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<Bytes>>>, // queue of bytes.
+    queue: Arc<Mutex<VecDeque<BytesMut>>>, // 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=Bytes>>(&mut self, iterator: I) {
+    fn extend<I: IntoIterator<Item=BytesMut>>(&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<Bytes>) {
+    fn drain_into(&mut self, vec: &mut Vec<BytesMut>) {
         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,16 +1,16 @@
 //! A large binary allocation for writing and sharing.
 
-use timely_bytes::arc::Bytes;
+use timely_bytes::arc::BytesMut;
 
 /// A large binary allocation for writing and sharing.
 ///
-/// A bytes slab wraps a `Bytes` and maintains a valid (written) length, and supports writing after
-/// this valid length, and extracting `Bytes` up to this valid length. Extracted bytes are enqueued
+/// A bytes slab wraps a `BytesMut` and maintains a valid (written) length, and supports writing after
+/// this valid length, and extracting `BytesMut` up to this valid length. Extracted bytes are enqueued
 /// and checked for uniqueness in order to recycle them (once all shared references are dropped).
 pub struct BytesSlab {
-    buffer:         Bytes,                      // current working buffer.
-    in_progress:    Vec<Option<Bytes>>,         // buffers shared with workers.
-    stash:          Vec<Bytes>,                 // reclaimed and reusable buffers.
+    buffer:         BytesMut,                   // current working buffer.
+    in_progress:    Vec<Option<BytesMut>>,      // buffers shared with workers.
+    stash:          Vec<BytesMut>,              // reclaimed and reusable buffers.
     shift:          usize,                      // current buffer allocation size.
     valid:          usize,                      // buffer[..valid] are valid bytes.
 }
@@ -19,7 +19,7 @@ impl BytesSlab {
     /// Allocates a new `BytesSlab` with an initial size determined by a shift.
     pub fn new(shift: usize) -> Self {
         BytesSlab {
-            buffer: Bytes::from(vec![0u8; 1 << shift].into_boxed_slice()),
+            buffer: BytesMut::from(vec![0u8; 1 << shift].into_boxed_slice()),
             in_progress: Vec::new(),
             stash: Vec::new(),
             shift,
@@ -39,7 +39,7 @@ impl BytesSlab {
         self.valid += bytes;
     }
     /// Extracts the first `bytes` valid bytes.
-    pub fn extract(&mut self, bytes: usize) -> Bytes {
+    pub fn extract(&mut self, bytes: usize) -> BytesMut {
         debug_assert!(bytes <= self.valid);
         self.valid -= bytes;
         self.buffer.extract_to(bytes)
@@ -82,7 +82,7 @@ impl BytesSlab {
                 self.in_progress.retain(|x| x.is_some());
             }
 
-            let new_buffer = self.stash.pop().unwrap_or_else(|| Bytes::from(vec![0; 1 << self.shift].into_boxed_slice()));
+            let new_buffer = self.stash.pop().unwrap_or_else(|| BytesMut::from(vec![0; 1 << self.shift].into_boxed_slice()));
             let old_buffer = ::std::mem::replace(&mut self.buffer, new_buffer);
 
             self.buffer[.. self.valid].copy_from_slice(&old_buffer[.. self.valid]);

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::Bytes;
+use timely_bytes::arc::BytesMut;
 
 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<Bytes>>>,    // source of serialized buffers
+    receiver: Rc<RefCell<VecDeque<BytesMut>>>,    // 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<Bytes>>>, _canary: Canary) -> Puller<T> {
+    pub fn new(receiver: Rc<RefCell<VecDeque<BytesMut>>>, _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<Bytes>>>,     // source of serialized buffers
+    receiver: Rc<RefCell<VecDeque<BytesMut>>>,     // 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<Bytes>>>, _canary: Canary) -> Self {
+    pub fn new(inner: Box<dyn Pull<T>>, receiver: Rc<RefCell<VecDeque<BytesMut>>>, _canary: Canary) -> Self {
         PullerInner {
             inner,
             _canary,

communication/src/initialize.rs

@@ -189,7 +189,7 @@ impl Config {
 /// }
 ///
 /// impl Bytesable for Message {
-///     fn from_bytes(bytes: timely_bytes::arc::Bytes) -> Self {
+///     fn from_bytes(bytes: timely_bytes::arc::BytesMut) -> 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::Bytes) -> Self {
+///     fn from_bytes(bytes: timely_bytes::arc::BytesMut) -> 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::Bytes) -> Self {
+//!     fn from_bytes(bytes: timely_bytes::arc::BytesMut) -> 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::Bytes;
+use timely_bytes::arc::BytesMut;
 
-/// A type that can be serialized and deserialized through `Bytes`.
+/// A type that can be serialized and deserialized through `BytesMut`.
 pub trait Bytesable {
     /// Wrap bytes as `Self`.
-    fn from_bytes(bytes: Bytes) -> Self;
+    fn from_bytes(bytes: BytesMut) -> Self;
 
     /// The number of bytes required to serialize the data.
     fn length_in_bytes(&self) -> usize;