Rewrite Bytes / BytesMut core implementation

The previous implementation didn't factor in a single `Bytes` handle
being stored in an `Arc`. This new implementation correctly impelments
both `Bytes` and `BytesMut` such that both are `Sync`.

The rewrite also increases the number of bytes that can be stored
inline.

.travis.yml

@@ -1,30 +1,56 @@
 ---
+dist: trusty
 language: rust
-rust:
-  - nightly
-  - stable
-  # Oldest supported Rust version. Do not change this without a Github issue
-  # discussion.
-  - 1.10.0
-
-matrix:
-  allow_failures:
-    - rust: nightly
-
-before_script:
-  - pip install 'travis-cargo<0.2' --user && export PATH=$HOME/.local/bin:$PATH
-
-script:
-  - cargo build
-  - RUST_BACKTRACE=1 cargo test
-  - cargo doc --no-deps
-
-after_success:
-  - travis-cargo --only stable doc-upload
+services: docker
+sudo: required
+rust: stable
 
 env:
   global:
-    secure: "f17G5kb6uAQlAG9+GknFFYAmngGBqy9h+3FtNbp3mXTI0FOLltz00Ul5kGPysE4eagypm/dWOuvBkNjN01jhE6fCbekmInEsobIuanatrk6TvXT6caJqykxhPJC2cUoq8pKnMqEOuucEqPPUH6Qy6Hz4/2cRu5JV22Uv9dtS29Q="
+    - CRATE_NAME=bytes
+    # Default job
+    - TARGET=x86_64-unknown-linux-gnu
+    - secure: "f17G5kb6uAQlAG9+GknFFYAmngGBqy9h+3FtNbp3mXTI0FOLltz00Ul5kGPysE4eagypm/dWOuvBkNjN01jhE6fCbekmInEsobIuanatrk6TvXT6caJqykxhPJC2cUoq8pKnMqEOuucEqPPUH6Qy6Hz4/2cRu5JV22Uv9dtS29Q="
+
+matrix:
+  include:
+    # Run build on oldest supported rust version. Do not change the rust
+    # version without a Github issue first.
+    #
+    # This job will also build and deploy the docs to gh-pages.
+    - env: TARGET=x86_64-unknown-linux-gnu
+      rust: 1.10.0
+      after_success:
+        - |
+            pip install 'travis-cargo<0.2' --user &&
+            export PATH=$HOME/.local/bin:$PATH
+        - travis-cargo --only stable doc
+        - travis-cargo --only stable doc-upload
+
+    # Run tests on some extra platforms
+    - env: TARGET=i686-unknown-linux-gnu
+    - env: TARGET=armv7-unknown-linux-gnueabihf
+    - env: TARGET=powerpc-unknown-linux-gnu
+    - env: TARGET=powerpc64-unknown-linux-gnu
+
+before_install: set -e
+
+install:
+  - sh ci/install.sh
+  - source ~/.cargo/env || true
+
+script:
+  - bash ci/script.sh
+
+after_script: set +e
+
+before_deploy:
+  - sh ci/before_deploy.sh
+
+cache: cargo
+before_cache:
+  # Travis can't cache files that are not readable by "others"
+  - chmod -R a+r $HOME/.cargo
 
 notifications:
   email:

Cargo.toml

@@ -20,3 +20,6 @@ exclude       = [
 
 [dependencies]
 byteorder = "1.0.0"
+
+[dev-dependencies]
+tokio-core = "0.1.0"

benches/vs_easy_buf.rs

@@ -0,0 +1,210 @@
+#![feature(test)]
+
+extern crate tokio_core;
+extern crate bytes;
+extern crate test;
+
+mod bench_easy_buf {
+    use test::{self, Bencher};
+    use tokio_core::io::EasyBuf;
+
+    #[bench]
+    fn alloc_small(b: &mut Bencher) {
+        b.iter(|| {
+            for _ in 0..1024 {
+                test::black_box(EasyBuf::with_capacity(12));
+            }
+        })
+    }
+
+    #[bench]
+    fn alloc_mid(b: &mut Bencher) {
+        b.iter(|| {
+            test::black_box(EasyBuf::with_capacity(128));
+        })
+    }
+
+    #[bench]
+    fn alloc_big(b: &mut Bencher) {
+        b.iter(|| {
+            test::black_box(EasyBuf::with_capacity(4096));
+        })
+    }
+
+    #[bench]
+    fn deref_front(b: &mut Bencher) {
+        let mut buf = EasyBuf::with_capacity(4096);
+        buf.get_mut().extend_from_slice(&[0; 1024][..]);
+
+        b.iter(|| {
+            for _ in 0..1024 {
+                test::black_box(buf.as_slice());
+            }
+        })
+    }
+
+    #[bench]
+    fn deref_mid(b: &mut Bencher) {
+        let mut buf = EasyBuf::with_capacity(4096);
+        buf.get_mut().extend_from_slice(&[0; 1024][..]);
+        let _a = buf.drain_to(512);
+
+        b.iter(|| {
+            for _ in 0..1024 {
+                test::black_box(buf.as_slice());
+            }
+        })
+    }
+
+    #[bench]
+    fn alloc_write_drain_to_mid(b: &mut Bencher) {
+        b.iter(|| {
+            let mut buf = EasyBuf::with_capacity(128);
+            buf.get_mut().extend_from_slice(&[0u8; 64]);
+            test::black_box(buf.drain_to(64));
+        })
+    }
+
+    #[bench]
+    fn drain_write_drain(b: &mut Bencher) {
+        let data = [0u8; 128];
+
+        b.iter(|| {
+            let mut buf = EasyBuf::with_capacity(1024);
+            let mut parts = Vec::with_capacity(8);
+
+            for _ in 0..8 {
+                buf.get_mut().extend_from_slice(&data[..]);
+                parts.push(buf.drain_to(128));
+            }
+
+            test::black_box(parts);
+        })
+    }
+}
+
+mod bench_bytes {
+    use test::{self, Bencher};
+    use bytes::{BytesMut, BufMut};
+
+    #[bench]
+    fn alloc_small(b: &mut Bencher) {
+        b.iter(|| {
+            for _ in 0..1024 {
+                test::black_box(BytesMut::with_capacity(12));
+            }
+        })
+    }
+
+    #[bench]
+    fn alloc_mid(b: &mut Bencher) {
+        b.iter(|| {
+            test::black_box(BytesMut::with_capacity(128));
+        })
+    }
+
+    #[bench]
+    fn alloc_big(b: &mut Bencher) {
+        b.iter(|| {
+            test::black_box(BytesMut::with_capacity(4096));
+        })
+    }
+
+    #[bench]
+    fn deref_unique(b: &mut Bencher) {
+        let mut buf = BytesMut::with_capacity(4096);
+        buf.put(&[0u8; 1024][..]);
+
+        b.iter(|| {
+            for _ in 0..1024 {
+                test::black_box(&buf[..]);
+            }
+        })
+    }
+
+    #[bench]
+    fn deref_unique_unroll(b: &mut Bencher) {
+        let mut buf = BytesMut::with_capacity(4096);
+        buf.put(&[0u8; 1024][..]);
+
+        b.iter(|| {
+            for _ in 0..128 {
+                test::black_box(&buf[..]);
+                test::black_box(&buf[..]);
+                test::black_box(&buf[..]);
+                test::black_box(&buf[..]);
+                test::black_box(&buf[..]);
+                test::black_box(&buf[..]);
+                test::black_box(&buf[..]);
+                test::black_box(&buf[..]);
+            }
+        })
+    }
+
+    #[bench]
+    fn deref_shared(b: &mut Bencher) {
+        let mut buf = BytesMut::with_capacity(4096);
+        buf.put(&[0u8; 1024][..]);
+        let _b2 = buf.split_off(1024);
+
+        b.iter(|| {
+            for _ in 0..1024 {
+                test::black_box(&buf[..]);
+            }
+        })
+    }
+
+    #[bench]
+    fn deref_inline(b: &mut Bencher) {
+        let mut buf = BytesMut::with_capacity(8);
+        buf.put(&[0u8; 8][..]);
+
+        b.iter(|| {
+            for _ in 0..1024 {
+                test::black_box(&buf[..]);
+            }
+        })
+    }
+
+    #[bench]
+    fn deref_two(b: &mut Bencher) {
+        let mut buf1 = BytesMut::with_capacity(8);
+        buf1.put(&[0u8; 8][..]);
+
+        let mut buf2 = BytesMut::with_capacity(4096);
+        buf2.put(&[0u8; 1024][..]);
+
+        b.iter(|| {
+            for _ in 0..512 {
+                test::black_box(&buf1[..]);
+                test::black_box(&buf2[..]);
+            }
+        })
+    }
+
+    #[bench]
+    fn alloc_write_drain_to_mid(b: &mut Bencher) {
+        b.iter(|| {
+            let mut buf = BytesMut::with_capacity(128);
+            buf.put_slice(&[0u8; 64]);
+            test::black_box(buf.drain_to(64));
+        })
+    }
+
+    #[bench]
+    fn drain_write_drain(b: &mut Bencher) {
+        let data = [0u8; 128];
+
+        b.iter(|| {
+            let mut buf = BytesMut::with_capacity(1024);
+            let mut parts = Vec::with_capacity(8);
+
+            for _ in 0..8 {
+                buf.put(&data[..]);
+                parts.push(buf.drain_to(128));
+            }
+
+            test::black_box(parts);
+        })
+    }
+}

ci/before_deploy.ps1

@@ -0,0 +1,23 @@
+# This script takes care of packaging the build artifacts that will go in the
+# release zipfile
+
+$SRC_DIR = $PWD.Path
+$STAGE = [System.Guid]::NewGuid().ToString()
+
+Set-Location $ENV:Temp
+New-Item -Type Directory -Name $STAGE
+Set-Location $STAGE
+
+$ZIP = "$SRC_DIR\$($Env:CRATE_NAME)-$($Env:APPVEYOR_REPO_TAG_NAME)-$($Env:TARGET).zip"
+
+# TODO Update this to package the right artifacts
+Copy-Item "$SRC_DIR\target\$($Env:TARGET)\release\hello.exe" '.\'
+
+7z a "$ZIP" *
+
+Push-AppveyorArtifact "$ZIP"
+
+Remove-Item *.* -Force
+Set-Location ..
+Remove-Item $STAGE
+Set-Location $SRC_DIR

ci/before_deploy.sh

@@ -0,0 +1,33 @@
+# This script takes care of building your crate and packaging it for release
+
+set -ex
+
+main() {
+    local src=$(pwd) \
+          stage=
+
+    case $TRAVIS_OS_NAME in
+        linux)
+            stage=$(mktemp -d)
+            ;;
+        osx)
+            stage=$(mktemp -d -t tmp)
+            ;;
+    esac
+
+    test -f Cargo.lock || cargo generate-lockfile
+
+    # TODO Update this to build the artifacts that matter to you
+    cross rustc --bin hello --target $TARGET --release -- -C lto
+
+    # TODO Update this to package the right artifacts
+    cp target/$TARGET/release/hello $stage/
+
+    cd $stage
+    tar czf $src/$CRATE_NAME-$TRAVIS_TAG-$TARGET.tar.gz *
+    cd $src
+
+    rm -rf $stage
+}
+
+main

ci/install.sh

@@ -0,0 +1,31 @@
+set -ex
+
+main() {
+    curl https://sh.rustup.rs -sSf | \
+        sh -s -- -y --default-toolchain $TRAVIS_RUST_VERSION
+
+    local target=
+    if [ $TRAVIS_OS_NAME = linux ]; then
+        target=x86_64-unknown-linux-gnu
+        sort=sort
+    else
+        target=x86_64-apple-darwin
+        sort=gsort  # for `sort --sort-version`, from brew's coreutils.
+    fi
+
+    # This fetches latest stable release
+    local tag=$(git ls-remote --tags --refs --exit-code https://github.com/japaric/cross \
+                       | cut -d/ -f3 \
+                       | grep -E '^v[0-9.]+$' \
+                       | $sort --version-sort \
+                       | tail -n1)
+    echo cross version: $tag
+    curl -LSfs https://japaric.github.io/trust/install.sh | \
+        sh -s -- \
+           --force \
+           --git japaric/cross \
+           --tag $tag \
+           --target $target
+}
+
+main

ci/script.sh

@@ -0,0 +1,18 @@
+# This script takes care of testing your crate
+
+set -ex
+
+main() {
+    cross build --target $TARGET
+
+    if [ ! -z $DISABLE_TESTS ]; then
+        return
+    fi
+
+    cross test --target $TARGET
+}
+
+# we don't run the "test phase" when doing deploys
+if [ -z $TRAVIS_TAG ]; then
+    main
+fi

tests/test_bytes.rs

@@ -5,11 +5,10 @@ use bytes::{Bytes, BytesMut, BufMut};
 const LONG: &'static [u8] = b"mary had a little lamb, little lamb, little lamb";
 const SHORT: &'static [u8] = b"hello world";
 
-#[cfg(target_pointer_width = "64")]
-const INLINE_CAP: usize = 8 * 3;
-
-#[cfg(target_pointer_width = "32")]
-const INNER_CAP: usize = 4 * 3;
+fn inline_cap() -> usize {
+    use std::mem;
+    4 * mem::size_of::<usize>() - 1
+}
 
 fn is_sync<T: Sync>() {}
 fn is_send<T: Send>() {}
@@ -17,6 +16,7 @@ fn is_send<T: Send>() {}
 #[test]
 fn test_bounds() {
     is_sync::<Bytes>();
+    is_sync::<BytesMut>();
     is_send::<Bytes>();
     is_send::<BytesMut>();
 }
@@ -90,25 +90,25 @@ fn slice() {
 #[should_panic]
 fn slice_oob_1() {
     let a = Bytes::from(&b"hello world"[..]);
-    a.slice(5, 25);
+    a.slice(5, inline_cap() + 1);
 }
 
 #[test]
 #[should_panic]
 fn slice_oob_2() {
     let a = Bytes::from(&b"hello world"[..]);
-    a.slice(25, 30);
+    a.slice(inline_cap() + 1, inline_cap() + 5);
 }
 
 #[test]
 fn split_off() {
-    let hello = Bytes::from(&b"helloworld"[..]);
+    let mut hello = Bytes::from(&b"helloworld"[..]);
     let world = hello.split_off(5);
 
     assert_eq!(hello, &b"hello"[..]);
     assert_eq!(world, &b"world"[..]);
 
-    let hello = BytesMut::from(&b"helloworld"[..]);
+    let mut hello = BytesMut::from(&b"helloworld"[..]);
     let world = hello.split_off(5);
 
     assert_eq!(hello, &b"hello"[..]);
@@ -118,21 +118,14 @@ fn split_off() {
 #[test]
 #[should_panic]
 fn split_off_oob() {
-    let hello = Bytes::from(&b"helloworld"[..]);
-    hello.split_off(25);
-}
-
-#[test]
-#[should_panic]
-fn split_off_oob_mut() {
-    let hello = BytesMut::from(&b"helloworld"[..]);
-    hello.split_off(25);
+    let mut hello = Bytes::from(&b"helloworld"[..]);
+    hello.split_off(inline_cap() + 1);
 }
 
 #[test]
 fn split_off_uninitialized() {
     let mut bytes = BytesMut::with_capacity(1024);
-    let other = bytes.split_off_mut(128);
+    let other = bytes.split_off(128);
 
     assert_eq!(bytes.len(), 0);
     assert_eq!(bytes.capacity(), 128);
@@ -144,44 +137,55 @@ fn split_off_uninitialized() {
 #[test]
 fn drain_to_1() {
     // Inline
-    let a = Bytes::from(SHORT);
+    let mut a = Bytes::from(SHORT);
     let b = a.drain_to(4);
 
     assert_eq!(SHORT[4..], a);
     assert_eq!(SHORT[..4], b);
 
     // Allocated
-    let a = Bytes::from(LONG);
+    let mut a = Bytes::from(LONG);
     let b = a.drain_to(4);
 
     assert_eq!(LONG[4..], a);
     assert_eq!(LONG[..4], b);
 
-    let a = Bytes::from(LONG);
+    let mut a = Bytes::from(LONG);
     let b = a.drain_to(30);
 
     assert_eq!(LONG[30..], a);
     assert_eq!(LONG[..30], b);
 }
 
+#[test]
+fn drain_to_2() {
+    let mut a = Bytes::from(LONG);
+    assert_eq!(LONG, a);
+
+    let b = a.drain_to(1);
+
+    assert_eq!(LONG[1..], a);
+    drop(b);
+}
+
 #[test]
 #[should_panic]
 fn drain_to_oob() {
-    let hello = Bytes::from(&b"helloworld"[..]);
-    hello.drain_to(30);
+    let mut hello = Bytes::from(&b"helloworld"[..]);
+    hello.drain_to(inline_cap() + 1);
 }
 
 #[test]
 #[should_panic]
 fn drain_to_oob_mut() {
-    let hello = BytesMut::from(&b"helloworld"[..]);
-    hello.drain_to(30);
+    let mut hello = BytesMut::from(&b"helloworld"[..]);
+    hello.drain_to(inline_cap() + 1);
 }
 
 #[test]
 fn drain_to_uninitialized() {
     let mut bytes = BytesMut::with_capacity(1024);
-    let other = bytes.drain_to_mut(128);
+    let other = bytes.drain_to(128);
 
     assert_eq!(bytes.len(), 0);
     assert_eq!(bytes.capacity(), 896);
@@ -212,12 +216,12 @@ fn reserve() {
     assert_eq!(bytes, "hello");
 
     // Inline -> Inline
-    let mut bytes = BytesMut::with_capacity(INLINE_CAP);
+    let mut bytes = BytesMut::with_capacity(inline_cap());
     bytes.put("abcdefghijkl");
 
     let a = bytes.drain_to(10);
-    bytes.reserve(INLINE_CAP - 3);
-    assert_eq!(INLINE_CAP, bytes.capacity());
+    bytes.reserve(inline_cap() - 3);
+    assert_eq!(inline_cap(), bytes.capacity());
 
     assert_eq!(bytes, "kl");
     assert_eq!(a, "abcdefghij");
@@ -238,19 +242,19 @@ fn reserve() {
 }
 
 #[test]
-fn try_reclaim() {
+fn try_reclaim_1() {
     // Inline w/ start at zero
     let mut bytes = BytesMut::from(&SHORT[..]);
     assert!(bytes.try_reclaim());
-    assert_eq!(bytes.capacity(), INLINE_CAP);
+    assert_eq!(bytes.capacity(), inline_cap());
     assert_eq!(bytes, SHORT);
 
     // Inline w/ start not at zero
     let mut bytes = BytesMut::from(&SHORT[..]);
     let _ = bytes.drain_to(2);
-    assert_eq!(bytes.capacity(), INLINE_CAP - 2);
+    assert_eq!(bytes.capacity(), inline_cap());
     assert!(bytes.try_reclaim());
-    assert_eq!(bytes.capacity(), INLINE_CAP);
+    assert_eq!(bytes.capacity(), inline_cap());
     assert_eq!(bytes, &SHORT[2..]);
 
     // Arc
@@ -263,3 +267,65 @@ fn try_reclaim() {
     assert!(bytes.try_reclaim());
     assert_eq!(bytes.capacity(), LONG.len());
 }
+
+#[test]
+fn try_reclaim_2() {
+    let mut bytes = BytesMut::from(
+        "Lorem ipsum dolor sit amet, consectetur adipiscing elit.");
+
+    // Create a new handle to the shared memory region
+    let a = bytes.drain_to(5);
+
+    // Attempting to reclaim here will fail due to `a` still being in
+    // existence.
+    assert!(!bytes.try_reclaim());
+    assert_eq!(bytes.capacity(), 51);
+
+    // Dropping the handle will allow reclaim to succeed.
+    drop(a);
+    assert!(bytes.try_reclaim());
+    assert_eq!(bytes.capacity(), 56);
+}
+
+#[test]
+fn inline_storage() {
+    let mut bytes = BytesMut::with_capacity(inline_cap());
+    let zero = [0u8; 64];
+
+    bytes.put(&zero[0..inline_cap()]);
+    assert_eq!(*bytes, zero[0..inline_cap()]);
+}
+
+#[test]
+fn stress() {
+    // Tests promoting a buffer from a vec -> shared in a concurrent situation
+    use std::sync::{Arc, Barrier};
+    use std::thread;
+
+    const THREADS: usize = 8;
+    const ITERS: usize = 1_000;
+
+    for i in 0..ITERS {
+        let data = [i as u8; 256];
+        let buf = Arc::new(BytesMut::from(&data[..]));
+
+        let barrier = Arc::new(Barrier::new(THREADS));
+        let mut joins = Vec::with_capacity(THREADS);
+
+        for _ in 0..THREADS {
+            let c = barrier.clone();
+            let buf = buf.clone();
+
+            joins.push(thread::spawn(move || {
+                c.wait();
+                let _buf = buf.clone();
+            }));
+        }
+
+        for th in joins {
+            th.join().unwrap();
+        }
+
+        assert_eq!(*buf, data[..]);
+    }
+}