Using an unrolled linked list to halve memory usage

src/lib.rs

@@ -1,4 +1,5 @@
 use std::borrow::Cow;
+use std::sync::Arc;
 
 mod block_read_write;
 
@@ -28,6 +29,56 @@ impl<'a> Record<'a> {
     }
 }
 
+#[derive(Clone, Default, Debug, Ord, PartialOrd, Eq, PartialEq)]
+pub struct FileNumber {
+    file_number: Arc<u64>,
+}
+
+impl FileNumber {
+    fn new(file_number: u64) -> Self {
+        FileNumber {
+            file_number: Arc::new(file_number),
+        }
+    }
+
+    /// Returns whether there is no clone of this FileNumber in existance.
+    ///
+    /// /!\ care should be taken to not have some other code store a &FileNumber which could alias
+    /// with self as it might then be sementically incorrect to delete content based only on this
+    /// returning `true`.
+    pub fn can_be_deleted(&self) -> bool {
+        Arc::strong_count(&self.file_number) == 1
+    }
+
+    #[cfg(test)]
+    pub fn unroll(&self, tracker: &crate::rolling::FileTracker) -> Vec<u64> {
+        let mut file = self.clone();
+        let mut file_numbers = Vec::new();
+        loop {
+            file_numbers.push(file.file_number());
+            if let Some(next_file) = tracker.next(&file) {
+                file = next_file;
+            } else {
+                return file_numbers;
+            }
+        }
+    }
+
+    pub fn filename(&self) -> String {
+        format!("wal-{:020}", self.file_number)
+    }
+
+    #[cfg(test)]
+    pub fn file_number(&self) -> u64 {
+        *self.file_number
+    }
+
+    #[cfg(test)]
+    pub fn for_test(file_number: u64) -> Self {
+        FileNumber::new(file_number)
+    }
+}
+
 /// Resources used by mrecordlog
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct ResourceUsage {

src/mem/arena.rs

@@ -0,0 +1,110 @@
+pub const PAGE_SIZE: usize = 1 << 20;
+
+// TODO make it an array once we get a way to allocate array on the heap.
+pub type Page = Box<[u8]>;
+
+#[derive(Clone, Copy, Eq, PartialEq, Debug)]
+pub struct PageId(usize);
+
+/// An arena of fixed sized pages.
+#[derive(Default)]
+pub struct Arena {
+    /// We use an array to store the list of pages.
+    /// It can be seen as an efficient map from page id to pages.
+    ///
+    /// This map's len can-only grows. Its size is therefore the maximum number of pages
+    /// that was ever allocated. One page being 1MB long, this is not a problem.
+    ///
+    /// If a page is not allocated, the corresponding entry is `None`.
+    pages: Vec<Option<Page>>,
+    /// `free_slots` slots keeps track of the pages that are not allocated.
+    free_slots: Vec<PageId>,
+    /// `free_page_ids` keeps track of the allocated pages that are
+    /// available.
+    free_page_ids: Vec<PageId>,
+}
+
+impl Arena {
+    /// Returns an allocated page id.
+    pub fn get_page_id(&mut self) -> PageId {
+        if let Some(page_id) = self.free_page_ids.pop() {
+            assert!(self.pages[page_id.0].is_some());
+            return page_id;
+        }
+        let page: Page = vec![0u8; PAGE_SIZE].into_boxed_slice();
+        if let Some(free_slot) = self.free_slots.pop() {
+            let slot = &mut self.pages[free_slot.0];
+            assert!(slot.is_none());
+            *slot = Some(page);
+            return free_slot;
+        } else {
+            let new_page_id = self.pages.len();
+            self.pages.push(Some(page));
+            PageId(new_page_id)
+        }
+    }
+
+    #[inline]
+    pub fn page(&self, page_id: PageId) -> &[u8] {
+        self.pages[page_id.0].as_ref().unwrap()
+    }
+
+    #[inline]
+    pub fn page_mut(&mut self, page_id: PageId) -> &mut [u8] {
+        self.pages[page_id.0].as_mut().unwrap()
+    }
+
+    pub fn release_page(&mut self, page_id: PageId) {
+        self.free_page_ids.push(page_id);
+        assert!(self.pages[page_id.0].is_some());
+        if self.free_page_ids.len() > self.num_allocated_pages() {
+            self.gc();
+        }
+    }
+
+    /// `gc` releases memory by deallocating ALL of the free pages.
+    pub fn gc(&mut self) {
+        for free_page_id in self.free_page_ids.drain(..) {
+            self.pages[free_page_id.0] = None;
+            self.free_slots.push(free_page_id);
+        }
+    }
+
+    pub fn num_allocated_pages(&self) -> usize {
+        self.pages.len() - self.free_slots.len()
+    }
+
+    pub fn capacity(&self) -> usize {
+        self.num_allocated_pages() * PAGE_SIZE
+    }
+}
+
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_arena_simple() {
+        let mut arena = Arena::default();
+        assert_eq!(arena.capacity(), 0);
+        assert_eq!(arena.get_page_id(), PageId(0));
+        assert_eq!(arena.get_page_id(), PageId(1));
+        arena.release_page(PageId(0));
+        assert_eq!(arena.get_page_id(), PageId(0));
+    }
+
+    #[test]
+    fn test_arena_gc() {
+        let mut arena = Arena::default();
+        assert_eq!(arena.capacity(), 0);
+        assert_eq!(arena.get_page_id(), PageId(0));
+        assert_eq!(arena.get_page_id(), PageId(1));
+        arena.release_page(PageId(1));
+        assert_eq!(arena.num_allocated_pages(), 2);
+        arena.gc();
+        assert_eq!(arena.num_allocated_pages(), 1);
+        assert_eq!(arena.get_page_id(), PageId(1));
+        assert_eq!(arena.num_allocated_pages(), 2);
+    }
+}

src/mem/mod.rs

@@ -1,6 +1,9 @@
 mod queue;
 mod queues;
+mod arena;
 
+
+use self::arena::{Arena, PAGE_SIZE};
 pub(crate) use self::queue::MemQueue;
 pub(crate) use self::queues::MemQueues;