From 6138941e6c1225becf95e7457e3e979bc6b26b83 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Tue, 23 Apr 2024 10:53:07 +0000 Subject: [PATCH] WIP --- .../src/tenant/ephemeral_file/sandwich.rs | 13 ++-- .../ephemeral_file/size_tracker_borrowed.rs | 4 +- .../ephemeral_file/zero_padded_buffer.rs | 2 +- .../virtual_file/owned_buffers_io/write.rs | 63 +++++++++---------- 4 files changed, 39 insertions(+), 43 deletions(-) diff --git a/pageserver/src/tenant/ephemeral_file/sandwich.rs b/pageserver/src/tenant/ephemeral_file/sandwich.rs index de8e7f571b..318aff589b 100644 --- a/pageserver/src/tenant/ephemeral_file/sandwich.rs +++ b/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 { - 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 = 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() diff --git a/pageserver/src/tenant/ephemeral_file/size_tracker_borrowed.rs b/pageserver/src/tenant/ephemeral_file/size_tracker_borrowed.rs index eac8002690..77de0f66f7 100644 --- a/pageserver/src/tenant/ephemeral_file/size_tracker_borrowed.rs +++ b/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 { - buffered_writer: owned_buffers_io::write::BufferedWriter< + pub buffered_writer: owned_buffers_io::write::BufferedWriter< zero_padded_buffer::Buf, owned_buffers_io::util::size_tracking_writer::Writer, >, - bytes_amount: u64, + pub bytes_amount: u64, } impl Writer { diff --git a/pageserver/src/tenant/ephemeral_file/zero_padded_buffer.rs b/pageserver/src/tenant/ephemeral_file/zero_padded_buffer.rs index 36b66fa65b..a541529a42 100644 --- a/pageserver/src/tenant/ephemeral_file/zero_padded_buffer.rs +++ b/pageserver/src/tenant/ephemeral_file/zero_padded_buffer.rs @@ -39,7 +39,7 @@ unsafe impl tokio_epoll_uring::IoBuf for Buf { } fn bytes_init(&self) -> usize { - self.written + N } fn bytes_total(&self) -> usize { diff --git a/pageserver/src/virtual_file/owned_buffers_io/write.rs b/pageserver/src/virtual_file/owned_buffers_io/write.rs index 298abde9e2..2865ebb75b 100644 --- a/pageserver/src/virtual_file/owned_buffers_io/write.rs +++ b/pageserver/src/virtual_file/owned_buffers_io/write.rs @@ -10,61 +10,60 @@ pub trait OwnedAsyncWriter { } struct Buffer { - len: usize, - buf: B, + pending: usize, + iobuf: B, } impl Buffer 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 { 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 { - 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 { @@ -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(())