diff --git a/pageserver/src/tenant/blob_io.rs b/pageserver/src/tenant/blob_io.rs index 0d33100ead..6e90b3e8ff 100644 --- a/pageserver/src/tenant/blob_io.rs +++ b/pageserver/src/tenant/blob_io.rs @@ -121,7 +121,7 @@ impl BlobWriter { self.offset } - const CAPACITY: usize = if BUFFERED { PAGE_SZ } else { 0 }; + const CAPACITY: usize = if BUFFERED { 64 * 1024 } else { 0 }; /// Writes the given buffer directly to the underlying `VirtualFile`. /// You need to make sure that the internal buffer is empty, otherwise diff --git a/pageserver/src/tenant/ephemeral_file.rs b/pageserver/src/tenant/ephemeral_file.rs index e48b9e83bd..3154c28dfa 100644 --- a/pageserver/src/tenant/ephemeral_file.rs +++ b/pageserver/src/tenant/ephemeral_file.rs @@ -5,14 +5,12 @@ use crate::config::PageServerConf; use crate::context::RequestContext; use crate::page_cache::{self, PAGE_SZ}; use crate::tenant::block_io::{BlockCursor, BlockLease, BlockReader}; -use crate::virtual_file::{self, VirtualFile}; -use bytes::BytesMut; +use crate::virtual_file::owned_buffers_io::write::OwnedAsyncWriter; +use crate::virtual_file::{self, owned_buffers_io, VirtualFile}; use camino::Utf8PathBuf; use pageserver_api::shard::TenantShardId; -use std::cmp::min; -use std::io::{self, ErrorKind}; -use std::ops::DerefMut; +use std::io; use std::sync::atomic::AtomicU64; use tracing::*; use utils::id::TimelineId; @@ -22,18 +20,25 @@ pub struct EphemeralFile { _tenant_shard_id: TenantShardId, _timeline_id: TimelineId, - file: VirtualFile, - len: u64, - /// An ephemeral file is append-only. - /// We keep the last page, which can still be modified, in [`Self::mutable_tail`]. - /// The other pages, which can no longer be modified, are accessed through the page cache. + /// We sandwich the buffered writer between two size-tracking writers. + /// This allows us to "elegantly" track in-memory bytes vs flushed bytes, + /// enabling [`Self::read_blk`] to determine whether to read from the + /// buffered writer's buffer, versus going to the VirtualFile. /// - /// None <=> IO is ongoing. - /// Size is fixed to PAGE_SZ at creation time and must not be changed. - mutable_tail: Option, + /// TODO: longer-term, we probably wand to get rid of this in favor + /// of a double-buffering scheme. See this commit's commit message + /// and git history for what we had before this sandwich, it might be useful. + file: owned_buffers_io::util::size_tracking_writer::Writer< + owned_buffers_io::write::BufferedWriter< + { Self::TAIL_SZ }, + owned_buffers_io::util::size_tracking_writer::Writer, + >, + >, } impl EphemeralFile { + const TAIL_SZ: usize = 64 * 1024; + pub async fn create( conf: &PageServerConf, tenant_shard_id: TenantShardId, @@ -57,19 +62,20 @@ impl EphemeralFile { .create(true), ) .await?; + let file = owned_buffers_io::util::size_tracking_writer::Writer::new(file); + let file = owned_buffers_io::write::BufferedWriter::new(file); + let file = owned_buffers_io::util::size_tracking_writer::Writer::new(file); Ok(EphemeralFile { page_cache_file_id: page_cache::next_file_id(), _tenant_shard_id: tenant_shard_id, _timeline_id: timeline_id, file, - len: 0, - mutable_tail: Some(BytesMut::zeroed(PAGE_SZ)), }) } pub(crate) fn len(&self) -> u64 { - self.len + self.file.bytes_written() } pub(crate) async fn read_blk( @@ -77,8 +83,28 @@ impl EphemeralFile { blknum: u32, ctx: &RequestContext, ) -> Result { - let flushed_blknums = 0..self.len / PAGE_SZ as u64; - if flushed_blknums.contains(&(blknum as u64)) { + let buffered_offset = self.file.bytes_written(); + let flushed_offset = self.file.as_inner().as_inner().bytes_written(); + assert!(buffered_offset >= flushed_offset); + + assert_eq!( + flushed_offset % (PAGE_SZ as u64), + 0, + "we need this in the logic below, because it assumes the page isn't spread across flushed part and in-memory buffer" + ); + + let read_offset = (blknum as u64) * (PAGE_SZ as u64); + if read_offset + (PAGE_SZ as u64) > buffered_offset { + // TODO: handle out-of-bounds access, i.e., access past end of file currently panics + // but should probably return an IO error. Pre-existing issue before this patch. + todo!("return IO error: read past end of file") + } + + if read_offset < flushed_offset { + assert!( + read_offset + (PAGE_SZ as u64) <= flushed_offset, + "this impl can't deal with pages spread across flushed & buffered part" + ); let cache = page_cache::get(); match cache .read_immutable_buf(self.page_cache_file_id, blknum, ctx) @@ -89,7 +115,9 @@ impl EphemeralFile { // order path before error because error is anyhow::Error => might have many contexts format!( "ephemeral file: read immutable page #{}: {}: {:#}", - blknum, self.file.path, e, + blknum, + self.file.as_inner().as_inner().as_inner().path, + e, ), ) })? { @@ -99,6 +127,9 @@ impl EphemeralFile { page_cache::ReadBufResult::NotFound(write_guard) => { let write_guard = self .file + .as_inner() + .as_inner() + .as_inner() .read_exact_at_page(write_guard, blknum as u64 * PAGE_SZ as u64) .await?; let read_guard = write_guard.mark_valid(); @@ -106,13 +137,14 @@ impl EphemeralFile { } }; } else { - debug_assert_eq!(blknum as u64, self.len / PAGE_SZ as u64); + let buffer: &[u8; Self::TAIL_SZ] = self.file.as_inner().inspect_buffer(); + + let read_offset_in_buffer = read_offset - buffered_offset; + let read_offset_in_buffer = usize::try_from(read_offset_in_buffer).unwrap(); + let page = &buffer[read_offset_in_buffer..(read_offset_in_buffer + PAGE_SZ)]; Ok(BlockLease::EphemeralFileMutableTail( - self.mutable_tail - .as_deref() - .expect("we're not doing IO, it must be Some()") - .try_into() - .expect("we ensure that it's always PAGE_SZ"), + page.try_into() + .expect("the slice above got it as page-size slice"), )) } } @@ -120,139 +152,26 @@ impl EphemeralFile { pub(crate) async fn write_blob( &mut self, srcbuf: &[u8], - ctx: &RequestContext, + _ctx: &RequestContext, ) -> Result { - struct Writer<'a> { - ephemeral_file: &'a mut EphemeralFile, - /// The block to which the next [`push_bytes`] will write. - blknum: u32, - /// The offset inside the block identified by [`blknum`] to which [`push_bytes`] will write. - off: usize, - } - impl<'a> Writer<'a> { - fn new(ephemeral_file: &'a mut EphemeralFile) -> io::Result> { - Ok(Writer { - blknum: (ephemeral_file.len / PAGE_SZ as u64) as u32, - off: (ephemeral_file.len % PAGE_SZ as u64) as usize, - ephemeral_file, - }) - } - #[inline(always)] - async fn push_bytes( - &mut self, - src: &[u8], - ctx: &RequestContext, - ) -> Result<(), io::Error> { - let mut src_remaining = src; - while !src_remaining.is_empty() { - let dst_remaining = &mut self - .ephemeral_file - .mutable_tail - .as_deref_mut() - .expect("IO is not yet ongoing")[self.off..]; - let n = min(dst_remaining.len(), src_remaining.len()); - dst_remaining[..n].copy_from_slice(&src_remaining[..n]); - self.off += n; - src_remaining = &src_remaining[n..]; - if self.off == PAGE_SZ { - let mutable_tail = std::mem::take(&mut self.ephemeral_file.mutable_tail) - .expect("IO is not yet ongoing"); - let (mutable_tail, res) = self - .ephemeral_file - .file - .write_all_at(mutable_tail, self.blknum as u64 * PAGE_SZ as u64) - .await; - // TODO: If we panic before we can put the mutable_tail back, subsequent calls will fail. - // I.e., the IO isn't retryable if we panic. - self.ephemeral_file.mutable_tail = Some(mutable_tail); - match res { - Ok(_) => { - // Pre-warm the page cache with what we just wrote. - // This isn't necessary for coherency/correctness, but it's how we've always done it. - let cache = page_cache::get(); - match cache - .read_immutable_buf( - self.ephemeral_file.page_cache_file_id, - self.blknum, - ctx, - ) - .await - { - Ok(page_cache::ReadBufResult::Found(_guard)) => { - // This function takes &mut self, so, it shouldn't be possible to reach this point. - unreachable!("we just wrote blknum {} and this function takes &mut self, so, no concurrent read_blk is possible", self.blknum); - } - Ok(page_cache::ReadBufResult::NotFound(mut write_guard)) => { - let buf: &mut [u8] = write_guard.deref_mut(); - debug_assert_eq!(buf.len(), PAGE_SZ); - buf.copy_from_slice( - self.ephemeral_file - .mutable_tail - .as_deref() - .expect("IO is not ongoing"), - ); - let _ = write_guard.mark_valid(); - // pre-warm successful - } - Err(e) => { - error!("ephemeral_file write_blob failed to get immutable buf to pre-warm page cache: {e:?}"); - // fail gracefully, it's not the end of the world if we can't pre-warm the cache here - } - } - // Zero the buffer for re-use. - // Zeroing is critical for correcntess because the write_blob code below - // and similarly read_blk expect zeroed pages. - self.ephemeral_file - .mutable_tail - .as_deref_mut() - .expect("IO is not ongoing") - .fill(0); - // This block is done, move to next one. - self.blknum += 1; - self.off = 0; - } - Err(e) => { - return Err(std::io::Error::new( - ErrorKind::Other, - // order error before path because path is long and error is short - format!( - "ephemeral_file: write_blob: write-back full tail blk #{}: {:#}: {}", - self.blknum, - e, - self.ephemeral_file.file.path, - ), - )); - } - } - } - } - Ok(()) - } - } - - let pos = self.len; - let mut writer = Writer::new(self)?; + let pos = self.file.bytes_written(); // Write the length field if srcbuf.len() < 0x80 { // short one-byte length header let len_buf = [srcbuf.len() as u8]; - writer.push_bytes(&len_buf, ctx).await?; + + self.file.write_all_borrowed(&len_buf).await?; } else { let mut len_buf = u32::to_be_bytes(srcbuf.len() as u32); len_buf[0] |= 0x80; - writer.push_bytes(&len_buf, ctx).await?; + self.file.write_all_borrowed(&len_buf).await?; } // Write the payload - writer.push_bytes(srcbuf, ctx).await?; + self.file.write_all_borrowed(&srcbuf).await?; - if srcbuf.len() < 0x80 { - self.len += 1; - } else { - self.len += 4; - } - self.len += srcbuf.len() as u64; + // TODO: bring back pre-warming of page cache, using another sandwich layer Ok(pos) } @@ -273,7 +192,7 @@ impl Drop for EphemeralFile { // We leave them there, [`crate::page_cache::PageCache::find_victim`] will evict them when needed. // unlink the file - let res = std::fs::remove_file(&self.file.path); + let res = std::fs::remove_file(&self.file.as_inner().as_inner().as_inner().path); if let Err(e) = res { if e.kind() != std::io::ErrorKind::NotFound { // just never log the not found errors, we cannot do anything for them; on detach @@ -282,7 +201,8 @@ impl Drop for EphemeralFile { // not found files might also be related to https://github.com/neondatabase/neon/issues/2442 error!( "could not remove ephemeral file '{}': {}", - self.file.path, e + self.file.as_inner().as_inner().as_inner().path, + e ); } } diff --git a/pageserver/src/virtual_file.rs b/pageserver/src/virtual_file.rs index 6d4774cf75..82437b6ce4 100644 --- a/pageserver/src/virtual_file.rs +++ b/pageserver/src/virtual_file.rs @@ -34,6 +34,24 @@ pub(crate) use io_engine::IoEngineKind; pub(crate) use metadata::Metadata; pub(crate) use open_options::*; +use self::owned_buffers_io::write::OwnedAsyncWriter; + +pub(crate) mod owned_buffers_io { + //! Abstractions for IO with owned buffers. + //! + //! Not actually tied to [`crate::virtual_file`] specifically, but, it's the primary + //! reason we need this abstraction. + //! + //! Over time, this could move into the `tokio-epoll-uring` crate, maybe `uring-common`, + //! but for the time being we're proving out the primitives in the neon.git repo + //! for faster iteration. + + pub(crate) mod write; + pub(crate) mod util { + pub(crate) mod size_tracking_writer; + } +} + /// /// A virtual file descriptor. You can use this just like std::fs::File, but internally /// the underlying file is closed if the system is low on file descriptors, @@ -1064,6 +1082,23 @@ impl Drop for VirtualFile { } } +impl OwnedAsyncWriter for VirtualFile { + #[inline(always)] + async fn write_all, Buf: IoBuf + Send>( + &mut self, + buf: B, + ) -> std::io::Result<(usize, B::Buf)> { + let (buf, res) = VirtualFile::write_all(self, buf).await; + res.map(move |v| (v, buf)) + } + + #[inline(always)] + async fn write_all_borrowed(&mut self, _buf: &[u8]) -> std::io::Result { + // TODO: ensure this through the type system + panic!("this should not happen"); + } +} + impl OpenFiles { fn new(num_slots: usize) -> OpenFiles { let mut slots = Box::new(Vec::with_capacity(num_slots)); diff --git a/pageserver/src/virtual_file/owned_buffers_io/util/size_tracking_writer.rs b/pageserver/src/virtual_file/owned_buffers_io/util/size_tracking_writer.rs new file mode 100644 index 0000000000..6f123bc195 --- /dev/null +++ b/pageserver/src/virtual_file/owned_buffers_io/util/size_tracking_writer.rs @@ -0,0 +1,49 @@ +use crate::virtual_file::owned_buffers_io::write::OwnedAsyncWriter; +use tokio_epoll_uring::{BoundedBuf, IoBuf}; + +pub struct Writer { + dst: W, + bytes_amount: u64, +} + +impl Writer { + pub fn new(dst: W) -> Self { + Self { + dst, + bytes_amount: 0, + } + } + pub fn bytes_written(&self) -> u64 { + self.bytes_amount + } + pub fn as_inner(&self) -> &W { + &self.dst + } + /// Returns the wrapped `VirtualFile` object as well as the number + /// of bytes that were written to it through this object. + pub fn into_inner(self) -> (u64, W) { + (self.bytes_amount, self.dst) + } +} + +impl OwnedAsyncWriter for Writer +where + W: OwnedAsyncWriter, +{ + #[inline(always)] + async fn write_all, Buf: IoBuf + Send>( + &mut self, + buf: B, + ) -> std::io::Result<(usize, B::Buf)> { + let (nwritten, buf) = self.dst.write_all(buf).await?; + self.bytes_amount += u64::try_from(nwritten).unwrap(); + Ok((nwritten, buf)) + } + + #[inline(always)] + async fn write_all_borrowed(&mut self, buf: &[u8]) -> std::io::Result { + let nwritten = self.dst.write_all_borrowed(buf).await?; + self.bytes_amount += u64::try_from(nwritten).unwrap(); + Ok(nwritten) + } +} diff --git a/pageserver/src/virtual_file/owned_buffers_io/write.rs b/pageserver/src/virtual_file/owned_buffers_io/write.rs new file mode 100644 index 0000000000..52dcf2e6c0 --- /dev/null +++ b/pageserver/src/virtual_file/owned_buffers_io/write.rs @@ -0,0 +1,296 @@ +use bytes::BytesMut; +use tokio_epoll_uring::{BoundedBuf, IoBuf, Slice}; + +/// A trait for doing owned-buffer write IO. +/// Think [`tokio::io::AsyncWrite`] but with owned buffers. +pub trait OwnedAsyncWriter { + async fn write_all, Buf: IoBuf + Send>( + &mut self, + buf: B, + ) -> std::io::Result<(usize, B::Buf)>; + async fn write_all_borrowed(&mut self, buf: &[u8]) -> std::io::Result; +} + +/// A wrapper aorund an [`OwnedAsyncWriter`] that batches smaller writers +/// into `BUFFER_SIZE`-sized writes. +/// +/// # Passthrough Of Large Writers +/// +/// Buffered writes larger than the `BUFFER_SIZE` cause the internal +/// buffer to be flushed, even if it is not full yet. Then, the large +/// buffered write is passed through to the unerlying [`OwnedAsyncWriter`]. +/// +/// This pass-through is generally beneficial for throughput, but if +/// the storage backend of the [`OwnedAsyncWriter`] is a shared resource, +/// unlimited large writes may cause latency or fairness issues. +/// +/// In such cases, a different implementation that always buffers in memory +/// may be preferable. +pub struct BufferedWriter { + writer: W, + // invariant: always remains Some(buf) + // with buf.capacity() == BUFFER_SIZE except + // - while IO is ongoing => goes back to Some() once the IO completed successfully + // - after an IO error => stays `None` forever + // In these exceptional cases, it's `None`. + buf: Option, +} + +impl BufferedWriter +where + W: OwnedAsyncWriter, +{ + pub fn new(writer: W) -> Self { + Self { + writer, + buf: Some(BytesMut::with_capacity(BUFFER_SIZE)), + } + } + + pub fn as_inner(&self) -> &W { + &self.writer + } + + /// panics if used after an error + pub fn inspect_buffer(&self) -> &[u8; BUFFER_SIZE] { + self.buf + .as_deref() + .expect("must not use after an error on the write path") + .try_into() + .unwrap() + } + + pub async fn flush_and_into_inner(mut self) -> std::io::Result { + self.flush().await?; + let Self { buf, writer } = self; + assert!(buf.is_some()); + Ok(writer) + } + + pub async fn write_buffered(&mut self, chunk: Slice) -> std::io::Result<(usize, B)> + where + B: IoBuf + Send, + { + let chunk_len = chunk.len(); + // avoid memcpy for the middle of the chunk + if chunk.len() >= BUFFER_SIZE { + self.flush().await?; + // do a big write, bypassing `buf` + assert_eq!( + self.buf + .as_ref() + .expect("must not use after an error") + .len(), + 0 + ); + let (nwritten, chunk) = self.writer.write_all(chunk).await?; + assert_eq!(nwritten, chunk_len); + return Ok((nwritten, chunk)); + } + // in-memory copy the < BUFFER_SIZED tail of the chunk + assert!(chunk.len() < BUFFER_SIZE); + let mut slice = &chunk[..]; + while !slice.is_empty() { + let buf = self.buf.as_mut().expect("must not use after an error"); + let need = BUFFER_SIZE - buf.len(); + let have = slice.len(); + let n = std::cmp::min(need, have); + buf.extend_from_slice(&slice[..n]); + slice = &slice[n..]; + if buf.len() >= BUFFER_SIZE { + assert_eq!(buf.len(), BUFFER_SIZE); + self.flush().await?; + } + } + assert!(slice.is_empty(), "by now we should have drained the chunk"); + Ok((chunk_len, chunk.into_inner())) + } + + /// Always goes through the internal buffer. + /// Guaranteed to never invoke [`OwnedAsyncWrite::write_all_borrowed`] on the underlying. + pub async fn write_all_borrowed(&mut self, mut chunk: &[u8]) -> std::io::Result { + let chunk_len = chunk.len(); + while !chunk.is_empty() { + let buf = self.buf.as_mut().expect("must not use after an error"); + let need = BUFFER_SIZE - buf.len(); + let have = chunk.len(); + let n = std::cmp::min(need, have); + buf.extend_from_slice(&chunk[..n]); + chunk = &chunk[n..]; + if buf.len() >= BUFFER_SIZE { + assert_eq!(buf.len(), BUFFER_SIZE); + self.flush().await?; + } + } + Ok(chunk_len) + } + + async fn flush(&mut self) -> std::io::Result<()> { + let buf = self.buf.take().expect("must not use after an error"); + if buf.is_empty() { + self.buf = Some(buf); + return std::io::Result::Ok(()); + } + let buf_len = buf.len(); + let (nwritten, mut buf) = self.writer.write_all(buf).await?; + assert_eq!(nwritten, buf_len); + buf.clear(); + self.buf = Some(buf); + Ok(()) + } +} + +impl OwnedAsyncWriter + for BufferedWriter +{ + #[inline(always)] + async fn write_all, Buf: IoBuf + Send>( + &mut self, + buf: B, + ) -> std::io::Result<(usize, B::Buf)> { + let nbytes = buf.bytes_init(); + if nbytes == 0 { + return Ok((0, Slice::into_inner(buf.slice_full()))); + } + let slice = buf.slice(0..nbytes); + BufferedWriter::write_buffered(self, slice).await + } + #[inline(always)] + async fn write_all_borrowed(&mut self, buf: &[u8]) -> std::io::Result { + BufferedWriter::write_all_borrowed(self, buf).await + } +} + +impl OwnedAsyncWriter for Vec { + async fn write_all, Buf: IoBuf + Send>( + &mut self, + buf: B, + ) -> std::io::Result<(usize, B::Buf)> { + let nbytes = buf.bytes_init(); + if nbytes == 0 { + return Ok((0, Slice::into_inner(buf.slice_full()))); + } + let buf = buf.slice(0..nbytes); + self.extend_from_slice(&buf[..]); + Ok((buf.len(), Slice::into_inner(buf))) + } + + async fn write_all_borrowed(&mut self, buf: &[u8]) -> std::io::Result { + self.extend_from_slice(&buf[..]); + Ok(buf.len()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(Default)] + struct RecorderWriter { + writes: Vec>, + } + impl OwnedAsyncWriter for RecorderWriter { + async fn write_all, Buf: IoBuf + Send>( + &mut self, + buf: B, + ) -> std::io::Result<(usize, B::Buf)> { + let nbytes = buf.bytes_init(); + if nbytes == 0 { + self.writes.push(vec![]); + return Ok((0, Slice::into_inner(buf.slice_full()))); + } + let buf = buf.slice(0..nbytes); + self.writes.push(Vec::from(&buf[..])); + Ok((buf.len(), Slice::into_inner(buf))) + } + + async fn write_all_borrowed(&mut self, buf: &[u8]) -> std::io::Result { + self.writes.push(Vec::from(buf)); + Ok(buf.len()) + } + } + + macro_rules! write { + ($writer:ident, $data:literal) => {{ + $writer + .write_buffered(::bytes::Bytes::from_static($data).slice_full()) + .await?; + }}; + } + + #[tokio::test] + async fn test_buffered_writes_only() -> std::io::Result<()> { + let recorder = RecorderWriter::default(); + let mut writer = BufferedWriter::<2, _>::new(recorder); + write!(writer, b"a"); + write!(writer, b"b"); + write!(writer, b"c"); + write!(writer, b"d"); + write!(writer, b"e"); + let recorder = writer.flush_and_into_inner().await?; + assert_eq!( + recorder.writes, + vec![Vec::from(b"ab"), Vec::from(b"cd"), Vec::from(b"e")] + ); + Ok(()) + } + + #[tokio::test] + async fn test_passthrough_writes_only() -> std::io::Result<()> { + let recorder = RecorderWriter::default(); + let mut writer = BufferedWriter::<2, _>::new(recorder); + write!(writer, b"abc"); + write!(writer, b"de"); + write!(writer, b""); + write!(writer, b"fghijk"); + let recorder = writer.flush_and_into_inner().await?; + assert_eq!( + recorder.writes, + vec![Vec::from(b"abc"), Vec::from(b"de"), Vec::from(b"fghijk")] + ); + Ok(()) + } + + #[tokio::test] + async fn test_passthrough_write_with_nonempty_buffer() -> std::io::Result<()> { + let recorder = RecorderWriter::default(); + let mut writer = BufferedWriter::<2, _>::new(recorder); + write!(writer, b"a"); + write!(writer, b"bc"); + write!(writer, b"d"); + write!(writer, b"e"); + let recorder = writer.flush_and_into_inner().await?; + assert_eq!( + recorder.writes, + vec![Vec::from(b"a"), Vec::from(b"bc"), Vec::from(b"de")] + ); + Ok(()) + } + + #[tokio::test] + async fn test_write_all_borrowed_always_goes_through_buffer() -> std::io::Result<()> { + let recorder = RecorderWriter::default(); + let mut writer = BufferedWriter::<2, _>::new(recorder); + + writer.write_all_borrowed(b"abc").await?; + writer.write_all_borrowed(b"d").await?; + writer.write_all_borrowed(b"e").await?; + writer.write_all_borrowed(b"fg").await?; + writer.write_all_borrowed(b"hi").await?; + writer.write_all_borrowed(b"j").await?; + writer.write_all_borrowed(b"klmno").await?; + + let recorder = writer.flush_and_into_inner().await?; + assert_eq!( + recorder.writes, + { + let expect: &[&[u8]] = &[b"ab", b"cd", b"ef", b"gh", b"ij", b"kl", b"mn", b"o"]; + expect + } + .into_iter() + .map(|v| v[..].to_vec()) + .collect::>() + ); + Ok(()) + } +}