Repository: bearcove/loona (language: rust)

Issue:
Simplify buffer interface greatly
`AggBuf` shouldn't be that complicated. Advancing should split the buffer, always, and return an `AggSlice` (maybe those should be renamed `AggMut` and `Agg`). `AggSlice` should not hold a `Rc` to `AggBuf` / `AggBufInner`, it should have a `SmallVec<[Buf; 1]>` and that's it.

This should remove the need for `borrow_mut` in a bunch of places.


Discussion / hints:
This actually might make parsing harder, so I'm not so sure about it anymore..

In the current system, `AggSlice`s have an index into the initialized portion of the original `AggBuf`, which allows slicing it so we can get an `AggBuf` of the "rest" (part not used when parsing). If `AggSlice` were a `SmallVec` of `Buf`, we'd need to keep track of that index separately.

Maybe it's still worth it?

Fix patch:
diff --git a/.gitignore b/.gitignore
index 1c985fd9..fe3043a1 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,3 +1,4 @@
 target
 Cargo.lock
 *.lcov
+lcov.info
diff --git a/.vscode/settings.json b/.vscode/settings.json
index 0d52d55a..ceeb0799 100644
--- a/.vscode/settings.json
+++ b/.vscode/settings.json
@@ -1,4 +1,7 @@
 {
 	"rust-analyzer.inlayHints.lifetimeElisionHints.enable": "always",
-	"rust-analyzer.inlayHints.lifetimeElisionHints.useParameterNames": true
+	"rust-analyzer.inlayHints.lifetimeElisionHints.useParameterNames": true,
+	"coverage-gutters.showLineCoverage": true,
+	"coverage-gutters.showRulerCoverage": true,
+	"coverage-gutters.coverageReportFileName": "target/llvm-cov/html/index.html"
 }
\ No newline at end of file
diff --git a/Cargo.toml b/Cargo.toml
index 9f4fb92d..e9ef6314 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -12,6 +12,7 @@ An HTTP implementation on top of io_uring
 
 [dependencies]
 eyre = { version = "0.6.8", default-features = false }
+memchr = "2.5.0"
 memmap2 = { version = "0.5.7", default-features = false }
 nom = { version = "7.1.1", default-features = false }
 pretty-hex = { version = "0.3.0", default-features = false }
diff --git a/Justfile b/Justfile
index 54f1b047..a03e2c5c 100644
--- a/Justfile
+++ b/Justfile
@@ -10,6 +10,10 @@ ci-test:
 	cargo llvm-cov nextest --lcov --output-path coverage.lcov
 	codecov
 
+cov:
+	cargo llvm-cov nextest --lcov --output-path lcov.info
+	cargo llvm-cov report --html
+
 # Run all tests with cargo nextest
 test *args:
 	just build-testbed
diff --git a/src/buffet/io_chunk.rs b/src/buffet/io_chunk.rs
new file mode 100644
index 00000000..ddf22556
--- /dev/null
+++ b/src/buffet/io_chunk.rs
@@ -0,0 +1,126 @@
+//! Types for performing vectored I/O.
+
+use tokio_uring::buf::IoBuf;
+
+use crate::Roll;
+
+pub enum IoChunk {
+    Static(&'static [u8]),
+    Vec(Vec<u8>),
+    Roll(Roll),
+}
+
+impl From<&'static [u8]> for IoChunk {
+    fn from(slice: &'static [u8]) -> Self {
+        IoChunk::Static(slice)
+    }
+}
+
+impl From<&'static str> for IoChunk {
+    fn from(slice: &'static str) -> Self {
+        IoChunk::Static(slice.as_bytes())
+    }
+}
+
+impl From<Vec<u8>> for IoChunk {
+    fn from(vec: Vec<u8>) -> Self {
+        IoChunk::Vec(vec)
+    }
+}
+
+impl From<Roll> for IoChunk {
+    fn from(roll: Roll) -> Self {
+        IoChunk::Roll(roll)
+    }
+}
+
+impl AsRef<[u8]> for IoChunk {
+    fn as_ref(&self) -> &[u8] {
+        match self {
+            IoChunk::Static(slice) => slice,
+            IoChunk::Vec(vec) => vec.as_ref(),
+            IoChunk::Roll(roll) => roll.as_ref(),
+        }
+    }
+}
+
+impl IoChunk {
+    #[inline(always)]
+    pub fn as_io_buf(&self) -> &dyn IoBuf {
+        match self {
+            IoChunk::Static(slice) => slice,
+            IoChunk::Vec(vec) => vec,
+            IoChunk::Roll(roll) => roll,
+        }
+    }
+}
+
+unsafe impl IoBuf for IoChunk {
+    #[inline(always)]
+    fn stable_ptr(&self) -> *const u8 {
+        IoBuf::stable_ptr(self.as_io_buf())
+    }
+
+    fn bytes_init(&self) -> usize {
+        IoBuf::bytes_init(self.as_io_buf())
+    }
+
+    fn bytes_total(&self) -> usize {
+        IoBuf::bytes_total(self.as_io_buf())
+    }
+}
+
+impl IoChunk {
+    #[inline(always)]
+    pub fn len(&self) -> usize {
+        match self {
+            IoChunk::Static(slice) => slice.len(),
+            IoChunk::Vec(vec) => vec.len(),
+            IoChunk::Roll(roll) => roll.len(),
+        }
+    }
+
+    #[inline(always)]
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+}
+
+#[derive(Default)]
+pub struct IoChunkList {
+    chunks: Vec<IoChunk>,
+}
+
+impl IoChunkList {
+    /// Add a single chunk to the list
+    pub fn push(&mut self, chunk: impl Into<IoChunk>) {
+        self.chunks.push(chunk.into());
+    }
+
+    /// Returns total length
+    pub fn len(&self) -> usize {
+        self.chunks.iter().map(|c| c.len()).sum()
+    }
+
+    pub fn num_chunks(&self) -> usize {
+        self.chunks.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.chunks.is_empty() || self.len() == 0
+    }
+
+    pub fn clear(&mut self) {
+        self.chunks.clear();
+    }
+
+    pub fn into_vec(self) -> Vec<IoChunk> {
+        self.chunks
+    }
+}
+
+impl From<Vec<IoChunk>> for IoChunkList {
+    fn from(chunks: Vec<IoChunk>) -> Self {
+        Self { chunks }
+    }
+}
diff --git a/src/bufpool/mod.rs b/src/buffet/mod.rs
similarity index 80%
rename from src/bufpool/mod.rs
rename to src/buffet/mod.rs
index 02261f79..fb140ea2 100644
--- a/src/bufpool/mod.rs
+++ b/src/buffet/mod.rs
@@ -1,5 +1,5 @@
-mod agg;
-pub use agg::*;
+mod roll;
+pub use roll::*;
 
 mod io_chunk;
 pub use io_chunk::*;
@@ -8,7 +8,7 @@ use std::{
     cell::{RefCell, RefMut},
     collections::VecDeque,
     marker::PhantomData,
-    ops::{self, Range},
+    ops::{self, Bound, RangeBounds},
 };
 
 use memmap2::MmapMut;
@@ -140,7 +140,7 @@ impl BufPool {
 
 /// A mutable buffer. Cannot be cloned, but can be written to
 pub struct BufMut {
-    index: u32,
+    pub(crate) index: u32,
     off: u16,
     len: u16,
 
@@ -149,18 +149,6 @@ pub struct BufMut {
 }
 
 impl BufMut {
-    /// Clone this buffer. This is only pub(crate) because it's used
-    /// by `AggBuf`.
-    pub(crate) fn dangerous_clone(&self) -> Self {
-        BUF_POOL.inc(1); // in fact, increase it by 1
-        BufMut {
-            index: self.index,
-            off: self.off,
-            len: self.len,
-            _non_send: PhantomData,
-        }
-    }
-
     #[inline(always)]
     pub fn alloc() -> Result<BufMut, Error> {
         BUF_POOL.alloc()
@@ -195,7 +183,7 @@ impl BufMut {
 
     /// Dangerous: freeze a slice of this. Must only be used if you can
     /// guarantee this portion won't be written to anymore.
-    pub(crate) fn freeze_slice(&self, range: Range<u16>) -> Buf {
+    pub(crate) fn freeze_slice(&self, range: impl RangeBounds<usize>) -> Buf {
         let b = Buf {
             index: self.index,
             off: self.off,
@@ -234,6 +222,15 @@ impl BufMut {
 
         (left, right)
     }
+
+    /// Skip over the first `n` bytes, panics if out of bound
+    pub fn skip(&mut self, n: usize) {
+        assert!(n <= self.len as usize);
+
+        let u16_n: u16 = n.try_into().unwrap();
+        self.off += u16_n;
+        self.len -= u16_n;
+    }
 }
 
 impl ops::Deref for BufMut {
@@ -301,7 +298,7 @@ impl Drop for BufMut {
 
 /// A read-only buffer. Can be cloned, but cannot be written to.
 pub struct Buf {
-    index: u32,
+    pub(crate) index: u32,
     off: u16,
     len: u16,
 
@@ -320,19 +317,57 @@ impl Buf {
         self.len == 0
     }
 
-    /// Slice this buffer
-    pub fn slice(&self, range: Range<u16>) -> Self {
-        assert!(range.end >= range.start);
-        assert!(range.end <= self.len);
+    /// Take an owned slice of this
+    pub fn slice(mut self, range: impl RangeBounds<usize>) -> Self {
+        let mut new_start = 0;
+        let mut new_end = self.len();
+
+        match range.start_bound() {
+            Bound::Included(&n) => new_start = n,
+            Bound::Excluded(&n) => new_start = n + 1,
+            Bound::Unbounded => {}
+        }
+
+        match range.end_bound() {
+            Bound::Included(&n) => new_end = n + 1,
+            Bound::Excluded(&n) => new_end = n,
+            Bound::Unbounded => {}
+        }
+
+        assert!(new_start <= new_end);
+        assert!(new_end <= self.len());
+
+        self.off += new_start as u16;
+        self.len = (new_end - new_start) as u16;
+        self
+    }
+
+    /// Split this buffer in twain.
+    /// Panics if `at` is out of bounds.
+    #[inline]
+    pub fn split_at(self, at: usize) -> (Self, Self) {
+        assert!(at <= self.len as usize);
 
-        BUF_POOL.inc(1);
-        Buf {
+        let left = Buf {
             index: self.index,
-            off: self.off + range.start,
-            len: (range.end - range.start) as _,
+            off: self.off,
+            len: at as _,
 
             _non_send: PhantomData,
-        }
+        };
+
+        let right = Buf {
+            index: self.index,
+            off: self.off + at as u16,
+            len: (self.len - at as u16),
+
+            _non_send: PhantomData,
+        };
+
+        std::mem::forget(self); // don't decrease ref count
+        BUF_POOL.inc(left.index); // in fact, increase it by 1
+
+        (left, right)
     }
 }
 
@@ -384,14 +419,34 @@ impl Drop for Buf {
 
 #[cfg(test)]
 mod tests {
-    use crate::bufpool::{Buf, BUF_POOL};
+    use std::rc::Rc;
+
+    use crate::buffet::{Buf, BUF_POOL};
 
     use super::BufMut;
 
     #[test]
-    fn align_test() {
-        assert_eq!(4, std::mem::align_of::<BufMut>());
-        assert_eq!(4, std::mem::align_of::<Buf>());
+    fn size_test() {
+        assert_eq!(8, std::mem::size_of::<BufMut>());
+        assert_eq!(8, std::mem::size_of::<Buf>());
+        assert_eq!(16, std::mem::size_of::<Box<[u8]>>());
+
+        assert_eq!(16, std::mem::size_of::<&[u8]>());
+
+        #[allow(dead_code)]
+        enum BufOrBox {
+            Buf(Buf),
+            Box((Rc<Box<[u8]>>, u32, u32)),
+        }
+        assert_eq!(16, std::mem::size_of::<BufOrBox>());
+
+        #[allow(dead_code)]
+        enum Chunk {
+            Buf(Buf),
+            Box(Box<[u8]>),
+            Static(&'static [u8]),
+        }
+        assert_eq!(24, std::mem::size_of::<Chunk>());
     }
 
     #[test]
diff --git a/src/buffet/roll.rs b/src/buffet/roll.rs
new file mode 100644
index 00000000..69441a69
--- /dev/null
+++ b/src/buffet/roll.rs
@@ -0,0 +1,1192 @@
+use std::{
+    borrow::Cow,
+    cell::UnsafeCell,
+    fmt::{Debug, Formatter},
+    iter::Enumerate,
+    ops::{Bound, Deref, RangeBounds},
+    rc::Rc,
+};
+
+use nom::{
+    Compare, CompareResult, FindSubstring, InputIter, InputLength, InputTake, InputTakeAtPosition,
+    Needed,
+};
+use tokio_uring::buf::{IoBuf, IoBufMut};
+
+use crate::{Buf, BufMut, ReadOwned, BUF_SIZE};
+
+/// A "rolling buffer". Uses either one [BufMut] or a `Box<[u8]>` for storage.
+/// This buffer never grows, but it can be split, and it can be reallocated so
+/// it regains its initical capacity, minus the length of the filled part.
+pub struct RollMut {
+    storage: StorageMut,
+    len: u32,
+}
+
+enum StorageMut {
+    Buf(BufMut),
+    Box(BoxStorage),
+}
+
+impl StorageMut {
+    #[inline(always)]
+    fn cap(&self) -> usize {
+        match self {
+            StorageMut::Buf(_) => BUF_SIZE as usize,
+            StorageMut::Box(b) => b.cap(),
+        }
+    }
+
+    #[inline(always)]
+    fn len(&self) -> usize {
+        match self {
+            StorageMut::Buf(b) => b.len(),
+            StorageMut::Box(b) => b.len(),
+        }
+    }
+
+    unsafe fn as_mut_ptr(&mut self) -> *mut u8 {
+        match self {
+            StorageMut::Buf(b) => b.as_mut_ptr(),
+            StorageMut::Box(b) => b.as_mut_ptr(),
+        }
+    }
+}
+
+#[derive(Clone)]
+struct BoxStorage {
+    buf: Rc<UnsafeCell<Box<[u8]>>>,
+    off: u32,
+}
+
+impl BoxStorage {
+    #[inline(always)]
+    fn len(&self) -> usize {
+        let buf = self.buf.get();
+        let len = unsafe { (*buf).len() };
+        len - self.off as usize
+    }
+
+    fn as_ptr(&self) -> *const u8 {
+        let buf = self.buf.get();
+        unsafe { (*buf).as_ptr().add(self.off as usize) }
+    }
+
+    unsafe fn as_mut_ptr(&self) -> *mut u8 {
+        let buf = self.buf.get();
+        (*buf).as_mut_ptr().add(self.off as usize)
+    }
+
+    /// Returns a slice of bytes into this buffer, of the specified length
+    /// Panics if the length is larger than the buffer.
+    fn slice(&self, len: u32) -> &[u8] {
+        let buf = self.buf.get();
+        unsafe { &(*buf)[self.off as usize..][..len as usize] }
+    }
+
+    /// Returns a mutable slice of bytes into this buffer, of the specified
+    /// length Panics if the length is larger than the buffer.
+    fn slice_mut(&mut self, len: u32) -> &mut [u8] {
+        let buf = self.buf.get();
+        unsafe { &mut (*buf)[self.off as usize..][..len as usize] }
+    }
+
+    fn cap(&self) -> usize {
+        let buf = self.buf.get();
+        unsafe { (*buf).len() }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+#[error("slice does not fit into this RollMut")]
+pub struct DoesNotFit;
+
+impl RollMut {
+    /// Allocate, using a single [BufMut] for storage.
+    pub fn alloc() -> eyre::Result<Self> {
+        Ok(Self {
+            storage: StorageMut::Buf(BufMut::alloc()?),
+            len: 0,
+        })
+    }
+
+    /// Double the capacity of this buffer by reallocating it, copying the
+    /// filled part into the new buffer. This method always uses a `Box<[u8]>`
+    /// for storage.
+    ///
+    /// This method is somewhat expensive.
+    pub fn grow(&mut self) {
+        let old_cap = self.storage.cap();
+        let new_cap = old_cap * 2;
+        // TODO: optimize via `MaybeUninit`?
+        let b = vec![0; new_cap].into_boxed_slice();
+        let mut bs = BoxStorage {
+            buf: Rc::new(UnsafeCell::new(b)),
+            off: 0,
+        };
+        let src_slice = self.borrow_filled();
+        let dst_slice = bs.slice_mut(src_slice.len() as u32);
+        dst_slice.copy_from_slice(src_slice);
+        let next_storage = StorageMut::Box(bs);
+
+        self.storage = next_storage;
+    }
+
+    /// Reallocates the backing storage for this buffer, copying the filled
+    /// portion into it. Panics if `len() == storage_size()`, in which case
+    /// reallocating won't do much good
+    pub fn realloc(&mut self) -> eyre::Result<()> {
+        assert!(self.len() != self.storage_size());
+
+        let src_slice = self.borrow_filled();
+        let next_storage = match &self.storage {
+            StorageMut::Buf(_) => {
+                let mut next_b = BufMut::alloc()?;
+                next_b[..src_slice.len()].copy_from_slice(src_slice);
+                StorageMut::Buf(next_b)
+            }
+            StorageMut::Box(b) => {
+                if src_slice.len() > BUF_SIZE as usize {
+                    // TODO: optimize via `MaybeUninit`?
+                    let mut next_b = vec![0; b.cap()].into_boxed_slice();
+                    next_b[..src_slice.len()].copy_from_slice(src_slice);
+                    let next_b = BoxStorage {
+                        buf: Rc::new(UnsafeCell::new(next_b)),
+                        off: 0,
+                    };
+                    StorageMut::Box(next_b)
+                } else {
+                    let mut next_b = BufMut::alloc()?;
+                    next_b[..src_slice.len()].copy_from_slice(src_slice);
+                    StorageMut::Buf(next_b)
+                }
+            }
+        };
+
+        self.storage = next_storage;
+
+        Ok(())
+    }
+
+    /// Reserve more capacity for this buffer if this buffer is full.
+    /// If this buffer's size matches the underlying storage size,
+    /// this is equivalent to `grow`. Otherwise, it's equivalent
+    /// to `realloc`.
+    pub fn reserve(&mut self) -> eyre::Result<()> {
+        if self.len() < self.cap() {
+            return Ok(());
+        }
+
+        if self.len() < self.storage_size() {
+            // we don't need to go up a buffer size
+            self.realloc()?
+        } else {
+            self.grow()
+        }
+
+        Ok(())
+    }
+
+    /// The length (filled portion) of this buffer, that can be read
+    #[inline(always)]
+    pub fn len(&self) -> usize {
+        self.len as usize
+    }
+
+    /// Returns true if this is empty
+    #[inline(always)]
+    pub fn is_empty(&self) -> bool {
+        self.len == 0
+    }
+
+    /// The capacity of this buffer, that can be written to
+    #[inline(always)]
+    pub fn cap(&self) -> usize {
+        self.storage.len() - self.len as usize
+    }
+
+    /// The size of the underlying storage (to know whether
+    /// to reallocate or grow)
+    pub fn storage_size(&self) -> usize {
+        self.storage.cap()
+    }
+
+    /// Read at most `limit` bytes from [ReadOwned] into this buffer.
+    ///
+    /// This method takes ownership of `self` because it submits an io_uring
+    /// operation, where the kernel owns the read buffer - the only way to
+    /// gain ownership of `self` again is to complete the read operation.
+    ///
+    /// Panics if `cap` is zero
+    pub async fn read_into(
+        self,
+        limit: usize,
+        r: &impl ReadOwned,
+    ) -> (std::io::Result<usize>, Self) {
+        let read_cap = std::cmp::min(limit, self.cap());
+        assert!(read_cap > 0, "refusing to do empty read");
+
+        let read_off = self.len;
+        let read_into = ReadInto {
+            buf: self,
+            off: read_off,
+            cap: read_cap.try_into().unwrap(),
+            init: 0,
+        };
+        let (res, mut read_into) = r.read(read_into).await;
+        read_into.buf.len += read_into.init;
+        (res, read_into.buf)
+    }
+
+    /// Put a slice into this buffer, fails if the slice doesn't fit in the buffer's capacity
+    pub fn put(&mut self, s: impl AsRef<[u8]>) -> Result<(), DoesNotFit> {
+        let s = s.as_ref();
+
+        let len = s.len();
+        if len > self.cap() {
+            return Err(DoesNotFit);
+        }
+        unsafe {
+            let ptr = self.storage.as_mut_ptr().add(self.len as usize);
+            std::ptr::copy_nonoverlapping(s.as_ptr(), ptr, len);
+        }
+        let u32_len: u32 = len.try_into().unwrap();
+        self.len += u32_len;
+        Ok(())
+    }
+
+    /// Get a [Roll] corresponding to the filled portion of this buffer
+    pub fn filled(&self) -> Roll {
+        match &self.storage {
+            StorageMut::Buf(b) => b.freeze_slice(0..self.len()).into(),
+            StorageMut::Box(b) => RollBox {
+                b: b.clone(),
+                len: self.len,
+            }
+            .into(),
+        }
+    }
+
+    /// Borrow the filled portion of this buffer as a slice
+    pub fn borrow_filled(&self) -> &[u8] {
+        match &self.storage {
+            StorageMut::Buf(b) => &b[..self.len as usize],
+            StorageMut::Box(b) => b.slice(self.len),
+        }
+    }
+
+    /// Split this [RollMut] at the given index.
+    /// Panics if `at > len()`. If `at < len()`, the filled portion will carry
+    /// over in the new [RollMut].
+    pub fn skip(&mut self, n: usize) {
+        let u32_n: u32 = n.try_into().unwrap();
+        assert!(u32_n <= self.len);
+
+        match &mut self.storage {
+            StorageMut::Buf(b) => b.skip(n),
+            StorageMut::Box(b) => b.off += u32_n,
+        }
+        self.len -= u32_n;
+    }
+
+    /// Takes the first `n` bytes (up to `len`) as a `Roll`, and advances
+    /// this buffer. Returns `None` if `len` is zero. Panics if `n` is
+    /// zero.
+    pub fn take_at_most(&mut self, n: usize) -> Option<Roll> {
+        assert!(n != 0, "refusing to do empty take_at_most");
+
+        if self.len == 0 {
+            return None;
+        }
+
+        let n = std::cmp::min(n, self.len as usize);
+        let roll = self.filled().slice(..n);
+        self.skip(n);
+        Some(roll)
+    }
+
+    /// Takes the whole `filled` part. Panics if empty, since this is probably
+    /// a misuse.
+    pub fn take_all(&mut self) -> Roll {
+        let roll = self.filled();
+        assert!(
+            !roll.is_empty(),
+            "take_all is pointless if the filled part is empty, check len first"
+        );
+        self.skip(roll.len());
+        roll
+    }
+
+    /// Advance this buffer, keeping the filled part only starting with
+    /// the given roll. (Useful after parsing something with nom)
+    ///
+    /// Panics if the roll is not from this buffer
+    pub fn keep(&mut self, roll: Roll) {
+        match (&mut self.storage, &roll.inner) {
+            (StorageMut::Buf(ours), RollInner::Buf(theirs)) => {
+                assert_eq!(ours.index, theirs.index, "roll must be from same buffer");
+                assert!(theirs.off >= ours.off, "roll must start within buffer");
+                let skipped = theirs.off - ours.off;
+                self.len -= skipped as u32;
+                ours.off = theirs.off;
+            }
+            (StorageMut::Box(ours), RollInner::Box(theirs)) => {
+                assert_eq!(
+                    ours.buf.get(),
+                    theirs.b.buf.get(),
+                    "roll must be from same buffer"
+                );
+                assert!(theirs.b.off >= ours.off, "roll must start within buffer");
+                let skipped = theirs.b.off - ours.off;
+                self.len -= skipped;
+                ours.off = theirs.b.off;
+            }
+            _ => {
+                panic!("roll must be from same buffer");
+            }
+        }
+    }
+}
+
+struct ReadInto {
+    buf: RollMut,
+    off: u32,
+    cap: u32,
+    init: u32,
+}
+
+unsafe impl IoBuf for ReadInto {
+    #[inline(always)]
+    fn stable_ptr(&self) -> *const u8 {
+        unreachable!("ReadInto should never be used as a write buffer")
+    }
+
+    #[inline(always)]
+    fn bytes_init(&self) -> usize {
+        unreachable!("ReadInto should never be used as a write buffer")
+    }
+
+    #[inline(always)]
+    fn bytes_total(&self) -> usize {
+        self.cap as _
+    }
+}
+
+unsafe impl IoBufMut for ReadInto {
+    #[inline(always)]
+    fn stable_mut_ptr(&mut self) -> *mut u8 {
+        unsafe { self.buf.storage.as_mut_ptr().add(self.off as usize) }
+    }
+
+    #[inline(always)]
+    unsafe fn set_init(&mut self, pos: usize) {
+        self.init = pos.try_into().expect("reads should be < 4GiB");
+    }
+}
+
+/// An immutable view into a [RollMut]
+#[derive(Clone)]
+pub struct Roll {
+    inner: RollInner,
+}
+
+impl Debug for Roll {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        Debug::fmt(&self[..], f)
+    }
+}
+
+unsafe impl IoBuf for Roll {
+    #[inline(always)]
+    fn stable_ptr(&self) -> *const u8 {
+        match &self.inner {
+            RollInner::Buf(b) => b.stable_ptr(),
+            RollInner::Box(b) => b.b.as_ptr(),
+        }
+    }
+
+    #[inline(always)]
+    fn bytes_init(&self) -> usize {
+        self.len()
+    }
+
+    #[inline(always)]
+    fn bytes_total(&self) -> usize {
+        self.len()
+    }
+}
+
+impl<T> PartialEq<T> for Roll
+where
+    T: AsRef<[u8]>,
+{
+    fn eq(&self, other: &T) -> bool {
+        &self[..] == other.as_ref()
+    }
+}
+
+impl Eq for Roll {}
+
+impl From<RollInner> for Roll {
+    fn from(inner: RollInner) -> Self {
+        Self { inner }
+    }
+}
+
+impl From<RollBox> for Roll {
+    fn from(b: RollBox) -> Self {
+        RollInner::Box(b).into()
+    }
+}
+
+impl From<Buf> for Roll {
+    fn from(b: Buf) -> Self {
+        RollInner::Buf(b).into()
+    }
+}
+
+#[derive(Clone)]
+enum RollInner {
+    Buf(Buf),
+    Box(RollBox),
+}
+
+#[derive(Clone)]
+struct RollBox {
+    b: BoxStorage,
+    len: u32,
+}
+
+impl RollBox {
+    #[inline(always)]
+    fn split_at(self, at: usize) -> (Self, Self) {
+        let at: u32 = at.try_into().unwrap();
+        assert!(at <= self.len);
+
+        let left = Self {
+            b: self.b.clone(),
+            len: at,
+        };
+
+        let mut right = Self {
+            b: self.b,
+            len: self.len - at,
+        };
+        right.b.off += at;
+
+        (left, right)
+    }
+
+    #[inline(always)]
+    fn len(&self) -> usize {
+        self.len as usize
+    }
+
+    fn slice(mut self, range: impl RangeBounds<usize>) -> Self {
+        let mut new_start = 0;
+        let mut new_end = self.len();
+
+        match range.start_bound() {
+            Bound::Included(&n) => new_start = n,
+            Bound::Excluded(&n) => new_start = n + 1,
+            Bound::Unbounded => {}
+        }
+
+        match range.end_bound() {
+            Bound::Included(&n) => new_end = n + 1,
+            Bound::Excluded(&n) => new_end = n,
+            Bound::Unbounded => {}
+        }
+
+        assert!(new_start <= new_end);
+        assert!(new_end <= self.len());
+
+        self.b.off += new_start as u32;
+        self.len = (new_end - new_start) as u32;
+        self
+    }
+}
+
+impl AsRef<[u8]> for RollBox {
+    #[inline(always)]
+    fn as_ref(&self) -> &[u8] {
+        self.b.slice(self.len)
+    }
+}
+
+impl AsRef<[u8]> for Roll {
+    #[inline(always)]
+    fn as_ref(&self) -> &[u8] {
+        match &self.inner {
+            RollInner::Buf(b) => b.as_ref(),
+            RollInner::Box(b) => b.as_ref(),
+        }
+    }
+}
+
+impl Deref for Roll {
+    type Target = [u8];
+
+    #[inline(always)]
+    fn deref(&self) -> &Self::Target {
+        self.as_ref()
+    }
+}
+
+impl RollInner {
+    #[inline(always)]
+    fn split_at(self, at: usize) -> (Self, Self) {
+        match self {
+            RollInner::Buf(b) => {
+                let (left, right) = b.split_at(at);
+                (RollInner::Buf(left), RollInner::Buf(right))
+            }
+            RollInner::Box(b) => {
+                let (left, right) = b.split_at(at);
+                (RollInner::Box(left), RollInner::Box(right))
+            }
+        }
+    }
+}
+
+impl Roll {
+    /// Returns the length of this roll
+    #[inline(always)]
+    pub fn len(&self) -> usize {
+        match &self.inner {
+            RollInner::Buf(b) => b.len(),
+            RollInner::Box(b) => b.len(),
+        }
+    }
+
+    /// Returns true if this roll is empty
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    pub fn split_at(self, at: usize) -> (Roll, Roll) {
+        let (left, right) = self.inner.split_at(at);
+        (left.into(), right.into())
+    }
+
+    pub fn slice(self, range: impl RangeBounds<usize>) -> Self {
+        match self.inner {
+            RollInner::Buf(b) => b.slice(range).into(),
+            RollInner::Box(b) => b.slice(range).into(),
+        }
+    }
+
+    pub fn iter(&self) -> RollIter {
+        RollIter {
+            roll: self.clone(),
+            pos: 0,
+        }
+    }
+
+    pub fn to_string_lossy(&self) -> Cow<'_, str> {
+        String::from_utf8_lossy(self)
+    }
+}
+
+impl InputIter for Roll {
+    type Item = u8;
+    type Iter = Enumerate<Self::IterElem>;
+    type IterElem = RollIter;
+
+    #[inline]
+    fn iter_indices(&self) -> Self::Iter {
+        self.iter_elements().enumerate()
+    }
+    #[inline]
+    fn iter_elements(&self) -> Self::IterElem {
+        self.iter()
+    }
+    #[inline]
+    fn position<P>(&self, predicate: P) -> Option<usize>
+    where
+        P: Fn(Self::Item) -> bool,
+    {
+        self.iter().position(predicate)
+    }
+    #[inline]
+    fn slice_index(&self, count: usize) -> Result<usize, Needed> {
+        if self.len() >= count {
+            Ok(count)
+        } else {
+            Err(Needed::new(count - self.len()))
+        }
+    }
+}
+
+/// An iterator over [Roll]
+pub struct RollIter {
+    roll: Roll,
+    pos: usize,
+}
+
+impl Iterator for RollIter {
+    type Item = u8;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.pos >= self.roll.len() {
+            return None;
+        }
+
+        let c = self.roll[self.pos];
+        self.pos += 1;
+        Some(c)
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let remaining = self.roll.len() - self.pos;
+        (remaining, Some(remaining))
+    }
+}
+
+impl InputTake for Roll {
+    #[inline]
+    fn take(&self, count: usize) -> Self {
+        self.clone().slice(..count)
+    }
+    #[inline]
+    fn take_split(&self, count: usize) -> (Self, Self) {
+        let (prefix, suffix) = self.clone().split_at(count);
+        (suffix, prefix)
+    }
+}
+
+impl InputTakeAtPosition for Roll {
+    type Item = u8;
+
+    fn split_at_position<P, E: nom::error::ParseError<Self>>(
+        &self,
+        predicate: P,
+    ) -> nom::IResult<Self, Self, E>
+    where
+        P: Fn(Self::Item) -> bool,
+    {
+        match self.iter().position(predicate) {
+            Some(i) => Ok(self.clone().take_split(i)),
+            None => Err(nom::Err::Incomplete(nom::Needed::new(1))),
+        }
+    }
+
+    fn split_at_position1<P, E: nom::error::ParseError<Self>>(
+        &self,
+        predicate: P,
+        e: nom::error::ErrorKind,
+    ) -> nom::IResult<Self, Self, E>
+    where
+        P: Fn(Self::Item) -> bool,
+    {
+        match self.iter().position(predicate) {
+            Some(0) => Err(nom::Err::Error(E::from_error_kind(self.clone(), e))),
+            Some(i) => Ok(self.take_split(i)),
+            None => Err(nom::Err::Incomplete(nom::Needed::new(1))),
+        }
+    }
+
+    fn split_at_position_complete<P, E: nom::error::ParseError<Self>>(
+        &self,
+        predicate: P,
+    ) -> nom::IResult<Self, Self, E>
+    where
+        P: Fn(Self::Item) -> bool,
+    {
+        match self.iter().position(predicate) {
+            Some(i) => Ok(self.take_split(i)),
+            None => Ok(self.take_split(self.input_len())),
+        }
+    }
+
+    fn split_at_position1_complete<P, E: nom::error::ParseError<Self>>(
+        &self,
+        predicate: P,
+        e: nom::error::ErrorKind,
+    ) -> nom::IResult<Self, Self, E>
+    where
+        P: Fn(Self::Item) -> bool,
+    {
+        match self.iter().position(predicate) {
+            Some(0) => Err(nom::Err::Error(E::from_error_kind(self.clone(), e))),
+            Some(i) => Ok(self.take_split(i)),
+            None => {
+                if self.is_empty() {
+                    Err(nom::Err::Error(E::from_error_kind(self.clone(), e)))
+                } else {
+                    Ok(self.take_split(self.input_len()))
+                }
+            }
+        }
+    }
+}
+
+impl FindSubstring<&[u8]> for Roll {
+    fn find_substring(&self, substr: &[u8]) -> Option<usize> {
+        if substr.len() > self.len() {
+            return None;
+        }
+
+        let (&substr_first, substr_rest) = match substr.split_first() {
+            Some(split) => split,
+            // an empty substring is found at position 0
+            // This matches the behavior of str.find("").
+            None => return Some(0),
+        };
+
+        if substr_rest.is_empty() {
+            return memchr::memchr(substr_first, self);
+        }
+
+        let mut offset = 0;
+        let haystack = &self[..self.len() - substr_rest.len()];
+
+        while let Some(position) = memchr::memchr(substr_first, &haystack[offset..]) {
+            offset += position;
+            let next_offset = offset + 1;
+            if &self[next_offset..][..substr_rest.len()] == substr_rest {
+                return Some(offset);
+            }
+
+            offset = next_offset;
+        }
+
+        None
+    }
+}
+
+impl Compare<&[u8]> for Roll {
+    #[inline(always)]
+    fn compare(&self, t: &[u8]) -> CompareResult {
+        let pos = self.iter().zip(t.iter()).position(|(a, b)| a != *b);
+
+        match pos {
+            Some(_) => CompareResult::Error,
+            None => {
+                if self.len() >= t.len() {
+                    CompareResult::Ok
+                } else {
+                    CompareResult::Incomplete
+                }
+            }
+        }
+    }
+
+    #[inline(always)]
+    fn compare_no_case(&self, t: &[u8]) -> CompareResult {
+        if self
+            .iter()
+            .zip(t)
+            .any(|(a, b)| lowercase_byte(a) != lowercase_byte(*b))
+        {
+            CompareResult::Error
+        } else if self.len() < t.len() {
+            CompareResult::Incomplete
+        } else {
+            CompareResult::Ok
+        }
+    }
+}
+
+fn lowercase_byte(c: u8) -> u8 {
+    match c {
+        b'A'..=b'Z' => c - b'A' + b'a',
+        _ => c,
+    }
+}
+
+impl InputLength for Roll {
+    #[inline]
+    fn input_len(&self) -> usize {
+        self.len()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use nom::IResult;
+    use tokio_uring::net::{TcpListener, TcpStream};
+
+    use crate::{ChanRead, Roll, RollMut, BUF_SIZE};
+
+    #[test]
+    fn test_roll_put() {
+        fn test_roll_put_inner(mut rm: RollMut) {
+            let initial_size = rm.cap();
+
+            rm.put(b"hello").unwrap();
+            assert_eq!(rm.cap(), initial_size - 5);
+
+            let filled = rm.filled();
+            assert_eq!(&filled[..], b"hello");
+
+            rm.skip(5);
+            assert_eq!(rm.len(), 0);
+            assert_eq!(rm.cap(), initial_size - 5);
+        }
+
+        let rm = RollMut::alloc().unwrap();
+        assert_eq!(rm.cap(), BUF_SIZE as usize);
+        test_roll_put_inner(rm);
+
+        let mut rm = RollMut::alloc().unwrap();
+        rm.grow();
+        test_roll_put_inner(rm);
+
+        let mut rm = RollMut::alloc().unwrap();
+        rm.grow();
+        rm.grow();
+        test_roll_put_inner(rm);
+    }
+
+    #[test]
+    fn test_roll_put_does_not_fit() {
+        let mut rm = RollMut::alloc().unwrap();
+        rm.put(" ".repeat(rm.cap())).unwrap();
+
+        let err = rm.put("drop").unwrap_err();
+        assert!(format!("{err:?}").contains("DoesNotFit"));
+        assert!(format!("{err}").contains("does not fit"));
+    }
+
+    #[test]
+    fn test_roll_realloc() {
+        fn test_roll_realloc_inner(mut rm: RollMut) {
+            let init_cap = rm.cap();
+            rm.put("hello").unwrap();
+            rm.take_all();
+            assert_eq!(rm.cap(), init_cap - 5);
+
+            rm.realloc().unwrap();
+            assert_eq!(rm.cap(), BUF_SIZE as usize);
+        }
+
+        let rm = RollMut::alloc().unwrap();
+        test_roll_realloc_inner(rm);
+
+        let mut rm = RollMut::alloc().unwrap();
+        rm.grow();
+        test_roll_realloc_inner(rm);
+    }
+
+    #[test]
+    fn test_roll_realloc_big() {
+        let mut rm = RollMut::alloc().unwrap();
+        rm.grow();
+
+        let put = "x".repeat(rm.cap() * 2 / 3);
+        rm.put(&put).unwrap();
+        rm.realloc().unwrap();
+
+        assert_eq!(rm.storage_size(), BUF_SIZE as usize * 2);
+        assert_eq!(rm.len(), put.len());
+        assert_eq!(rm.borrow_filled(), put.as_bytes());
+    }
+
+    #[test]
+    fn test_roll_reserve() {
+        let mut rm = RollMut::alloc().unwrap();
+        assert_eq!(rm.cap(), BUF_SIZE as usize);
+        assert_eq!(rm.len(), 0);
+        rm.reserve().unwrap();
+        assert_eq!(rm.cap(), BUF_SIZE as usize);
+        assert_eq!(rm.len(), 0);
+
+        rm.put("hello").unwrap();
+        rm.take_all();
+
+        assert_eq!(rm.cap(), BUF_SIZE as usize - 5);
+        assert_eq!(rm.len(), 0);
+        rm.reserve().unwrap();
+        assert_eq!(rm.cap(), BUF_SIZE as usize - 5);
+        assert_eq!(rm.len(), 0);
+
+        let old_cap = rm.cap();
+        rm.put(b" ".repeat(old_cap)).unwrap();
+        assert_eq!(rm.cap(), 0);
+        assert_eq!(rm.len(), old_cap);
+
+        rm.reserve().unwrap();
+        assert_eq!(rm.cap(), 5);
+        assert_eq!(rm.len(), old_cap);
+
+        rm.put("hello").unwrap();
+        rm.reserve().unwrap();
+        assert_eq!(rm.cap(), BUF_SIZE as usize);
+        assert_eq!(rm.len(), BUF_SIZE as usize);
+    }
+
+    #[test]
+    fn test_roll_put_then_grow() {
+        let mut rm = RollMut::alloc().unwrap();
+        assert_eq!(rm.cap(), BUF_SIZE as usize);
+
+        let input = b"I am pretty long";
+
+        rm.put(input).unwrap();
+        assert_eq!(rm.len(), input.len());
+        assert_eq!(rm.borrow_filled(), input);
+
+        assert_eq!(rm.cap(), BUF_SIZE as usize - input.len());
+
+        rm.grow();
+        assert_eq!(rm.cap(), 2 * (BUF_SIZE as usize) - input.len());
+        assert_eq!(rm.borrow_filled(), input);
+
+        rm.skip(5);
+        assert_eq!(rm.borrow_filled(), b"pretty long");
+    }
+
+    #[test]
+    fn test_roll_readfrom_start() {
+        tokio_uring::start(async move {
+            let mut rm = RollMut::alloc().unwrap();
+
+            let (send, read) = ChanRead::new();
+            tokio_uring::spawn(async move {
+                send.send("123456").await.unwrap();
+            });
+
+            let mut res;
+            (res, rm) = rm.read_into(3, &read).await;
+            res.unwrap();
+
+            assert_eq!(rm.len(), 3);
+            assert_eq!(rm.filled().as_ref(), b"123");
+
+            (res, rm) = rm.read_into(3, &read).await;
+            res.unwrap();
+
+            assert_eq!(rm.len(), 6);
+            assert_eq!(rm.filled().as_ref(), b"123456");
+        });
+    }
+
+    #[test]
+    fn test_roll_keep() {
+        fn test_roll_keep_inner(mut rm: RollMut) {
+            rm.put(b"helloworld").unwrap();
+            assert_eq!(rm.borrow_filled(), b"helloworld");
+
+            {
+                let roll = rm.filled().slice(3..=5);
+                assert_eq!(roll, b"low");
+            }
+
+            {
+                let roll = rm.filled().slice(..5);
+                assert_eq!(roll, b"hello");
+            }
+
+            let roll = rm.filled().slice(5..);
+            assert_eq!(roll, b"world");
+
+            rm.keep(roll);
+            assert_eq!(rm.borrow_filled(), b"world");
+        }
+
+        let rm = RollMut::alloc().unwrap();
+        test_roll_keep_inner(rm);
+
+        let mut rm = RollMut::alloc().unwrap();
+        rm.grow();
+        test_roll_keep_inner(rm);
+    }
+
+    #[test]
+    #[should_panic(expected = "roll must be from same buffer")]
+    fn test_roll_keep_different_buf() {
+        let mut rm1 = RollMut::alloc().unwrap();
+        rm1.put("hello").unwrap();
+
+        let mut rm2 = RollMut::alloc().unwrap();
+        rm2.put("hello").unwrap();
+        let roll2 = rm2.take_all();
+
+        rm1.keep(roll2);
+    }
+
+    #[test]
+    #[should_panic(expected = "roll must be from same buffer")]
+    fn test_roll_keep_different_box() {
+        let mut rm1 = RollMut::alloc().unwrap();
+        rm1.grow();
+        rm1.put("hello").unwrap();
+
+        let mut rm2 = RollMut::alloc().unwrap();
+        rm2.grow();
+        rm2.put("hello").unwrap();
+        let roll2 = rm2.take_all();
+
+        rm1.keep(roll2);
+    }
+
+    #[test]
+    #[should_panic(expected = "roll must be from same buffer")]
+    fn test_roll_keep_different_type() {
+        let mut rm1 = RollMut::alloc().unwrap();
+        rm1.grow();
+        rm1.put("hello").unwrap();
+
+        let mut rm2 = RollMut::alloc().unwrap();
+        rm2.put("hello").unwrap();
+        let roll2 = rm2.take_all();
+
+        rm1.keep(roll2);
+    }
+
+    #[test]
+    #[should_panic(expected = "roll must start within buffer")]
+    fn test_roll_keep_before_buf() {
+        let mut rm1 = RollMut::alloc().unwrap();
+        rm1.put("hello").unwrap();
+        let roll = rm1.filled();
+        rm1.skip(5);
+        rm1.keep(roll);
+    }
+
+    #[test]
+    #[should_panic(expected = "roll must start within buffer")]
+    fn test_roll_keep_before_box() {
+        let mut rm1 = RollMut::alloc().unwrap();
+        rm1.grow();
+        rm1.put("hello").unwrap();
+        let roll = rm1.filled();
+        rm1.skip(5);
+        rm1.keep(roll);
+    }
+
+    #[test]
+    fn test_roll_iter() {
+        let mut rm = RollMut::alloc().unwrap();
+        rm.put(b"hello").unwrap();
+        let roll = rm.filled();
+        let v = roll.iter().collect::<Vec<_>>();
+        assert_eq!(v, b"hello");
+
+        assert_eq!(roll.to_string_lossy(), "hello");
+    }
+
+    #[test]
+    fn test_roll_iobuf() {
+        async fn test_roll_iobuf_inner(mut rm: RollMut) -> eyre::Result<()> {
+            rm.put(b"hello").unwrap();
+            let roll = rm.take_all();
+
+            let ln = TcpListener::bind("[::]:0".parse()?)?;
+            let local_addr = ln.local_addr()?;
+
+            let send_fut = async move {
+                let stream = TcpStream::connect(local_addr).await?;
+                let res;
+                (res, _) = stream.write_all(roll).await;
+                res?;
+                Ok::<_, eyre::Report>(())
+            };
+
+            let recv_fut = async move {
+                let (stream, addr) = ln.accept().await?;
+                println!("Accepted connection from {addr}");
+
+                let mut buf = Vec::with_capacity(1024);
+                let res;
+                (res, buf) = stream.read(buf).await;
+                res?;
+
+                assert_eq!(buf, b"hello");
+
+                Ok::<_, eyre::Report>(())
+            };
+
+            tokio::try_join!(send_fut, recv_fut)?;
+            Ok(())
+        }
+
+        tokio_uring::start(async move {
+            let rm = RollMut::alloc().unwrap();
+            test_roll_iobuf_inner(rm).await.unwrap();
+
+            let mut rm = RollMut::alloc().unwrap();
+            rm.grow();
+            test_roll_iobuf_inner(rm).await.unwrap();
+        });
+    }
+
+    #[test]
+    fn test_roll_take_at_most() {
+        let mut rm = RollMut::alloc().unwrap();
+        rm.put(b"hello").unwrap();
+        let roll = rm.take_at_most(4).unwrap();
+        assert_eq!(roll, b"hell");
+
+        let mut rm = RollMut::alloc().unwrap();
+        rm.put(b"hello").unwrap();
+        let roll = rm.take_at_most(12).unwrap();
+        assert_eq!(roll, b"hello");
+
+        let mut rm = RollMut::alloc().unwrap();
+        assert!(rm.take_at_most(12).is_none());
+    }
+
+    #[test]
+    fn test_roll_take_all() {
+        let mut rm = RollMut::alloc().unwrap();
+        rm.put(b"hello").unwrap();
+        let roll = rm.take_all();
+        assert_eq!(roll, b"hello");
+    }
+
+    #[test]
+    #[should_panic(expected = "take_all is pointless if the filled part is empty")]
+    fn test_roll_take_all_empty() {
+        let mut rm = RollMut::alloc().unwrap();
+        rm.take_all();
+    }
+
+    #[test]
+    #[should_panic(expected = "refusing to do empty take_at_most")]
+    fn test_roll_take_at_most_panic() {
+        let mut rm = RollMut::alloc().unwrap();
+        rm.take_at_most(0);
+    }
+
+    #[test]
+    fn test_roll_nom_sample() {
+        fn parse(i: Roll) -> IResult<Roll, Roll> {
+            nom::bytes::streaming::tag(&b"HTTP/1.1 200 OK"[..])(i)
+        }
+
+        let mut buf = RollMut::alloc().unwrap();
+
+        let input = b"HTTP/1.1 200 OK".repeat(1000);
+        let mut pending = &input[..];
+
+        loop {
+            let (rest, version) = match parse(buf.filled()) {
+                Ok(t) => t,
+                Err(e) => {
+                    if e.is_incomplete() {
+                        {
+                            if pending.is_empty() {
+                                println!("ran out of input");
+                                break;
+                            }
+
+                            let n = std::cmp::min(buf.cap(), pending.len());
+                            buf.put(&pending[..n]).unwrap();
+                            pending = &pending[n..];
+
+                            println!("advanced by {n}, {} remaining", pending.len());
+                        }
+
+                        continue;
+                    }
+                    panic!("parsing error: {e}");
+                }
+            };
+            assert_eq!(version, b"HTTP/1.1 200 OK");
+
+            buf.keep(rest);
+        }
+    }
+}
diff --git a/src/bufpool/agg.rs b/src/bufpool/agg.rs
deleted file mode 100644
index 8faa6e20..00000000
--- a/src/bufpool/agg.rs
+++ /dev/null
@@ -1,1111 +0,0 @@
-use std::{
-    cell::{RefCell, RefMut},
-    fmt,
-    iter::Enumerate,
-    ops::Range,
-    rc::Rc,
-};
-
-use nom::{
-    Compare, CompareResult, FindSubstring, InputIter, InputLength, InputTake, InputTakeAtPosition,
-};
-use smallvec::SmallVec;
-
-use crate::{
-    bufpool::{self, BufMut},
-    Buf,
-};
-
-use super::{IoChunk, IoChunkable};
-
-macro_rules! dbg2 {
-    ($($arg:tt)*) => {
-        #[cfg(debug_assertions)]
-        {
-            // 🐉 uncomment to debug tests:
-            // dbg!($($arg)*);
-        }
-    };
-}
-
-/// An "aggregate buffer", uses one or more [BufMut]s for storage. Allows
-/// writing to uninitialized data, and borrowing ref-counted [AggSlice] of
-/// initialized data.
-///
-/// ```text
-///     +-------------------------------+---------------------------------+
-///     |             block 0           |            block 1              |
-///     +-------------------------------+---------------------------------+
-///     |                  |                 |                            |
-///     |<----- off ------>|<----------------+- capacity ---------------->|
-///     |                  |                 |                            |
-///     | used by previous |<----- len ----->|<---------- avail --------->|
-///     | bufs and slices. |                 |                            |
-///                        |   filled and    |     can be written to      |
-///                        |    readable     |                            |
-/// ```
-///
-/// A non-zero `off` indicates that this [AggBuf] was split from another
-/// one, re-using `avail` bytes.
-///
-/// [AggSlice] offsets are relative to the global offset.
-pub struct AggBuf {
-    inner: Rc<RefCell<AggBufInner>>,
-}
-
-/// The inner representation of an [AggBuf].
-#[derive(Default)]
-pub struct AggBufInner {
-    /// storage
-    blocks: SmallVec<[BufMut; 5]>,
-
-    /// size of each block - typically a compile-time constant but let's
-    /// have a field anyway for now.
-    block_size: u32,
-
-    /// global offset: how many bytes to ignore from the first block, in case
-    /// this aggregate buffer is re-used from a previous operation.
-    off: u32,
-
-    /// length: number of bytes that are initialized (were we read into, and can
-    /// be sliced).
-    len: u32,
-}
-
-pub struct AggBufRead<'a> {
-    handle: &'a Rc<RefCell<AggBufInner>>,
-    borrow: std::cell::Ref<'a, AggBufInner>,
-}
-
-impl Default for AggBuf {
-    /// Create an empty [AggBuf].
-    ///
-    /// [AggBuf::grow_if_needed] must be called before writing to it.
-    fn default() -> Self {
-        let inner = AggBufInner::new();
-        Self {
-            inner: Rc::new(RefCell::new(inner)),
-        }
-    }
-}
-
-impl AggBuf {
-    /// Borrow this aggregate buffer immutably
-    pub fn read(&self) -> AggBufRead<'_> {
-        let handle = &self.inner;
-        let borrow = handle.borrow();
-        AggBufRead { handle, borrow }
-    }
-
-    /// Borrow this aggregate buffer mutably
-    pub fn write(&self) -> RefMut<AggBufInner> {
-        self.inner.borrow_mut()
-    }
-
-    /// Split off at the end of the filled portion
-    pub fn split(self) -> Self {
-        let slice = {
-            let read = self.read();
-            let len = read.len();
-            self.read().slice(len..len)
-        };
-        self.split_at(slice)
-    }
-
-    /// Split off at the start of the given slice, re-using the space in the
-    /// given slice, and any unfilled space.
-    ///
-    /// ```text
-    /// Before/after (1 block)
-    ///
-    ///                                               <-- rest -->
-    /// [.............AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABBBBBBBBBBBB...........]
-    /// <--- off ----><-------------------- len -----------------><-- avail-->
-    ///
-    /// [.............................................BBBBBBBBBBBB...........]
-    /// <------------------- off --------------------><--- len --><-- avail-->
-    ///
-    /// Before/after (2+ blocks)
-    ///
-    ///                                               <-- rest -->
-    /// [.............AAAAAAAAAAAAAAAAAAAA][AAAAAAAAAABBBBBBBBBBBB...........]
-    /// <--- off ----><-------------------- len -----------------><-- avail-->
-    ///
-    /// [..........BBBBBBBBBBBB...........]
-    /// <-- off --><--- len --><-- avail-->
-    /// ```
-    pub fn split_at(self, rest: AggSlice) -> Self {
-        let inner = self.inner.borrow();
-        let block_size = inner.block_size;
-        dbg2!("split_at", inner.off, inner.len, inner.capacity());
-
-        let abs_filled_end = inner.len + inner.off;
-        let abs_rest_start = rest.off + inner.off;
-        let rest_first_block = abs_rest_start / block_size;
-        let abs_block_start = rest_first_block * block_size;
-
-        dbg2!("split_at", abs_block_start, abs_filled_end, abs_rest_start);
-
-        // we now have (1 block)
-        //
-        // [.............AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABBBBBBBBBBBB...........]
-        // |                                             |          |
-        // +- abs_block_start                            |          + abs_filled_end
-        //                                               + abs_rest_start
-        //
-        // or (2+ blocks)
-        //
-        // [.............AAAAAAAAAAAAAAAAAAAA][AAAAAAAAAABBBBBBBBBBBB...........]
-        //                                    |          |          |
-        //                    abs_block_start +          |          + abs_filled_end
-        //                                               + abs_rest_start
-
-        assert!(
-            abs_block_start <= abs_rest_start,
-            "rest must start into last block. abs_block_start = {abs_block_start} < abs_rest_start = {abs_rest_start}",
-        );
-        assert!(
-            abs_rest_start <= abs_filled_end,
-            "rest must start before end of filled part"
-        );
-
-        let rel_filled_end = abs_filled_end - abs_block_start;
-        let rel_rest_start = abs_rest_start - abs_block_start;
-
-        // we now have (1 block)
-        //
-        // <------------------ offset ------------------><--- len --><-- avail-->
-        // [.............AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABBBBBBBBBBBB...........]
-        //                                               |          |
-        //                                               |          + rel_filled_end
-        //                                               + rel_rest_start
-        //
-        // or (2+ blocks)
-        //
-        // <- offset -><-- len --><-- avail-->
-        // [AAAAAAAAAABBBBBBBBBBBB...........]
-        //            |          |
-        //            |          + rel_filled_end
-        //            + rel_rest_start
-
-        if rel_rest_start == inner.block_size {
-            // there's nothing to re-use here, just return an empty buffer
-            return Default::default();
-        }
-
-        // let's clone get that block and make everything relative to its start
-        let reused_blocks = inner
-            .blocks
-            .iter()
-            .skip(rest_first_block as usize)
-            .map(|block| block.dangerous_clone())
-            .collect();
-
-        let new_inner = AggBufInner {
-            blocks: reused_blocks,
-            block_size: inner.block_size,
-            off: rel_rest_start,
-            len: rel_filled_end - rel_rest_start,
-        };
-        Self {
-            inner: Rc::new(RefCell::new(new_inner)),
-        }
-    }
-
-    /// Return a write slice appropriate for a io_uring read.
-    ///
-    /// This panics if the write_slice would be empty, to avoid
-    /// misuse / no-op reads.
-    pub fn write_slice(self) -> AggWriteSlice {
-        let ptr;
-        let len;
-
-        {
-            let mut inner = self.inner.borrow_mut();
-            if inner.blocks.is_empty() {
-                panic!(
-                    "holding AggBuf wrong: please call grow_if_needed before calling write_slice"
-                );
-            }
-
-            let (block_index, block_range) = inner.contiguous_range(inner.len..inner.capacity());
-            ptr = unsafe {
-                inner.blocks[block_index]
-                    .as_mut_ptr()
-                    .add(block_range.start)
-            };
-            len = block_range.len();
-        }
-
-        AggWriteSlice {
-            buf: self,
-            ptr,
-            len: len as _,
-            pos: 0,
-        }
-    }
-
-    /// Take at most `n` bytes from the filled portion of this buffer and
-    /// advance this buffer's offset. If this returns `None`, it means
-    /// the filled portion is empty.
-    pub fn take_contiguous_at_most(&mut self, n: u32) -> Option<Buf> {
-        let inner = self.inner.borrow();
-        if inner.len == 0 {
-            return None;
-        }
-
-        let start = 0;
-        let len = std::cmp::min(n, inner.len);
-        let end = start + len;
-
-        dbg2!("take_contiguous_at_most", start, len, end);
-
-        let (block_index, block_range) = inner.contiguous_range(start..end);
-        dbg2!("take_contiguous_at_most", block_index, &block_range);
-
-        let block = inner.blocks[block_index].dangerous_clone();
-        let u16_block_range: Range<u16> =
-            (block_range.start.try_into().unwrap())..(block_range.end.try_into().unwrap());
-        let ret = block.freeze_slice(u16_block_range);
-        let consumed = ret.len() as u32;
-
-        let next_inner = if block_range.end == inner.block_size as usize {
-            // we consumed a whole buf
-            dbg2!("take_contiguous_at_most", "consumed whole buf");
-            AggBufInner {
-                block_size: inner.block_size,
-                blocks: inner
-                    .blocks
-                    .iter()
-                    .skip(1)
-                    .map(|b| b.dangerous_clone())
-                    .collect(),
-                off: inner.off + consumed - inner.block_size,
-                len: inner.len - consumed,
-            }
-        } else {
-            // we still have part of a buf
-            dbg2!("take_contiguous_at_most", "still have part of a buf");
-            AggBufInner {
-                block_size: inner.block_size,
-                blocks: inner.blocks.iter().map(|b| b.dangerous_clone()).collect(),
-                off: inner.off + consumed,
-                len: inner.len - consumed,
-            }
-        };
-        drop(inner);
-
-        *self = Self {
-            inner: Rc::new(RefCell::new(next_inner)),
-        };
-        Some(ret)
-    }
-}
-
-/// A write slice of an [AggBuf] suitable for an io_uring read/write
-pub struct AggWriteSlice {
-    buf: AggBuf,
-    ptr: *mut u8,
-    pos: u32,
-    len: u32,
-}
-
-impl AggWriteSlice {
-    pub fn into_inner(self) -> AggBuf {
-        self.buf.inner.borrow_mut().len += self.pos;
-        self.buf
-    }
-
-    /// Limit how many bytes can be read. If the slice was slower
-    /// in the first place, don't do anything.
-    pub fn limit(mut self, len: u64) -> Self {
-        if len < self.len as u64 {
-            self.len = len as u32;
-        }
-        self
-    }
-}
-
-unsafe impl tokio_uring::buf::IoBuf for AggWriteSlice {
-    fn stable_ptr(&self) -> *const u8 {
-        self.ptr
-    }
-
-    fn bytes_init(&self) -> usize {
-        self.pos as _
-    }
-
-    fn bytes_total(&self) -> usize {
-        self.len as _
-    }
-}
-
-unsafe impl tokio_uring::buf::IoBufMut for AggWriteSlice {
-    fn stable_mut_ptr(&mut self) -> *mut u8 {
-        self.ptr
-    }
-
-    unsafe fn set_init(&mut self, pos: usize) {
-        self.pos = pos as _
-    }
-}
-
-impl AggBufInner {
-    fn new() -> Self {
-        Self {
-            blocks: Default::default(),
-            block_size: crate::bufpool::BUF_SIZE as _,
-            off: 0,
-            len: 0,
-        }
-    }
-
-    /// Returns the size of each buffer in the pool
-    #[inline(always)]
-    pub fn block_size(&self) -> u32 {
-        self.block_size
-    }
-
-    /// Returns the total capacity of the buffer
-    #[inline(always)]
-    pub fn capacity(&self) -> u32 {
-        self.blocks.len() as u32 * self.block_size - self.off
-    }
-
-    /// Returns the (filled) length of the buffer
-    #[inline(always)]
-    pub fn len(&self) -> u32 {
-        self.len
-    }
-
-    /// Return true if this aggregate buf is empty
-    #[inline(always)]
-    pub fn is_empty(&self) -> bool {
-        self.len == 0
-    }
-
-    /// Returns a block index and a slice into it, given a slice into the
-    /// aggregate buffer. Doesn't check for the `filled` region
-    fn contiguous_range(&self, wanted: Range<u32>) -> (usize, Range<usize>) {
-        dbg2!("contiguous_range", &wanted);
-        let (start, end) = (wanted.start, wanted.end);
-        assert!(start <= end);
-
-        if wanted.start == wanted.end {
-            // special case: empty slice
-            return (0, 0..0);
-        }
-
-        let wanted = end - start;
-
-        // take the global offset into account when indexing into bufs
-        let start = start + self.off;
-        let block_index = (start / self.block_size) as usize;
-
-        debug_assert!(block_index < self.blocks.len());
-
-        let block_offset = start % self.block_size;
-        dbg2!("contiguous_range", block_index, block_offset);
-        let avail = self.block_size - block_offset;
-        let given = std::cmp::min(avail, wanted);
-        dbg2!("contiguous_range", avail, given);
-
-        let block_offset = block_offset as usize;
-        let given = given as usize;
-
-        (block_index, block_offset..block_offset + given)
-    }
-
-    /// If `len == capacity` (ie. the `unfilled_mut` slice would be empty), try
-    /// to add a block to this aggregate buffer. This is fallible, as we might
-    /// be out of memory.
-    pub fn grow_if_needed(&mut self) -> Result<(), bufpool::Error> {
-        if self.len < self.capacity() {
-            return Ok(());
-        }
-
-        let block = BufMut::alloc()?;
-        self.blocks.push(block);
-        Ok(())
-    }
-
-    /// Writes a slice into the buffer, growing it if needed.
-    pub fn put(&mut self, s: impl AsRef<[u8]>) -> Result<(), bufpool::Error> {
-        let mut s = s.as_ref();
-        while !s.is_empty() {
-            self.grow_if_needed()?;
-
-            {
-                let unfilled = self.unfilled_mut();
-                let unfilled_len = unfilled.len();
-                let to_copy = std::cmp::min(unfilled_len, s.len());
-                unfilled[..to_copy].copy_from_slice(&s[..to_copy]);
-                self.advance(to_copy as u32);
-                s = &s[to_copy..];
-            }
-        }
-        Ok(())
-    }
-
-    /// Writes an [AggSlice] into this buffer, growing it if needed.
-    pub fn put_agg(&mut self, s: &AggSlice) -> Result<(), bufpool::Error> {
-        // TODO: this is very, very inefficient
-        self.put(&s.to_vec())
-    }
-
-    /// Gives a mutable slice that can be written to.
-    /// Must call `advance` after writing to the returned slice.
-    pub fn unfilled_mut(&mut self) -> &mut [u8] {
-        if self.blocks.is_empty() {
-            return &mut [];
-        }
-
-        let (block_index, range) = self.contiguous_range(self.len()..self.capacity());
-        &mut self.blocks[block_index][range]
-    }
-
-    /// Called after writing to `unfilled_mut`. Panics if adding `n` brings the
-    /// buffer over capacity.
-    ///
-    /// This isn't unsafe because [BufMut] are zeroed. But you will get
-    /// incorrect results if you advance past the end of what's been filled.
-    pub fn advance(&mut self, n: u32) {
-        assert!(self.len + n <= self.capacity());
-        self.len += n;
-    }
-}
-
-impl<'a> std::ops::Deref for AggBufRead<'a> {
-    type Target = AggBufInner;
-
-    fn deref(&self) -> &Self::Target {
-        &self.borrow
-    }
-}
-
-impl AggBufRead<'_> {
-    /// Return a slice for the whole length of this buffer
-    pub fn read_slice(&self) -> AggSlice {
-        self.slice(0..self.len())
-    }
-
-    /// Take a slice out of the filled portion of this buffer. Panics if
-    /// it is outside the filled portion.
-    pub fn slice(&self, range: Range<u32>) -> AggSlice {
-        assert!(range.start <= range.end);
-        assert!(range.end <= self.len);
-
-        AggSlice {
-            parent: AggBuf {
-                inner: self.handle.clone(),
-            },
-            off: range.start as _,
-            len: (range.end - range.start) as _,
-        }
-    }
-
-    /// Returns the biggest continuous slice we can get a the given offset.
-    /// Panics if `wanted` is out of bounds. If the requested range spans
-    /// multiple buffers, the returned slice will be smaller than requested.
-    pub fn filled(&self, wanted: Range<u32>) -> &[u8] {
-        // FIXME: this probably shouldn't be pub
-
-        assert!(wanted.end <= self.len);
-
-        let (block_index, given) = self.contiguous_range(wanted);
-        &self.blocks[block_index][given]
-    }
-}
-
-/// A slice of an [AggBuf]. This is a read-only view, it's clonable,
-/// it holds a reference to the underlying [AggBuf], so holding it
-/// will keep the _whole_ [AggBuf] alive.
-pub struct AggSlice {
-    parent: AggBuf,
-    off: u32,
-    len: u32,
-}
-
-impl Clone for AggSlice {
-    fn clone(&self) -> Self {
-        Self {
-            parent: AggBuf {
-                inner: self.parent.inner.clone(),
-            },
-            off: self.off,
-            len: self.len,
-        }
-    }
-}
-
-impl fmt::Debug for AggSlice {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("AggSlice")
-            .field("off", &self.off)
-            .field("len", &self.len)
-            .finish()
-    }
-}
-
-impl IoChunkable for AggSlice {
-    fn len(&self) -> usize {
-        self.len as _
-    }
-
-    /// Returns next contiguous slice
-    fn next_chunk(&self, offset: u32) -> Option<IoChunk> {
-        if offset == self.len {
-            return None;
-        }
-
-        let parent = self.parent.read();
-        let (block_index, range) = parent.contiguous_range(self.off + offset..self.off + self.len);
-
-        // FIXME: use try_into?
-        let range = (range.start as u16)..(range.end as u16);
-        Some(parent.blocks[block_index].freeze_slice(range).into())
-    }
-}
-
-impl AggSlice {
-    /// Returns an iterator over the bytes in this slice
-    #[inline]
-    pub fn iter(&self) -> AggSliceIter {
-        AggSliceIter {
-            slice: self.clone(),
-            pos: 0,
-        }
-    }
-
-    /// Returns as a vector. This allocates a lot.
-    pub fn to_vec(&self) -> Vec<u8> {
-        self.iter().collect()
-    }
-
-    /// Returns as a string. This allocates a lot.
-    pub fn to_string_lossy(&self) -> String {
-        String::from_utf8_lossy(&self.to_vec()).to_string()
-    }
-
-    /// Returns the length of this slice
-    pub fn len(&self) -> usize {
-        self.len as _
-    }
-
-    /// Returns true if this is empty
-    pub fn is_empty(&self) -> bool {
-        self.len == 0
-    }
-
-    /// Returns true if this slice equals `slice`, ignoring ASCII case
-    pub fn eq_ignore_ascii_case(&self, slice: impl AsRef<[u8]>) -> bool {
-        let slice = slice.as_ref();
-
-        if self.len() != slice.len() {
-            return false;
-        }
-
-        // FIXME: this is very naive and could be optimized
-        self.iter()
-            .zip(slice.iter())
-            .all(|(l, r)| l.eq_ignore_ascii_case(r))
-    }
-}
-
-impl<T> PartialEq<T> for AggSlice
-where
-    T: AsRef<[u8]>,
-{
-    fn eq(&self, slice: &T) -> bool {
-        let slice = slice.as_ref();
-
-        if self.len() != slice.len() {
-            return false;
-        }
-
-        // FIXME: this is very naive and could be optimized
-        self.iter().eq(slice.iter().copied())
-    }
-}
-
-impl InputLength for AggSlice {
-    fn input_len(&self) -> usize {
-        self.len as _
-    }
-}
-
-impl InputTake for AggSlice {
-    fn take(&self, count: usize) -> Self {
-        let count: u32 = count.try_into().unwrap();
-        if count > self.len {
-            panic!("take: count > self.len");
-        }
-
-        Self {
-            parent: AggBuf {
-                inner: self.parent.inner.clone(),
-            },
-            off: self.off,
-            len: count,
-        }
-    }
-
-    fn take_split(&self, count: usize) -> (Self, Self) {
-        let count: u32 = count.try_into().unwrap();
-        if count > self.len {
-            panic!("take_split: count > self.len");
-        }
-
-        let prefix = Self {
-            parent: AggBuf {
-                inner: self.parent.inner.clone(),
-            },
-            off: self.off,
-            len: count,
-        };
-        let suffix = Self {
-            parent: AggBuf {
-                inner: self.parent.inner.clone(),
-            },
-            off: self.off + count,
-            len: self.len - count,
-        };
-        (suffix, prefix)
-    }
-}
-
-impl Compare<&[u8]> for AggSlice {
-    #[inline(always)]
-    fn compare(&self, t: &[u8]) -> CompareResult {
-        let pos = self.iter().zip(t.iter()).position(|(a, b)| a != *b);
-
-        match pos {
-            Some(_) => CompareResult::Error,
-            None => {
-                if self.len() >= t.len() {
-                    CompareResult::Ok
-                } else {
-                    CompareResult::Incomplete
-                }
-            }
-        }
-    }
-
-    #[inline(always)]
-    fn compare_no_case(&self, t: &[u8]) -> CompareResult {
-        if self
-            .iter()
-            .zip(t)
-            .any(|(a, b)| lowercase_byte(a) != lowercase_byte(*b))
-        {
-            CompareResult::Error
-        } else if self.len() < t.len() {
-            CompareResult::Incomplete
-        } else {
-            CompareResult::Ok
-        }
-    }
-}
-
-impl InputTakeAtPosition for AggSlice {
-    type Item = u8;
-
-    fn split_at_position<P, E: nom::error::ParseError<Self>>(
-        &self,
-        predicate: P,
-    ) -> nom::IResult<Self, Self, E>
-    where
-        P: Fn(Self::Item) -> bool,
-    {
-        match self.iter().position(predicate) {
-            Some(i) => Ok(self.clone().take_split(i)),
-            None => Err(nom::Err::Incomplete(nom::Needed::new(1))),
-        }
-    }
-
-    fn split_at_position1<P, E: nom::error::ParseError<Self>>(
-        &self,
-        predicate: P,
-        e: nom::error::ErrorKind,
-    ) -> nom::IResult<Self, Self, E>
-    where
-        P: Fn(Self::Item) -> bool,
-    {
-        match self.iter().position(predicate) {
-            Some(0) => Err(nom::Err::Error(E::from_error_kind(self.clone(), e))),
-            Some(i) => Ok(self.take_split(i)),
-            None => Err(nom::Err::Incomplete(nom::Needed::new(1))),
-        }
-    }
-
-    fn split_at_position_complete<P, E: nom::error::ParseError<Self>>(
-        &self,
-        predicate: P,
-    ) -> nom::IResult<Self, Self, E>
-    where
-        P: Fn(Self::Item) -> bool,
-    {
-        match self.iter().position(predicate) {
-            Some(i) => Ok(self.take_split(i)),
-            None => Ok(self.take_split(self.input_len())),
-        }
-    }
-
-    fn split_at_position1_complete<P, E: nom::error::ParseError<Self>>(
-        &self,
-        predicate: P,
-        e: nom::error::ErrorKind,
-    ) -> nom::IResult<Self, Self, E>
-    where
-        P: Fn(Self::Item) -> bool,
-    {
-        match self.iter().position(predicate) {
-            Some(0) => Err(nom::Err::Error(E::from_error_kind(self.clone(), e))),
-            Some(i) => Ok(self.take_split(i)),
-            None => {
-                if self.is_empty() {
-                    Err(nom::Err::Error(E::from_error_kind(self.clone(), e)))
-                } else {
-                    Ok(self.take_split(self.input_len()))
-                }
-            }
-        }
-    }
-}
-
-impl FindSubstring<&[u8]> for AggSlice {
-    fn find_substring(&self, substr: &[u8]) -> Option<usize> {
-        let mut offset = None;
-        let mut curr_substr = substr;
-        for (i, c) in self.iter().enumerate() {
-            if c == curr_substr[0] {
-                if offset.is_none() {
-                    offset = Some(i);
-                }
-                curr_substr = &curr_substr[1..];
-                if curr_substr.is_empty() {
-                    return offset;
-                }
-            } else {
-                curr_substr = substr;
-            }
-        }
-
-        None
-    }
-}
-
-impl InputIter for AggSlice {
-    type Item = u8;
-    type Iter = Enumerate<AggSliceIter>;
-    type IterElem = AggSliceIter;
-
-    #[inline]
-    fn iter_indices(&self) -> Self::Iter {
-        self.iter().enumerate()
-    }
-
-    #[inline]
-    fn iter_elements(&self) -> Self::IterElem {
-        self.iter()
-    }
-
-    #[inline]
-    fn position<P>(&self, predicate: P) -> Option<usize>
-    where
-        P: Fn(Self::Item) -> bool,
-    {
-        self.iter().position(predicate)
-    }
-
-    #[inline]
-    fn slice_index(&self, count: usize) -> Result<usize, nom::Needed> {
-        if self.len() >= count {
-            Ok(count)
-        } else {
-            Err(nom::Needed::new(count - self.len()))
-        }
-    }
-}
-
-fn lowercase_byte(c: u8) -> u8 {
-    match c {
-        b'A'..=b'Z' => c - b'A' + b'a',
-        _ => c,
-    }
-}
-
-pub struct AggSliceIter {
-    slice: AggSlice,
-    pos: u32,
-}
-
-impl Iterator for AggSliceIter {
-    type Item = u8;
-
-    fn next(&mut self) -> Option<Self::Item> {
-        if self.pos >= self.slice.len {
-            return None;
-        }
-
-        dbg2!("AggSliceIter::next", self.pos, self.slice.len);
-
-        // FIXME: this implementation is extremely naive and not efficient at
-        // all. we shouldn't have to borrow here or do block math on every
-        // iteration: just until we run out of the current block.
-        let inner = self.slice.parent.inner.borrow();
-        let global_off = self.pos + self.slice.off;
-        let (block_index, range) = inner.contiguous_range(global_off..global_off + 1);
-        self.pos += 1;
-        Some(inner.blocks[block_index][range][0])
-    }
-
-    fn size_hint(&self) -> (usize, Option<usize>) {
-        let remaining = self.slice.len - self.pos;
-        (remaining as _, Some(remaining as _))
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use crate::bufpool::{AggBuf, AggBufInner, AggSlice};
-    use nom::IResult;
-    use pretty_assertions::assert_eq;
-
-    #[test]
-    fn agg_inner_size() {
-        assert_eq!(std::mem::size_of::<AggBufInner>(), 64);
-    }
-
-    #[test]
-    fn agg_slice_size() {
-        assert_eq!(std::mem::size_of::<AggSlice>(), 16);
-    }
-
-    #[test]
-    fn agg_fill() {
-        let buf: AggBuf = Default::default();
-
-        let block_size;
-        let agg_len;
-        {
-            let mut buf = buf.write();
-            block_size = buf.block_size();
-
-            {
-                println!("allocating first block");
-                buf.grow_if_needed().unwrap();
-            }
-
-            {
-                println!("filling first block");
-                let slice = buf.unfilled_mut();
-                let len = slice.len();
-                assert_eq!(len, block_size as usize);
-                slice.fill(1);
-                buf.advance(len as _);
-                assert_eq!(buf.len(), block_size as _);
-            }
-
-            {
-                println!("unfilled should be empty");
-                let slice = buf.unfilled_mut();
-                assert_eq!(slice.len(), 0);
-            }
-
-            {
-                println!("allocating second block");
-                buf.grow_if_needed().unwrap();
-            }
-
-            {
-                println!("filling second block");
-                let slice = buf.unfilled_mut();
-                let len = slice.len();
-                assert_eq!(len, block_size as usize);
-                slice.fill(2);
-                buf.advance(len as _);
-                assert_eq!(buf.len(), block_size * 2);
-            }
-
-            {
-                println!("unfilled should be empty again");
-                let slice = buf.unfilled_mut();
-                assert_eq!(slice.len(), 0);
-            }
-
-            agg_len = buf.len();
-        }
-
-        {
-            let buf = buf.read();
-
-            let slice = buf.filled(0..agg_len);
-            assert_eq!(slice.len(), block_size as usize);
-            for b in slice {
-                assert_eq!(*b, 1)
-            }
-
-            let slice = buf.filled(15..agg_len);
-            assert_eq!(slice.len(), block_size as usize - 15);
-            for b in slice {
-                assert_eq!(*b, 1)
-            }
-
-            let slice = buf.filled((agg_len / 2)..agg_len);
-            assert_eq!(slice.len(), block_size as usize);
-            for b in slice {
-                assert_eq!(*b, 2)
-            }
-        }
-    }
-
-    #[test]
-    fn agg_nom_traits() {
-        use nom::{Compare, InputLength, InputTake};
-
-        let buf: AggBuf = Default::default();
-        let hello = "hello";
-        let world = "world";
-
-        {
-            let mut buf = buf.write();
-            buf.grow_if_needed().unwrap();
-
-            {
-                let dst = buf.unfilled_mut();
-                let dst_len = dst.len();
-                let mut src = b"#".repeat(dst_len);
-                src[(dst_len - hello.len())..].copy_from_slice(hello.as_bytes());
-                dst.copy_from_slice(&src);
-
-                buf.advance(dst_len as _);
-            }
-
-            buf.grow_if_needed().unwrap();
-
-            {
-                let dst = buf.unfilled_mut();
-                dst[..world.len()].copy_from_slice(world.as_bytes());
-                buf.advance(world.len() as _);
-            }
-        }
-
-        let block_size = buf.write().block_size;
-        let start = block_size - hello.len() as u32;
-        let end = start + hello.len() as u32 + world.len() as u32;
-        let slice = buf.read().slice(start..end);
-        assert_eq!(slice.input_len(), hello.len() + world.len());
-
-        eprintln!("to_vec + compare owned");
-        assert_eq!(slice.to_string_lossy(), "helloworld");
-
-        assert_eq!(slice.compare(b"that's not it"), nom::CompareResult::Error);
-        assert_eq!(slice.compare(b"hello"), nom::CompareResult::Ok);
-        eprintln!("helloworld nom::Compare");
-        assert_eq!(slice.compare(b"helloworld"), nom::CompareResult::Ok);
-        assert_eq!(
-            slice.compare(b"helloworldwoops"),
-            nom::CompareResult::Incomplete
-        );
-
-        {
-            let hello_slice = slice.take(5);
-            dbg2!(hello_slice.to_string_lossy());
-            assert_eq!(hello_slice.compare(b"hello"), nom::CompareResult::Ok);
-        }
-
-        {
-            let (world_slice, hello_slice) = slice.take_split(5);
-            dbg2!(hello_slice.to_string_lossy());
-            dbg2!(world_slice.to_string_lossy());
-            assert_eq!(hello_slice.compare(b"hello"), nom::CompareResult::Ok);
-            assert_eq!(world_slice.compare(b"world"), nom::CompareResult::Ok);
-        }
-
-        // TODO: test `compare_no_case`
-    }
-
-    #[test]
-    fn agg_nom_sample() {
-        fn parse(i: AggSlice) -> IResult<AggSlice, AggSlice> {
-            nom::bytes::streaming::tag(&b"HTTP/1.1 200 OK"[..])(i)
-        }
-
-        let mut buf: AggBuf = Default::default();
-
-        let input = b"HTTP/1.1 200 OK".repeat(1000);
-        let mut pending = &input[..];
-
-        loop {
-            let slice = buf.read().slice(0..buf.read().len());
-            let (rest, version) = match parse(slice) {
-                Ok(t) => t,
-                Err(e) => {
-                    if e.is_incomplete() {
-                        {
-                            if pending.is_empty() {
-                                println!("ran out of input");
-                                break;
-                            }
-
-                            let mut buf = buf.write();
-                            buf.grow_if_needed().unwrap();
-                            let unfilled = buf.unfilled_mut();
-                            let n = std::cmp::min(unfilled.len(), pending.len());
-                            unfilled[..n].copy_from_slice(&pending[..n]);
-                            pending = &pending[n..];
-                            buf.advance(n as _);
-                            println!("advanced by {n}, {} remaining", pending.len());
-                        }
-
-                        continue;
-                    }
-                    panic!("parsing error: {e}");
-                }
-            };
-            assert_eq!(version.to_string_lossy(), "HTTP/1.1 200 OK");
-
-            buf = buf.split_at(rest);
-        }
-    }
-
-    #[test]
-    fn agg_take_contiguous_at_most_simple() -> eyre::Result<()> {
-        let mut buf = AggBuf::default();
-        let input = "hello world";
-
-        buf.write().put(input)?;
-        let contig = buf.take_contiguous_at_most(input.len() as u32).unwrap();
-        assert_eq!(&contig[..], input.as_bytes());
-
-        Ok(())
-    }
-
-    #[test]
-    fn agg_take_continuous_at_most_spread() -> eyre::Result<()> {
-        let mut buf = AggBuf::default();
-        buf.write().grow_if_needed()?;
-        assert_eq!(buf.read().capacity(), 4096);
-
-        let input = b"hello world";
-        buf.write().put(" ".repeat(4096 - 5))?;
-        buf.write().put(input)?;
-
-        println!("take padding (should still have rest of block)");
-        let contig = buf.take_contiguous_at_most(4096 - 5).unwrap();
-        assert_eq!(contig.len(), 4096 - 5);
-
-        let mut remain = input.len() as u32;
-        let mut needle_remain = &input[..];
-        while remain > 0 {
-            println!("take part of needle (remain {remain})");
-            let contig = buf.take_contiguous_at_most(remain).unwrap();
-            assert_eq!(&contig[..], &needle_remain[..contig.len()]);
-            remain -= contig.len() as u32;
-            needle_remain = &needle_remain[contig.len()..];
-        }
-
-        Ok(())
-    }
-}
diff --git a/src/bufpool/io_chunk.rs b/src/bufpool/io_chunk.rs
deleted file mode 100644
index 6e04593f..00000000
--- a/src/bufpool/io_chunk.rs
+++ /dev/null
@@ -1,200 +0,0 @@
-//! Types for performing vectored I/O.
-
-use tokio_uring::buf::IoBuf;
-
-use super::Buf;
-
-pub enum IoChunk {
-    Static(&'static [u8]),
-    Vec(Vec<u8>),
-    Buf(Buf),
-}
-
-impl From<&'static [u8]> for IoChunk {
-    fn from(slice: &'static [u8]) -> Self {
-        IoChunk::Static(slice)
-    }
-}
-
-impl From<Vec<u8>> for IoChunk {
-    fn from(vec: Vec<u8>) -> Self {
-        IoChunk::Vec(vec)
-    }
-}
-
-impl From<Buf> for IoChunk {
-    fn from(buf: Buf) -> Self {
-        IoChunk::Buf(buf)
-    }
-}
-
-impl AsRef<[u8]> for IoChunk {
-    fn as_ref(&self) -> &[u8] {
-        match self {
-            IoChunk::Static(slice) => slice,
-            IoChunk::Vec(vec) => vec.as_ref(),
-            IoChunk::Buf(buf) => buf.as_ref(),
-        }
-    }
-}
-
-impl IoChunk {
-    #[inline(always)]
-    pub fn as_io_buf(&self) -> &dyn IoBuf {
-        match self {
-            IoChunk::Static(slice) => slice,
-            IoChunk::Vec(vec) => vec,
-            IoChunk::Buf(buf) => buf,
-        }
-    }
-}
-
-unsafe impl IoBuf for IoChunk {
-    #[inline(always)]
-    fn stable_ptr(&self) -> *const u8 {
-        IoBuf::stable_ptr(self.as_io_buf())
-    }
-
-    fn bytes_init(&self) -> usize {
-        IoBuf::bytes_init(self.as_io_buf())
-    }
-
-    fn bytes_total(&self) -> usize {
-        IoBuf::bytes_total(self.as_io_buf())
-    }
-}
-
-pub trait IoChunkable {
-    /// Returns the total length of all chunks
-    fn len(&self) -> usize;
-
-    /// Returns true if the chunkable is empty
-    fn is_empty(&self) -> bool {
-        self.len() == 0
-    }
-
-    /// Returns next chunk that can be written
-    fn next_chunk(&self, offset: u32) -> Option<IoChunk>;
-
-    /// Append this chunkable to the given list
-    fn append_to(&self, list: &mut Vec<IoChunk>) {
-        let mut offset = 0;
-        while let Some(chunk) = self.next_chunk(offset) {
-            offset += chunk.len() as u32;
-            list.push(chunk);
-        }
-    }
-}
-
-impl IoChunk {
-    #[inline(always)]
-    pub fn len(&self) -> usize {
-        match self {
-            IoChunk::Static(slice) => slice.len(),
-            IoChunk::Vec(vec) => vec.len(),
-            IoChunk::Buf(buf) => buf.len(),
-        }
-    }
-
-    #[inline(always)]
-    pub fn is_empty(&self) -> bool {
-        self.len() == 0
-    }
-}
-
-impl IoChunkable for &'static [u8] {
-    fn len(&self) -> usize {
-        (self as &'static [u8]).len()
-    }
-
-    fn next_chunk(&self, offset: u32) -> Option<IoChunk> {
-        if offset == 0 {
-            Some(IoChunk::Static(self))
-        } else {
-            None
-        }
-    }
-}
-
-impl IoChunkable for &'static str {
-    fn len(&self) -> usize {
-        (self as &'static str).len()
-    }
-
-    fn next_chunk(&self, offset: u32) -> Option<IoChunk> {
-        IoChunkable::next_chunk(&self.as_bytes(), offset)
-    }
-}
-
-impl IoChunkable for Vec<u8> {
-    fn len(&self) -> usize {
-        self.len()
-    }
-
-    fn next_chunk(&self, offset: u32) -> Option<IoChunk> {
-        if offset == 0 {
-            // FIXME: ooh that's bad, shouldn't need to clone here
-            Some(IoChunk::Vec(self.clone()))
-        } else {
-            None
-        }
-    }
-}
-
-impl IoChunkable for Buf {
-    fn len(&self) -> usize {
-        self.len()
-    }
-
-    fn next_chunk(&self, offset: u32) -> Option<IoChunk> {
-        if offset == 0 {
-            Some(IoChunk::Buf(self.clone()))
-        } else {
-            None
-        }
-    }
-}
-
-#[derive(Default)]
-pub struct IoChunkList {
-    chunks: Vec<IoChunk>,
-}
-
-impl IoChunkList {
-    /// Add a chunkable to the list (this may result in multiple chunks)
-    pub fn push(&mut self, chunkable: impl IoChunkable) {
-        chunkable.append_to(&mut self.chunks);
-    }
-
-    /// Add a single chunk to the list
-    pub fn push_chunk(&mut self, chunk: IoChunk) {
-        self.chunks.push(chunk);
-    }
-
-    /// Returns total length
-    pub fn len(&self) -> usize {
-        self.chunks.iter().map(|c| c.len()).sum()
-    }
-
-    pub fn num_chunks(&self) -> usize {
-        self.chunks.len()
-    }
-
-    pub fn is_empty(&self) -> bool {
-        self.chunks.is_empty() || self.len() == 0
-    }
-
-    pub fn clear(&mut self) {
-        self.chunks.clear();
-    }
-
-    pub fn into_vec(self) -> Vec<IoChunk> {
-        self.chunks
-    }
-}
-
-impl From<Vec<IoChunk>> for IoChunkList {
-    fn from(chunks: Vec<IoChunk>) -> Self {
-        Self { chunks }
-    }
-}
diff --git a/src/h1/body.rs b/src/h1/body.rs
index de7c0018..03e96e7e 100644
--- a/src/h1/body.rs
+++ b/src/h1/body.rs
@@ -1,18 +1,16 @@
 use std::{fmt, rc::Rc};
 
-use pretty_hex::PrettyHex;
-use tracing::{debug, trace};
+use tracing::debug;
 
 use crate::{
-    bufpool::{AggBuf, IoChunkList},
     util::{read_and_parse, write_all_list},
-    Body, BodyChunk, BodyErrorReason, IoChunk, ReadOwned, WriteOwned,
+    Body, BodyChunk, BodyErrorReason, IoChunk, IoChunkList, ReadOwned, RollMut, WriteOwned,
 };
 
 /// An HTTP/1.1 body, either chunked or content-length.
 pub(crate) struct H1Body<T> {
     transport: Rc<T>,
-    buf: Option<AggBuf>,
+    buf: Option<RollMut>,
     state: Decoder,
 }
 
@@ -52,7 +50,7 @@ impl<T> fmt::Debug for H1Body<T> {
 }
 
 impl<T: ReadOwned> H1Body<T> {
-    pub(crate) fn new(transport: Rc<T>, buf: AggBuf, kind: H1BodyKind) -> Self {
+    pub(crate) fn new(transport: Rc<T>, buf: RollMut, kind: H1BodyKind) -> Self {
         let state = match kind {
             H1BodyKind::Chunked => Decoder::Chunked(ChunkedDecoder::ReadingChunkHeader),
             H1BodyKind::ContentLength(len) => {
@@ -68,7 +66,7 @@ impl<T: ReadOwned> H1Body<T> {
 
     /// Returns the inner buffer, but only if the body has been
     /// fully read.
-    pub(crate) fn into_buf(self) -> Option<AggBuf> {
+    pub(crate) fn into_buf(self) -> Option<RollMut> {
         if !self.eof() {
             return None;
         }
@@ -114,7 +112,7 @@ impl<T: ReadOwned> Body for H1Body<T> {
 impl ContentLengthDecoder {
     async fn next_chunk(
         &mut self,
-        buf_slot: &mut Option<AggBuf>,
+        buf_slot: &mut Option<RollMut>,
         transport: &impl ReadOwned,
     ) -> eyre::Result<BodyChunk> {
         let remain = self.len - self.read;
@@ -128,28 +126,20 @@ impl ContentLengthDecoder {
             .take()
             .ok_or_else(|| BodyErrorReason::CalledNextChunkAfterError.as_err())?;
 
-        if buf.read().len() == 0 {
-            buf.write().grow_if_needed()?;
-            let mut slice = buf.write_slice();
+        if buf.is_empty() {
+            buf.reserve()?;
+
             let res;
-            (res, slice) = transport.read(slice).await;
-            buf = slice.into_inner();
+            (res, buf) = buf.read_into(usize::MAX, transport).await;
             res.map_err(|e| BodyErrorReason::ErrorWhileReadingChunkData.with_cx(e))?;
         }
 
-        // FIXME: integer demotion
-        let chunk = buf.take_contiguous_at_most(remain as u32);
-        match chunk {
-            Some(chunk) => {
-                trace!("got chunk {}", chunk.hex_dump());
-                self.read += chunk.len() as u64;
-                buf_slot.replace(buf);
-                Ok(BodyChunk::Chunk(chunk.into()))
-            }
-            None => Err(BodyErrorReason::ClosedWhileReadingContentLength
-                .as_err()
-                .into()),
-        }
+        let chunk = buf
+            .take_at_most(remain as usize)
+            .ok_or_else(|| BodyErrorReason::ClosedWhileReadingContentLength.as_err())?;
+        self.read += chunk.len() as u64;
+        buf_slot.replace(buf);
+        Ok(BodyChunk::Chunk(chunk.into()))
     }
 
     fn eof(&self) -> bool {
@@ -160,7 +150,7 @@ impl ContentLengthDecoder {
 impl ChunkedDecoder {
     async fn next_chunk(
         &mut self,
-        buf_slot: &mut Option<AggBuf>,
+        buf_slot: &mut Option<RollMut>,
         transport: &impl ReadOwned,
     ) -> eyre::Result<BodyChunk> {
         loop {
@@ -217,17 +207,15 @@ impl ChunkedDecoder {
                     continue;
                 }
 
-                if buf.read().len() == 0 {
-                    buf.write().grow_if_needed()?;
-                    let mut slice = buf.write_slice();
+                if buf.is_empty() {
+                    buf.reserve()?;
+
                     let res;
-                    (res, slice) = transport.read(slice).await;
-                    buf = slice.into_inner();
+                    (res, buf) = buf.read_into(*remain as usize, transport).await;
                     res.map_err(|e| BodyErrorReason::ErrorWhileReadingChunkData.with_cx(e))?;
                 }
 
-                // FIXME: integer demotion
-                let chunk = buf.take_contiguous_at_most(*remain as u32);
+                let chunk = buf.take_at_most(*remain as usize);
                 match chunk {
                     Some(chunk) => {
                         *remain -= chunk.len() as u64;
@@ -284,7 +272,7 @@ pub(crate) async fn write_h1_body_chunk(
         BodyWriteMode::Chunked => {
             let mut list = IoChunkList::default();
             list.push(format!("{:x}\r\n", chunk.len()).into_bytes());
-            list.push_chunk(chunk);
+            list.push(chunk);
             list.push("\r\n");
 
             let list = write_all_list(transport, list).await?;
diff --git a/src/h1/client.rs b/src/h1/client.rs
index e3552f60..662dce02 100644
--- a/src/h1/client.rs
+++ b/src/h1/client.rs
@@ -4,11 +4,11 @@ use eyre::Context;
 use tracing::debug;
 
 use crate::{
-    bufpool::{AggBuf, IoChunkList},
+    buffet::IoChunkList,
     io::ReadWriteOwned,
     types::Request,
     util::{read_and_parse, write_all_list},
-    Body, Response,
+    Body, Response, RollMut,
 };
 
 use super::{
@@ -77,7 +77,7 @@ where
     let recv_res_fut = {
         let transport = transport.clone();
         async move {
-            let buf = AggBuf::default();
+            let buf = RollMut::alloc()?;
             let (buf, res) = read_and_parse(
                 super::parse::response,
                 transport.as_ref(),
diff --git a/src/h1/encode.rs b/src/h1/encode.rs
index 78e0f795..e103b527 100644
--- a/src/h1/encode.rs
+++ b/src/h1/encode.rs
@@ -1,5 +1,5 @@
 use crate::{
-    bufpool::IoChunkList,
+    buffet::IoChunkList,
     types::{Headers, Request, Response},
 };
 
diff --git a/src/h1/parse.rs b/src/h1/parse.rs
index 501df31d..e4cbda7d 100644
--- a/src/h1/parse.rs
+++ b/src/h1/parse.rs
@@ -11,24 +11,24 @@ use nom::{
 };
 
 use crate::{
-    bufpool::AggSlice,
     types::{Header, Headers, Request, Response},
+    Roll,
 };
 
 const CRLF: &[u8] = b"\r\n";
 
 /// Parses a chunked transfer coding chunk size (hex text followed by CRLF)
-pub fn chunk_size(i: AggSlice) -> IResult<AggSlice, u64> {
+pub fn chunk_size(i: Roll) -> IResult<Roll, u64> {
     terminated(u64_text_hex, tag(CRLF))(i)
 }
 
-pub fn crlf(i: AggSlice) -> IResult<AggSlice, ()> {
+pub fn crlf(i: Roll) -> IResult<Roll, ()> {
     let (i, _) = tag(CRLF)(i)?;
     Ok((i, ()))
 }
 
 // Looks like `GET /path HTTP/1.1\r\n`, then headers
-pub fn request(i: AggSlice) -> IResult<AggSlice, Request> {
+pub fn request(i: Roll) -> IResult<Roll, Request> {
     let (i, method) = take_until_and_consume(b" ")(i)?;
     let (i, path) = take_until_and_consume(b" ")(i)?;
     let (i, version) = terminated(http_version, tag(CRLF))(i)?;
@@ -44,7 +44,7 @@ pub fn request(i: AggSlice) -> IResult<AggSlice, Request> {
 }
 
 // Looks like `HTTP/1.1 200 OK\r\n` or `HTTP/1.1 404 Not Found\r\n`, then headers
-pub fn response(i: AggSlice) -> IResult<AggSlice, Response> {
+pub fn response(i: Roll) -> IResult<Roll, Response> {
     let (i, version) = terminated(http_version, space1)(i)?;
     let (i, code) = terminated(u16_text, space1)(i)?;
     let (i, reason) = terminated(take_until(CRLF), tag(CRLF))(i)?;
@@ -60,26 +60,26 @@ pub fn response(i: AggSlice) -> IResult<AggSlice, Response> {
 }
 
 /// Parses text as a u16
-fn u16_text(i: AggSlice) -> IResult<AggSlice, u16> {
+fn u16_text(i: Roll) -> IResult<Roll, u16> {
     // TODO: limit how many digits we read
     let f = take_while1(nom::character::is_digit);
     // FIXME: this is inefficient (calling `to_vec` just to parse)
-    let f = nom::combinator::map_res(f, |s: AggSlice| String::from_utf8(s.to_vec()));
+    let f = nom::combinator::map_res(f, |s: Roll| String::from_utf8(s.to_vec()));
     let mut f = nom::combinator::map_res(f, |s| s.parse());
     f(i)
 }
 
 /// Parses text as a hex u64
-fn u64_text_hex(i: AggSlice) -> IResult<AggSlice, u64> {
+fn u64_text_hex(i: Roll) -> IResult<Roll, u64> {
     // TODO: limit how many digits we read
     let f = take_while1(nom::character::is_hex_digit);
     // FIXME: this is inefficient (calling `to_vec` just to parse)
-    let f = nom::combinator::map_res(f, |s: AggSlice| String::from_utf8(s.to_vec()));
+    let f = nom::combinator::map_res(f, |s: Roll| String::from_utf8(s.to_vec()));
     let mut f = nom::combinator::map_res(f, |s| u64::from_str_radix(&s, 16));
     f(i)
 }
 
-pub fn http_version(i: AggSlice) -> IResult<AggSlice, u8> {
+pub fn http_version(i: Roll) -> IResult<Roll, u8> {
     let (i, _) = tag(&b"HTTP/1."[..])(i)?;
     let (i, version) = take(1usize)(i)?;
     let version = match version.iter().next().unwrap() {
@@ -97,7 +97,7 @@ pub fn http_version(i: AggSlice) -> IResult<AggSlice, u8> {
     Ok((i, version))
 }
 
-pub fn headers_and_crlf(mut i: AggSlice) -> IResult<AggSlice, Headers> {
+pub fn headers_and_crlf(mut i: Roll) -> IResult<Roll, Headers> {
     let mut headers = Headers::default();
     loop {
         if let (i, Some(_)) = opt(tag(CRLF))(i.clone())? {
@@ -112,7 +112,7 @@ pub fn headers_and_crlf(mut i: AggSlice) -> IResult<AggSlice, Headers> {
 }
 
 /// Parse a single header line
-fn header(i: AggSlice) -> IResult<AggSlice, (AggSlice, AggSlice)> {
+fn header(i: Roll) -> IResult<Roll, (Roll, Roll)> {
     let (i, name) = take_until_and_consume(b":")(i)?;
     let (i, value) = preceded(space1, take_until_and_consume(CRLF))(i)?;
 
@@ -120,14 +120,12 @@ fn header(i: AggSlice) -> IResult<AggSlice, (AggSlice, AggSlice)> {
 }
 
 /// Parse at least one SP character
-fn space1(i: AggSlice) -> IResult<AggSlice, ()> {
+fn space1(i: Roll) -> IResult<Roll, ()> {
     let (i, _) = take_while1(|c| c == b' ')(i)?;
     Ok((i, ()))
 }
 
 /// Parse until the given tag, then skip the tag
-fn take_until_and_consume(
-    needle: &[u8],
-) -> impl FnMut(AggSlice) -> IResult<AggSlice, AggSlice> + '_ {
+fn take_until_and_consume(needle: &[u8]) -> impl FnMut(Roll) -> IResult<Roll, Roll> + '_ {
     terminated(take_until(needle), tag(needle))
 }
diff --git a/src/h1/server.rs b/src/h1/server.rs
index 9a2857aa..da87f419 100644
--- a/src/h1/server.rs
+++ b/src/h1/server.rs
@@ -4,10 +4,10 @@ use eyre::Context;
 use tracing::debug;
 
 use crate::{
-    bufpool::{AggBuf, IoChunkList},
+    buffet::IoChunkList,
     io::WriteOwned,
     util::{read_and_parse, write_all_list, SemanticError},
-    Body, Headers, IoChunk, IoChunkable, ReadWriteOwned, Request, Response,
+    Body, Headers, IoChunk, ReadWriteOwned, Request, Response, RollMut,
 };
 
 use super::{
@@ -17,13 +17,13 @@ use super::{
 
 pub struct ServerConf {
     /// Max length of the request line + HTTP headers
-    pub max_http_header_len: u32,
+    pub max_http_header_len: usize,
 
     /// Max length of a single header record, e.g. `user-agent: foobar`
-    pub max_header_record_len: u32,
+    pub max_header_record_len: usize,
 
     /// Max number of header records
-    pub max_header_records: u32,
+    pub max_header_records: usize,
 }
 
 impl Default for ServerConf {
@@ -57,7 +57,7 @@ pub enum ServeOutcome {
 pub async fn serve(
     transport: impl ReadWriteOwned,
     conf: Rc<ServerConf>,
-    mut client_buf: AggBuf,
+    mut client_buf: RollMut,
     driver: impl ServerDriver,
 ) -> eyre::Result<ServeOutcome> {
     let transport = Rc::new(transport);
@@ -221,20 +221,6 @@ impl<T> Responder<T, ExpectResponseBody>
 where
     T: WriteOwned,
 {
-    /// Send multiple response body chunks (as many as there are in the chunkable).
-    pub async fn write_chunkable(
-        self,
-        chunkable: impl IoChunkable,
-    ) -> eyre::Result<Responder<T, ExpectResponseBody>> {
-        let mut this = self;
-        let mut offset = 0;
-        while let Some(chunk) = chunkable.next_chunk(offset) {
-            offset += chunk.len() as u32;
-            this = this.write_chunk(chunk).await?;
-        }
-        Ok(this)
-    }
-
     /// Send a response body chunk. Errors out if sending more than the
     /// announced content-length.
     pub async fn write_chunk(
diff --git a/src/lib.rs b/src/lib.rs
index 52ca2aae..c8a399b9 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -6,8 +6,8 @@
 
 mod util;
 
-mod bufpool;
-pub use bufpool::*;
+mod buffet;
+pub use buffet::*;
 
 mod io;
 pub use io::*;
diff --git a/src/types/headers.rs b/src/types/headers.rs
index 7d1d533f..d2d9ed7f 100644
--- a/src/types/headers.rs
+++ b/src/types/headers.rs
@@ -2,7 +2,7 @@
 
 use smallvec::SmallVec;
 
-use crate::bufpool::AggSlice;
+use crate::Roll;
 
 const HEADERS_SMALLVEC_CAPACITY: usize = 32;
 
@@ -49,16 +49,9 @@ impl Headers {
     /// Returns the content-length header
     pub fn content_length(&self) -> Option<u64> {
         for h in self {
-            if h.name.eq_ignore_ascii_case("content-length") {
-                // FIXME: this is really wasteful. maybe there's something to be
-                // done where: 1) it's probably contiguous aynway, so just use that,
-                // or: 2) if it's not, just copy it to a stack-allocated slice.
-                //
-                // this could be a method of AggSlice that lends a `&[u8]`
-                // to a closure and errors out if it's too big. it could take
-                // const generics.
-                let value = h.value.to_vec();
-                if let Ok(s) = std::str::from_utf8(&value[..]) {
+            if h.name.eq_ignore_ascii_case(b"content-length") {
+                // TODO: introduce RollStr
+                if let Ok(s) = std::str::from_utf8(&h.value[..]) {
                     if let Ok(l) = s.parse() {
                         return Some(l);
                     }
@@ -88,6 +81,6 @@ impl IntoIterator for Headers {
 }
 
 pub struct Header {
-    pub name: AggSlice,
-    pub value: AggSlice,
+    pub name: Roll,
+    pub value: Roll,
 }
diff --git a/src/types/mod.rs b/src/types/mod.rs
index 47608981..0c38726b 100644
--- a/src/types/mod.rs
+++ b/src/types/mod.rs
@@ -4,14 +4,14 @@ use std::fmt::{self, Debug};
 pub use headers::*;
 use tracing::debug;
 
-use crate::{bufpool::AggSlice, IoChunk};
+use crate::{IoChunk, Roll};
 
 /// An HTTP request
 pub struct Request {
-    pub method: AggSlice,
+    pub method: Roll,
 
     /// Requested entity
-    pub path: AggSlice,
+    pub path: Roll,
 
     /// The 'b' in 'HTTP/1.b'
     pub version: u8,
@@ -38,7 +38,7 @@ pub struct Response {
     pub code: u16,
 
     /// Human-readable string following the status code
-    pub reason: AggSlice,
+    pub reason: Roll,
 
     /// Response headers
     pub headers: Headers,
diff --git a/src/util.rs b/src/util.rs
index bbc36f04..f5b7919f 100644
--- a/src/util.rs
+++ b/src/util.rs
@@ -1,49 +1,54 @@
 use eyre::Context;
 use nom::IResult;
+use pretty_hex::PrettyHex;
 use tracing::debug;
 
 use crate::{
-    bufpool::{AggBuf, AggSlice, IoChunkList},
+    buffet::IoChunkList,
     io::{ReadOwned, WriteOwned},
+    Roll, RollMut,
 };
 
 /// Returns `None` on EOF, error if partially parsed message.
 pub(crate) async fn read_and_parse<Parser, Output>(
     parser: Parser,
     stream: &impl ReadOwned,
-    mut buf: AggBuf,
-    max_len: u32,
-) -> eyre::Result<Option<(AggBuf, Output)>>
+    mut buf: RollMut,
+    max_len: usize,
+) -> eyre::Result<Option<(RollMut, Output)>>
 where
-    Parser: Fn(AggSlice) -> IResult<AggSlice, Output>,
+    Parser: Fn(Roll) -> IResult<Roll, Output>,
 {
     loop {
-        debug!("reading+parsing ({} bytes so far)", buf.read().len());
-        let slice = buf.read().read_slice();
+        debug!("reading+parsing ({} bytes so far)", buf.len());
+        let filled = buf.filled();
 
-        let (rest, req) = match parser(slice) {
-            Ok(t) => t,
+        match parser(filled) {
+            Ok((rest, output)) => {
+                buf.keep(rest);
+                return Ok(Some((buf, output)));
+            }
             Err(err) => {
                 if err.is_incomplete() {
-                    debug!(
-                        "incomplete request, need more data. start of buffer: {:?}",
-                        buf.read()
-                            .slice(0..std::cmp::min(buf.read().len(), 128))
-                            .to_string_lossy()
-                    );
+                    {
+                        let filled = buf.borrow_filled();
+                        debug!(
+                            "incomplete request, need more data. start of buffer: {:?}",
+                            filled[0..std::cmp::min(filled.len(), 128)].hex_dump()
+                        );
+                    }
 
-                    if buf.read().len() >= max_len {
+                    let res;
+                    let read_limit = max_len - buf.len();
+                    if buf.len() >= max_len {
                         return Err(SemanticError::BufferLimitReachedWhileParsing.into());
                     }
 
-                    buf.write().grow_if_needed()?;
+                    (res, buf) = buf.read_into(read_limit, stream).await;
 
-                    let (res, buf_s) = stream.read(buf.write_slice()).await;
                     let n = res.wrap_err("reading request headers from downstream")?;
-                    buf = buf_s.into_inner();
-
                     if n == 0 {
-                        if !buf.read().is_empty() {
+                        if !buf.is_empty() {
                             return Err(eyre::eyre!("unexpected EOF"));
                         } else {
                             return Ok(None);
@@ -60,8 +65,6 @@ where
                 }
             }
         };
-
-        return Ok(Some((buf.split_at(rest), req)));
     }
 }
 


Test patch:
diff --git a/tests/integration_test.rs b/tests/integration_test.rs
index 288e639b..fd3c0fa0 100644
--- a/tests/integration_test.rs
+++ b/tests/integration_test.rs
@@ -6,7 +6,7 @@ mod helpers;
 
 use bytes::BytesMut;
 use hring::{
-    h1, AggBuf, Body, BodyChunk, ChanRead, ChanWrite, Headers, ReadWritePair, Request, Response,
+    h1, Body, BodyChunk, ChanRead, ChanWrite, Headers, ReadWritePair, Request, Response, RollMut,
     WriteOwned,
 };
 use httparse::{Status, EMPTY_HEADER};
@@ -56,11 +56,10 @@ fn serve_api() {
                 _req_body: &mut impl Body,
                 res: h1::Responder<T, h1::ExpectResponseHeaders>,
             ) -> eyre::Result<h1::Responder<T, h1::ResponseDone>> {
-                let mut buf = AggBuf::default();
+                let mut buf = RollMut::alloc()?;
 
-                buf.write().put(b"Continue")?;
-                let reason = buf.read().read_slice();
-                buf = buf.split();
+                buf.put(b"Continue")?;
+                let reason = buf.take_all();
 
                 let res = res
                     .write_interim_response(Response {
@@ -71,9 +70,8 @@ fn serve_api() {
                     })
                     .await?;
 
-                buf.write().put(b"OK")?;
-                let reason = buf.read().read_slice();
-                buf = buf.split();
+                buf.put(b"OK")?;
+                let reason = buf.take_all();
 
                 _ = buf;
 
@@ -95,7 +93,7 @@ fn serve_api() {
         let (tx, read) = ChanRead::new();
         let (mut rx, write) = ChanWrite::new();
         let transport = ReadWritePair(read, write);
-        let client_buf = AggBuf::default();
+        let client_buf = RollMut::alloc()?;
         let driver = TestDriver {};
         let serve_fut = tokio_uring::spawn(h1::serve(transport, conf, client_buf, driver));
 
@@ -145,15 +143,13 @@ fn request_api() {
         let (mut rx, write) = ChanWrite::new();
         let transport = ReadWritePair(read, write);
 
-        let mut buf = AggBuf::default();
+        let mut buf = RollMut::alloc()?;
 
-        buf.write().put(b"GET")?;
-        let method = buf.read().read_slice();
-        buf = buf.split();
+        buf.put(b"GET")?;
+        let method = buf.take_all();
 
-        buf.write().put(b"/")?;
-        let path = buf.read().read_slice();
-        buf = buf.split();
+        buf.put(b"/")?;
+        let path = buf.take_all();
 
         _ = buf;
 
diff --git a/tests/proxy.rs b/tests/proxy.rs
index b3754047..43aa5ec2 100644
--- a/tests/proxy.rs
+++ b/tests/proxy.rs
@@ -3,7 +3,7 @@
 #![feature(type_alias_impl_trait)]
 #![feature(async_fn_in_trait)]
 
-use hring::{h1, Body, BodyChunk, Response, WriteOwned};
+use hring::{h1, Body, BodyChunk, Response, RollMut, WriteOwned};
 use std::{cell::RefCell, future::Future, net::SocketAddr, rc::Rc};
 use tracing::debug;
 
@@ -135,7 +135,7 @@ pub async fn start(
                             upstream_addr,
                             pool,
                         };
-                        h1::serve(transport, conf, Default::default(), driver)
+                        h1::serve(transport, conf, RollMut::alloc().unwrap(), driver)
                             .await
                             .unwrap();
                         debug!("Done serving h1 connection");