WIP: more implementation of LinearStore

Checkpoint:
 - Renamed ReadOnlyLinearStore to ReadLinearStore since
   ReadWriteLinearStore must also implement it
 - Renamed ReadWriteLinearStore to WriteLinearStore since all writeable
   linear stores must implement ReadLinearStore
 - Required NodeStore to have a ReadLinearStore type of LinearStore.
   This was necessary because we had to call the stream_from method
 - The LayeredReader now compiles but needs more work

firewood/src/storage/linear/committed.rs

@@ -4,13 +4,13 @@
 use std::collections::BTreeMap;
 use std::sync::Arc;
 
-use super::LinearStore;
+use super::{LinearStore, ReadLinearStore};
 
 /// A linear store used for historical revisions
 ///
 /// A [Committed] [LinearStore] supports read operations only
 #[derive(Debug)]
-struct Committed<P> {
+struct Committed<P: ReadLinearStore> {
     old: BTreeMap<u64, Box<[u8]>>,
     parent: Arc<LinearStore<P>>,
 }

firewood/src/storage/linear/filebacked.rs

@@ -9,28 +9,28 @@
 // object. Instead, we probably should use an IO system that can perform multiple
 // read/write operations at once
 
-use super::{ReadOnlyLinearStore, ReadWriteLinearStore};
+use super::{ReadLinearStore, WriteLinearStore};
 use std::fs::File;
 use std::io::{Error, Read, Seek};
 use std::os::unix::fs::FileExt;
 use std::path::PathBuf;
 use std::sync::Mutex;
 
 #[derive(Debug)]
-struct FileBacked {
+pub(super) struct FileBacked {
     path: PathBuf,
     fd: Mutex<File>,
 }
 
-impl ReadOnlyLinearStore for FileBacked {
+impl ReadLinearStore for FileBacked {
     fn stream_from(&self, addr: u64) -> Result<impl Read, Error> {
         let mut fd = self.fd.lock().expect("p");
         fd.seek(std::io::SeekFrom::Start(addr))?;
         Ok(fd.try_clone().expect("poisoned lock"))
     }
 }
 
-impl ReadWriteLinearStore for FileBacked {
+impl WriteLinearStore for FileBacked {
     fn write(&mut self, offset: u64, object: &[u8]) -> Result<usize, Error> {
         self.fd
             .lock()

firewood/src/storage/linear/mod.rs

@@ -4,8 +4,8 @@
 // TODO: remove this once we have code that uses it
 #![allow(dead_code)]
 
-//! A [LinearStore] provides a view of a set of bytes at
-//! a given time. A [LinearStore] has three different types,
+//! A LinearStore provides a view of a set of bytes at
+//! a given time. A LinearStore has three different types,
 //! which refer to another base type, as follows:
 //! ```mermaid
 //! stateDiagram-v2
@@ -18,12 +18,12 @@
 //! Each type is described in more detail below.
 
 use std::fmt::Debug;
-use std::io::{Cursor, Error, Read};
+use std::io::{Error, Read};
 
 mod committed;
 /// A linear store used for proposals
 ///
-/// A [Proposed] [LinearStore] supports read operations which look for the
+/// A Proposed LinearStore supports read operations which look for the
 /// changed bytes in the `new` member, and if not present, delegate to
 /// their parent.
 ///
@@ -38,59 +38,51 @@ mod committed;
 /// not. Mutable proposals implement the [ReadWriteLinearStore] trait
 ///
 /// The possible combinations are:
-///  - `Proposed<Proposed, ReadWrite>`
-///  - `Proposed<Proposed, ReadOnly>`
-///  - `Proposed<FileBacked, ReadWrite>`
-///  - `Proposed<FileBacked, ReadOnly>`
+///  - `Proposed<Proposed, ReadWrite>` (in-progress nested proposal)
+///  - `Proposed<Proposed, ReadOnly>` (completed nested proposal)
+///  - `Proposed<FileBacked, ReadWrite>` (first proposal on base revision)
+///  - `Proposed<FileBacked, ReadOnly>` (completed first proposal on base)
 ///
 /// Transitioning from ReadWrite to ReadOnly just prevents future mutations to
-/// the proposal maps
+/// the proposal maps. ReadWrite proposals only exist during the application of
+/// a Batch to the proposal, and are subsequently changed to ReadOnly
 ///
-// As an example of how this works, lets say that you have a [FileBased] Layer
-// base: [0, 1, 2, 3, 4, 5, 6, 7, 8] <-- this is on disk AKA R2
-//   us: [0, 1, 2, 3, 4, 6, 6, 7, 9, 10] <-- part of a proposal AKA P1 this is the "virtual" state. Not stored in entirety in this struct.
-// changes: {old: 5:[5] 8:[8]}, new: {5:[6] : 8:[9 10]}   <-- this is what is actually is stored in this struct (in `changes`).
-// start streaming from index 4
-// first byte I read comes from base
-// second byte I read comes from change
-// third byte I read comes from base
+/// # How a commit works
+/// 
+/// Lets assume we have the following:
+///  - bytes "on disk":   [0, 1, 2] `LinearStore<FileBacked>`
+///  - bytes in proposal: [   3   ] `LinearStore<Proposed<FileBacked, ReadOnly>>`
+/// that is, we're changing the second byte (1) to (3)
+/// 
+/// To commit:
+///  - Convert the `LinearStore<FileBacked>` to `LinearStore<Committed>` taking the
+/// old pages from the `LinearStore<Proposed<FileBacked, Readonly>>`
+///  - Change any direct child proposals from `LinearStore<Proposed<Proposed, Readonly>>`
+/// into `LinearStore<FileBacked>`
+///  - Invalidate any other `LinearStore` that is a child of `LinearStore<FileBacked>`
+///  - Flush all the `Proposed<FileBacked, ReadOnly>::new` bytes to disk
+///  - Convert the `LinearStore<Proposed<FileBacked, Readonly>>` to `LinearStore<FileBacked>`
 
-// commit means: what is on disk is no longer actually R2 and vice versa
-// commit means: what is about to be on disk is in the changes list of P1 (specificially new regions)
-// commit means: what was on disk is in the changes list of P1 for changed byte indices of P1 (specificaly old regions)
-//  Every byte index is one of:
-// * Unchanged in P1
-// * Changed in P1
-
-// read 1 byte from R1:
-//  -- check and see if R2 has an "old" change at this index, if so, use that
-//  -- check and see if P1 (R3) has an "old" change at this index, if so use that
-//  -- not: read from disk (cache)
-
-// physical structure - set of linear bytes
-//   -- set of physical changes A
-// logical structure - set of nodes interconnected
-//   -- set of logical changes B
 mod filebacked;
 mod proposed;
 
 #[derive(Debug)]
-pub(super) struct LinearStore<T> {
-    state: T,
+pub(super) struct LinearStore<S: ReadLinearStore> {
+    state: S,
 }
 
 /// All linearstores support reads
-pub(super) trait ReadOnlyLinearStore: Debug {
+pub(super) trait ReadLinearStore: Debug {
     fn stream_from(&self, addr: u64) -> Result<impl Read, Error>;
 }
 
 /// Some linear stores support updates
-pub(super) trait ReadWriteLinearStore: Debug {
+pub(super) trait WriteLinearStore: Debug {
     fn write(&mut self, offset: u64, object: &[u8]) -> Result<usize, Error>;
     fn size(&self) -> Result<u64, Error>;
 }
 
-impl<ReadWrite: Debug> ReadWriteLinearStore for LinearStore<ReadWrite> {
+impl<ReadWrite: ReadLinearStore + Debug> WriteLinearStore for LinearStore<ReadWrite> {
     fn write(&mut self, _offset: u64, _bytes: &[u8]) -> Result<usize, Error> {
         todo!()
     }
@@ -100,9 +92,8 @@ impl<ReadWrite: Debug> ReadWriteLinearStore for LinearStore<ReadWrite> {
     }
 }
 
-/// TODO: remove this once all the ReadOnlyLinearStore implementations are complete
-impl<T: std::fmt::Debug> ReadOnlyLinearStore for LinearStore<T> {
-    fn stream_from(&self, _addr: u64) -> Result<impl Read, Error> {
-        Ok(Cursor::new([]))
+impl<S: ReadLinearStore> ReadLinearStore for LinearStore<S> {
+    fn stream_from(&self, addr: u64) -> Result<impl Read, Error> {
+        self.state.stream_from(addr)
     }
 }

firewood/src/storage/linear/proposed.rs

@@ -1,31 +1,39 @@
 use std::collections::BTreeMap;
-use std::io::Read;
+use std::fmt::Debug;
+use std::io::{Error, Read};
 use std::marker::PhantomData;
 use std::sync::Arc;
 
-use super::LinearStore;
+use super::{LinearStore, ReadLinearStore};
 
+/// [Proposed] is a [LinearStore] state that contains a copy of the old and new data.
+/// The P type parameter indicates the state of the linear store for it's parent,
+/// which could be a another [Proposed] or is [FileBacked](super::filebacked::FileBacked)
+/// The M type parameter indicates the mutability of the proposal, either read-write or readonly
 #[derive(Debug)]
-struct Proposed<P, T> {
+struct Proposed<P: ReadLinearStore, M> {
     new: BTreeMap<u64, Box<[u8]>>,
     old: BTreeMap<u64, Box<[u8]>>,
     parent: Arc<LinearStore<P>>,
-    phantom: PhantomData<T>,
+    phantom: PhantomData<M>,
 }
 
+
+/// A [LayeredReader] is obtained by calling [Proposed::stream_from]
+/// The P type parameter refers to the type of the parent of this layer
+/// The M type parameter is not specified here, but should always be
+/// read-only, since we do not support mutating parents of another
+/// proposal
 #[derive(Debug)]
-struct LayeredReader<'a, T> {
+struct LayeredReader<'a, P: ReadLinearStore, M> {
     offset: u64,
     state: LayeredReaderState<'a>,
-    layer: &'a LinearStore<T>,
-}
-
-impl<'a, T> Read for LayeredReader<'a, T> {
-    fn read(&mut self, _buf: &mut [u8]) -> std::io::Result<usize> {
-        todo!()
-    }
+    layer: &'a Proposed<P, M>,
 }
 
+/// A [LayeredReaderState] keeps track of when the next transition
+/// happens for a layer. If you attempt to read bytes past the
+/// transition, you'll get what is left, and the state will change
 #[derive(Debug)]
 enum LayeredReaderState<'a> {
     InitialState,
@@ -40,32 +48,55 @@ enum LayeredReaderState<'a> {
     NoMoreModifiedParts,
 }
 
-/*
-impl<'a, P, T> Read for LayeredReader<'a, Proposed<P, T>> {
+impl<P: ReadLinearStore, M: Debug> ReadLinearStore for Proposed<P, M> {
+    fn stream_from(&self, addr: u64) -> Result<impl Read, Error> {
+        Ok(LayeredReader {
+            offset: addr,
+            state: LayeredReaderState::InitialState,
+            layer: self,
+        })
+    }
+}
+
+// TODO: DiffResolver should also be implemented by Committed
+/// The [DiffResolver] trait indicates which field is used from the
+/// state of a [LinearStore] based on it's state.
+trait DiffResolver<'a> {
+    fn diffs(&'a self) -> &'a BTreeMap<u64, Box<[u8]>>;
+}
+
+impl<'a, P: ReadLinearStore, M: Debug> DiffResolver<'a> for Proposed<P, M> {
+    fn diffs(&'a self) -> &'a BTreeMap<u64, Box<[u8]>> {
+        &self.new
+    }
+}
+
+// TODO: This is not fully implemented as of this commit
+// it needs to work across Committed and Proposed types
+impl<'a, P: ReadLinearStore, M: Debug> Read for LayeredReader<'a, P, M> {
     fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
         match self.state {
             LayeredReaderState::InitialState => {
-
                 // figure out which of these cases is true:
                 //  a. We are inside a delta [LayeredReaderState::InsideModifiedPart]
                 //  b. We are before an upcoming delta [LayeredReaderState::BeforeModifiedPart]
                 //  c. We are past the last delta [LayeredReaderState::NoMoreModifiedParts]
                 self.state = 'state: {
                     // check for (a) - find the delta in front of (or at) our bytes offset
-                    if let Some(delta) = self.layer.state.new.range(..=self.offset).next_back() {
+                    if let Some(delta) = self.layer.diffs().range(..=self.offset).next_back() {
                         // see if the length of the change is inside our address
                         let delta_start = *delta.0;
                         let delta_end = delta_start + delta.1.len() as u64;
                         if self.offset >= delta_start && self.offset < delta_end {
                             // yes, we are inside a modified part
                             break 'state LayeredReaderState::InsideModifiedPart {
-                                part: &delta.1,
+                                part: delta.1,
                                 offset_within: (self.offset - delta_start) as usize,
                             };
                         }
                     }
                     // check for (b) - find the next delta and record it
-                    if let Some(delta) = self.layer.state.new.range(self.offset..).next() {
+                    if let Some(delta) = self.layer.diffs().range(self.offset..).next() {
                         // case (b) is true
                         LayeredReaderState::BeforeModifiedPart {
                             next_offset: *delta.0,
@@ -80,7 +111,7 @@ impl<'a, P, T> Read for LayeredReader<'a, Proposed<P, T>> {
             }
             LayeredReaderState::BeforeModifiedPart {
                 next_offset,
-                next_modified_part,
+                next_modified_part: _,
             } => {
                 // see how much we can read until we get to the next modified part
                 let remaining_passthrough: usize = (next_offset - self.offset) as usize;
@@ -94,25 +125,10 @@ impl<'a, P, T> Read for LayeredReader<'a, Proposed<P, T>> {
                 }
             }
             LayeredReaderState::InsideModifiedPart {
-                part,
-                offset_within,
+                part: _,
+                offset_within: _,
             } => todo!(),
             LayeredReaderState::NoMoreModifiedParts => todo!(),
         }
     }
 }
-
-
-
-impl<Type: Debug> ReadOnlyLinearStore for LinearStore<Type>
-where for<'a> LayeredReader<'a, Type>: Read + Debug {
-    fn stream_from(&self, addr: u64) -> Result<impl Read, Error> {
-        // Find out whether a change in this store touches
-        Ok(LayeredReader {
-            offset: addr,
-            state: LayeredReaderState::InitialState,
-            layer: self,
-        })
-    }
-}
-*/

firewood/src/storage/mod.rs

@@ -22,8 +22,8 @@ use tokio::sync::{mpsc::error::SendError, oneshot::error::RecvError};
 use typed_builder::TypedBuilder;
 
 pub mod buffer;
-pub mod node;
 pub mod linear;
+pub mod node;
 
 pub(crate) const PAGE_SIZE_NBIT: u64 = 12;
 pub(crate) const PAGE_SIZE: u64 = 1 << PAGE_SIZE_NBIT;

firewood/src/storage/node.rs

@@ -7,15 +7,14 @@
 /// free space management of nodes in the page store. It lays out the format
 /// of the [PageStore]. More specifically, it places a [FileIdentifyingMagic]
 /// and a [FreeSpaceHeader] at the beginning
-use std::fmt::Debug;
 use std::io::{Error, ErrorKind, Read, Write};
 use std::num::NonZeroU64;
 use std::sync::Arc;
 
 use enum_as_inner::EnumAsInner;
 use serde::{Deserialize, Serialize};
 
-use super::linear::{LinearStore, ReadOnlyLinearStore, ReadWriteLinearStore};
+use super::linear::{LinearStore, ReadLinearStore, WriteLinearStore};
 
 /// Either a branch or leaf node
 #[derive(PartialEq, Eq, Clone, Debug, EnumAsInner, Deserialize, Serialize)]
@@ -44,12 +43,12 @@ struct Leaf {
 }
 
 #[derive(Debug)]
-struct NodeStore<T> {
+struct NodeStore<T: ReadLinearStore> {
     header: FreeSpaceManagementHeader,
     page_store: LinearStore<T>,
 }
 
-impl<T: ReadOnlyLinearStore + std::fmt::Debug> NodeStore<T> {
+impl<T: ReadLinearStore> NodeStore<T> {
     /// Read a node from the provided [DiskAddress]
     ///
     /// A node on disk will consist of a header which both identifies the
@@ -74,7 +73,7 @@ impl<T: ReadOnlyLinearStore + std::fmt::Debug> NodeStore<T> {
     }
 }
 
-impl<T: ReadWriteLinearStore + ReadOnlyLinearStore + std::fmt::Debug> NodeStore<T> {
+impl<T: WriteLinearStore + ReadLinearStore> NodeStore<T> {
     /// Allocate space for a [Node] in the [PageStore]
     fn create(&mut self, node: &Node) -> Result<DiskAddress, Error> {
         let serialized =