Moving toward a paging-based recordlog

Cargo.toml

@@ -12,9 +12,10 @@ rust-version = "1.68"                         # 1.67 contains an UB we would tri
 [dependencies]
 bytes = "1"
 crc32fast = "1.2"
-serde = { version = "1", features = ["derive"] }
 thiserror = "2"
 tracing = "0.1.37"
+arc-swap = "1.7"
+rclite = "0.2"
 
 [dev-dependencies]
 criterion = "0.5"
@@ -26,3 +27,7 @@ tempfile = "3"
 [[bench]]
 name = "bench"
 harness = false
+
+# Uncomment for profiling
+# [profile.bench]
+# debug=true

TODO

@@ -0,0 +1,12 @@
+we do not need to record position anymore. Keeping the last record is sufficient.
+Of course it would be nicer to store that in a separate way to avoid fragmentation.
+
+Record refcount
+
+No GC however means on restart we will often have to read way more than what is actually required.
+
+We need a way to stop reading the past.
+
+Truncating a non existing queue.
+
+just truncate to record position.

src/block_read_write.rs

@@ -4,14 +4,22 @@ use crate::PersistAction;
 
 pub const BLOCK_NUM_BYTES: usize = 32_768;
 
+/// A block read is supposed to be positioned on a block at its initialization.
+///
+/// In other words, it is not necessary to call `next_block` a first time
+/// before calling `block()`.
 pub trait BlockRead {
+    type Session;
+
+    fn start_session(&self) -> Self::Session;
+
     /// Loads the next block.
     /// If `Ok(true)` is returned, the new block is available through
     /// `.block()`.
     ///
     /// If `Ok(false)` is returned, the end of the `BlockReader`
     /// has been reached and the content of `block()` could be anything.
-    fn next_block(&mut self) -> io::Result<bool>;
+    fn next_block(&mut self, read_session: &mut Self::Session) -> io::Result<bool>;
 
     /// A `BlockReader` is always position on a specific block.
     ///
@@ -25,8 +33,17 @@ pub trait BlockRead {
 }
 
 pub trait BlockWrite {
+    type Session;
+
+    fn start_write_session(&mut self) -> io::Result<Self::Session>;
+
+    fn make_room(&mut self, num_bytes: u64) -> io::Result<()>;
+
     /// Must panic if buf is larger than `num_bytes_remaining_in_block`.
-    fn write(&mut self, buf: &[u8]) -> io::Result<()>;
+    /// Not that this trait does not have next_block() method.
+    ///
+    /// We automatically go to the next block after the current block has been entirely written.
+    fn write(&mut self, buf: &[u8], write_session: &mut Self::Session) -> io::Result<()>;
     /// Persist the data following the `persist_action`.
     fn persist(&mut self, persist_action: PersistAction) -> io::Result<()>;
     /// Number of bytes that can be added in the block.
@@ -49,7 +66,11 @@ impl<'a> From<&'a [u8]> for ArrayReader<'a> {
 }
 
 impl BlockRead for ArrayReader<'_> {
-    fn next_block(&mut self) -> io::Result<bool> {
+    type Session = ();
+
+    fn start_session(&self) -> Self::Session {}
+
+    fn next_block(&mut self, _session: &mut Self::Session) -> io::Result<bool> {
         if self.data.len() < BLOCK_NUM_BYTES {
             return Ok(false);
         }
@@ -81,7 +102,14 @@ impl From<VecBlockWriter> for Vec<u8> {
 }
 
 impl BlockWrite for VecBlockWriter {
-    fn write(&mut self, buf: &[u8]) -> io::Result<()> {
+    type Session = ();
+    fn make_room(&mut self, num_bytes: u64) -> io::Result<()> {
+        // TODO consider just doubling for performance.
+        let new_len = ceil_to_block(self.cursor + num_bytes as usize);
+        self.buffer.resize(new_len, 0u8);
+        Ok(())
+    }
+    fn write(&mut self, buf: &[u8], _session: &mut Self::Session) -> io::Result<()> {
         assert!(buf.len() <= self.num_bytes_remaining_in_block());
         if self.cursor + buf.len() > self.buffer.len() {
             let new_len = ceil_to_block((self.cursor + buf.len()) * 2 + 1);
@@ -99,4 +127,8 @@ impl BlockWrite for VecBlockWriter {
     fn num_bytes_remaining_in_block(&self) -> usize {
         BLOCK_NUM_BYTES - (self.cursor % BLOCK_NUM_BYTES)
     }
+
+    fn start_write_session(&mut self) -> io::Result<Self::Session> {
+        Ok(())
+    }
 }

src/error.rs

@@ -91,3 +91,13 @@ pub enum ReadRecordError {
     #[error("Corruption")]
     Corruption,
 }
+
+#[derive(Error, Debug)]
+pub enum HeaderError {
+    #[error("invalid magic number: found {magic_number}")]
+    InvalidMagicNumber { magic_number: u32 },
+    #[error("invalid checksum")]
+    InvalidChecksum,
+    #[error("unsupported version: {version}")]
+    UnsupportedVersion { version: u32 },
+}

src/frame/reader.rs

@@ -2,15 +2,14 @@ use std::io;
 
 use thiserror::Error;
 
-use crate::frame::{FrameType, FrameWriter, Header, HEADER_LEN};
-use crate::rolling::{RollingReader, RollingWriter};
+use crate::frame::{FrameType, Header, HEADER_LEN};
 use crate::{BlockRead, BLOCK_NUM_BYTES};
 
 pub struct FrameReader<R> {
-    reader: R,
+    pub(crate) reader: R,
 
     /// In block cursor
-    cursor: usize,
+    pub(crate) cursor: usize,
 
     // The current block is corrupted.
     block_corrupted: bool,
@@ -35,8 +34,8 @@ impl<R: BlockRead + Unpin> FrameReader<R> {
         }
     }
 
-    pub fn read(&self) -> &R {
-        &self.reader
+    pub fn start_session(&self) -> R::Session {
+        self.reader.start_session()
     }
 
     // Returns the number of bytes remaining into
@@ -47,13 +46,16 @@ impl<R: BlockRead + Unpin> FrameReader<R> {
         crate::BLOCK_NUM_BYTES - self.cursor
     }
 
-    fn go_to_next_block_if_necessary(&mut self) -> Result<(), ReadFrameError> {
+    fn go_to_next_block_if_necessary(
+        &mut self,
+        session: &mut R::Session,
+    ) -> Result<(), ReadFrameError> {
         let num_bytes_to_end_of_block = self.num_bytes_to_end_of_block();
         let need_to_skip_block = self.block_corrupted || num_bytes_to_end_of_block < HEADER_LEN;
         if !need_to_skip_block {
             return Ok(());
         }
-        if !self.reader.next_block()? {
+        if !self.reader.next_block(session)? {
             return Err(ReadFrameError::NotAvailable);
         }
 
@@ -79,8 +81,11 @@ impl<R: BlockRead + Unpin> FrameReader<R> {
     }
 
     // Reads the next frame.
-    pub fn read_frame(&mut self) -> Result<(FrameType, &[u8]), ReadFrameError> {
-        self.go_to_next_block_if_necessary()?;
+    pub fn read_frame(
+        &mut self,
+        session: &mut R::Session,
+    ) -> Result<(FrameType, &[u8]), ReadFrameError> {
+        self.go_to_next_block_if_necessary(session)?;
         let header = self.get_frame_header()?;
         self.cursor += HEADER_LEN;
         if self.cursor + header.len() > BLOCK_NUM_BYTES {
@@ -103,11 +108,3 @@ impl<R: BlockRead + Unpin> FrameReader<R> {
         Ok((header.frame_type(), frame_payload))
     }
 }
-
-impl FrameReader<RollingReader> {
-    pub fn into_writer(self) -> io::Result<FrameWriter<RollingWriter>> {
-        let mut rolling_writer: RollingWriter = self.reader.into_writer()?;
-        rolling_writer.forward(self.cursor)?;
-        Ok(FrameWriter::create(rolling_writer))
-    }
-}

src/frame/tests.rs

@@ -3,39 +3,42 @@
 use crate::block_read_write::{ArrayReader, VecBlockWriter};
 use crate::frame::header::{FrameType, HEADER_LEN};
 use crate::frame::{FrameReader, FrameWriter, ReadFrameError};
-use crate::{PersistAction, BLOCK_NUM_BYTES};
+use crate::{BlockWrite as _, PersistAction, BLOCK_NUM_BYTES};
 
 #[test]
 fn test_frame_simple() {
     let block_writer = {
-        let wrt: VecBlockWriter = VecBlockWriter::default();
+        let mut wrt: VecBlockWriter = VecBlockWriter::default();
+        let mut session = wrt.start_write_session().unwrap();
         let mut frame_writer = FrameWriter::create(wrt);
+
         frame_writer
-            .write_frame(FrameType::First, &b"abc"[..])
+            .write_frame(FrameType::First, &b"abc"[..], &mut session)
             .unwrap();
         frame_writer
-            .write_frame(FrameType::Middle, &b"de"[..])
+            .write_frame(FrameType::Middle, &b"de"[..], &mut session)
             .unwrap();
         frame_writer
-            .write_frame(FrameType::Last, &b"fgh"[..])
+            .write_frame(FrameType::Last, &b"fgh"[..], &mut session)
             .unwrap();
         frame_writer.persist(PersistAction::Flush).unwrap();
         frame_writer.into_writer()
     };
     let buffer: Vec<u8> = block_writer.into();
     let mut frame_reader = FrameReader::open(ArrayReader::from(&buffer[..]));
-    let read_frame_res = frame_reader.read_frame();
+    let mut session = frame_reader.start_session();
+    let read_frame_res = frame_reader.read_frame(&mut session);
     assert_eq!(read_frame_res.unwrap(), (FrameType::First, &b"abc"[..]));
     assert_eq!(
-        frame_reader.read_frame().unwrap(),
+        frame_reader.read_frame(&mut session).unwrap(),
         (FrameType::Middle, &b"de"[..])
     );
     assert_eq!(
-        frame_reader.read_frame().unwrap(),
+        frame_reader.read_frame(&mut session).unwrap(),
         (FrameType::Last, &b"fgh"[..])
     );
     assert!(matches!(
-        frame_reader.read_frame().unwrap_err(),
+        frame_reader.read_frame(&mut session).unwrap_err(),
         ReadFrameError::NotAvailable
     ));
 }
@@ -44,29 +47,34 @@
 fn test_frame_corruption_in_payload() -> io::Result<()> {
     let mut buf: Vec<u8> = {
         let mut frame_writer = FrameWriter::create(VecBlockWriter::default());
-        frame_writer.write_frame(FrameType::First, &b"abc"[..])?;
+        let mut session = frame_writer.start_session()?;
+        frame_writer.write_frame(FrameType::First, &b"abc"[..], &mut session)?;
         frame_writer.persist(PersistAction::Flush)?;
-        frame_writer.write_frame(FrameType::Middle, &b"de"[..])?;
+        frame_writer.write_frame(FrameType::Middle, &b"de"[..], &mut session)?;
         frame_writer.persist(PersistAction::Flush)?;
         frame_writer.into_writer().into()
     };
     buf[8] = 0u8;
     let mut frame_reader = FrameReader::open(ArrayReader::from(&buf[..]));
+    let mut session = frame_reader.start_session();
     assert!(matches!(
-        frame_reader.read_frame(),
+        frame_reader.read_frame(&mut session),
         Err(ReadFrameError::Corruption)
     ));
     assert!(matches!(
-        frame_reader.read_frame(),
+        frame_reader.read_frame(&mut session),
         Ok((FrameType::Middle, b"de"))
     ));
     Ok(())
 }
 
 fn repeat_empty_frame_util(repeat: usize) -> Vec<u8> {
     let mut frame_writer = FrameWriter::create(VecBlockWriter::default());
+    let mut session = frame_writer.start_session().unwrap();
     for _ in 0..repeat {
-        frame_writer.write_frame(FrameType::Full, &b""[..]).unwrap();
+        frame_writer
+            .write_frame(FrameType::Full, &b""[..], &mut session)
+            .unwrap();
     }
     frame_writer.persist(PersistAction::Flush).unwrap();
     frame_writer.into_writer().into()
@@ -77,12 +85,13 @@
     let num_frames = 1 + BLOCK_NUM_BYTES / HEADER_LEN;
     let buffer = repeat_empty_frame_util(num_frames);
     let mut frame_reader = FrameReader::open(ArrayReader::from(&buffer[..]));
+    let mut session = frame_reader.start_session();
     for _ in 0..num_frames {
-        let read_frame_res = frame_reader.read_frame();
+        let read_frame_res = frame_reader.read_frame(&mut session);
         assert!(matches!(read_frame_res, Ok((FrameType::Full, &[]))));
     }
     assert!(matches!(
-        frame_reader.read_frame(),
+        frame_reader.read_frame(&mut &mut &mut &mut &mut &mut &mut &mut session),
         Err(ReadFrameError::NotAvailable)
     ));
     Ok(())
@@ -96,20 +105,21 @@
     let mut buffer = repeat_empty_frame_util(num_frames);
     buffer[2000 * HEADER_LEN + 5] = 255u8;
     let mut frame_reader = FrameReader::open(ArrayReader::from(&buffer[..]));
+    let mut session = frame_reader.start_session();
     for _ in 0..2000 {
-        let read_frame_res = frame_reader.read_frame();
+        let read_frame_res = frame_reader.read_frame(&mut session);
         assert!(matches!(read_frame_res, Ok((FrameType::Full, &[]))));
     }
     assert!(matches!(
-        frame_reader.read_frame(),
+        frame_reader.read_frame(&mut session),
         Err(ReadFrameError::Corruption)
     ));
     assert!(matches!(
-        frame_reader.read_frame(),
+        frame_reader.read_frame(&mut session),
         Ok((FrameType::Full, &[]))
     ));
     assert!(matches!(
-        frame_reader.read_frame(),
+        frame_reader.read_frame(&mut session),
         Err(ReadFrameError::NotAvailable)
     ));
     Ok(())