chore: prepare bytes v1.9.0 (#748)

Signed-off-by: Alice Ryhl <[email protected]>

CHANGELOG.md

@@ -1,3 +1,22 @@
+# 1.9.0 (November 27, 2024)
+
+### Added
+
+- Add `Bytes::from_owner` to enable externally-allocated memory (#742)
+
+### Documented
+
+- Fix typo in Buf::chunk() comment (#744)
+
+### Internal changes
+
+- Replace BufMut::put with BufMut::put_slice in Writer impl (#745)
+- Rename hex_impl! to fmt_impl! and reuse it for fmt::Debug (#743)
+
+# 1.8.0 (October 21, 2024)
+
+- Guarantee address in `split_off`/`split_to` for empty slices (#740)
+
 # 1.7.2 (September 17, 2024)
 
 ### Fixed
@@ -10,7 +29,7 @@
 
 ### Internal changes
 
-- Ensure BytesMut::advance reduces capacity (#728) 
+- Ensure BytesMut::advance reduces capacity (#728)
 
 # 1.7.1 (August 1, 2024)
 

Cargo.toml

@@ -4,7 +4,7 @@ name = "bytes"
 # When releasing to crates.io:
 # - Update CHANGELOG.md.
 # - Create "v1.x.y" git tag.
-version = "1.7.2"
+version = "1.9.0"
 edition = "2018"
 rust-version = "1.39"
 license = "MIT"

src/buf/buf_impl.rs

@@ -125,8 +125,8 @@ pub trait Buf {
     fn remaining(&self) -> usize;
 
     /// Returns a slice starting at the current position and of length between 0
-    /// and `Buf::remaining()`. Note that this *can* return shorter slice (this allows
-    /// non-continuous internal representation).
+    /// and `Buf::remaining()`. Note that this *can* return a shorter slice (this
+    /// allows non-continuous internal representation).
     ///
     /// This is a lower level function. Most operations are done with other
     /// functions.

src/buf/writer.rs

@@ -78,7 +78,7 @@ impl<B: BufMut + Sized> io::Write for Writer<B> {
     fn write(&mut self, src: &[u8]) -> io::Result<usize> {
         let n = cmp::min(self.buf.remaining_mut(), src.len());
 
-        self.buf.put(&src[0..n]);
+        self.buf.put_slice(&src[..n]);
         Ok(n)
     }
 

src/bytes.rs

@@ -1,6 +1,7 @@
 use core::iter::FromIterator;
 use core::mem::{self, ManuallyDrop};
 use core::ops::{Deref, RangeBounds};
+use core::ptr::NonNull;
 use core::{cmp, fmt, hash, ptr, slice, usize};
 
 use alloc::{
@@ -184,6 +185,110 @@ impl Bytes {
         }
     }
 
+    /// Creates a new `Bytes` with length zero and the given pointer as the address.
+    fn new_empty_with_ptr(ptr: *const u8) -> Self {
+        debug_assert!(!ptr.is_null());
+
+        // Detach this pointer's provenance from whichever allocation it came from, and reattach it
+        // to the provenance of the fake ZST [u8;0] at the same address.
+        let ptr = without_provenance(ptr as usize);
+
+        Bytes {
+            ptr,
+            len: 0,
+            data: AtomicPtr::new(ptr::null_mut()),
+            vtable: &STATIC_VTABLE,
+        }
+    }
+
+    /// Create [Bytes] with a buffer whose lifetime is controlled
+    /// via an explicit owner.
+    ///
+    /// A common use case is to zero-copy construct from mapped memory.
+    ///
+    /// ```
+    /// # struct File;
+    /// #
+    /// # impl File {
+    /// #     pub fn open(_: &str) -> Result<Self, ()> {
+    /// #         Ok(Self)
+    /// #     }
+    /// # }
+    /// #
+    /// # mod memmap2 {
+    /// #     pub struct Mmap;
+    /// #
+    /// #     impl Mmap {
+    /// #         pub unsafe fn map(_file: &super::File) -> Result<Self, ()> {
+    /// #             Ok(Self)
+    /// #         }
+    /// #     }
+    /// #
+    /// #     impl AsRef<[u8]> for Mmap {
+    /// #         fn as_ref(&self) -> &[u8] {
+    /// #             b"buf"
+    /// #         }
+    /// #     }
+    /// # }
+    /// use bytes::Bytes;
+    /// use memmap2::Mmap;
+    ///
+    /// # fn main() -> Result<(), ()> {
+    /// let file = File::open("upload_bundle.tar.gz")?;
+    /// let mmap = unsafe { Mmap::map(&file) }?;
+    /// let b = Bytes::from_owner(mmap);
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// The `owner` will be transferred to the constructed [Bytes] object, which
+    /// will ensure it is dropped once all remaining clones of the constructed
+    /// object are dropped. The owner will then be responsible for dropping the
+    /// specified region of memory as part of its [Drop] implementation.
+    ///
+    /// Note that converting [Bytes] constructed from an owner into a [BytesMut]
+    /// will always create a deep copy of the buffer into newly allocated memory.
+    pub fn from_owner<T>(owner: T) -> Self
+    where
+        T: AsRef<[u8]> + Send + 'static,
+    {
+        // Safety & Miri:
+        // The ownership of `owner` is first transferred to the `Owned` wrapper and `Bytes` object.
+        // This ensures that the owner is pinned in memory, allowing us to call `.as_ref()` safely
+        // since the lifetime of the owner is controlled by the lifetime of the new `Bytes` object,
+        // and the lifetime of the resulting borrowed `&[u8]` matches that of the owner.
+        // Note that this remains safe so long as we only call `.as_ref()` once.
+        //
+        // There are some additional special considerations here:
+        //   * We rely on Bytes's Drop impl to clean up memory should `.as_ref()` panic.
+        //   * Setting the `ptr` and `len` on the bytes object last (after moving the owner to
+        //     Bytes) allows Miri checks to pass since it avoids obtaining the `&[u8]` slice
+        //     from a stack-owned Box.
+        // More details on this: https://github.com/tokio-rs/bytes/pull/742/#discussion_r1813375863
+        //                  and: https://github.com/tokio-rs/bytes/pull/742/#discussion_r1813316032
+
+        let owned = Box::into_raw(Box::new(Owned {
+            lifetime: OwnedLifetime {
+                ref_cnt: AtomicUsize::new(1),
+                drop: owned_box_and_drop::<T>,
+            },
+            owner,
+        }));
+
+        let mut ret = Bytes {
+            ptr: NonNull::dangling().as_ptr(),
+            len: 0,
+            data: AtomicPtr::new(owned.cast()),
+            vtable: &OWNED_VTABLE,
+        };
+
+        let buf = unsafe { &*owned }.owner.as_ref();
+        ret.ptr = buf.as_ptr();
+        ret.len = buf.len();
+
+        ret
+    }
+
     /// Returns the number of bytes contained in this `Bytes`.
     ///
     /// # Examples
@@ -214,14 +319,16 @@ impl Bytes {
         self.len == 0
     }
 
-    /// Returns true if this is the only reference to the data.
+    /// Returns true if this is the only reference to the data and
+    /// `Into<BytesMut>` would avoid cloning the underlying buffer.
     ///
-    /// Always returns false if the data is backed by a static slice.
+    /// Always returns false if the data is backed by a [static slice](Bytes::from_static),
+    /// or an [owner](Bytes::from_owner).
     ///
     /// The result of this method may be invalidated immediately if another
     /// thread clones this value while this is being called. Ensure you have
     /// unique access to this value (`&mut Bytes`) first if you need to be
-    /// certain the result is valid (i.e. for safety reasons)
+    /// certain the result is valid (i.e. for safety reasons).
     /// # Examples
     ///
     /// ```
@@ -366,7 +473,9 @@ impl Bytes {
     /// Splits the bytes into two at the given index.
     ///
     /// Afterwards `self` contains elements `[0, at)`, and the returned `Bytes`
-    /// contains elements `[at, len)`.
+    /// contains elements `[at, len)`. It's guaranteed that the memory does not
+    /// move, that is, the address of `self` does not change, and the address of
+    /// the returned slice is `at` bytes after that.
     ///
     /// This is an `O(1)` operation that just increases the reference count and
     /// sets a few indices.
@@ -389,11 +498,11 @@ impl Bytes {
     #[must_use = "consider Bytes::truncate if you don't need the other half"]
     pub fn split_off(&mut self, at: usize) -> Self {
         if at == self.len() {
-            return Bytes::new();
+            return Bytes::new_empty_with_ptr(self.ptr.wrapping_add(at));
         }
 
         if at == 0 {
-            return mem::replace(self, Bytes::new());
+            return mem::replace(self, Bytes::new_empty_with_ptr(self.ptr));
         }
 
         assert!(
@@ -438,11 +547,12 @@ impl Bytes {
     #[must_use = "consider Bytes::advance if you don't need the other half"]
     pub fn split_to(&mut self, at: usize) -> Self {
         if at == self.len() {
-            return mem::replace(self, Bytes::new());
+            let end_ptr = self.ptr.wrapping_add(at);
+            return mem::replace(self, Bytes::new_empty_with_ptr(end_ptr));
         }
 
         if at == 0 {
-            return Bytes::new();
+            return Bytes::new_empty_with_ptr(self.ptr);
         }
 
         assert!(
@@ -517,6 +627,9 @@ impl Bytes {
     /// If `self` is not unique for the entire original buffer, this will fail
     /// and return self.
     ///
+    /// This will also always fail if the buffer was constructed via either
+    /// [from_owner](Bytes::from_owner) or [from_static](Bytes::from_static).
+    ///
     /// # Examples
     ///
     /// ```
@@ -993,6 +1106,83 @@ unsafe fn static_drop(_: &mut AtomicPtr<()>, _: *const u8, _: usize) {
     // nothing to drop for &'static [u8]
 }
 
+// ===== impl OwnedVtable =====
+
+#[repr(C)]
+struct OwnedLifetime {
+    ref_cnt: AtomicUsize,
+    drop: unsafe fn(*mut ()),
+}
+
+#[repr(C)]
+struct Owned<T> {
+    lifetime: OwnedLifetime,
+    owner: T,
+}
+
+unsafe fn owned_box_and_drop<T>(ptr: *mut ()) {
+    let b: Box<Owned<T>> = Box::from_raw(ptr as _);
+    drop(b);
+}
+
+unsafe fn owned_clone(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Bytes {
+    let owned = data.load(Ordering::Relaxed);
+    let ref_cnt = &(*owned.cast::<OwnedLifetime>()).ref_cnt;
+    let old_cnt = ref_cnt.fetch_add(1, Ordering::Relaxed);
+    if old_cnt > usize::MAX >> 1 {
+        crate::abort()
+    }
+
+    Bytes {
+        ptr,
+        len,
+        data: AtomicPtr::new(owned as _),
+        vtable: &OWNED_VTABLE,
+    }
+}
+
+unsafe fn owned_to_vec(_data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Vec<u8> {
+    let slice = slice::from_raw_parts(ptr, len);
+    slice.to_vec()
+}
+
+unsafe fn owned_to_mut(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> BytesMut {
+    let bytes_mut = BytesMut::from_vec(owned_to_vec(data, ptr, len));
+    owned_drop_impl(data.load(Ordering::Relaxed));
+    bytes_mut
+}
+
+unsafe fn owned_is_unique(_data: &AtomicPtr<()>) -> bool {
+    false
+}
+
+unsafe fn owned_drop_impl(owned: *mut ()) {
+    let lifetime = owned.cast::<OwnedLifetime>();
+    let ref_cnt = &(*lifetime).ref_cnt;
+
+    let old_cnt = ref_cnt.fetch_sub(1, Ordering::Release);
+    if old_cnt != 1 {
+        return;
+    }
+    ref_cnt.load(Ordering::Acquire);
+
+    let drop_fn = &(*lifetime).drop;
+    drop_fn(owned)
+}
+
+unsafe fn owned_drop(data: &mut AtomicPtr<()>, _ptr: *const u8, _len: usize) {
+    let owned = data.load(Ordering::Relaxed);
+    owned_drop_impl(owned);
+}
+
+static OWNED_VTABLE: Vtable = Vtable {
+    clone: owned_clone,
+    to_vec: owned_to_vec,
+    to_mut: owned_to_mut,
+    is_unique: owned_is_unique,
+    drop: owned_drop,
+};
+
 // ===== impl PromotableVtable =====
 
 static PROMOTABLE_EVEN_VTABLE: Vtable = Vtable {
@@ -1426,6 +1616,10 @@ where
     new_addr as *mut u8
 }
 
+fn without_provenance(ptr: usize) -> *const u8 {
+    core::ptr::null::<u8>().wrapping_add(ptr)
+}
+
 // compile-fails
 
 /// ```compile_fail

src/bytes_mut.rs

@@ -291,7 +291,9 @@ impl BytesMut {
     /// Splits the bytes into two at the given index.
     ///
     /// Afterwards `self` contains elements `[0, at)`, and the returned
-    /// `BytesMut` contains elements `[at, capacity)`.
+    /// `BytesMut` contains elements `[at, capacity)`. It's guaranteed that the
+    /// memory does not move, that is, the address of `self` does not change,
+    /// and the address of the returned slice is `at` bytes after that.
     ///
     /// This is an `O(1)` operation that just increases the reference count
     /// and sets a few indices.

src/fmt/debug.rs

@@ -36,14 +36,5 @@ impl Debug for BytesRef<'_> {
     }
 }
 
-impl Debug for Bytes {
-    fn fmt(&self, f: &mut Formatter<'_>) -> Result {
-        Debug::fmt(&BytesRef(self.as_ref()), f)
-    }
-}
-
-impl Debug for BytesMut {
-    fn fmt(&self, f: &mut Formatter<'_>) -> Result {
-        Debug::fmt(&BytesRef(self.as_ref()), f)
-    }
-}
+fmt_impl!(Debug, Bytes);
+fmt_impl!(Debug, BytesMut);

src/fmt/hex.rs

@@ -21,17 +21,7 @@ impl UpperHex for BytesRef<'_> {
     }
 }
 
-macro_rules! hex_impl {
-    ($tr:ident, $ty:ty) => {
-        impl $tr for $ty {
-            fn fmt(&self, f: &mut Formatter<'_>) -> Result {
-                $tr::fmt(&BytesRef(self.as_ref()), f)
-            }
-        }
-    };
-}
-
-hex_impl!(LowerHex, Bytes);
-hex_impl!(LowerHex, BytesMut);
-hex_impl!(UpperHex, Bytes);
-hex_impl!(UpperHex, BytesMut);
+fmt_impl!(LowerHex, Bytes);
+fmt_impl!(LowerHex, BytesMut);
+fmt_impl!(UpperHex, Bytes);
+fmt_impl!(UpperHex, BytesMut);

src/fmt/mod.rs

@@ -1,3 +1,13 @@
+macro_rules! fmt_impl {
+    ($tr:ident, $ty:ty) => {
+        impl $tr for $ty {
+            fn fmt(&self, f: &mut Formatter<'_>) -> Result {
+                $tr::fmt(&BytesRef(self.as_ref()), f)
+            }
+        }
+    };
+}
+
 mod debug;
 mod hex;
 

tests/test_bytes.rs

@@ -1,7 +1,10 @@
 #![warn(rust_2018_idioms)]
 
 use bytes::{Buf, BufMut, Bytes, BytesMut};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
 
+use std::panic::{self, AssertUnwindSafe};
 use std::usize;
 
 const LONG: &[u8] = b"mary had a little lamb, little lamb, little lamb";
@@ -1399,3 +1402,232 @@ fn try_reclaim_arc() {
     buf.advance(2);
     assert_eq!(true, buf.try_reclaim(6));
 }
+
+#[test]
+fn split_off_empty_addr() {
+    let mut buf = Bytes::from(vec![0; 1024]);
+
+    let ptr_start = buf.as_ptr();
+    let ptr_end = ptr_start.wrapping_add(1024);
+
+    let empty_end = buf.split_off(1024);
+    assert_eq!(empty_end.len(), 0);
+    assert_eq!(empty_end.as_ptr(), ptr_end);
+
+    let _ = buf.split_off(0);
+    assert_eq!(buf.len(), 0);
+    assert_eq!(buf.as_ptr(), ptr_start);
+
+    // Is miri happy about the provenance?
+    let _ = &empty_end[..];
+    let _ = &buf[..];
+}
+
+#[test]
+fn split_to_empty_addr() {
+    let mut buf = Bytes::from(vec![0; 1024]);
+
+    let ptr_start = buf.as_ptr();
+    let ptr_end = ptr_start.wrapping_add(1024);
+
+    let empty_start = buf.split_to(0);
+    assert_eq!(empty_start.len(), 0);
+    assert_eq!(empty_start.as_ptr(), ptr_start);
+
+    let _ = buf.split_to(1024);
+    assert_eq!(buf.len(), 0);
+    assert_eq!(buf.as_ptr(), ptr_end);
+
+    // Is miri happy about the provenance?
+    let _ = &empty_start[..];
+    let _ = &buf[..];
+}
+
+#[test]
+fn split_off_empty_addr_mut() {
+    let mut buf = BytesMut::from([0; 1024].as_slice());
+
+    let ptr_start = buf.as_ptr();
+    let ptr_end = ptr_start.wrapping_add(1024);
+
+    let empty_end = buf.split_off(1024);
+    assert_eq!(empty_end.len(), 0);
+    assert_eq!(empty_end.as_ptr(), ptr_end);
+
+    let _ = buf.split_off(0);
+    assert_eq!(buf.len(), 0);
+    assert_eq!(buf.as_ptr(), ptr_start);
+
+    // Is miri happy about the provenance?
+    let _ = &empty_end[..];
+    let _ = &buf[..];
+}
+
+#[test]
+fn split_to_empty_addr_mut() {
+    let mut buf = BytesMut::from([0; 1024].as_slice());
+
+    let ptr_start = buf.as_ptr();
+    let ptr_end = ptr_start.wrapping_add(1024);
+
+    let empty_start = buf.split_to(0);
+    assert_eq!(empty_start.len(), 0);
+    assert_eq!(empty_start.as_ptr(), ptr_start);
+
+    let _ = buf.split_to(1024);
+    assert_eq!(buf.len(), 0);
+    assert_eq!(buf.as_ptr(), ptr_end);
+
+    // Is miri happy about the provenance?
+    let _ = &empty_start[..];
+    let _ = &buf[..];
+}
+
+#[derive(Clone)]
+struct SharedAtomicCounter(Arc<AtomicUsize>);
+
+impl SharedAtomicCounter {
+    pub fn new() -> Self {
+        SharedAtomicCounter(Arc::new(AtomicUsize::new(0)))
+    }
+
+    pub fn increment(&self) {
+        self.0.fetch_add(1, Ordering::AcqRel);
+    }
+
+    pub fn get(&self) -> usize {
+        self.0.load(Ordering::Acquire)
+    }
+}
+
+#[derive(Clone)]
+struct OwnedTester<const L: usize> {
+    buf: [u8; L],
+    drop_count: SharedAtomicCounter,
+    pub panic_as_ref: bool,
+}
+
+impl<const L: usize> OwnedTester<L> {
+    fn new(buf: [u8; L], drop_count: SharedAtomicCounter) -> Self {
+        Self {
+            buf,
+            drop_count,
+            panic_as_ref: false,
+        }
+    }
+}
+
+impl<const L: usize> AsRef<[u8]> for OwnedTester<L> {
+    fn as_ref(&self) -> &[u8] {
+        if self.panic_as_ref {
+            panic!("test-triggered panic in `AsRef<[u8]> for OwnedTester`");
+        }
+        self.buf.as_slice()
+    }
+}
+
+impl<const L: usize> Drop for OwnedTester<L> {
+    fn drop(&mut self) {
+        self.drop_count.increment();
+    }
+}
+
+#[test]
+fn owned_is_unique_always_false() {
+    let b1 = Bytes::from_owner([1, 2, 3, 4, 5, 6, 7]);
+    assert!(!b1.is_unique()); // even if ref_cnt == 1
+    let b2 = b1.clone();
+    assert!(!b1.is_unique());
+    assert!(!b2.is_unique());
+    drop(b1);
+    assert!(!b2.is_unique()); // even if ref_cnt == 1
+}
+
+#[test]
+fn owned_buf_sharing() {
+    let buf = [1, 2, 3, 4, 5, 6, 7];
+    let b1 = Bytes::from_owner(buf);
+    let b2 = b1.clone();
+    assert_eq!(&buf[..], &b1[..]);
+    assert_eq!(&buf[..], &b2[..]);
+    assert_eq!(b1.as_ptr(), b2.as_ptr());
+    assert_eq!(b1.len(), b2.len());
+    assert_eq!(b1.len(), buf.len());
+}
+
+#[test]
+fn owned_buf_slicing() {
+    let b1 = Bytes::from_owner(SHORT);
+    assert_eq!(SHORT, &b1[..]);
+    let b2 = b1.slice(1..(b1.len() - 1));
+    assert_eq!(&SHORT[1..(SHORT.len() - 1)], b2);
+    assert_eq!(unsafe { SHORT.as_ptr().add(1) }, b2.as_ptr());
+    assert_eq!(SHORT.len() - 2, b2.len());
+}
+
+#[test]
+fn owned_dropped_exactly_once() {
+    let buf: [u8; 5] = [1, 2, 3, 4, 5];
+    let drop_counter = SharedAtomicCounter::new();
+    let owner = OwnedTester::new(buf, drop_counter.clone());
+    let b1 = Bytes::from_owner(owner);
+    let b2 = b1.clone();
+    assert_eq!(drop_counter.get(), 0);
+    drop(b1);
+    assert_eq!(drop_counter.get(), 0);
+    let b3 = b2.slice(1..b2.len() - 1);
+    drop(b2);
+    assert_eq!(drop_counter.get(), 0);
+    drop(b3);
+    assert_eq!(drop_counter.get(), 1);
+}
+
+#[test]
+fn owned_to_mut() {
+    let buf: [u8; 10] = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9];
+    let drop_counter = SharedAtomicCounter::new();
+    let owner = OwnedTester::new(buf, drop_counter.clone());
+    let b1 = Bytes::from_owner(owner);
+
+    // Holding an owner will fail converting to a BytesMut,
+    // even when the bytes instance has a ref_cnt == 1.
+    let b1 = b1.try_into_mut().unwrap_err();
+
+    // That said, it's still possible, just not cheap.
+    let bm1: BytesMut = b1.into();
+    let new_buf = &bm1[..];
+    assert_eq!(new_buf, &buf[..]);
+
+    // `.into::<BytesMut>()` has correctly dropped the owner
+    assert_eq!(drop_counter.get(), 1);
+}
+
+#[test]
+fn owned_to_vec() {
+    let buf: [u8; 10] = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9];
+    let drop_counter = SharedAtomicCounter::new();
+    let owner = OwnedTester::new(buf, drop_counter.clone());
+    let b1 = Bytes::from_owner(owner);
+
+    let v1 = b1.to_vec();
+    assert_eq!(&v1[..], &buf[..]);
+    assert_eq!(&v1[..], &b1[..]);
+
+    drop(b1);
+    assert_eq!(drop_counter.get(), 1);
+}
+
+#[test]
+fn owned_safe_drop_on_as_ref_panic() {
+    let buf: [u8; 10] = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9];
+    let drop_counter = SharedAtomicCounter::new();
+    let mut owner = OwnedTester::new(buf, drop_counter.clone());
+    owner.panic_as_ref = true;
+
+    let result = panic::catch_unwind(AssertUnwindSafe(|| {
+        let _ = Bytes::from_owner(owner);
+    }));
+
+    assert!(result.is_err());
+    assert_eq!(drop_counter.get(), 1);
+}