WIP

pageserver/src/tenant/ephemeral_file/sandwich.rs

@@ -1,4 +1,6 @@
-use crate::{page_cache::PAGE_SZ, virtual_file::VirtualFile};
+use crate::{
+    page_cache::PAGE_SZ, tenant::ephemeral_file::zero_padded_buffer, virtual_file::VirtualFile,
+};
 
 use super::size_tracker_borrowed;
 
@@ -34,9 +36,9 @@ impl Sandwich {
     }
 
     pub(crate) async fn read_blk(&self, blknum: u32) -> Result<ReadResult, std::io::Error> {
-        let buffered_offset = self.sandwich.buffered_offset();
         let flushed_offset = self.sandwich.flushed_offset();
-        assert!(buffered_offset >= flushed_offset);
+        let buffer = self.sandwich.buffered_writer.inspect_buffer();
+        let buffered_offset = flushed_offset + buffer.pending();
         let read_offset = (blknum as u64) * (PAGE_SZ as u64);
 
         assert_eq!(
@@ -73,9 +75,8 @@ impl Sandwich {
                 .checked_sub(flushed_offset)
                 .expect("would have taken `if` branch instead of this one");
             let read_offset_in_buffer = usize::try_from(read_offset_in_buffer).unwrap();
-            let page = self
-                .sandwich
-                .buffered_zero_padded_page_at(read_offset_in_buffer);
+            let zero_padded_buffer: &zero_padded_buffer::Buf<TAIL_SZ> = buffer.raw_iobuf();
+            let page = zero_padded_buffer[read_offset_in_buffer..(read_offset_in_buffer + PAGE_SZ)];
             Ok(ReadResult::ServedFromZeroPaddedMutableTail {
                 buffer: page
                     .try_into()

pageserver/src/tenant/ephemeral_file/size_tracker_borrowed.rs

@@ -3,11 +3,11 @@ use crate::virtual_file::{owned_buffers_io, VirtualFile};
 use super::zero_padded_buffer;
 
 pub struct Writer<const TAIL_SZ: usize> {
-    buffered_writer: owned_buffers_io::write::BufferedWriter<
+    pub buffered_writer: owned_buffers_io::write::BufferedWriter<
         zero_padded_buffer::Buf<TAIL_SZ>,
         owned_buffers_io::util::size_tracking_writer::Writer<VirtualFile>,
     >,
-    bytes_amount: u64,
+    pub bytes_amount: u64,
 }
 
 impl<const TAIL_SZ: usize> Writer<TAIL_SZ> {

pageserver/src/tenant/ephemeral_file/zero_padded_buffer.rs

@@ -39,7 +39,7 @@ unsafe impl<const N: usize> tokio_epoll_uring::IoBuf for Buf<N> {
     }
 
     fn bytes_init(&self) -> usize {
-        self.written
+        N
     }
 
     fn bytes_total(&self) -> usize {

pageserver/src/virtual_file/owned_buffers_io/write.rs

@@ -10,61 +10,60 @@ pub trait OwnedAsyncWriter {
 }
 
 struct Buffer<B> {
-    len: usize,
-    buf: B,
+    pending: usize,
+    iobuf: B,
 }
 
 impl<B> Buffer<B>
 where
     B: tokio_epoll_uring::IoBufMut,
 {
-    pub fn new(buf: B) -> Self {
-        Self { len: 0, buf }
+    pub fn new(iobuf: B) -> Self {
+        Self { pending: 0, iobuf }
     }
 
     #[inline(always)]
     fn invariants(&self) {
-        debug_assert!(self.len <= self.buf.bytes_total());
-        debug_assert!(self.len <= self.buf.bytes_init());
+        debug_assert!(self.pending <= self.iobuf.bytes_total());
+        debug_assert!(self.pending <= self.iobuf.bytes_init());
     }
 
     #[inline(always)]
     pub fn cap(&self) -> usize {
-        self.buf.bytes_total()
+        self.iobuf.bytes_total()
     }
 
     #[inline(always)]
-    pub fn len(&self) -> usize {
-        self.len
+    pub fn pending(&self) -> usize {
+        self.pending
     }
 
-    pub fn clear(&mut self) {
-        self.invariants();
-        self.len = 0;
+    pub fn flush(self) -> Slice<B> {
         self.invariants();
+        self.iobuf.slice(0..self.pending)
     }
 
     pub fn extend_from_slice(&mut self, other: &[u8]) {
         self.invariants();
-        let remaining = self.buf.bytes_total() - self.len;
+        let remaining = self.iobuf.bytes_total() - self.pending;
         if other.len() > remaining {
-            panic!("extend_from_slice() would extend beyond buffer capacity; cap={} self.len={} other.len()={}", self.buf.bytes_total(), self.len, other.len());
+            panic!("extend_from_slice() would extend beyond buffer capacity; cap={} self.len={} other.len()={}", self.iobuf.bytes_total(), self.pending, other.len());
         }
         // SAFETY: we did bounds-checking above; non-overlapping is guaranteed by Rust borrowing
         // (self is borrowed mut, so `other` can't reference self or anything in it).
         unsafe {
-            self.buf
+            self.iobuf
                 .stable_mut_ptr()
-                .add(self.len)
+                .add(self.pending)
                 .copy_from_nonoverlapping(other.as_ptr(), other.len());
-            self.buf.set_init(self.len + other.len());
+            self.iobuf.set_init(self.pending + other.len());
         }
-        self.len += other.len();
+        self.pending += other.len();
         self.invariants();
     }
 
-    pub fn as_inner(&self) -> &B {
-        &self.buf
+    pub fn raw_iobuf(&self) -> &B {
+        &self.iobuf
     }
 }
 
@@ -111,7 +110,7 @@ where
 
     /// Panics if used after any of the write paths returned an error
     pub fn inspect_buffer(&self) -> &Buffer<B> {
-        self.buf.as_ref().expect("must not use after an error")
+        self.buf()
     }
 
     pub async fn flush_and_into_inner(mut self) -> std::io::Result<W> {
@@ -141,7 +140,7 @@ where
                 self.buf
                     .as_ref()
                     .expect("must not use after an error")
-                    .len(),
+                    .pending(),
                 0
             );
             let (nwritten, chunk) = self.writer.write_all(chunk).await?;
@@ -153,13 +152,13 @@ where
         let mut slice = &chunk[..];
         while !slice.is_empty() {
             let buf = self.buf.as_mut().expect("must not use after an error");
-            let need = buf.cap() - buf.len();
+            let need = buf.cap() - buf.pending();
             let have = slice.len();
             let n = std::cmp::min(need, have);
             buf.extend_from_slice(&slice[..n]);
             slice = &slice[n..];
-            if buf.len() >= buf.cap() {
-                assert_eq!(buf.len(), buf.cap());
+            if buf.pending() >= buf.cap() {
+                assert_eq!(buf.pending(), buf.cap());
                 self.flush().await?;
             }
         }
@@ -176,13 +175,13 @@ where
         let chunk_len = chunk.len();
         while !chunk.is_empty() {
             let buf = self.buf.as_mut().expect("must not use after an error");
-            let need = buf.cap() - buf.len();
+            let need = buf.cap() - buf.pending();
             let have = chunk.len();
             let n = std::cmp::min(need, have);
             buf.extend_from_slice(&chunk[..n]);
             chunk = &chunk[n..];
-            if buf.len() >= buf.cap() {
-                assert_eq!(buf.len(), buf.cap());
+            if buf.pending() >= buf.cap() {
+                assert_eq!(buf.pending(), buf.cap());
                 self.flush().await?;
             }
         }
@@ -191,12 +190,8 @@ where
 
     async fn flush(&mut self) -> std::io::Result<()> {
         let buf = self.buf.take().expect("must not use after an error");
-        if buf.len() == 0 {
-            self.buf = Some(buf);
-            return std::io::Result::Ok(());
-        }
-        let buf_len = buf.len();
-        let (nwritten, mut io_buf) = self.writer.write_all(buf.buf.slice(0..buf_len)).await?;
+        let buf_len = buf.pending();
+        let (nwritten, mut io_buf) = self.writer.write_all(buf.flush()).await?;
         assert_eq!(nwritten, buf_len);
         self.buf = Some(Buffer::new(io_buf));
         Ok(())