Merge branch 'v0.4.x'

.travis.yml

@@ -36,6 +36,9 @@ matrix:
     # Serde implementation
     - env: EXTRA_ARGS="--features serde"
 
+    # 128 bit numbers
+    - env: EXTRA_ARGS="--features i128"
+
     # WASM support
     - rust: beta
       script:

Cargo.toml

@@ -19,10 +19,16 @@ exclude       = [
 ]
 categories = ["network-programming", "data-structures"]
 
+[package.metadata.docs.rs]
+features = ["i128"]
+
 [dependencies]
-byteorder = "1.0.0"
+byteorder = "1.1.0"
 iovec = { git = "https://github.com/carllerche/iovec" }
 serde = { version = "1.0", optional = true }
 
 [dev-dependencies]
 serde_test = "1.0"
+
+[features]
+i128 = ["byteorder/i128"]

benches/bytes.rs

@@ -113,6 +113,39 @@ fn deref_two(b: &mut Bencher) {
     })
 }
 
+#[bench]
+fn clone_inline(b: &mut Bencher) {
+    let bytes = Bytes::from_static(b"hello world");
+
+    b.iter(|| {
+        for _ in 0..1024 {
+            test::black_box(&bytes.clone());
+        }
+    })
+}
+
+#[bench]
+fn clone_static(b: &mut Bencher) {
+    let bytes = Bytes::from_static("hello world 1234567890 and have a good byte 0987654321".as_bytes());
+
+    b.iter(|| {
+        for _ in 0..1024 {
+            test::black_box(&bytes.clone());
+        }
+    })
+}
+
+#[bench]
+fn clone_arc(b: &mut Bencher) {
+    let bytes = Bytes::from("hello world 1234567890 and have a good byte 0987654321".as_bytes());
+
+    b.iter(|| {
+        for _ in 0..1024 {
+            test::black_box(&bytes.clone());
+        }
+    })
+}
+
 #[bench]
 fn alloc_write_split_to_mid(b: &mut Bencher) {
     b.iter(|| {

ci/tsan

@@ -16,6 +16,6 @@ race:test::run_tests_console::*closure
 # Probably more fences in std.
 race:__call_tls_dtors
 
-# `is_inline` is explicitly called concurrently without synchronization. The
+# `is_inline_or_static` is explicitly called concurrently without synchronization.
-# safety explanation can be found in a comment.
+# The safety explanation can be found in a comment.
-race:Inner::is_inline
+race:Inner::is_inline_or_static

src/buf/buf.rs

@@ -557,6 +557,98 @@ pub trait Buf {
         buf_get_impl!(self, 8, LittleEndian::read_i64);
     }
 
+    /// Gets an unsigned 128 bit integer from `self` in big-endian byte order.
+    ///
+    /// **NOTE:** This method requires the `i128` feature.
+    /// The current position is advanced by 16.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bytes::Buf;
+    /// use std::io::Cursor;
+    ///
+    /// let mut buf = Cursor::new(b"\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16 hello");
+    /// assert_eq!(0x01020304050607080910111213141516, buf.get_u128());
+    /// ```
+    ///
+    /// # Panics
+    ///
+    /// This function panics if there is not enough remaining data in `self`.
+    #[cfg(feature = "i128")]
+    fn get_u128(&mut self) -> u128 {
+        buf_get_impl!(self, 16, BigEndian::read_u128);
+    }
+
+    /// Gets an unsigned 128 bit integer from `self` in little-endian byte order.
+    ///
+    /// **NOTE:** This method requires the `i128` feature.
+    /// The current position is advanced by 16.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bytes::Buf;
+    /// use std::io::Cursor;
+    ///
+    /// let mut buf = Cursor::new(b"\x16\x15\x14\x13\x12\x11\x10\x09\x08\x07\x06\x05\x04\x03\x02\x01 hello");
+    /// assert_eq!(0x01020304050607080910111213141516, buf.get_u128_le());
+    /// ```
+    ///
+    /// # Panics
+    ///
+    /// This function panics if there is not enough remaining data in `self`.
+    #[cfg(feature = "i128")]
+    fn get_u128_le(&mut self) -> u128 {
+        buf_get_impl!(self, 16, LittleEndian::read_u128);
+    }
+
+    /// Gets a signed 128 bit integer from `self` in big-endian byte order.
+    ///
+    /// **NOTE:** This method requires the `i128` feature.
+    /// The current position is advanced by 16.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bytes::Buf;
+    /// use std::io::Cursor;
+    ///
+    /// let mut buf = Cursor::new(b"\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16 hello");
+    /// assert_eq!(0x01020304050607080910111213141516, buf.get_i128());
+    /// ```
+    ///
+    /// # Panics
+    ///
+    /// This function panics if there is not enough remaining data in `self`.
+    #[cfg(feature = "i128")]
+    fn get_i128(&mut self) -> i128 {
+        buf_get_impl!(self, 16, BigEndian::read_i128);
+    }
+
+    /// Gets a signed 128 bit integer from `self` in little-endian byte order.
+    ///
+    /// **NOTE:** This method requires the `i128` feature.
+    /// The current position is advanced by 16.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bytes::Buf;
+    /// use std::io::Cursor;
+    ///
+    /// let mut buf = Cursor::new(b"\x16\x15\x14\x13\x12\x11\x10\x09\x08\x07\x06\x05\x04\x03\x02\x01 hello");
+    /// assert_eq!(0x01020304050607080910111213141516, buf.get_i128_le());
+    /// ```
+    ///
+    /// # Panics
+    ///
+    /// This function panics if there is not enough remaining data in `self`.
+    #[cfg(feature = "i128")]
+    fn get_i128_le(&mut self) -> i128 {
+        buf_get_impl!(self, 16, LittleEndian::read_i128);
+    }
+
     /// Gets an unsigned n-byte integer from `self` in big-endian byte order.
     ///
     /// The current position is advanced by `nbytes`.

src/buf/buf_mut.rs

@@ -626,6 +626,110 @@ pub trait BufMut {
         self.put_slice(&buf)
     }
 
+    /// Writes an unsigned 128 bit integer to `self` in the big-endian byte order.
+    ///
+    /// **NOTE:** This method requires the `i128` feature.
+    /// The current position is advanced by 16.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bytes::BufMut;
+    ///
+    /// let mut buf = vec![];
+    /// buf.put_u128(0x01020304050607080910111213141516);
+    /// assert_eq!(buf, b"\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16");
+    /// ```
+    ///
+    /// # Panics
+    ///
+    /// This function panics if there is not enough remaining capacity in
+    /// `self`.
+    #[cfg(feature = "i128")]
+    fn put_u128(&mut self, n: u128) {
+        let mut buf = [0; 16];
+        BigEndian::write_u128(&mut buf, n);
+        self.put_slice(&buf)
+    }
+
+    /// Writes an unsigned 128 bit integer to `self` in little-endian byte order.
+    ///
+    /// **NOTE:** This method requires the `i128` feature.
+    /// The current position is advanced by 16.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bytes::BufMut;
+    ///
+    /// let mut buf = vec![];
+    /// buf.put_u128_le(0x01020304050607080910111213141516);
+    /// assert_eq!(buf, b"\x16\x15\x14\x13\x12\x11\x10\x09\x08\x07\x06\x05\x04\x03\x02\x01");
+    /// ```
+    ///
+    /// # Panics
+    ///
+    /// This function panics if there is not enough remaining capacity in
+    /// `self`.
+    #[cfg(feature = "i128")]
+    fn put_u128_le(&mut self, n: u128) {
+        let mut buf = [0; 16];
+        LittleEndian::write_u128(&mut buf, n);
+        self.put_slice(&buf)
+    }
+
+    /// Writes a signed 128 bit integer to `self` in the big-endian byte order.
+    ///
+    /// **NOTE:** This method requires the `i128` feature.
+    /// The current position is advanced by 16.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bytes::BufMut;
+    ///
+    /// let mut buf = vec![];
+    /// buf.put_i128(0x01020304050607080910111213141516);
+    /// assert_eq!(buf, b"\x01\x02\x03\x04\x05\x06\x07\x08\x09\x10\x11\x12\x13\x14\x15\x16");
+    /// ```
+    ///
+    /// # Panics
+    ///
+    /// This function panics if there is not enough remaining capacity in
+    /// `self`.
+    #[cfg(feature = "i128")]
+    fn put_i128(&mut self, n: i128) {
+        let mut buf = [0; 16];
+        BigEndian::write_i128(&mut buf, n);
+        self.put_slice(&buf)
+    }
+
+    /// Writes a signed 128 bit integer to `self` in little-endian byte order.
+    ///
+    /// **NOTE:** This method requires the `i128` feature.
+    /// The current position is advanced by 16.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bytes::BufMut;
+    ///
+    /// let mut buf = vec![];
+    /// buf.put_i128_le(0x01020304050607080910111213141516);
+    /// assert_eq!(buf, b"\x16\x15\x14\x13\x12\x11\x10\x09\x08\x07\x06\x05\x04\x03\x02\x01");
+    /// ```
+    ///
+    /// # Panics
+    ///
+    /// This function panics if there is not enough remaining capacity in
+    /// `self`.
+    #[cfg(feature = "i128")]
+    fn put_i128_le(&mut self, n: i128) {
+        let mut buf = [0; 16];
+        LittleEndian::write_i128(&mut buf, n);
+        self.put_slice(&buf)
+    }
+
     /// Writes an unsigned n-byte integer to `self` in big-endian byte order.
     ///
     /// The current position is advanced by `nbytes`.

src/bytes.rs

@@ -2139,18 +2139,27 @@ impl Inner {
     unsafe fn shallow_clone(&self, mut_self: bool) -> Inner {
         // Always check `inline` first, because if the handle is using inline
         // data storage, all of the `Inner` struct fields will be gibberish.
-        if self.is_inline() {
-            // In this case, a shallow_clone still involves copying the data.
         //
-            // TODO: Just copy the fields
+        // Additionally, if kind is STATIC, then Arc is *never* changed, making
-            let mut inner: Inner = mem::uninitialized();
+        // it safe and faster to check for it now before an atomic acquire.
-            let len = self.inline_len();
 
-            inner.arc = AtomicPtr::new(KIND_INLINE as *mut Shared);
+        if self.is_inline_or_static() {
-            inner.set_inline_len(len);
+            // In this case, a shallow_clone still involves copying the data.
-            inner.as_raw()[0..len].copy_from_slice(self.as_ref());
+            let mut inner: Inner = mem::uninitialized();
+            ptr::copy_nonoverlapping(
+                self,
+                &mut inner,
+                1,
+            );
             inner
         } else {
+            self.shallow_clone_sync(mut_self)
+        }
+    }
+
+
+    #[cold]
+    unsafe fn shallow_clone_sync(&self, mut_self: bool) -> Inner {
         // The function requires `&self`, this means that `shallow_clone`
         // could be called concurrently.
         //
@@ -2159,12 +2168,40 @@ impl Inner {
         // `compare_and_swap` that comes later in this function. The goal is
         // to ensure that if `arc` is currently set to point to a `Shared`,
         // that the current thread acquires the associated memory.
-            let mut arc = self.arc.load(Acquire);
+        let arc = self.arc.load(Acquire);
+        let kind = arc as usize & KIND_MASK;
 
+        if kind == KIND_ARC {
+            self.shallow_clone_arc(arc)
+        } else {
+            assert!(kind == KIND_VEC);
+            self.shallow_clone_vec(arc as usize, mut_self)
+        }
+    }
+
+    unsafe fn shallow_clone_arc(&self, arc: *mut Shared) -> Inner {
+        debug_assert!(arc as usize & KIND_MASK == KIND_ARC);
+
+        let old_size = (*arc).ref_count.fetch_add(1, Relaxed);
+
+        if old_size == usize::MAX {
+            abort();
+        }
+
+        Inner {
+            arc: AtomicPtr::new(arc),
+            .. *self
+        }
+    }
+
+    #[cold]
+    unsafe fn shallow_clone_vec(&self, arc: usize, mut_self: bool) -> Inner {
         // If  the buffer is still tracked in a `Vec<u8>`. It is time to
         // promote the vec to an `Arc`. This could potentially be called
         // concurrently, so some care must be taken.
-            if arc as usize & KIND_MASK == KIND_VEC {
+
+        debug_assert!(arc & KIND_MASK == KIND_VEC);
+
         let original_capacity_repr =
             (arc as usize & ORIGINAL_CAPACITY_MASK) >> ORIGINAL_CAPACITY_OFFSET;
 
@@ -2213,9 +2250,9 @@ impl Inner {
         // ordering will synchronize with the `compare_and_swap`
         // that happened in the other thread and the `Shared`
         // pointed to by `actual` will be visible.
-                let actual = self.arc.compare_and_swap(arc, shared, AcqRel);
+        let actual = self.arc.compare_and_swap(arc as *mut Shared, shared, AcqRel);
 
-                if actual == arc {
+        if actual as usize == arc {
             // The upgrade was successful, the new handle can be
             // returned.
             return Inner {
@@ -2230,33 +2267,9 @@ impl Inner {
         let shared = Box::from_raw(shared);
         mem::forget(*shared);
 
-                // Update the `arc` local variable and fall through to a ref
-                // count update
-                arc = actual;
-            } else if arc as usize & KIND_MASK == KIND_STATIC {
-                // Static buffer
-                return Inner {
-                    arc: AtomicPtr::new(arc),
-                    .. *self
-                };
-            }
-
         // Buffer already promoted to shared storage, so increment ref
         // count.
-            //
+        self.shallow_clone_arc(actual)
-            // Relaxed ordering is acceptable as the memory has already been
-            // acquired via the `Acquire` load above.
-            let old_size = (*arc).ref_count.fetch_add(1, Relaxed);
-
-            if old_size == usize::MAX {
-                panic!(); // TODO: abort
-            }
-
-            Inner {
-                arc: AtomicPtr::new(arc),
-                .. *self
-            }
-        }
     }
 
     #[inline]
@@ -2415,6 +2428,18 @@ impl Inner {
         self.kind() == KIND_INLINE
     }
 
+    #[inline]
+    fn is_inline_or_static(&self) -> bool {
+        // The value returned by `kind` isn't itself safe, but the value could
+        // inform what operations to take, and unsafely do something without
+        // synchronization.
+        //
+        // KIND_INLINE and KIND_STATIC will *never* change, so branches on that
+        // information is safe.
+        let kind = self.kind();
+        kind == KIND_INLINE || kind == KIND_STATIC
+    }
+
     /// Used for `debug_assert` statements. &mut is used to guarantee that it is
     /// safe to check VEC_KIND
     #[inline]
@@ -2915,3 +2940,21 @@ impl PartialEq<Bytes> for BytesMut
         &other[..] == &self[..]
     }
 }
+
+// While there is `std::process:abort`, it's only available in Rust 1.17, and
+// our minimum supported version is currently 1.15. So, this acts as an abort
+// by triggering a double panic, which always aborts in Rust.
+struct Abort;
+
+impl Drop for Abort {
+    fn drop(&mut self) {
+        panic!();
+    }
+}
+
+#[inline(never)]
+#[cold]
+fn abort() {
+    let _a = Abort;
+    panic!();
+}