LCOV - code coverage report
Current view: top level - pageserver/src/tenant - blob_io.rs (source / functions) Coverage Total Hit
Test: 07bee600374ccd486c69370d0972d9035964fe68.info Lines: 92.7 % 385 357
Test Date: 2025-02-20 13:11:02 Functions: 91.3 % 103 94

            Line data    Source code
       1              : //!
       2              : //! Functions for reading and writing variable-sized "blobs".
       3              : //!
       4              : //! Each blob begins with a 1- or 4-byte length field, followed by the
       5              : //! actual data. If the length is smaller than 128 bytes, the length
       6              : //! is written as a one byte. If it's larger than that, the length
       7              : //! is written as a four-byte integer, in big-endian, with the high
       8              : //! bit set. This way, we can detect whether it's 1- or 4-byte header
       9              : //! by peeking at the first byte. For blobs larger than 128 bits,
      10              : //! we also specify three reserved bits, only one of the three bit
      11              : //! patterns is currently in use (0b011) and signifies compression
      12              : //! with zstd.
      13              : //!
      14              : //! len <  128: 0XXXXXXX
      15              : //! len >= 128: 1CCCXXXX XXXXXXXX XXXXXXXX XXXXXXXX
      16              : //!
      17              : use async_compression::Level;
      18              : use bytes::{BufMut, BytesMut};
      19              : use pageserver_api::models::ImageCompressionAlgorithm;
      20              : use tokio::io::AsyncWriteExt;
      21              : use tokio_epoll_uring::{BoundedBuf, IoBuf, Slice};
      22              : use tracing::warn;
      23              : 
      24              : use crate::context::RequestContext;
      25              : use crate::page_cache::PAGE_SZ;
      26              : use crate::tenant::block_io::BlockCursor;
      27              : use crate::virtual_file::owned_buffers_io::io_buf_ext::{FullSlice, IoBufExt};
      28              : use crate::virtual_file::VirtualFile;
      29              : use std::cmp::min;
      30              : use std::io::{Error, ErrorKind};
      31              : 
      32              : #[derive(Copy, Clone, Debug)]
      33              : pub struct CompressionInfo {
      34              :     pub written_compressed: bool,
      35              :     pub compressed_size: Option<usize>,
      36              : }
      37              : 
      38              : impl BlockCursor<'_> {
      39              :     /// Read a blob into a new buffer.
      40        16592 :     pub async fn read_blob(
      41        16592 :         &self,
      42        16592 :         offset: u64,
      43        16592 :         ctx: &RequestContext,
      44        16592 :     ) -> Result<Vec<u8>, std::io::Error> {
      45        16592 :         let mut buf = Vec::new();
      46        16592 :         self.read_blob_into_buf(offset, &mut buf, ctx).await?;
      47        16592 :         Ok(buf)
      48        16592 :     }
      49              :     /// Read blob into the given buffer. Any previous contents in the buffer
      50              :     /// are overwritten.
      51        16720 :     pub async fn read_blob_into_buf(
      52        16720 :         &self,
      53        16720 :         offset: u64,
      54        16720 :         dstbuf: &mut Vec<u8>,
      55        16720 :         ctx: &RequestContext,
      56        16720 :     ) -> Result<(), std::io::Error> {
      57        16720 :         let mut blknum = (offset / PAGE_SZ as u64) as u32;
      58        16720 :         let mut off = (offset % PAGE_SZ as u64) as usize;
      59              : 
      60        16720 :         let mut buf = self.read_blk(blknum, ctx).await?;
      61              : 
      62              :         // peek at the first byte, to determine if it's a 1- or 4-byte length
      63        16720 :         let first_len_byte = buf[off];
      64        16720 :         let len: usize = if first_len_byte < 0x80 {
      65              :             // 1-byte length header
      66         4448 :             off += 1;
      67         4448 :             first_len_byte as usize
      68              :         } else {
      69              :             // 4-byte length header
      70        12272 :             let mut len_buf = [0u8; 4];
      71        12272 :             let thislen = PAGE_SZ - off;
      72        12272 :             if thislen < 4 {
      73              :                 // it is split across two pages
      74            0 :                 len_buf[..thislen].copy_from_slice(&buf[off..PAGE_SZ]);
      75            0 :                 blknum += 1;
      76            0 :                 buf = self.read_blk(blknum, ctx).await?;
      77            0 :                 len_buf[thislen..].copy_from_slice(&buf[0..4 - thislen]);
      78            0 :                 off = 4 - thislen;
      79        12272 :             } else {
      80        12272 :                 len_buf.copy_from_slice(&buf[off..off + 4]);
      81        12272 :                 off += 4;
      82        12272 :             }
      83        12272 :             let bit_mask = if self.read_compressed {
      84           40 :                 !LEN_COMPRESSION_BIT_MASK
      85              :             } else {
      86        12232 :                 0x7f
      87              :             };
      88        12272 :             len_buf[0] &= bit_mask;
      89        12272 :             u32::from_be_bytes(len_buf) as usize
      90              :         };
      91        16720 :         let compression_bits = first_len_byte & LEN_COMPRESSION_BIT_MASK;
      92        16720 : 
      93        16720 :         let mut tmp_buf = Vec::new();
      94              :         let buf_to_write;
      95        16720 :         let compression = if compression_bits <= BYTE_UNCOMPRESSED || !self.read_compressed {
      96        16712 :             if compression_bits > BYTE_UNCOMPRESSED {
      97            0 :                 warn!("reading key above future limit ({len} bytes)");
      98        16712 :             }
      99        16712 :             buf_to_write = dstbuf;
     100        16712 :             None
     101            8 :         } else if compression_bits == BYTE_ZSTD {
     102            8 :             buf_to_write = &mut tmp_buf;
     103            8 :             Some(dstbuf)
     104              :         } else {
     105            0 :             let error = std::io::Error::new(
     106            0 :                 std::io::ErrorKind::InvalidData,
     107            0 :                 format!("invalid compression byte {compression_bits:x}"),
     108            0 :             );
     109            0 :             return Err(error);
     110              :         };
     111              : 
     112        16720 :         buf_to_write.clear();
     113        16720 :         buf_to_write.reserve(len);
     114        16720 : 
     115        16720 :         // Read the payload
     116        16720 :         let mut remain = len;
     117        58872 :         while remain > 0 {
     118        42152 :             let mut page_remain = PAGE_SZ - off;
     119        42152 :             if page_remain == 0 {
     120              :                 // continue on next page
     121        25512 :                 blknum += 1;
     122        25512 :                 buf = self.read_blk(blknum, ctx).await?;
     123        25512 :                 off = 0;
     124        25512 :                 page_remain = PAGE_SZ;
     125        16640 :             }
     126        42152 :             let this_blk_len = min(remain, page_remain);
     127        42152 :             buf_to_write.extend_from_slice(&buf[off..off + this_blk_len]);
     128        42152 :             remain -= this_blk_len;
     129        42152 :             off += this_blk_len;
     130              :         }
     131              : 
     132        16720 :         if let Some(dstbuf) = compression {
     133            8 :             if compression_bits == BYTE_ZSTD {
     134            8 :                 let mut decoder = async_compression::tokio::write::ZstdDecoder::new(dstbuf);
     135            8 :                 decoder.write_all(buf_to_write).await?;
     136            8 :                 decoder.flush().await?;
     137              :             } else {
     138            0 :                 unreachable!("already checked above")
     139              :             }
     140        16712 :         }
     141              : 
     142        16720 :         Ok(())
     143        16720 :     }
     144              : }
     145              : 
     146              : /// Reserved bits for length and compression
     147              : pub(super) const LEN_COMPRESSION_BIT_MASK: u8 = 0xf0;
     148              : 
     149              : /// The maximum size of blobs we support. The highest few bits
     150              : /// are reserved for compression and other further uses.
     151              : pub(crate) const MAX_SUPPORTED_BLOB_LEN: usize = 0x0fff_ffff;
     152              : 
     153              : pub(super) const BYTE_UNCOMPRESSED: u8 = 0x80;
     154              : pub(super) const BYTE_ZSTD: u8 = BYTE_UNCOMPRESSED | 0x10;
     155              : 
     156              : /// A wrapper of `VirtualFile` that allows users to write blobs.
     157              : ///
     158              : /// If a `BlobWriter` is dropped, the internal buffer will be
     159              : /// discarded. You need to call [`flush_buffer`](Self::flush_buffer)
     160              : /// manually before dropping.
     161              : pub struct BlobWriter<const BUFFERED: bool> {
     162              :     inner: VirtualFile,
     163              :     offset: u64,
     164              :     /// A buffer to save on write calls, only used if BUFFERED=true
     165              :     buf: Vec<u8>,
     166              :     /// We do tiny writes for the length headers; they need to be in an owned buffer;
     167              :     io_buf: Option<BytesMut>,
     168              : }
     169              : 
     170              : impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
     171         4196 :     pub fn new(inner: VirtualFile, start_offset: u64) -> Self {
     172         4196 :         Self {
     173         4196 :             inner,
     174         4196 :             offset: start_offset,
     175         4196 :             buf: Vec::with_capacity(Self::CAPACITY),
     176         4196 :             io_buf: Some(BytesMut::new()),
     177         4196 :         }
     178         4196 :     }
     179              : 
     180      4099620 :     pub fn size(&self) -> u64 {
     181      4099620 :         self.offset
     182      4099620 :     }
     183              : 
     184              :     const CAPACITY: usize = if BUFFERED { 64 * 1024 } else { 0 };
     185              : 
     186              :     /// Writes the given buffer directly to the underlying `VirtualFile`.
     187              :     /// You need to make sure that the internal buffer is empty, otherwise
     188              :     /// data will be written in wrong order.
     189              :     #[inline(always)]
     190      2203652 :     async fn write_all_unbuffered<Buf: IoBuf + Send>(
     191      2203652 :         &mut self,
     192      2203652 :         src_buf: FullSlice<Buf>,
     193      2203652 :         ctx: &RequestContext,
     194      2203652 :     ) -> (FullSlice<Buf>, Result<(), Error>) {
     195      2203652 :         let (src_buf, res) = self.inner.write_all(src_buf, ctx).await;
     196      2203652 :         let nbytes = match res {
     197      2203652 :             Ok(nbytes) => nbytes,
     198            0 :             Err(e) => return (src_buf, Err(e)),
     199              :         };
     200      2203652 :         self.offset += nbytes as u64;
     201      2203652 :         (src_buf, Ok(()))
     202      2203652 :     }
     203              : 
     204              :     #[inline(always)]
     205              :     /// Flushes the internal buffer to the underlying `VirtualFile`.
     206        24592 :     pub async fn flush_buffer(&mut self, ctx: &RequestContext) -> Result<(), Error> {
     207        24592 :         let buf = std::mem::take(&mut self.buf);
     208        24592 :         let (slice, res) = self.inner.write_all(buf.slice_len(), ctx).await;
     209        24592 :         res?;
     210        24592 :         let mut buf = slice.into_raw_slice().into_inner();
     211        24592 :         buf.clear();
     212        24592 :         self.buf = buf;
     213        24592 :         Ok(())
     214        24592 :     }
     215              : 
     216              :     #[inline(always)]
     217              :     /// Writes as much of `src_buf` into the internal buffer as it fits
     218     25939688 :     fn write_into_buffer(&mut self, src_buf: &[u8]) -> usize {
     219     25939688 :         let remaining = Self::CAPACITY - self.buf.len();
     220     25939688 :         let to_copy = src_buf.len().min(remaining);
     221     25939688 :         self.buf.extend_from_slice(&src_buf[..to_copy]);
     222     25939688 :         self.offset += to_copy as u64;
     223     25939688 :         to_copy
     224     25939688 :     }
     225              : 
     226              :     /// Internal, possibly buffered, write function
     227     28121840 :     async fn write_all<Buf: IoBuf + Send>(
     228     28121840 :         &mut self,
     229     28121840 :         src_buf: FullSlice<Buf>,
     230     28121840 :         ctx: &RequestContext,
     231     28121840 :     ) -> (FullSlice<Buf>, Result<(), Error>) {
     232     28121840 :         let src_buf = src_buf.into_raw_slice();
     233     28121840 :         let src_buf_bounds = src_buf.bounds();
     234     28121840 :         let restore = move |src_buf_slice: Slice<_>| {
     235     25918188 :             FullSlice::must_new(Slice::from_buf_bounds(
     236     25918188 :                 src_buf_slice.into_inner(),
     237     25918188 :                 src_buf_bounds,
     238     25918188 :             ))
     239     25918188 :         };
     240              : 
     241     28121840 :         if !BUFFERED {
     242      2203192 :             assert!(self.buf.is_empty());
     243      2203192 :             return self
     244      2203192 :                 .write_all_unbuffered(FullSlice::must_new(src_buf), ctx)
     245      2203192 :                 .await;
     246     25918648 :         }
     247     25918648 :         let remaining = Self::CAPACITY - self.buf.len();
     248     25918648 :         let src_buf_len = src_buf.bytes_init();
     249     25918648 :         if src_buf_len == 0 {
     250           48 :             return (restore(src_buf), Ok(()));
     251     25918600 :         }
     252     25918600 :         let mut src_buf = src_buf.slice(0..src_buf_len);
     253     25918600 :         // First try to copy as much as we can into the buffer
     254     25918600 :         if remaining > 0 {
     255     25918600 :             let copied = self.write_into_buffer(&src_buf);
     256     25918600 :             src_buf = src_buf.slice(copied..);
     257     25918600 :         }
     258              :         // Then, if the buffer is full, flush it out
     259     25918600 :         if self.buf.len() == Self::CAPACITY {
     260        21628 :             if let Err(e) = self.flush_buffer(ctx).await {
     261            0 :                 return (restore(src_buf), Err(e));
     262        21628 :             }
     263     25896972 :         }
     264              :         // Finally, write the tail of src_buf:
     265              :         // If it wholly fits into the buffer without
     266              :         // completely filling it, then put it there.
     267              :         // If not, write it out directly.
     268     25918600 :         let src_buf = if !src_buf.is_empty() {
     269        21548 :             assert_eq!(self.buf.len(), 0);
     270        21548 :             if src_buf.len() < Self::CAPACITY {
     271        21088 :                 let copied = self.write_into_buffer(&src_buf);
     272        21088 :                 // We just verified above that src_buf fits into our internal buffer.
     273        21088 :                 assert_eq!(copied, src_buf.len());
     274        21088 :                 restore(src_buf)
     275              :             } else {
     276          460 :                 let (src_buf, res) = self
     277          460 :                     .write_all_unbuffered(FullSlice::must_new(src_buf), ctx)
     278          460 :                     .await;
     279          460 :                 if let Err(e) = res {
     280            0 :                     return (src_buf, Err(e));
     281          460 :                 }
     282          460 :                 src_buf
     283              :             }
     284              :         } else {
     285     25897052 :             restore(src_buf)
     286              :         };
     287     25918600 :         (src_buf, Ok(()))
     288     28121840 :     }
     289              : 
     290              :     /// Write a blob of data. Returns the offset that it was written to,
     291              :     /// which can be used to retrieve the data later.
     292        20696 :     pub async fn write_blob<Buf: IoBuf + Send>(
     293        20696 :         &mut self,
     294        20696 :         srcbuf: FullSlice<Buf>,
     295        20696 :         ctx: &RequestContext,
     296        20696 :     ) -> (FullSlice<Buf>, Result<u64, Error>) {
     297        20696 :         let (buf, res) = self
     298        20696 :             .write_blob_maybe_compressed(srcbuf, ctx, ImageCompressionAlgorithm::Disabled)
     299        20696 :             .await;
     300        20696 :         (buf, res.map(|(off, _compression_info)| off))
     301        20696 :     }
     302              : 
     303              :     /// Write a blob of data. Returns the offset that it was written to,
     304              :     /// which can be used to retrieve the data later.
     305     14060920 :     pub(crate) async fn write_blob_maybe_compressed<Buf: IoBuf + Send>(
     306     14060920 :         &mut self,
     307     14060920 :         srcbuf: FullSlice<Buf>,
     308     14060920 :         ctx: &RequestContext,
     309     14060920 :         algorithm: ImageCompressionAlgorithm,
     310     14060920 :     ) -> (FullSlice<Buf>, Result<(u64, CompressionInfo), Error>) {
     311     14060920 :         let offset = self.offset;
     312     14060920 :         let mut compression_info = CompressionInfo {
     313     14060920 :             written_compressed: false,
     314     14060920 :             compressed_size: None,
     315     14060920 :         };
     316     14060920 : 
     317     14060920 :         let len = srcbuf.len();
     318     14060920 : 
     319     14060920 :         let mut io_buf = self.io_buf.take().expect("we always put it back below");
     320     14060920 :         io_buf.clear();
     321     14060920 :         let mut compressed_buf = None;
     322     14060920 :         let ((io_buf_slice, hdr_res), srcbuf) = async {
     323     14060920 :             if len < 128 {
     324              :                 // Short blob. Write a 1-byte length header
     325     13988184 :                 io_buf.put_u8(len as u8);
     326     13988184 :                 (self.write_all(io_buf.slice_len(), ctx).await, srcbuf)
     327              :             } else {
     328              :                 // Write a 4-byte length header
     329        72736 :                 if len > MAX_SUPPORTED_BLOB_LEN {
     330            0 :                     return (
     331            0 :                         (
     332            0 :                             io_buf.slice_len(),
     333            0 :                             Err(Error::new(
     334            0 :                                 ErrorKind::Other,
     335            0 :                                 format!("blob too large ({len} bytes)"),
     336            0 :                             )),
     337            0 :                         ),
     338            0 :                         srcbuf,
     339            0 :                     );
     340        72736 :                 }
     341        72736 :                 let (high_bit_mask, len_written, srcbuf) = match algorithm {
     342        20104 :                     ImageCompressionAlgorithm::Zstd { level } => {
     343        20104 :                         let mut encoder = if let Some(level) = level {
     344        20104 :                             async_compression::tokio::write::ZstdEncoder::with_quality(
     345        20104 :                                 Vec::new(),
     346        20104 :                                 Level::Precise(level.into()),
     347        20104 :                             )
     348              :                         } else {
     349            0 :                             async_compression::tokio::write::ZstdEncoder::new(Vec::new())
     350              :                         };
     351        20104 :                         encoder.write_all(&srcbuf[..]).await.unwrap();
     352        20104 :                         encoder.shutdown().await.unwrap();
     353        20104 :                         let compressed = encoder.into_inner();
     354        20104 :                         compression_info.compressed_size = Some(compressed.len());
     355        20104 :                         if compressed.len() < len {
     356           12 :                             compression_info.written_compressed = true;
     357           12 :                             let compressed_len = compressed.len();
     358           12 :                             compressed_buf = Some(compressed);
     359           12 :                             (BYTE_ZSTD, compressed_len, srcbuf)
     360              :                         } else {
     361        20092 :                             (BYTE_UNCOMPRESSED, len, srcbuf)
     362              :                         }
     363              :                     }
     364        52632 :                     ImageCompressionAlgorithm::Disabled => (BYTE_UNCOMPRESSED, len, srcbuf),
     365              :                 };
     366        72736 :                 let mut len_buf = (len_written as u32).to_be_bytes();
     367        72736 :                 assert_eq!(len_buf[0] & 0xf0, 0);
     368        72736 :                 len_buf[0] |= high_bit_mask;
     369        72736 :                 io_buf.extend_from_slice(&len_buf[..]);
     370        72736 :                 (self.write_all(io_buf.slice_len(), ctx).await, srcbuf)
     371              :             }
     372     14060920 :         }
     373     14060920 :         .await;
     374     14060920 :         self.io_buf = Some(io_buf_slice.into_raw_slice().into_inner());
     375     14060920 :         match hdr_res {
     376     14060920 :             Ok(_) => (),
     377            0 :             Err(e) => return (srcbuf, Err(e)),
     378              :         }
     379     14060920 :         let (srcbuf, res) = if let Some(compressed_buf) = compressed_buf {
     380           12 :             let (_buf, res) = self.write_all(compressed_buf.slice_len(), ctx).await;
     381           12 :             (srcbuf, res)
     382              :         } else {
     383     14060908 :             self.write_all(srcbuf, ctx).await
     384              :         };
     385     14060920 :         (srcbuf, res.map(|_| (offset, compression_info)))
     386     14060920 :     }
     387              : }
     388              : 
     389              : impl BlobWriter<true> {
     390              :     /// Access the underlying `VirtualFile`.
     391              :     ///
     392              :     /// This function flushes the internal buffer before giving access
     393              :     /// to the underlying `VirtualFile`.
     394         2884 :     pub async fn into_inner(mut self, ctx: &RequestContext) -> Result<VirtualFile, Error> {
     395         2884 :         self.flush_buffer(ctx).await?;
     396         2884 :         Ok(self.inner)
     397         2884 :     }
     398              : 
     399              :     /// Access the underlying `VirtualFile`.
     400              :     ///
     401              :     /// Unlike [`into_inner`](Self::into_inner), this doesn't flush
     402              :     /// the internal buffer before giving access.
     403           52 :     pub fn into_inner_no_flush(self) -> VirtualFile {
     404           52 :         self.inner
     405           52 :     }
     406              : }
     407              : 
     408              : impl BlobWriter<false> {
     409              :     /// Access the underlying `VirtualFile`.
     410         1180 :     pub fn into_inner(self) -> VirtualFile {
     411         1180 :         self.inner
     412         1180 :     }
     413              : }
     414              : 
     415              : #[cfg(test)]
     416              : pub(crate) mod tests {
     417              :     use super::*;
     418              :     use crate::{context::DownloadBehavior, task_mgr::TaskKind, tenant::block_io::BlockReaderRef};
     419              :     use camino::Utf8PathBuf;
     420              :     use camino_tempfile::Utf8TempDir;
     421              :     use rand::{Rng, SeedableRng};
     422              : 
     423           48 :     async fn round_trip_test<const BUFFERED: bool>(blobs: &[Vec<u8>]) -> Result<(), Error> {
     424           48 :         round_trip_test_compressed::<BUFFERED>(blobs, false).await
     425           48 :     }
     426              : 
     427           80 :     pub(crate) async fn write_maybe_compressed<const BUFFERED: bool>(
     428           80 :         blobs: &[Vec<u8>],
     429           80 :         compression: bool,
     430           80 :         ctx: &RequestContext,
     431           80 :     ) -> Result<(Utf8TempDir, Utf8PathBuf, Vec<u64>), Error> {
     432           80 :         let temp_dir = camino_tempfile::tempdir()?;
     433           80 :         let pathbuf = temp_dir.path().join("file");
     434           80 : 
     435           80 :         // Write part (in block to drop the file)
     436           80 :         let mut offsets = Vec::new();
     437              :         {
     438           80 :             let file = VirtualFile::create(pathbuf.as_path(), ctx).await?;
     439           80 :             let mut wtr = BlobWriter::<BUFFERED>::new(file, 0);
     440        24816 :             for blob in blobs.iter() {
     441        24816 :                 let (_, res) = if compression {
     442         4200 :                     let res = wtr
     443         4200 :                         .write_blob_maybe_compressed(
     444         4200 :                             blob.clone().slice_len(),
     445         4200 :                             ctx,
     446         4200 :                             ImageCompressionAlgorithm::Zstd { level: Some(1) },
     447         4200 :                         )
     448         4200 :                         .await;
     449         4200 :                     (res.0, res.1.map(|(off, _)| off))
     450              :                 } else {
     451        20616 :                     wtr.write_blob(blob.clone().slice_len(), ctx).await
     452              :                 };
     453        24816 :                 let offs = res?;
     454        24816 :                 offsets.push(offs);
     455              :             }
     456              :             // Write out one page worth of zeros so that we can
     457              :             // read again with read_blk
     458           80 :             let (_, res) = wtr.write_blob(vec![0; PAGE_SZ].slice_len(), ctx).await;
     459           80 :             let offs = res?;
     460           80 :             println!("Writing final blob at offs={offs}");
     461           80 :             wtr.flush_buffer(ctx).await?;
     462              :         }
     463           80 :         Ok((temp_dir, pathbuf, offsets))
     464           80 :     }
     465              : 
     466           64 :     async fn round_trip_test_compressed<const BUFFERED: bool>(
     467           64 :         blobs: &[Vec<u8>],
     468           64 :         compression: bool,
     469           64 :     ) -> Result<(), Error> {
     470           64 :         let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Error);
     471           64 :         let (_temp_dir, pathbuf, offsets) =
     472           64 :             write_maybe_compressed::<BUFFERED>(blobs, compression, &ctx).await?;
     473              : 
     474           64 :         let file = VirtualFile::open(pathbuf, &ctx).await?;
     475           64 :         let rdr = BlockReaderRef::VirtualFile(&file);
     476           64 :         let rdr = BlockCursor::new_with_compression(rdr, compression);
     477        16576 :         for (idx, (blob, offset)) in blobs.iter().zip(offsets.iter()).enumerate() {
     478        16576 :             let blob_read = rdr.read_blob(*offset, &ctx).await?;
     479        16576 :             assert_eq!(
     480        16576 :                 blob, &blob_read,
     481            0 :                 "mismatch for idx={idx} at offset={offset}"
     482              :             );
     483              :         }
     484           64 :         Ok(())
     485           64 :     }
     486              : 
     487        12316 :     pub(crate) fn random_array(len: usize) -> Vec<u8> {
     488        12316 :         let mut rng = rand::thread_rng();
     489    136303808 :         (0..len).map(|_| rng.gen()).collect::<_>()
     490        12316 :     }
     491              : 
     492              :     #[tokio::test]
     493            4 :     async fn test_one() -> Result<(), Error> {
     494            4 :         let blobs = &[vec![12, 21, 22]];
     495            4 :         round_trip_test::<false>(blobs).await?;
     496            4 :         round_trip_test::<true>(blobs).await?;
     497            4 :         Ok(())
     498            4 :     }
     499              : 
     500              :     #[tokio::test]
     501            4 :     async fn test_hello_simple() -> Result<(), Error> {
     502            4 :         let blobs = &[
     503            4 :             vec![0, 1, 2, 3],
     504            4 :             b"Hello, World!".to_vec(),
     505            4 :             Vec::new(),
     506            4 :             b"foobar".to_vec(),
     507            4 :         ];
     508            4 :         round_trip_test::<false>(blobs).await?;
     509            4 :         round_trip_test::<true>(blobs).await?;
     510            4 :         round_trip_test_compressed::<false>(blobs, true).await?;
     511            4 :         round_trip_test_compressed::<true>(blobs, true).await?;
     512            4 :         Ok(())
     513            4 :     }
     514              : 
     515              :     #[tokio::test]
     516            4 :     async fn test_really_big_array() -> Result<(), Error> {
     517            4 :         let blobs = &[
     518            4 :             b"test".to_vec(),
     519            4 :             random_array(10 * PAGE_SZ),
     520            4 :             b"hello".to_vec(),
     521            4 :             random_array(66 * PAGE_SZ),
     522            4 :             vec![0xf3; 24 * PAGE_SZ],
     523            4 :             b"foobar".to_vec(),
     524            4 :         ];
     525            4 :         round_trip_test::<false>(blobs).await?;
     526            4 :         round_trip_test::<true>(blobs).await?;
     527            4 :         round_trip_test_compressed::<false>(blobs, true).await?;
     528            4 :         round_trip_test_compressed::<true>(blobs, true).await?;
     529            4 :         Ok(())
     530            4 :     }
     531              : 
     532              :     #[tokio::test]
     533            4 :     async fn test_arrays_inc() -> Result<(), Error> {
     534            4 :         let blobs = (0..PAGE_SZ / 8)
     535         4096 :             .map(|v| random_array(v * 16))
     536            4 :             .collect::<Vec<_>>();
     537            4 :         round_trip_test::<false>(&blobs).await?;
     538            4 :         round_trip_test::<true>(&blobs).await?;
     539            4 :         Ok(())
     540            4 :     }
     541              : 
     542              :     #[tokio::test]
     543            4 :     async fn test_arrays_random_size() -> Result<(), Error> {
     544            4 :         let mut rng = rand::rngs::StdRng::seed_from_u64(42);
     545            4 :         let blobs = (0..1024)
     546         4096 :             .map(|_| {
     547         4096 :                 let mut sz: u16 = rng.gen();
     548         4096 :                 // Make 50% of the arrays small
     549         4096 :                 if rng.gen() {
     550         2064 :                     sz &= 63;
     551         2064 :                 }
     552         4096 :                 random_array(sz.into())
     553         4096 :             })
     554            4 :             .collect::<Vec<_>>();
     555            4 :         round_trip_test::<false>(&blobs).await?;
     556            4 :         round_trip_test::<true>(&blobs).await?;
     557            4 :         Ok(())
     558            4 :     }
     559              : 
     560              :     #[tokio::test]
     561            4 :     async fn test_arrays_page_boundary() -> Result<(), Error> {
     562            4 :         let blobs = &[
     563            4 :             random_array(PAGE_SZ - 4),
     564            4 :             random_array(PAGE_SZ - 4),
     565            4 :             random_array(PAGE_SZ - 4),
     566            4 :         ];
     567            4 :         round_trip_test::<false>(blobs).await?;
     568            4 :         round_trip_test::<true>(blobs).await?;
     569            4 :         Ok(())
     570            4 :     }
     571              : }
        

Generated by: LCOV version 2.1-beta