diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index a4f7b1d93..c0658a142 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -7,6 +7,8 @@ on:
   push:
     branches:
       - master
+  schedule:
+    - cron: '0 2 * * 0'
 
 env:
   RUSTFLAGS: -Dwarnings
@@ -23,11 +25,11 @@ jobs:
     name: rustfmt
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
     - name: Install Rust
-      run: rustup update stable && rustup default stable
+      run: rustup update stable
     - name: Check formatting
-      run: cargo fmt --all -- --check
+      run: cargo fmt --all --check
 
   # TODO
   # # Apply clippy lints
@@ -35,7 +37,7 @@ jobs:
   #   name: clippy
   #   runs-on: ubuntu-latest
   #   steps:
-  #   - uses: actions/checkout@v3
+  #   - uses: actions/checkout@v4
   #   - name: Apply clippy lints
   #     run: cargo clippy --all-features
 
@@ -48,11 +50,11 @@ jobs:
     name: minrust
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
-      - name: Install Rust
-        run: rustup update 1.39.0 && rustup default 1.39.0
+      - uses: actions/checkout@v4
+      - name: Install cargo-hack
+        uses: taiki-e/install-action@cargo-hack
       - name: Check
-        run: . ci/test-stable.sh check
+        run: cargo hack check --feature-powerset --optional-deps --rust-version
 
   # Stable
   stable:
@@ -65,23 +67,27 @@ jobs:
           - windows-latest
     runs-on: ${{ matrix.os }}
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Install Rust
         # --no-self-update is necessary because the windows environment cannot self-update rustup.exe.
-        run: rustup update stable --no-self-update && rustup default stable
+        run: rustup update stable --no-self-update
+      - name: Install cargo-hack
+        uses: taiki-e/install-action@cargo-hack
       - name: Test
-        run: . ci/test-stable.sh test
+        run: ci/test-stable.sh test
 
   # Nightly
   nightly:
     name: nightly
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Install Rust
         run: rustup update $nightly && rustup default $nightly
+      - name: Install cargo-hack
+        uses: taiki-e/install-action@cargo-hack
       - name: Test
-        run: . ci/test-stable.sh test
+        run: ci/test-stable.sh test
 
   # Run tests on some extra platforms
   cross:
@@ -96,13 +102,14 @@ jobs:
           - wasm32-unknown-unknown
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Install Rust
-        run: rustup update stable && rustup default stable
+        run: rustup update stable
+      - name: Install cross
+        uses: taiki-e/install-action@cross
+        if: matrix.target != 'wasm32-unknown-unknown'
       - name: cross build --target ${{ matrix.target }}
-        run: |
-          cargo install cross
-          cross build --target ${{ matrix.target }}
+        run: cross build --target ${{ matrix.target }}
         if: matrix.target != 'wasm32-unknown-unknown'
       # WASM support
       - name: cargo build --target ${{ matrix.target }}
@@ -116,18 +123,19 @@ jobs:
     name: tsan
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Install Rust
         run: rustup update $nightly && rustup default $nightly
       - name: Install rust-src
         run: rustup component add rust-src
       - name: ASAN / TSAN
-        run: . ci/tsan.sh
+        run: ci/tsan.sh
+
   miri:
     name: miri
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Miri
         run: ci/miri.sh
 
@@ -136,7 +144,7 @@ jobs:
     name: loom
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Install Rust
         run: rustup update $nightly && rustup default $nightly
       - name: Loom tests
@@ -155,7 +163,7 @@ jobs:
       - loom
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
       - name: Install Rust
         run: rustup update $nightly && rustup default $nightly
       - name: Build documentation
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 67b9f673a..93c1419cc 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,82 @@
+# 1.7.1 (August 1, 2024)
+
+This release reverts the following change due to a regression:
+
+- Reuse capacity when possible in `<BytesMut as Buf>::advance` impl (#698)
+
+The revert can be found at #726.
+
+# 1.7.0 (July 31, 2024)
+
+### Added
+
+- Add conversion from `Bytes` to `BytesMut` (#695, #710)
+- Add reclaim method without additional allocation (#686)
+
+### Documented
+
+- Clarify how `BytesMut::zeroed` works (#714)
+- Clarify the behavior of `Buf::chunk` (#717)
+
+### Changed
+
+- Change length condition of `BytesMut::truncate`
+- Reuse capacity when possible in `<BytesMut as Buf>::advance` impl (#698)
+- Improve `must_use` suggestion of `BytesMut::split` (#699)
+
+### Internal changes
+
+- Use `ManuallyDrop` instead of `mem::forget` (#678)
+- Don't set `len` in `BytesMut::reserve` (#682)
+- Optimize `Bytes::copy_to_bytes` (#688)
+- Refactor `BytesMut::truncate` (#694)
+- Refactor `BytesMut::resize` (#696)
+- Reorder assertion in `Bytes::split_to`, `Bytes::split_off` (#689, #693)
+- Use `offset_from` in more places (#705)
+- Correct the wrong usage of `IntoIter` (#707)
+
+# 1.6.1 (July 13, 2024)
+
+This release fixes a bug where `Bytes::is_unique` returns incorrect values when
+the `Bytes` originates from a shared `BytesMut`. (#718)
+
+# 1.6.0 (March 22, 2024)
+
+### Added
+
+- Add `Bytes::is_unique` (#643)
+
+### Documented
+
+- Fix changelog typo (#628)
+- Fix some spelling mistakes (#633)
+- Typo fix (#637)
+- Fix broken links (#639)
+- Add security policy (#649)
+
+### Internal changes
+
+- Move comment to correct constant (#629)
+- Various cleanup (#635)
+- Simplify `UninitSlice::as_uninit_slice_mut()` logic (#644)
+- Use `self.` instead of `Self::` (#642)
+- `BytesMut`: Assert alignment of `Shared` (#652)
+- Remove unnecessary namespace qualifier (#660)
+- Remove an unnecessary else branch (#662)
+- Remove unreachable else branch (#661)
+- make parameter mut in `From<Vec>` (#667)
+- Restore commented tests (#665)
+- Use `sub` instead of `offset` (#668)
+- Calculate original capacity only if necessary (#666)
+- `set_vec_pos` does not need a second parameter (#672)
+- `get_vec_pos`: use `&self` instead of `&mut self` (#670)
+- Refactor `split_at`/`split_to` (#663)
+- Use `Iterator` from the prelude (#673)
+- `copy_to_bytes`: Add panic section to docs (#676)
+- Remove redundant reserve call (#674)
+- Use `ManuallyDrop` instead of `mem::forget` (#675)
+
+
 # 1.5.0 (September 7, 2023)
 
 ### Added
diff --git a/Cargo.toml b/Cargo.toml
index 06b19e671..e07253942 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -4,7 +4,9 @@ name = "bytes"
 # When releasing to crates.io:
 # - Update CHANGELOG.md.
 # - Create "v1.x.y" git tag.
-version = "1.5.0"
+version = "1.7.1"
+edition = "2018"
+rust-version = "1.39"
 license = "MIT"
 authors = [
     "Carl Lerche <me@carllerche.com>",
@@ -15,7 +17,6 @@ repository = "https://github.com/tokio-rs/bytes"
 readme = "README.md"
 keywords = ["buffers", "zero-copy", "io"]
 categories = ["network-programming", "data-structures"]
-edition = "2018"
 
 [features]
 default = ["std"]
@@ -28,7 +29,7 @@ serde = { version = "1.0.60", optional = true, default-features = false, feature
 serde_test = "1.0"
 
 [target.'cfg(loom)'.dev-dependencies]
-loom = "0.5"
+loom = "0.7"
 
 [package.metadata.docs.rs]
 rustdoc-args = ["--cfg", "docsrs"]
diff --git a/SECURITY.md b/SECURITY.md
new file mode 100644
index 000000000..b74a831cb
--- /dev/null
+++ b/SECURITY.md
@@ -0,0 +1,9 @@
+# Security Policy
+
+Bytes is part of the Tokio project and uses the same security policy as [Tokio][tokio-security].
+
+## Report a security issue
+
+The process for reporting an issue is the same as for [Tokio][tokio-security]. This includes private reporting via security@tokio.rs.
+
+[tokio-security]: https://github.com/tokio-rs/tokio/security/policy
diff --git a/benches/bytes.rs b/benches/bytes.rs
index 61d1e832a..8782d0066 100644
--- a/benches/bytes.rs
+++ b/benches/bytes.rs
@@ -47,7 +47,7 @@ fn clone_static(b: &mut Bencher) {
 
     b.iter(|| {
         for _ in 0..1024 {
-            test::black_box(&bytes.clone());
+            test::black_box(test::black_box(&bytes).clone());
         }
     })
 }
@@ -58,7 +58,7 @@ fn clone_shared(b: &mut Bencher) {
 
     b.iter(|| {
         for _ in 0..1024 {
-            test::black_box(&bytes.clone());
+            test::black_box(test::black_box(&bytes).clone());
         }
     })
 }
@@ -70,7 +70,7 @@ fn clone_arc_vec(b: &mut Bencher) {
 
     b.iter(|| {
         for _ in 0..1024 {
-            test::black_box(&bytes.clone());
+            test::black_box(test::black_box(&bytes).clone());
         }
     })
 }
diff --git a/ci/test-stable.sh b/ci/test-stable.sh
old mode 100644
new mode 100755
index 4421f3a97..a8eaa3c65
--- a/ci/test-stable.sh
+++ b/ci/test-stable.sh
@@ -4,10 +4,6 @@ set -ex
 
 cmd="${1:-test}"
 
-# Install cargo-hack for feature flag test
-host=$(rustc -Vv | grep host | sed 's/host: //')
-curl -LsSf https://github.com/taiki-e/cargo-hack/releases/latest/download/cargo-hack-$host.tar.gz | tar xzf - -C ~/.cargo/bin
-
 # Run with each feature
 # * --each-feature includes both default/no-default features
 # * --optional-deps is needed for serde feature
@@ -15,8 +11,6 @@ cargo hack "${cmd}" --each-feature --optional-deps
 # Run with all features
 cargo "${cmd}" --all-features
 
-cargo doc --no-deps --all-features
-
 if [[ "${RUST_VERSION}" == "nightly"* ]]; then
     # Check benchmarks
     cargo check --benches
diff --git a/ci/tsan.sh b/ci/tsan.sh
old mode 100644
new mode 100755
diff --git a/src/buf/buf_impl.rs b/src/buf/buf_impl.rs
index b4ebf408a..9ef464075 100644
--- a/src/buf/buf_impl.rs
+++ b/src/buf/buf_impl.rs
@@ -66,6 +66,12 @@ macro_rules! buf_get_impl {
     }};
 }
 
+// https://en.wikipedia.org/wiki/Sign_extension
+fn sign_extend(val: u64, nbytes: usize) -> i64 {
+    let shift = (8 - nbytes) * 8;
+    (val << shift) as i64 >> shift
+}
+
 /// Read bytes from a buffer.
 ///
 /// A buffer stores bytes in memory such that read operations are infallible.
@@ -141,9 +147,11 @@ pub trait Buf {
     ///
     /// # Implementer notes
     ///
-    /// This function should never panic. Once the end of the buffer is reached,
-    /// i.e., `Buf::remaining` returns 0, calls to `chunk()` should return an
-    /// empty slice.
+    /// This function should never panic. `chunk()` should return an empty
+    /// slice **if and only if** `remaining()` returns 0. In other words,
+    /// `chunk()` returning an empty slice implies that `remaining()` will
+    /// return 0 and `remaining()` returning 0 implies that `chunk()` will
+    /// return an empty slice.
     // The `chunk` method was previously called `bytes`. This alias makes the rename
     // more easily discoverable.
     #[cfg_attr(docsrs, doc(alias = "bytes"))]
@@ -921,7 +929,7 @@ pub trait Buf {
     /// This function panics if there is not enough remaining data in `self`, or
     /// if `nbytes` is greater than 8.
     fn get_int(&mut self, nbytes: usize) -> i64 {
-        buf_get_impl!(be => self, i64, nbytes);
+        sign_extend(self.get_uint(nbytes), nbytes)
     }
 
     /// Gets a signed n-byte integer from `self` in little-endian byte order.
@@ -942,7 +950,7 @@ pub trait Buf {
     /// This function panics if there is not enough remaining data in `self`, or
     /// if `nbytes` is greater than 8.
     fn get_int_le(&mut self, nbytes: usize) -> i64 {
-        buf_get_impl!(le => self, i64, nbytes);
+        sign_extend(self.get_uint_le(nbytes), nbytes)
     }
 
     /// Gets a signed n-byte integer from `self` in native-endian byte order.
@@ -991,7 +999,7 @@ pub trait Buf {
     ///
     /// This function panics if there is not enough remaining data in `self`.
     fn get_f32(&mut self) -> f32 {
-        f32::from_bits(Self::get_u32(self))
+        f32::from_bits(self.get_u32())
     }
 
     /// Gets an IEEE754 single-precision (4 bytes) floating point number from
@@ -1012,7 +1020,7 @@ pub trait Buf {
     ///
     /// This function panics if there is not enough remaining data in `self`.
     fn get_f32_le(&mut self) -> f32 {
-        f32::from_bits(Self::get_u32_le(self))
+        f32::from_bits(self.get_u32_le())
     }
 
     /// Gets an IEEE754 single-precision (4 bytes) floating point number from
@@ -1036,7 +1044,7 @@ pub trait Buf {
     ///
     /// This function panics if there is not enough remaining data in `self`.
     fn get_f32_ne(&mut self) -> f32 {
-        f32::from_bits(Self::get_u32_ne(self))
+        f32::from_bits(self.get_u32_ne())
     }
 
     /// Gets an IEEE754 double-precision (8 bytes) floating point number from
@@ -1057,7 +1065,7 @@ pub trait Buf {
     ///
     /// This function panics if there is not enough remaining data in `self`.
     fn get_f64(&mut self) -> f64 {
-        f64::from_bits(Self::get_u64(self))
+        f64::from_bits(self.get_u64())
     }
 
     /// Gets an IEEE754 double-precision (8 bytes) floating point number from
@@ -1078,7 +1086,7 @@ pub trait Buf {
     ///
     /// This function panics if there is not enough remaining data in `self`.
     fn get_f64_le(&mut self) -> f64 {
-        f64::from_bits(Self::get_u64_le(self))
+        f64::from_bits(self.get_u64_le())
     }
 
     /// Gets an IEEE754 double-precision (8 bytes) floating point number from
@@ -1102,7 +1110,7 @@ pub trait Buf {
     ///
     /// This function panics if there is not enough remaining data in `self`.
     fn get_f64_ne(&mut self) -> f64 {
-        f64::from_bits(Self::get_u64_ne(self))
+        f64::from_bits(self.get_u64_ne())
     }
 
     /// Consumes `len` bytes inside self and returns new instance of `Bytes`
@@ -1120,6 +1128,10 @@ pub trait Buf {
     /// let bytes = (&b"hello world"[..]).copy_to_bytes(5);
     /// assert_eq!(&bytes[..], &b"hello"[..]);
     /// ```
+    ///
+    /// # Panics
+    ///
+    /// This function panics if `len > self.remaining()`.
     fn copy_to_bytes(&mut self, len: usize) -> crate::Bytes {
         use super::BufMut;
 
diff --git a/src/buf/buf_mut.rs b/src/buf/buf_mut.rs
index 304e11b13..2a3b5e9ee 100644
--- a/src/buf/buf_mut.rs
+++ b/src/buf/buf_mut.rs
@@ -165,7 +165,7 @@ pub unsafe trait BufMut {
     ///
     /// # Implementer notes
     ///
-    /// This function should never panic. `chunk_mut` should return an empty
+    /// This function should never panic. `chunk_mut()` should return an empty
     /// slice **if and only if** `remaining_mut()` returns 0. In other words,
     /// `chunk_mut()` returning an empty slice implies that `remaining_mut()` will
     /// return 0 and `remaining_mut()` returning 0 implies that `chunk_mut()` will
@@ -1107,7 +1107,7 @@ pub unsafe trait BufMut {
         }
     }
 
-    /// Writes  an IEEE754 single-precision (4 bytes) floating point number to
+    /// Writes an IEEE754 single-precision (4 bytes) floating point number to
     /// `self` in big-endian byte order.
     ///
     /// The current position is advanced by 4.
@@ -1131,7 +1131,7 @@ pub unsafe trait BufMut {
         self.put_u32(n.to_bits());
     }
 
-    /// Writes  an IEEE754 single-precision (4 bytes) floating point number to
+    /// Writes an IEEE754 single-precision (4 bytes) floating point number to
     /// `self` in little-endian byte order.
     ///
     /// The current position is advanced by 4.
@@ -1183,7 +1183,7 @@ pub unsafe trait BufMut {
         self.put_u32_ne(n.to_bits());
     }
 
-    /// Writes  an IEEE754 double-precision (8 bytes) floating point number to
+    /// Writes an IEEE754 double-precision (8 bytes) floating point number to
     /// `self` in big-endian byte order.
     ///
     /// The current position is advanced by 8.
@@ -1207,7 +1207,7 @@ pub unsafe trait BufMut {
         self.put_u64(n.to_bits());
     }
 
-    /// Writes  an IEEE754 double-precision (8 bytes) floating point number to
+    /// Writes an IEEE754 double-precision (8 bytes) floating point number to
     /// `self` in little-endian byte order.
     ///
     /// The current position is advanced by 8.
@@ -1231,7 +1231,7 @@ pub unsafe trait BufMut {
         self.put_u64_le(n.to_bits());
     }
 
-    /// Writes  an IEEE754 double-precision (8 bytes) floating point number to
+    /// Writes an IEEE754 double-precision (8 bytes) floating point number to
     /// `self` in native-endian byte order.
     ///
     /// The current position is advanced by 8.
diff --git a/src/buf/uninit_slice.rs b/src/buf/uninit_slice.rs
index 0715ad233..aea096ae6 100644
--- a/src/buf/uninit_slice.rs
+++ b/src/buf/uninit_slice.rs
@@ -110,7 +110,7 @@ impl UninitSlice {
         unsafe { self[index..].as_mut_ptr().write(byte) }
     }
 
-    /// Copies bytes  from `src` into `self`.
+    /// Copies bytes from `src` into `self`.
     ///
     /// The length of `src` must be the same as `self`.
     ///
@@ -185,7 +185,7 @@ impl UninitSlice {
     /// ```
     #[inline]
     pub unsafe fn as_uninit_slice_mut(&mut self) -> &mut [MaybeUninit<u8>] {
-        &mut *(self as *mut _ as *mut [MaybeUninit<u8>])
+        &mut self.0
     }
 
     /// Returns the number of bytes in the slice.
diff --git a/src/bytes.rs b/src/bytes.rs
index 9fed3d287..7c3b72dd9 100644
--- a/src/bytes.rs
+++ b/src/bytes.rs
@@ -1,6 +1,7 @@
 use core::iter::FromIterator;
+use core::mem::{self, ManuallyDrop};
 use core::ops::{Deref, RangeBounds};
-use core::{cmp, fmt, hash, mem, ptr, slice, usize};
+use core::{cmp, fmt, hash, ptr, slice, usize};
 
 use alloc::{
     alloc::{dealloc, Layout},
@@ -14,7 +15,7 @@ use crate::buf::IntoIter;
 #[allow(unused)]
 use crate::loom::sync::atomic::AtomicMut;
 use crate::loom::sync::atomic::{AtomicPtr, AtomicUsize, Ordering};
-use crate::Buf;
+use crate::{offset_from, Buf, BytesMut};
 
 /// A cheaply cloneable and sliceable chunk of contiguous memory.
 ///
@@ -112,6 +113,9 @@ pub(crate) struct Vtable {
     ///
     /// takes `Bytes` to value
     pub to_vec: unsafe fn(&AtomicPtr<()>, *const u8, usize) -> Vec<u8>,
+    pub to_mut: unsafe fn(&AtomicPtr<()>, *const u8, usize) -> BytesMut,
+    /// fn(data)
+    pub is_unique: unsafe fn(&AtomicPtr<()>) -> bool,
     /// fn(data, ptr, len)
     pub drop: unsafe fn(&mut AtomicPtr<()>, *const u8, usize),
 }
@@ -208,6 +212,28 @@ impl Bytes {
         self.len == 0
     }
 
+    /// Returns true if this is the only reference to the data.
+    ///
+    /// Always returns false if the data is backed by a static slice.
+    ///
+    /// 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)
+    /// # Examples
+    ///
+    /// ```
+    /// use bytes::Bytes;
+    ///
+    /// let a = Bytes::from(vec![1, 2, 3]);
+    /// assert!(a.is_unique());
+    /// let b = a.clone();
+    /// assert!(!a.is_unique());
+    /// ```
+    pub fn is_unique(&self) -> bool {
+        unsafe { (self.vtable.is_unique)(&self.data) }
+    }
+
     /// Creates `Bytes` instance from slice, by copying it.
     pub fn copy_from_slice(data: &[u8]) -> Self {
         data.to_vec().into()
@@ -360,13 +386,6 @@ impl Bytes {
     /// Panics if `at > len`.
     #[must_use = "consider Bytes::truncate if you don't need the other half"]
     pub fn split_off(&mut self, at: usize) -> Self {
-        assert!(
-            at <= self.len(),
-            "split_off out of bounds: {:?} <= {:?}",
-            at,
-            self.len(),
-        );
-
         if at == self.len() {
             return Bytes::new();
         }
@@ -375,6 +394,13 @@ impl Bytes {
             return mem::replace(self, Bytes::new());
         }
 
+        assert!(
+            at <= self.len(),
+            "split_off out of bounds: {:?} <= {:?}",
+            at,
+            self.len(),
+        );
+
         let mut ret = self.clone();
 
         self.len = at;
@@ -409,13 +435,6 @@ impl Bytes {
     /// Panics if `at > len`.
     #[must_use = "consider Bytes::advance if you don't need the other half"]
     pub fn split_to(&mut self, at: usize) -> Self {
-        assert!(
-            at <= self.len(),
-            "split_to out of bounds: {:?} <= {:?}",
-            at,
-            self.len(),
-        );
-
         if at == self.len() {
             return mem::replace(self, Bytes::new());
         }
@@ -424,6 +443,13 @@ impl Bytes {
             return Bytes::new();
         }
 
+        assert!(
+            at <= self.len(),
+            "split_to out of bounds: {:?} <= {:?}",
+            at,
+            self.len(),
+        );
+
         let mut ret = self.clone();
 
         unsafe { self.inc_start(at) };
@@ -482,6 +508,29 @@ impl Bytes {
         self.truncate(0);
     }
 
+    /// Try to convert self into `BytesMut`.
+    ///
+    /// If `self` is unique for the entire original buffer, this will succeed
+    /// and return a `BytesMut` with the contents of `self` without copying.
+    /// If `self` is not unique for the entire original buffer, this will fail
+    /// and return self.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bytes::{Bytes, BytesMut};
+    ///
+    /// let bytes = Bytes::from(b"hello".to_vec());
+    /// assert_eq!(bytes.try_into_mut(), Ok(BytesMut::from(&b"hello"[..])));
+    /// ```
+    pub fn try_into_mut(self) -> Result<BytesMut, Bytes> {
+        if self.is_unique() {
+            Ok(self.into())
+        } else {
+            Err(self)
+        }
+    }
+
     #[inline]
     pub(crate) unsafe fn with_vtable(
         ptr: *const u8,
@@ -531,6 +580,21 @@ impl Clone for Bytes {
     }
 }
 
+macro_rules! buf_get_impl {
+    ($this:ident, $typ:tt::$conv:tt) => {{
+        const SIZE: usize = mem::size_of::<$typ>();
+
+        assert!($this.len >= SIZE);
+        // slice.get() returns None on failing bounds check, resulting in a panic, but skips the unnecessary code of the
+        // default buf impl that needs to account for non-contiguous memory
+        let ret = unsafe { $typ::$conv(($this.ptr as *const $typ).read_unaligned()) };
+
+        // if the direct conversion was possible, advance and return
+        $this.advance(SIZE);
+        return ret;
+    }};
+}
+
 impl Buf for Bytes {
     #[inline]
     fn remaining(&self) -> usize {
@@ -556,14 +620,138 @@ impl Buf for Bytes {
         }
     }
 
-    fn copy_to_bytes(&mut self, len: usize) -> crate::Bytes {
-        if len == self.remaining() {
-            core::mem::replace(self, Bytes::new())
-        } else {
-            let ret = self.slice(..len);
-            self.advance(len);
-            ret
-        }
+    fn copy_to_bytes(&mut self, len: usize) -> Self {
+        self.split_to(len)
+    }
+
+    #[inline]
+    fn get_u8(&mut self) -> u8 {
+        buf_get_impl!(self, u8::from)
+    }
+
+    #[inline]
+    fn get_i8(&mut self) -> i8 {
+        buf_get_impl!(self, i8::from)
+    }
+
+    #[inline]
+    fn get_u16(&mut self) -> u16 {
+        buf_get_impl!(self, u16::from_be);
+    }
+
+    #[inline]
+    fn get_u16_le(&mut self) -> u16 {
+        buf_get_impl!(self, u16::from_le);
+    }
+
+    #[inline]
+    fn get_u16_ne(&mut self) -> u16 {
+        buf_get_impl!(self, u16::from);
+    }
+
+    #[inline]
+    fn get_i16(&mut self) -> i16 {
+        buf_get_impl!(self, i16::from_be);
+    }
+
+    #[inline]
+    fn get_i16_le(&mut self) -> i16 {
+        buf_get_impl!(self, i16::from_le);
+    }
+
+    #[inline]
+    fn get_i16_ne(&mut self) -> i16 {
+        buf_get_impl!(self, i16::from);
+    }
+
+    #[inline]
+    fn get_u32(&mut self) -> u32 {
+        buf_get_impl!(self, u32::from_be);
+    }
+
+    #[inline]
+    fn get_u32_le(&mut self) -> u32 {
+        buf_get_impl!(self, u32::from_le);
+    }
+
+    #[inline]
+    fn get_u32_ne(&mut self) -> u32 {
+        buf_get_impl!(self, u32::from);
+    }
+
+    #[inline]
+    fn get_i32(&mut self) -> i32 {
+        buf_get_impl!(self, i32::from_be);
+    }
+
+    #[inline]
+    fn get_i32_le(&mut self) -> i32 {
+        buf_get_impl!(self, i32::from_le);
+    }
+
+    #[inline]
+    fn get_i32_ne(&mut self) -> i32 {
+        buf_get_impl!(self, i32::from);
+    }
+
+    #[inline]
+    fn get_u64(&mut self) -> u64 {
+        buf_get_impl!(self, u64::from_be);
+    }
+
+    #[inline]
+    fn get_u64_le(&mut self) -> u64 {
+        buf_get_impl!(self, u64::from_le);
+    }
+
+    #[inline]
+    fn get_u64_ne(&mut self) -> u64 {
+        buf_get_impl!(self, u64::from);
+    }
+
+    #[inline]
+    fn get_i64(&mut self) -> i64 {
+        buf_get_impl!(self, i64::from_be);
+    }
+
+    #[inline]
+    fn get_i64_le(&mut self) -> i64 {
+        buf_get_impl!(self, i64::from_le);
+    }
+
+    #[inline]
+    fn get_i64_ne(&mut self) -> i64 {
+        buf_get_impl!(self, i64::from);
+    }
+
+    #[inline]
+    fn get_u128(&mut self) -> u128 {
+        buf_get_impl!(self, u128::from_be);
+    }
+
+    #[inline]
+    fn get_u128_le(&mut self) -> u128 {
+        buf_get_impl!(self, u128::from_le);
+    }
+
+    #[inline]
+    fn get_u128_ne(&mut self) -> u128 {
+        buf_get_impl!(self, u128::from);
+    }
+
+    #[inline]
+    fn get_i128(&mut self) -> i128 {
+        buf_get_impl!(self, i128::from_be);
+    }
+
+    #[inline]
+    fn get_i128_le(&mut self) -> i128 {
+        buf_get_impl!(self, i128::from_le);
+    }
+
+    #[inline]
+    fn get_i128_ne(&mut self) -> i128 {
+        buf_get_impl!(self, i128::from);
     }
 }
 
@@ -805,13 +993,14 @@ impl From<&'static str> for Bytes {
 
 impl From<Vec<u8>> for Bytes {
     fn from(vec: Vec<u8>) -> Bytes {
-        let mut vec = vec;
+        let mut vec = ManuallyDrop::new(vec);
         let ptr = vec.as_mut_ptr();
         let len = vec.len();
         let cap = vec.capacity();
 
         // Avoid an extra allocation if possible.
         if len == cap {
+            let vec = ManuallyDrop::into_inner(vec);
             return Bytes::from(vec.into_boxed_slice());
         }
 
@@ -820,7 +1009,6 @@ impl From<Vec<u8>> for Bytes {
             cap,
             ref_cnt: AtomicUsize::new(1),
         });
-        mem::forget(vec);
 
         let shared = Box::into_raw(shared);
         // The pointer should be aligned, so this assert should
@@ -869,6 +1057,28 @@ impl From<Box<[u8]>> for Bytes {
     }
 }
 
+impl From<Bytes> for BytesMut {
+    /// Convert self into `BytesMut`.
+    ///
+    /// If `bytes` is unique for the entire original buffer, this will return a
+    /// `BytesMut` with the contents of `bytes` without copying.
+    /// If `bytes` is not unique for the entire original buffer, this will make
+    /// a copy of `bytes` subset of the original buffer in a new `BytesMut`.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bytes::{Bytes, BytesMut};
+    ///
+    /// let bytes = Bytes::from(b"hello".to_vec());
+    /// assert_eq!(BytesMut::from(bytes), BytesMut::from(&b"hello"[..]));
+    /// ```
+    fn from(bytes: Bytes) -> Self {
+        let bytes = ManuallyDrop::new(bytes);
+        unsafe { (bytes.vtable.to_mut)(&bytes.data, bytes.ptr, bytes.len) }
+    }
+}
+
 impl From<String> for Bytes {
     fn from(s: String) -> Bytes {
         Bytes::from(s.into_bytes())
@@ -877,7 +1087,7 @@ impl From<String> for Bytes {
 
 impl From<Bytes> for Vec<u8> {
     fn from(bytes: Bytes) -> Vec<u8> {
-        let bytes = mem::ManuallyDrop::new(bytes);
+        let bytes = ManuallyDrop::new(bytes);
         unsafe { (bytes.vtable.to_vec)(&bytes.data, bytes.ptr, bytes.len) }
     }
 }
@@ -898,6 +1108,8 @@ impl fmt::Debug for Vtable {
 const STATIC_VTABLE: Vtable = Vtable {
     clone: static_clone,
     to_vec: static_to_vec,
+    to_mut: static_to_mut,
+    is_unique: static_is_unique,
     drop: static_drop,
 };
 
@@ -911,6 +1123,15 @@ unsafe fn static_to_vec(_: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Vec<u8
     slice.to_vec()
 }
 
+unsafe fn static_to_mut(_: &AtomicPtr<()>, ptr: *const u8, len: usize) -> BytesMut {
+    let slice = slice::from_raw_parts(ptr, len);
+    BytesMut::from(slice)
+}
+
+fn static_is_unique(_: &AtomicPtr<()>) -> bool {
+    false
+}
+
 unsafe fn static_drop(_: &mut AtomicPtr<()>, _: *const u8, _: usize) {
     // nothing to drop for &'static [u8]
 }
@@ -920,12 +1141,16 @@ unsafe fn static_drop(_: &mut AtomicPtr<()>, _: *const u8, _: usize) {
 static PROMOTABLE_EVEN_VTABLE: Vtable = Vtable {
     clone: promotable_even_clone,
     to_vec: promotable_even_to_vec,
+    to_mut: promotable_even_to_mut,
+    is_unique: promotable_is_unique,
     drop: promotable_even_drop,
 };
 
 static PROMOTABLE_ODD_VTABLE: Vtable = Vtable {
     clone: promotable_odd_clone,
     to_vec: promotable_odd_to_vec,
+    to_mut: promotable_odd_to_mut,
+    is_unique: promotable_is_unique,
     drop: promotable_odd_drop,
 };
 
@@ -959,7 +1184,7 @@ unsafe fn promotable_to_vec(
 
         let buf = f(shared);
 
-        let cap = (ptr as usize - buf as usize) + len;
+        let cap = offset_from(ptr, buf) + len;
 
         // Copy back buffer
         ptr::copy(ptr, buf, len);
@@ -968,12 +1193,47 @@ unsafe fn promotable_to_vec(
     }
 }
 
+unsafe fn promotable_to_mut(
+    data: &AtomicPtr<()>,
+    ptr: *const u8,
+    len: usize,
+    f: fn(*mut ()) -> *mut u8,
+) -> BytesMut {
+    let shared = data.load(Ordering::Acquire);
+    let kind = shared as usize & KIND_MASK;
+
+    if kind == KIND_ARC {
+        shared_to_mut_impl(shared.cast(), ptr, len)
+    } else {
+        // KIND_VEC is a view of an underlying buffer at a certain offset.
+        // The ptr + len always represents the end of that buffer.
+        // Before truncating it, it is first promoted to KIND_ARC.
+        // Thus, we can safely reconstruct a Vec from it without leaking memory.
+        debug_assert_eq!(kind, KIND_VEC);
+
+        let buf = f(shared);
+        let off = offset_from(ptr, buf);
+        let cap = off + len;
+        let v = Vec::from_raw_parts(buf, cap, cap);
+
+        let mut b = BytesMut::from_vec(v);
+        b.advance_unchecked(off);
+        b
+    }
+}
+
 unsafe fn promotable_even_to_vec(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Vec<u8> {
     promotable_to_vec(data, ptr, len, |shared| {
         ptr_map(shared.cast(), |addr| addr & !KIND_MASK)
     })
 }
 
+unsafe fn promotable_even_to_mut(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> BytesMut {
+    promotable_to_mut(data, ptr, len, |shared| {
+        ptr_map(shared.cast(), |addr| addr & !KIND_MASK)
+    })
+}
+
 unsafe fn promotable_even_drop(data: &mut AtomicPtr<()>, ptr: *const u8, len: usize) {
     data.with_mut(|shared| {
         let shared = *shared;
@@ -1005,6 +1265,10 @@ unsafe fn promotable_odd_to_vec(data: &AtomicPtr<()>, ptr: *const u8, len: usize
     promotable_to_vec(data, ptr, len, |shared| shared.cast())
 }
 
+unsafe fn promotable_odd_to_mut(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> BytesMut {
+    promotable_to_mut(data, ptr, len, |shared| shared.cast())
+}
+
 unsafe fn promotable_odd_drop(data: &mut AtomicPtr<()>, ptr: *const u8, len: usize) {
     data.with_mut(|shared| {
         let shared = *shared;
@@ -1020,8 +1284,20 @@ unsafe fn promotable_odd_drop(data: &mut AtomicPtr<()>, ptr: *const u8, len: usi
     });
 }
 
+unsafe fn promotable_is_unique(data: &AtomicPtr<()>) -> bool {
+    let shared = data.load(Ordering::Acquire);
+    let kind = shared as usize & KIND_MASK;
+
+    if kind == KIND_ARC {
+        let ref_cnt = (*shared.cast::<Shared>()).ref_cnt.load(Ordering::Relaxed);
+        ref_cnt == 1
+    } else {
+        true
+    }
+}
+
 unsafe fn free_boxed_slice(buf: *mut u8, offset: *const u8, len: usize) {
-    let cap = (offset as usize - buf as usize) + len;
+    let cap = offset_from(offset, buf) + len;
     dealloc(buf, Layout::from_size_align(cap, 1).unwrap())
 }
 
@@ -1049,6 +1325,8 @@ const _: [(); 0 - mem::align_of::<Shared>() % 2] = []; // Assert that the alignm
 static SHARED_VTABLE: Vtable = Vtable {
     clone: shared_clone,
     to_vec: shared_to_vec,
+    to_mut: shared_to_mut,
+    is_unique: shared_is_unique,
     drop: shared_drop,
 };
 
@@ -1073,11 +1351,11 @@ unsafe fn shared_to_vec_impl(shared: *mut Shared, ptr: *const u8, len: usize) ->
         .compare_exchange(1, 0, Ordering::AcqRel, Ordering::Relaxed)
         .is_ok()
     {
-        let buf = (*shared).buf;
-        let cap = (*shared).cap;
-
-        // Deallocate Shared
-        drop(Box::from_raw(shared as *mut mem::ManuallyDrop<Shared>));
+        // Deallocate the `Shared` instance without running its destructor.
+        let shared = *Box::from_raw(shared);
+        let shared = ManuallyDrop::new(shared);
+        let buf = shared.buf;
+        let cap = shared.cap;
 
         // Copy back buffer
         ptr::copy(ptr, buf, len);
@@ -1094,6 +1372,51 @@ unsafe fn shared_to_vec(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> Vec
     shared_to_vec_impl(data.load(Ordering::Relaxed).cast(), ptr, len)
 }
 
+unsafe fn shared_to_mut_impl(shared: *mut Shared, ptr: *const u8, len: usize) -> BytesMut {
+    // The goal is to check if the current handle is the only handle
+    // that currently has access to the buffer. This is done by
+    // checking if the `ref_cnt` is currently 1.
+    //
+    // The `Acquire` ordering synchronizes with the `Release` as
+    // part of the `fetch_sub` in `release_shared`. The `fetch_sub`
+    // operation guarantees that any mutations done in other threads
+    // are ordered before the `ref_cnt` is decremented. As such,
+    // this `Acquire` will guarantee that those mutations are
+    // visible to the current thread.
+    //
+    // Otherwise, we take the other branch, copy the data and call `release_shared`.
+    if (*shared).ref_cnt.load(Ordering::Acquire) == 1 {
+        // Deallocate the `Shared` instance without running its destructor.
+        let shared = *Box::from_raw(shared);
+        let shared = ManuallyDrop::new(shared);
+        let buf = shared.buf;
+        let cap = shared.cap;
+
+        // Rebuild Vec
+        let off = offset_from(ptr, buf);
+        let v = Vec::from_raw_parts(buf, len + off, cap);
+
+        let mut b = BytesMut::from_vec(v);
+        b.advance_unchecked(off);
+        b
+    } else {
+        // Copy the data from Shared in a new Vec, then release it
+        let v = slice::from_raw_parts(ptr, len).to_vec();
+        release_shared(shared);
+        BytesMut::from_vec(v)
+    }
+}
+
+unsafe fn shared_to_mut(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> BytesMut {
+    shared_to_mut_impl(data.load(Ordering::Relaxed).cast(), ptr, len)
+}
+
+pub(crate) unsafe fn shared_is_unique(data: &AtomicPtr<()>) -> bool {
+    let shared = data.load(Ordering::Acquire);
+    let ref_cnt = (*shared.cast::<Shared>()).ref_cnt.load(Ordering::Relaxed);
+    ref_cnt == 1
+}
+
 unsafe fn shared_drop(data: &mut AtomicPtr<()>, _ptr: *const u8, _len: usize) {
     data.with_mut(|shared| {
         release_shared(shared.cast());
@@ -1123,7 +1446,7 @@ unsafe fn shallow_clone_vec(
     offset: *const u8,
     len: usize,
 ) -> Bytes {
-    // If  the buffer is still tracked in a `Vec<u8>`. It is time to
+    // 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.
 
@@ -1136,7 +1459,7 @@ unsafe fn shallow_clone_vec(
     // vector.
     let shared = Box::new(Shared {
         buf,
-        cap: (offset as usize - buf as usize) + len,
+        cap: offset_from(offset, buf) + len,
         // Initialize refcount to 2. One for this reference, and one
         // for the new clone that will be returned from
         // `shallow_clone`.
diff --git a/src/bytes_mut.rs b/src/bytes_mut.rs
index 57fd33e4b..ec74c4e97 100644
--- a/src/bytes_mut.rs
+++ b/src/bytes_mut.rs
@@ -1,4 +1,4 @@
-use core::iter::{FromIterator, Iterator};
+use core::iter::FromIterator;
 use core::mem::{self, ManuallyDrop, MaybeUninit};
 use core::ops::{Deref, DerefMut};
 use core::ptr::{self, NonNull};
@@ -17,7 +17,7 @@ use crate::bytes::Vtable;
 #[allow(unused)]
 use crate::loom::sync::atomic::AtomicMut;
 use crate::loom::sync::atomic::{AtomicPtr, AtomicUsize, Ordering};
-use crate::{Buf, BufMut, Bytes};
+use crate::{offset_from, Buf, BufMut, Bytes};
 
 /// A unique reference to a contiguous slice of memory.
 ///
@@ -80,6 +80,12 @@ struct Shared {
     ref_count: AtomicUsize,
 }
 
+// Assert that the alignment of `Shared` is divisible by 2.
+// This is a necessary invariant since we depend on allocating `Shared` a
+// shared object to implicitly carry the `KIND_ARC` flag in its pointer.
+// This flag is set when the LSB is 0.
+const _: [(); 0 - mem::align_of::<Shared>() % 2] = []; // Assert that the alignment of `Shared` is divisible by 2.
+
 // Buffer storage strategy flags.
 const KIND_ARC: usize = 0b0;
 const KIND_VEC: usize = 0b1;
@@ -237,29 +243,35 @@ impl BytesMut {
     /// th.join().unwrap();
     /// ```
     #[inline]
-    pub fn freeze(mut self) -> Bytes {
-        if self.kind() == KIND_VEC {
+    pub fn freeze(self) -> Bytes {
+        let bytes = ManuallyDrop::new(self);
+        if bytes.kind() == KIND_VEC {
             // Just re-use `Bytes` internal Vec vtable
             unsafe {
-                let (off, _) = self.get_vec_pos();
-                let vec = rebuild_vec(self.ptr.as_ptr(), self.len, self.cap, off);
-                mem::forget(self);
+                let off = bytes.get_vec_pos();
+                let vec = rebuild_vec(bytes.ptr.as_ptr(), bytes.len, bytes.cap, off);
                 let mut b: Bytes = vec.into();
                 b.advance(off);
                 b
             }
         } else {
-            debug_assert_eq!(self.kind(), KIND_ARC);
+            debug_assert_eq!(bytes.kind(), KIND_ARC);
 
-            let ptr = self.ptr.as_ptr();
-            let len = self.len;
-            let data = AtomicPtr::new(self.data.cast());
-            mem::forget(self);
+            let ptr = bytes.ptr.as_ptr();
+            let len = bytes.len;
+            let data = AtomicPtr::new(bytes.data.cast());
             unsafe { Bytes::with_vtable(ptr, len, data, &SHARED_VTABLE) }
         }
     }
 
-    /// Creates a new `BytesMut`, which is initialized with zero.
+    /// Creates a new `BytesMut` containing `len` zeros.
+    ///
+    /// The resulting object has a length of `len` and a capacity greater
+    /// than or equal to `len`. The entire length of the object will be filled
+    /// with zeros.
+    ///
+    /// On some platforms or allocators this function may be faster than
+    /// a manual implementation.
     ///
     /// # Examples
     ///
@@ -268,6 +280,7 @@ impl BytesMut {
     ///
     /// let zeros = BytesMut::zeroed(42);
     ///
+    /// assert!(zeros.capacity() >= 42);
     /// assert_eq!(zeros.len(), 42);
     /// zeros.into_iter().for_each(|x| assert_eq!(x, 0));
     /// ```
@@ -311,8 +324,10 @@ impl BytesMut {
         );
         unsafe {
             let mut other = self.shallow_clone();
-            other.set_start(at);
-            self.set_end(at);
+            // SAFETY: We've checked that `at` <= `self.capacity()` above.
+            other.advance_unchecked(at);
+            self.cap = at;
+            self.len = cmp::min(self.len, at);
             other
         }
     }
@@ -342,7 +357,7 @@ impl BytesMut {
     ///
     /// assert_eq!(other, b"hello world"[..]);
     /// ```
-    #[must_use = "consider BytesMut::advance(len()) if you don't need the other half"]
+    #[must_use = "consider BytesMut::clear if you don't need the other half"]
     pub fn split(&mut self) -> BytesMut {
         let len = self.len();
         self.split_to(len)
@@ -385,8 +400,11 @@ impl BytesMut {
 
         unsafe {
             let mut other = self.shallow_clone();
-            other.set_end(at);
-            self.set_start(at);
+            // SAFETY: We've checked that `at` <= `self.len()` and we know that `self.len()` <=
+            // `self.capacity()`.
+            self.advance_unchecked(at);
+            other.cap = at;
+            other.len = at;
             other
         }
     }
@@ -413,9 +431,8 @@ impl BytesMut {
     /// ```
     pub fn truncate(&mut self, len: usize) {
         if len <= self.len() {
-            unsafe {
-                self.set_len(len);
-            }
+            // SAFETY: Shrinking the buffer cannot expose uninitialized bytes.
+            unsafe { self.set_len(len) };
         }
     }
 
@@ -431,7 +448,8 @@ impl BytesMut {
     /// assert!(buf.is_empty());
     /// ```
     pub fn clear(&mut self) {
-        self.truncate(0);
+        // SAFETY: Setting the length to zero cannot expose uninitialized bytes.
+        unsafe { self.set_len(0) };
     }
 
     /// Resizes the buffer so that `len` is equal to `new_len`.
@@ -457,18 +475,26 @@ impl BytesMut {
     /// assert_eq!(&buf[..], &[0x1, 0x1, 0x3, 0x3]);
     /// ```
     pub fn resize(&mut self, new_len: usize, value: u8) {
-        let len = self.len();
-        if new_len > len {
-            let additional = new_len - len;
-            self.reserve(additional);
-            unsafe {
-                let dst = self.chunk_mut().as_mut_ptr();
-                ptr::write_bytes(dst, value, additional);
-                self.set_len(new_len);
-            }
+        let additional = if let Some(additional) = new_len.checked_sub(self.len()) {
+            additional
         } else {
             self.truncate(new_len);
+            return;
+        };
+
+        if additional == 0 {
+            return;
         }
+
+        self.reserve(additional);
+        let dst = self.spare_capacity_mut().as_mut_ptr();
+        // SAFETY: `spare_capacity_mut` returns a valid, properly aligned pointer and we've
+        // reserved enough space to write `additional` bytes.
+        unsafe { ptr::write_bytes(dst, value, additional) };
+
+        // SAFETY: There are at least `new_len` initialized bytes in the buffer so no
+        // uninitialized bytes are being exposed.
+        unsafe { self.set_len(new_len) };
     }
 
     /// Sets the length of the buffer.
@@ -571,12 +597,13 @@ impl BytesMut {
             return;
         }
 
-        self.reserve_inner(additional);
+        // will always succeed
+        let _ = self.reserve_inner(additional, true);
     }
 
-    // In separate function to allow the short-circuits in `reserve` to
-    // be inline-able. Significant helps performance.
-    fn reserve_inner(&mut self, additional: usize) {
+    // In separate function to allow the short-circuits in `reserve` and `try_reclaim` to
+    // be inline-able. Significantly helps performance. Returns false if it did not succeed.
+    fn reserve_inner(&mut self, additional: usize, allocate: bool) -> bool {
         let len = self.len();
         let kind = self.kind();
 
@@ -590,7 +617,7 @@ impl BytesMut {
             // We need to make sure that this optimization does not kill the
             // amortized runtimes of BytesMut's operations.
             unsafe {
-                let (off, prev) = self.get_vec_pos();
+                let off = self.get_vec_pos();
 
                 // Only reuse space if we can satisfy the requested additional space.
                 //
@@ -611,16 +638,19 @@ impl BytesMut {
                     //
                     // Just move the pointer back to the start after copying
                     // data back.
-                    let base_ptr = self.ptr.as_ptr().offset(-(off as isize));
+                    let base_ptr = self.ptr.as_ptr().sub(off);
                     // Since `off >= self.len()`, the two regions don't overlap.
                     ptr::copy_nonoverlapping(self.ptr.as_ptr(), base_ptr, self.len);
                     self.ptr = vptr(base_ptr);
-                    self.set_vec_pos(0, prev);
+                    self.set_vec_pos(0);
 
                     // Length stays constant, but since we moved backwards we
                     // can gain capacity back.
                     self.cap += off;
                 } else {
+                    if !allocate {
+                        return false;
+                    }
                     // Not enough space, or reusing might be too much overhead:
                     // allocate more space!
                     let mut v =
@@ -629,11 +659,11 @@ impl BytesMut {
 
                     // Update the info
                     self.ptr = vptr(v.as_mut_ptr().add(off));
-                    self.len = v.len() - off;
                     self.cap = v.capacity() - off;
+                    debug_assert_eq!(self.len, v.len() - off);
                 }
 
-                return;
+                return true;
             }
         }
 
@@ -644,15 +674,13 @@ impl BytesMut {
         // allocating a new vector with the requested capacity.
         //
         // Compute the new capacity
-        let mut new_cap = len.checked_add(additional).expect("overflow");
-
-        let original_capacity;
-        let original_capacity_repr;
+        let mut new_cap = match len.checked_add(additional) {
+            Some(new_cap) => new_cap,
+            None if !allocate => return false,
+            None => panic!("overflow"),
+        };
 
         unsafe {
-            original_capacity_repr = (*shared).original_capacity_repr;
-            original_capacity = original_capacity_from_repr(original_capacity_repr);
-
             // First, try to reclaim the buffer. This is possible if the current
             // handle is the only outstanding handle pointing to the buffer.
             if (*shared).is_unique() {
@@ -681,6 +709,9 @@ impl BytesMut {
                     self.ptr = vptr(ptr);
                     self.cap = v.capacity();
                 } else {
+                    if !allocate {
+                        return false;
+                    }
                     // calculate offset
                     let off = (self.ptr.as_ptr() as usize) - (v.as_ptr() as usize);
 
@@ -719,11 +750,17 @@ impl BytesMut {
                     self.cap = v.capacity() - off;
                 }
 
-                return;
-            } else {
-                new_cap = cmp::max(new_cap, original_capacity);
+                return true;
             }
         }
+        if !allocate {
+            return false;
+        }
+
+        let original_capacity_repr = unsafe { (*shared).original_capacity_repr };
+        let original_capacity = original_capacity_from_repr(original_capacity_repr);
+
+        new_cap = cmp::max(new_cap, original_capacity);
 
         // Create a new vector to store the data
         let mut v = ManuallyDrop::new(Vec::with_capacity(new_cap));
@@ -739,8 +776,69 @@ impl BytesMut {
         let data = (original_capacity_repr << ORIGINAL_CAPACITY_OFFSET) | KIND_VEC;
         self.data = invalid_ptr(data);
         self.ptr = vptr(v.as_mut_ptr());
-        self.len = v.len();
         self.cap = v.capacity();
+        debug_assert_eq!(self.len, v.len());
+        return true;
+    }
+
+    /// Attempts to cheaply reclaim already allocated capacity for at least `additional` more
+    /// bytes to be inserted into the given `BytesMut` and returns `true` if it succeeded.
+    ///
+    /// `try_reclaim` behaves exactly like `reserve`, except that it never allocates new storage
+    /// and returns a `bool` indicating whether it was successful in doing so:
+    ///
+    /// `try_reclaim` returns false under these conditions:
+    ///  - The spare capacity left is less than `additional` bytes AND
+    ///  - The existing allocation cannot be reclaimed cheaply or it was less than
+    ///    `additional` bytes in size
+    ///
+    /// Reclaiming the allocation cheaply is possible if the `BytesMut` has no outstanding
+    /// references through other `BytesMut`s or `Bytes` which point to the same underlying
+    /// storage.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use bytes::BytesMut;
+    ///
+    /// let mut buf = BytesMut::with_capacity(64);
+    /// assert_eq!(true, buf.try_reclaim(64));
+    /// assert_eq!(64, buf.capacity());
+    ///
+    /// buf.extend_from_slice(b"abcd");
+    /// let mut split = buf.split();
+    /// assert_eq!(60, buf.capacity());
+    /// assert_eq!(4, split.capacity());
+    /// assert_eq!(false, split.try_reclaim(64));
+    /// assert_eq!(false, buf.try_reclaim(64));
+    /// // The split buffer is filled with "abcd"
+    /// assert_eq!(false, split.try_reclaim(4));
+    /// // buf is empty and has capacity for 60 bytes
+    /// assert_eq!(true, buf.try_reclaim(60));
+    ///
+    /// drop(buf);
+    /// assert_eq!(false, split.try_reclaim(64));
+    ///
+    /// split.clear();
+    /// assert_eq!(4, split.capacity());
+    /// assert_eq!(true, split.try_reclaim(64));
+    /// assert_eq!(64, split.capacity());
+    /// ```
+    // I tried splitting out try_reclaim_inner after the short circuits, but it was inlined
+    // regardless with Rust 1.78.0 so probably not worth it
+    #[inline]
+    #[must_use = "consider BytesMut::reserve if you need an infallible reservation"]
+    pub fn try_reclaim(&mut self, additional: usize) -> bool {
+        let len = self.len();
+        let rem = self.capacity() - len;
+
+        if additional <= rem {
+            // The handle can already store at least `additional` more bytes, so
+            // there is no further work needed to be done.
+            return true;
+        }
+
+        self.reserve_inner(additional, false)
     }
 
     /// Appends given bytes to this `BytesMut`.
@@ -821,11 +919,11 @@ impl BytesMut {
     // internal change could make a simple pattern (`BytesMut::from(vec)`)
     // suddenly a lot more expensive.
     #[inline]
-    pub(crate) fn from_vec(mut vec: Vec<u8>) -> BytesMut {
+    pub(crate) fn from_vec(vec: Vec<u8>) -> BytesMut {
+        let mut vec = ManuallyDrop::new(vec);
         let ptr = vptr(vec.as_mut_ptr());
         let len = vec.len();
         let cap = vec.capacity();
-        mem::forget(vec);
 
         let original_capacity_repr = original_capacity_to_repr(cap);
         let data = (original_capacity_repr << ORIGINAL_CAPACITY_OFFSET) | KIND_VEC;
@@ -848,14 +946,19 @@ impl BytesMut {
         unsafe { slice::from_raw_parts_mut(self.ptr.as_ptr(), self.len) }
     }
 
-    unsafe fn set_start(&mut self, start: usize) {
+    /// Advance the buffer without bounds checking.
+    ///
+    /// # SAFETY
+    ///
+    /// The caller must ensure that `count` <= `self.cap`.
+    pub(crate) unsafe fn advance_unchecked(&mut self, count: usize) {
         // Setting the start to 0 is a no-op, so return early if this is the
         // case.
-        if start == 0 {
+        if count == 0 {
             return;
         }
 
-        debug_assert!(start <= self.cap, "internal: set_start out of bounds");
+        debug_assert!(count <= self.cap, "internal: set_start out of bounds");
 
         let kind = self.kind();
 
@@ -864,11 +967,10 @@ impl BytesMut {
             // complicated. First, we have to track how far ahead the
             // "start" of the byte buffer from the beginning of the vec. We
             // also have to ensure that we don't exceed the maximum shift.
-            let (mut pos, prev) = self.get_vec_pos();
-            pos += start;
+            let pos = self.get_vec_pos() + count;
 
             if pos <= MAX_VEC_POS {
-                self.set_vec_pos(pos, prev);
+                self.set_vec_pos(pos);
             } else {
                 // The repr must be upgraded to ARC. This will never happen
                 // on 64 bit systems and will only happen on 32 bit systems
@@ -881,23 +983,9 @@ impl BytesMut {
         // Updating the start of the view is setting `ptr` to point to the
         // new start and updating the `len` field to reflect the new length
         // of the view.
-        self.ptr = vptr(self.ptr.as_ptr().add(start));
-
-        if self.len >= start {
-            self.len -= start;
-        } else {
-            self.len = 0;
-        }
-
-        self.cap -= start;
-    }
-
-    unsafe fn set_end(&mut self, end: usize) {
-        debug_assert_eq!(self.kind(), KIND_ARC);
-        assert!(end <= self.cap, "set_end out of bounds");
-
-        self.cap = end;
-        self.len = cmp::min(self.len, end);
+        self.ptr = vptr(self.ptr.as_ptr().add(count));
+        self.len = self.len.checked_sub(count).unwrap_or(0);
+        self.cap -= count;
     }
 
     fn try_unsplit(&mut self, other: BytesMut) -> Result<(), BytesMut> {
@@ -976,19 +1064,18 @@ impl BytesMut {
     }
 
     #[inline]
-    unsafe fn get_vec_pos(&mut self) -> (usize, usize) {
+    unsafe fn get_vec_pos(&self) -> usize {
         debug_assert_eq!(self.kind(), KIND_VEC);
 
-        let prev = self.data as usize;
-        (prev >> VEC_POS_OFFSET, prev)
+        self.data as usize >> VEC_POS_OFFSET
     }
 
     #[inline]
-    unsafe fn set_vec_pos(&mut self, pos: usize, prev: usize) {
+    unsafe fn set_vec_pos(&mut self, pos: usize) {
         debug_assert_eq!(self.kind(), KIND_VEC);
         debug_assert!(pos <= MAX_VEC_POS);
 
-        self.data = invalid_ptr((pos << VEC_POS_OFFSET) | (prev & NOT_VEC_POS_MASK));
+        self.data = invalid_ptr((pos << VEC_POS_OFFSET) | (self.data as usize & NOT_VEC_POS_MASK));
     }
 
     /// Returns the remaining spare capacity of the buffer as a slice of `MaybeUninit<u8>`.
@@ -1037,7 +1124,7 @@ impl Drop for BytesMut {
 
         if kind == KIND_VEC {
             unsafe {
-                let (off, _) = self.get_vec_pos();
+                let off = self.get_vec_pos();
 
                 // Vector storage, free the vector
                 let _ = rebuild_vec(self.ptr.as_ptr(), self.len, self.cap, off);
@@ -1068,11 +1155,13 @@ impl Buf for BytesMut {
             self.remaining(),
         );
         unsafe {
-            self.set_start(cnt);
+            // SAFETY: We've checked that `cnt` <= `self.remaining()` and we know that
+            // `self.remaining()` <= `self.cap`.
+            self.advance_unchecked(cnt);
         }
     }
 
-    fn copy_to_bytes(&mut self, len: usize) -> crate::Bytes {
+    fn copy_to_bytes(&mut self, len: usize) -> Bytes {
         self.split_to(len).freeze()
     }
 }
@@ -1104,7 +1193,7 @@ unsafe impl BufMut for BytesMut {
     // Specialize these methods so they can skip checking `remaining_mut`
     // and `advance_mut`.
 
-    fn put<T: crate::Buf>(&mut self, mut src: T)
+    fn put<T: Buf>(&mut self, mut src: T)
     where
         Self: Sized,
     {
@@ -1284,9 +1373,7 @@ impl Extend<u8> for BytesMut {
 
         // TODO: optimize
         // 1. If self.kind() == KIND_VEC, use Vec::extend
-        // 2. Make `reserve` inline-able
         for b in iter {
-            self.reserve(1);
             self.put_u8(b);
         }
     }
@@ -1403,56 +1490,59 @@ fn original_capacity_from_repr(repr: usize) -> usize {
     1 << (repr + (MIN_ORIGINAL_CAPACITY_WIDTH - 1))
 }
 
-/*
-#[test]
-fn test_original_capacity_to_repr() {
-    assert_eq!(original_capacity_to_repr(0), 0);
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_original_capacity_to_repr() {
+        assert_eq!(original_capacity_to_repr(0), 0);
 
-    let max_width = 32;
+        let max_width = 32;
 
-    for width in 1..(max_width + 1) {
-        let cap = 1 << width - 1;
+        for width in 1..(max_width + 1) {
+            let cap = 1 << width - 1;
 
-        let expected = if width < MIN_ORIGINAL_CAPACITY_WIDTH {
-            0
-        } else if width < MAX_ORIGINAL_CAPACITY_WIDTH {
-            width - MIN_ORIGINAL_CAPACITY_WIDTH
-        } else {
-            MAX_ORIGINAL_CAPACITY_WIDTH - MIN_ORIGINAL_CAPACITY_WIDTH
-        };
+            let expected = if width < MIN_ORIGINAL_CAPACITY_WIDTH {
+                0
+            } else if width < MAX_ORIGINAL_CAPACITY_WIDTH {
+                width - MIN_ORIGINAL_CAPACITY_WIDTH
+            } else {
+                MAX_ORIGINAL_CAPACITY_WIDTH - MIN_ORIGINAL_CAPACITY_WIDTH
+            };
 
-        assert_eq!(original_capacity_to_repr(cap), expected);
+            assert_eq!(original_capacity_to_repr(cap), expected);
 
-        if width > 1 {
-            assert_eq!(original_capacity_to_repr(cap + 1), expected);
-        }
+            if width > 1 {
+                assert_eq!(original_capacity_to_repr(cap + 1), expected);
+            }
 
-        //  MIN_ORIGINAL_CAPACITY_WIDTH must be bigger than 7 to pass tests below
-        if width == MIN_ORIGINAL_CAPACITY_WIDTH + 1 {
-            assert_eq!(original_capacity_to_repr(cap - 24), expected - 1);
-            assert_eq!(original_capacity_to_repr(cap + 76), expected);
-        } else if width == MIN_ORIGINAL_CAPACITY_WIDTH + 2 {
-            assert_eq!(original_capacity_to_repr(cap - 1), expected - 1);
-            assert_eq!(original_capacity_to_repr(cap - 48), expected - 1);
+            //  MIN_ORIGINAL_CAPACITY_WIDTH must be bigger than 7 to pass tests below
+            if width == MIN_ORIGINAL_CAPACITY_WIDTH + 1 {
+                assert_eq!(original_capacity_to_repr(cap - 24), expected - 1);
+                assert_eq!(original_capacity_to_repr(cap + 76), expected);
+            } else if width == MIN_ORIGINAL_CAPACITY_WIDTH + 2 {
+                assert_eq!(original_capacity_to_repr(cap - 1), expected - 1);
+                assert_eq!(original_capacity_to_repr(cap - 48), expected - 1);
+            }
         }
     }
-}
 
-#[test]
-fn test_original_capacity_from_repr() {
-    assert_eq!(0, original_capacity_from_repr(0));
+    #[test]
+    fn test_original_capacity_from_repr() {
+        assert_eq!(0, original_capacity_from_repr(0));
 
-    let min_cap = 1 << MIN_ORIGINAL_CAPACITY_WIDTH;
+        let min_cap = 1 << MIN_ORIGINAL_CAPACITY_WIDTH;
 
-    assert_eq!(min_cap, original_capacity_from_repr(1));
-    assert_eq!(min_cap * 2, original_capacity_from_repr(2));
-    assert_eq!(min_cap * 4, original_capacity_from_repr(3));
-    assert_eq!(min_cap * 8, original_capacity_from_repr(4));
-    assert_eq!(min_cap * 16, original_capacity_from_repr(5));
-    assert_eq!(min_cap * 32, original_capacity_from_repr(6));
-    assert_eq!(min_cap * 64, original_capacity_from_repr(7));
+        assert_eq!(min_cap, original_capacity_from_repr(1));
+        assert_eq!(min_cap * 2, original_capacity_from_repr(2));
+        assert_eq!(min_cap * 4, original_capacity_from_repr(3));
+        assert_eq!(min_cap * 8, original_capacity_from_repr(4));
+        assert_eq!(min_cap * 16, original_capacity_from_repr(5));
+        assert_eq!(min_cap * 32, original_capacity_from_repr(6));
+        assert_eq!(min_cap * 64, original_capacity_from_repr(7));
+    }
 }
-*/
 
 unsafe impl Send for BytesMut {}
 unsafe impl Sync for BytesMut {}
@@ -1614,15 +1704,16 @@ impl PartialEq<Bytes> for BytesMut {
 }
 
 impl From<BytesMut> for Vec<u8> {
-    fn from(mut bytes: BytesMut) -> Self {
+    fn from(bytes: BytesMut) -> Self {
         let kind = bytes.kind();
+        let bytes = ManuallyDrop::new(bytes);
 
         let mut vec = if kind == KIND_VEC {
             unsafe {
-                let (off, _) = bytes.get_vec_pos();
+                let off = bytes.get_vec_pos();
                 rebuild_vec(bytes.ptr.as_ptr(), bytes.len, bytes.cap, off)
             }
-        } else if kind == KIND_ARC {
+        } else {
             let shared = bytes.data as *mut Shared;
 
             if unsafe { (*shared).is_unique() } {
@@ -1632,10 +1723,8 @@ impl From<BytesMut> for Vec<u8> {
 
                 vec
             } else {
-                return bytes.deref().to_vec();
+                return ManuallyDrop::into_inner(bytes).deref().to_vec();
             }
-        } else {
-            return bytes.deref().to_vec();
         };
 
         let len = bytes.len;
@@ -1645,8 +1734,6 @@ impl From<BytesMut> for Vec<u8> {
             vec.set_len(len);
         }
 
-        mem::forget(bytes);
-
         vec
     }
 }
@@ -1672,25 +1759,8 @@ fn invalid_ptr<T>(addr: usize) -> *mut T {
     ptr.cast::<T>()
 }
 
-/// Precondition: dst >= original
-///
-/// The following line is equivalent to:
-///
-/// ```rust,ignore
-/// self.ptr.as_ptr().offset_from(ptr) as usize;
-/// ```
-///
-/// But due to min rust is 1.39 and it is only stabilized
-/// in 1.47, we cannot use it.
-#[inline]
-fn offset_from(dst: *mut u8, original: *mut u8) -> usize {
-    debug_assert!(dst >= original);
-
-    dst as usize - original as usize
-}
-
 unsafe fn rebuild_vec(ptr: *mut u8, mut len: usize, mut cap: usize, off: usize) -> Vec<u8> {
-    let ptr = ptr.offset(-(off as isize));
+    let ptr = ptr.sub(off);
     len += off;
     cap += off;
 
@@ -1702,6 +1772,8 @@ unsafe fn rebuild_vec(ptr: *mut u8, mut len: usize, mut cap: usize, off: usize)
 static SHARED_VTABLE: Vtable = Vtable {
     clone: shared_v_clone,
     to_vec: shared_v_to_vec,
+    to_mut: shared_v_to_mut,
+    is_unique: shared_v_is_unique,
     drop: shared_v_drop,
 };
 
@@ -1735,6 +1807,41 @@ unsafe fn shared_v_to_vec(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> V
     }
 }
 
+unsafe fn shared_v_to_mut(data: &AtomicPtr<()>, ptr: *const u8, len: usize) -> BytesMut {
+    let shared: *mut Shared = data.load(Ordering::Relaxed).cast();
+
+    if (*shared).is_unique() {
+        let shared = &mut *shared;
+
+        // The capacity is always the original capacity of the buffer
+        // minus the offset from the start of the buffer
+        let v = &mut shared.vec;
+        let v_capacity = v.capacity();
+        let v_ptr = v.as_mut_ptr();
+        let offset = offset_from(ptr as *mut u8, v_ptr);
+        let cap = v_capacity - offset;
+
+        let ptr = vptr(ptr as *mut u8);
+
+        BytesMut {
+            ptr,
+            len,
+            cap,
+            data: shared,
+        }
+    } else {
+        let v = slice::from_raw_parts(ptr, len).to_vec();
+        release_shared(shared);
+        BytesMut::from_vec(v)
+    }
+}
+
+unsafe fn shared_v_is_unique(data: &AtomicPtr<()>) -> bool {
+    let shared = data.load(Ordering::Acquire);
+    let ref_count = (*shared.cast::<Shared>()).ref_count.load(Ordering::Relaxed);
+    ref_count == 1
+}
+
 unsafe fn shared_v_drop(data: &mut AtomicPtr<()>, _ptr: *const u8, _len: usize) {
     data.with_mut(|shared| {
         release_shared(*shared as *mut Shared);
diff --git a/src/lib.rs b/src/lib.rs
index 1b3e6fc40..7ddd2205b 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,3 +1,4 @@
+#![allow(unknown_lints, unexpected_cfgs)]
 #![warn(missing_docs, missing_debug_implementations, rust_2018_idioms)]
 #![doc(test(
     no_crate_inject,
@@ -147,3 +148,18 @@ fn panic_does_not_fit(size: usize, nbytes: usize) -> ! {
         size, nbytes
     );
 }
+
+/// Precondition: dst >= original
+///
+/// The following line is equivalent to:
+///
+/// ```rust,ignore
+/// self.ptr.as_ptr().offset_from(ptr) as usize;
+/// ```
+///
+/// But due to min rust is 1.39 and it is only stabilized
+/// in 1.47, we cannot use it.
+#[inline]
+fn offset_from(dst: *const u8, original: *const u8) -> usize {
+    dst as usize - original as usize
+}
diff --git a/tests/test_buf.rs b/tests/test_buf.rs
index 3940f9247..5aadea43e 100644
--- a/tests/test_buf.rs
+++ b/tests/test_buf.rs
@@ -36,6 +36,19 @@ fn test_get_u16() {
     assert_eq!(0x5421, buf.get_u16_le());
 }
 
+#[test]
+fn test_get_int() {
+    let mut buf = &b"\xd6zomg"[..];
+    assert_eq!(-42, buf.get_int(1));
+    let mut buf = &b"\xd6zomg"[..];
+    assert_eq!(-42, buf.get_int_le(1));
+
+    let mut buf = &b"\xfe\x1d\xc0zomg"[..];
+    assert_eq!(0xffffffffffc01dfeu64 as i64, buf.get_int_le(3));
+    let mut buf = &b"\xfe\x1d\xc0zomg"[..];
+    assert_eq!(0xfffffffffffe1dc0u64 as i64, buf.get_int(3));
+}
+
 #[test]
 #[should_panic]
 fn test_get_u16_buffer_underflow() {
diff --git a/tests/test_bytes.rs b/tests/test_bytes.rs
index 5ec60a5b0..03780f1d9 100644
--- a/tests/test_bytes.rs
+++ b/tests/test_bytes.rs
@@ -598,6 +598,28 @@ fn extend_mut_from_bytes() {
     assert_eq!(*bytes, LONG[..]);
 }
 
+#[test]
+fn extend_past_lower_limit_of_size_hint() {
+    // See https://github.com/tokio-rs/bytes/pull/674#pullrequestreview-1913035700
+    struct Iter<I>(I);
+
+    impl<I: Iterator<Item = u8>> Iterator for Iter<I> {
+        type Item = u8;
+
+        fn next(&mut self) -> Option<Self::Item> {
+            self.0.next()
+        }
+
+        fn size_hint(&self) -> (usize, Option<usize>) {
+            (5, None)
+        }
+    }
+
+    let mut bytes = BytesMut::with_capacity(5);
+    bytes.extend(Iter(std::iter::repeat(0).take(10)));
+    assert_eq!(bytes.len(), 10);
+}
+
 #[test]
 fn extend_mut_without_size_hint() {
     let mut bytes = BytesMut::with_capacity(0);
@@ -654,6 +676,43 @@ fn advance_bytes_mut() {
     assert_eq!(a, b"d zomg wat wat"[..]);
 }
 
+// Ensures BytesMut::advance reduces always capacity
+//
+// See https://github.com/tokio-rs/bytes/issues/725
+#[test]
+fn advance_bytes_mut_remaining_capacity() {
+    // reduce the search space under miri
+    let max_capacity = if cfg!(miri) { 16 } else { 256 };
+    for capacity in 0..=max_capacity {
+        for len in 0..=capacity {
+            for advance in 0..=len {
+                eprintln!("testing capacity={capacity}, len={len}, advance={advance}");
+                let mut buf = BytesMut::with_capacity(capacity);
+
+                buf.resize(len, 42);
+                assert_eq!(buf.len(), len, "resize should write `len` bytes");
+                assert_eq!(
+                    buf.remaining(),
+                    len,
+                    "Buf::remaining() should equal BytesMut::len"
+                );
+
+                buf.advance(advance);
+                assert_eq!(
+                    buf.remaining(),
+                    len - advance,
+                    "Buf::advance should reduce the remaining len"
+                );
+                assert_eq!(
+                    buf.capacity(),
+                    capacity - advance,
+                    "Buf::advance should reduce the remaining capacity"
+                );
+            }
+        }
+    }
+}
+
 #[test]
 #[should_panic]
 fn advance_past_len() {
@@ -710,97 +769,6 @@ fn partial_eq_bytesmut() {
     assert!(bytesmut != bytes2);
 }
 
-/*
-#[test]
-fn bytes_unsplit_basic() {
-    let buf = Bytes::from(&b"aaabbbcccddd"[..]);
-
-    let splitted = buf.split_off(6);
-    assert_eq!(b"aaabbb", &buf[..]);
-    assert_eq!(b"cccddd", &splitted[..]);
-
-    buf.unsplit(splitted);
-    assert_eq!(b"aaabbbcccddd", &buf[..]);
-}
-
-#[test]
-fn bytes_unsplit_empty_other() {
-    let buf = Bytes::from(&b"aaabbbcccddd"[..]);
-
-    // empty other
-    let other = Bytes::new();
-
-    buf.unsplit(other);
-    assert_eq!(b"aaabbbcccddd", &buf[..]);
-}
-
-#[test]
-fn bytes_unsplit_empty_self() {
-    // empty self
-    let mut buf = Bytes::new();
-
-    let mut other = Bytes::with_capacity(64);
-    other.extend_from_slice(b"aaabbbcccddd");
-
-    buf.unsplit(other);
-    assert_eq!(b"aaabbbcccddd", &buf[..]);
-}
-
-#[test]
-fn bytes_unsplit_arc_different() {
-    let mut buf = Bytes::with_capacity(64);
-    buf.extend_from_slice(b"aaaabbbbeeee");
-
-    buf.split_off(8); //arc
-
-    let mut buf2 = Bytes::with_capacity(64);
-    buf2.extend_from_slice(b"ccccddddeeee");
-
-    buf2.split_off(8); //arc
-
-    buf.unsplit(buf2);
-    assert_eq!(b"aaaabbbbccccdddd", &buf[..]);
-}
-
-#[test]
-fn bytes_unsplit_arc_non_contiguous() {
-    let mut buf = Bytes::with_capacity(64);
-    buf.extend_from_slice(b"aaaabbbbeeeeccccdddd");
-
-    let mut buf2 = buf.split_off(8); //arc
-
-    let buf3 = buf2.split_off(4); //arc
-
-    buf.unsplit(buf3);
-    assert_eq!(b"aaaabbbbccccdddd", &buf[..]);
-}
-
-#[test]
-fn bytes_unsplit_two_split_offs() {
-    let mut buf = Bytes::with_capacity(64);
-    buf.extend_from_slice(b"aaaabbbbccccdddd");
-
-    let mut buf2 = buf.split_off(8); //arc
-    let buf3 = buf2.split_off(4); //arc
-
-    buf2.unsplit(buf3);
-    buf.unsplit(buf2);
-    assert_eq!(b"aaaabbbbccccdddd", &buf[..]);
-}
-
-#[test]
-fn bytes_unsplit_overlapping_references() {
-    let mut buf = Bytes::with_capacity(64);
-    buf.extend_from_slice(b"abcdefghijklmnopqrstuvwxyz");
-    let mut buf0010 = buf.slice(0..10);
-    let buf1020 = buf.slice(10..20);
-    let buf0515 = buf.slice(5..15);
-    buf0010.unsplit(buf1020);
-    assert_eq!(b"abcdefghijklmnopqrst", &buf0010[..]);
-    assert_eq!(b"fghijklmno", &buf0515[..]);
-}
-*/
-
 #[test]
 fn bytes_mut_unsplit_basic() {
     let mut buf = BytesMut::with_capacity(64);
@@ -1208,3 +1176,243 @@ fn test_bytes_capacity_len() {
         }
     }
 }
+
+#[test]
+fn static_is_unique() {
+    let b = Bytes::from_static(LONG);
+    assert!(!b.is_unique());
+}
+
+#[test]
+fn vec_is_unique() {
+    let v: Vec<u8> = LONG.to_vec();
+    let b = Bytes::from(v);
+    assert!(b.is_unique());
+}
+
+#[test]
+fn arc_is_unique() {
+    let v: Vec<u8> = LONG.to_vec();
+    let b = Bytes::from(v);
+    let c = b.clone();
+    assert!(!b.is_unique());
+    drop(c);
+    assert!(b.is_unique());
+}
+
+#[test]
+fn shared_is_unique() {
+    let v: Vec<u8> = LONG.to_vec();
+    let b = Bytes::from(v);
+    let c = b.clone();
+    assert!(!c.is_unique());
+    drop(b);
+    assert!(c.is_unique());
+}
+
+#[test]
+fn mut_shared_is_unique() {
+    let mut b = BytesMut::from(LONG);
+    let c = b.split().freeze();
+    assert!(!c.is_unique());
+    drop(b);
+    assert!(c.is_unique());
+}
+
+#[test]
+fn test_bytesmut_from_bytes_static() {
+    let bs = b"1b23exfcz3r";
+
+    // Test STATIC_VTABLE.to_mut
+    let bytes_mut = BytesMut::from(Bytes::from_static(bs));
+    assert_eq!(bytes_mut, bs[..]);
+}
+
+#[test]
+fn test_bytesmut_from_bytes_bytes_mut_vec() {
+    let bs = b"1b23exfcz3r";
+    let bs_long = b"1b23exfcz3r1b23exfcz3r";
+
+    // Test case where kind == KIND_VEC
+    let mut bytes_mut: BytesMut = bs[..].into();
+    bytes_mut = BytesMut::from(bytes_mut.freeze());
+    assert_eq!(bytes_mut, bs[..]);
+    bytes_mut.extend_from_slice(&bs[..]);
+    assert_eq!(bytes_mut, bs_long[..]);
+}
+
+#[test]
+fn test_bytesmut_from_bytes_bytes_mut_shared() {
+    let bs = b"1b23exfcz3r";
+
+    // Set kind to KIND_ARC so that after freeze, Bytes will use bytes_mut.SHARED_VTABLE
+    let mut bytes_mut: BytesMut = bs[..].into();
+    drop(bytes_mut.split_off(bs.len()));
+
+    let b1 = bytes_mut.freeze();
+    let b2 = b1.clone();
+
+    // shared.is_unique() = False
+    let mut b1m = BytesMut::from(b1);
+    assert_eq!(b1m, bs[..]);
+    b1m[0] = b'9';
+
+    // shared.is_unique() = True
+    let b2m = BytesMut::from(b2);
+    assert_eq!(b2m, bs[..]);
+}
+
+#[test]
+fn test_bytesmut_from_bytes_bytes_mut_offset() {
+    let bs = b"1b23exfcz3r";
+
+    // Test bytes_mut.SHARED_VTABLE.to_mut impl where offset != 0
+    let mut bytes_mut1: BytesMut = bs[..].into();
+    let bytes_mut2 = bytes_mut1.split_off(9);
+
+    let b1 = bytes_mut1.freeze();
+    let b2 = bytes_mut2.freeze();
+
+    let b1m = BytesMut::from(b1);
+    let b2m = BytesMut::from(b2);
+
+    assert_eq!(b2m, bs[9..]);
+    assert_eq!(b1m, bs[..9]);
+}
+
+#[test]
+fn test_bytesmut_from_bytes_promotable_even_vec() {
+    let vec = vec![33u8; 1024];
+
+    // Test case where kind == KIND_VEC
+    let b1 = Bytes::from(vec.clone());
+    let b1m = BytesMut::from(b1);
+    assert_eq!(b1m, vec);
+}
+
+#[test]
+fn test_bytesmut_from_bytes_promotable_even_arc_1() {
+    let vec = vec![33u8; 1024];
+
+    // Test case where kind == KIND_ARC, ref_cnt == 1
+    let b1 = Bytes::from(vec.clone());
+    drop(b1.clone());
+    let b1m = BytesMut::from(b1);
+    assert_eq!(b1m, vec);
+}
+
+#[test]
+fn test_bytesmut_from_bytes_promotable_even_arc_2() {
+    let vec = vec![33u8; 1024];
+
+    // Test case where kind == KIND_ARC, ref_cnt == 2
+    let b1 = Bytes::from(vec.clone());
+    let b2 = b1.clone();
+    let b1m = BytesMut::from(b1);
+    assert_eq!(b1m, vec);
+
+    // Test case where vtable = SHARED_VTABLE, kind == KIND_ARC, ref_cnt == 1
+    let b2m = BytesMut::from(b2);
+    assert_eq!(b2m, vec);
+}
+
+#[test]
+fn test_bytesmut_from_bytes_promotable_even_arc_offset() {
+    let vec = vec![33u8; 1024];
+
+    // Test case where offset != 0
+    let mut b1 = Bytes::from(vec.clone());
+    let b2 = b1.split_off(20);
+    let b1m = BytesMut::from(b1);
+    let b2m = BytesMut::from(b2);
+
+    assert_eq!(b2m, vec[20..]);
+    assert_eq!(b1m, vec[..20]);
+}
+
+#[test]
+fn try_reclaim_empty() {
+    let mut buf = BytesMut::new();
+    assert_eq!(false, buf.try_reclaim(6));
+    buf.reserve(6);
+    assert_eq!(true, buf.try_reclaim(6));
+    let cap = buf.capacity();
+    assert!(cap >= 6);
+    assert_eq!(false, buf.try_reclaim(cap + 1));
+
+    let mut buf = BytesMut::new();
+    buf.reserve(6);
+    let cap = buf.capacity();
+    assert!(cap >= 6);
+    let mut split = buf.split();
+    drop(buf);
+    assert_eq!(0, split.capacity());
+    assert_eq!(true, split.try_reclaim(6));
+    assert_eq!(false, split.try_reclaim(cap + 1));
+}
+
+#[test]
+fn try_reclaim_vec() {
+    let mut buf = BytesMut::with_capacity(6);
+    buf.put_slice(b"abc");
+    // Reclaiming a ludicrous amount of space should calmly return false
+    assert_eq!(false, buf.try_reclaim(usize::MAX));
+
+    assert_eq!(false, buf.try_reclaim(6));
+    buf.advance(2);
+    assert_eq!(4, buf.capacity());
+    // We can reclaim 5 bytes, because the byte in the buffer can be moved to the front. 6 bytes
+    // cannot be reclaimed because there is already one byte stored
+    assert_eq!(false, buf.try_reclaim(6));
+    assert_eq!(true, buf.try_reclaim(5));
+    buf.advance(1);
+    assert_eq!(true, buf.try_reclaim(6));
+    assert_eq!(6, buf.capacity());
+}
+
+#[test]
+fn try_reclaim_arc() {
+    let mut buf = BytesMut::with_capacity(6);
+    buf.put_slice(b"abc");
+    let x = buf.split().freeze();
+    buf.put_slice(b"def");
+    // Reclaiming a ludicrous amount of space should calmly return false
+    assert_eq!(false, buf.try_reclaim(usize::MAX));
+
+    let y = buf.split().freeze();
+    let z = y.clone();
+    assert_eq!(false, buf.try_reclaim(6));
+    drop(x);
+    drop(z);
+    assert_eq!(false, buf.try_reclaim(6));
+    drop(y);
+    assert_eq!(true, buf.try_reclaim(6));
+    assert_eq!(6, buf.capacity());
+    assert_eq!(0, buf.len());
+    buf.put_slice(b"abc");
+    buf.put_slice(b"def");
+    assert_eq!(6, buf.capacity());
+    assert_eq!(6, buf.len());
+    assert_eq!(false, buf.try_reclaim(6));
+    buf.advance(4);
+    assert_eq!(true, buf.try_reclaim(4));
+    buf.advance(2);
+    assert_eq!(true, buf.try_reclaim(6));
+}
+
+#[test]
+#[should_panic]
+fn test_bytes_overread() {
+    let mut b = Bytes::from_static(&[0, 1, 2]);
+    let _ = b.get_u32();
+}
+
+// running this test would result in a panic without `.read_unaligned()`
+// on x86 read_unaligned compiles down to a single `mov`, on platforms with no unaligned access,
+// it uses rust's `copy_nonoverlapping`
+#[test]
+fn test_bytes_misaligned() {
+    let mut b = Bytes::from_static(&[0, 1, 2, 3, 4, 5, 6, 7, 8]);
+    b.advance(2);
+    let _ = b.get_u32();
+}
diff --git a/tests/test_bytes_odd_alloc.rs b/tests/test_bytes_odd_alloc.rs
index 27ed87736..4758dc2f9 100644
--- a/tests/test_bytes_odd_alloc.rs
+++ b/tests/test_bytes_odd_alloc.rs
@@ -6,7 +6,7 @@
 use std::alloc::{GlobalAlloc, Layout, System};
 use std::ptr;
 
-use bytes::Bytes;
+use bytes::{Bytes, BytesMut};
 
 #[global_allocator]
 static ODD: Odd = Odd;
@@ -95,3 +95,53 @@ fn test_bytes_into_vec() {
     assert_eq!(Vec::from(b2), vec[20..]);
     assert_eq!(Vec::from(b1), vec[..20]);
 }
+
+#[test]
+fn test_bytesmut_from_bytes_vec() {
+    let vec = vec![33u8; 1024];
+
+    // Test case where kind == KIND_VEC
+    let b1 = Bytes::from(vec.clone());
+    let b1m = BytesMut::from(b1);
+    assert_eq!(b1m, vec);
+}
+
+#[test]
+fn test_bytesmut_from_bytes_arc_1() {
+    let vec = vec![33u8; 1024];
+
+    // Test case where kind == KIND_ARC, ref_cnt == 1
+    let b1 = Bytes::from(vec.clone());
+    drop(b1.clone());
+    let b1m = BytesMut::from(b1);
+    assert_eq!(b1m, vec);
+}
+
+#[test]
+fn test_bytesmut_from_bytes_arc_2() {
+    let vec = vec![33u8; 1024];
+
+    // Test case where kind == KIND_ARC, ref_cnt == 2
+    let b1 = Bytes::from(vec.clone());
+    let b2 = b1.clone();
+    let b1m = BytesMut::from(b1);
+    assert_eq!(b1m, vec);
+
+    // Test case where vtable = SHARED_VTABLE, kind == KIND_ARC, ref_cnt == 1
+    let b2m = BytesMut::from(b2);
+    assert_eq!(b2m, vec);
+}
+
+#[test]
+fn test_bytesmut_from_bytes_arc_offset() {
+    let vec = vec![33u8; 1024];
+
+    // Test case where offset != 0
+    let mut b1 = Bytes::from(vec.clone());
+    let b2 = b1.split_off(20);
+    let b1m = BytesMut::from(b1);
+    let b2m = BytesMut::from(b2);
+
+    assert_eq!(b2m, vec[20..]);
+    assert_eq!(b1m, vec[..20]);
+}
diff --git a/tests/test_iter.rs b/tests/test_iter.rs
index a5bfddddf..bad901860 100644
--- a/tests/test_iter.rs
+++ b/tests/test_iter.rs
@@ -1,11 +1,11 @@
 #![warn(rust_2018_idioms)]
 
-use bytes::Bytes;
+use bytes::{buf::IntoIter, Bytes};
 
 #[test]
 fn iter_len() {
     let buf = Bytes::from_static(b"hello world");
-    let iter = buf.iter();
+    let iter = IntoIter::new(buf);
 
     assert_eq!(iter.size_hint(), (11, Some(11)));
     assert_eq!(iter.len(), 11);
@@ -13,8 +13,8 @@ fn iter_len() {
 
 #[test]
 fn empty_iter_len() {
-    let buf = Bytes::from_static(b"");
-    let iter = buf.iter();
+    let buf = Bytes::new();
+    let iter = IntoIter::new(buf);
 
     assert_eq!(iter.size_hint(), (0, Some(0)));
     assert_eq!(iter.len(), 0);