Recovery

src/rust/config/src/seg.rs

@@ -8,6 +8,10 @@ use serde::{Deserialize, Serialize};
 
 const MB: usize = 1024 * 1024;
 
+// restore and graceful shutdown options
+const RESTORE: bool = false;
+const GRACEFUL_SHUTDOWN: bool = false;
+
 // defaults for hashtable
 const HASH_POWER: u8 = 16;
 const OVERFLOW_FACTOR: f64 = 1.0;
@@ -24,9 +28,18 @@ const COMPACT_TARGET: usize = 2;
 const MERGE_TARGET: usize = 4;
 const MERGE_MAX: usize = 8;
 
-// datapool
+// datapool (`Segments.data`)
 const DATAPOOL_PATH: Option<&str> = None;
 
+// `Segments` fields
+const SEGMENT_FIELDS_PATH: Option<&str> = None;
+
+// ttl buckets
+const TTL_BUCKETS_PATH: Option<&str> = None;
+
+// hashtable
+const HASHTABLE_PATH: Option<&str> = None;
+
 #[derive(Copy, Clone, Debug, Serialize, Deserialize)]
 pub enum Eviction {
     None,
@@ -39,6 +52,14 @@ pub enum Eviction {
 }
 
 // helper functions for default values
+fn restore() -> bool {
+    RESTORE
+}
+
+fn graceful_shutdown() -> bool {
+    GRACEFUL_SHUTDOWN
+}
+
 fn hash_power() -> u8 {
     HASH_POWER
 }
@@ -75,9 +96,25 @@ fn datapool_path() -> Option<String> {
     DATAPOOL_PATH.map(|v| v.to_string())
 }
 
+fn segments_fields_path() -> Option<String> {
+    SEGMENT_FIELDS_PATH.map(|v| v.to_string())
+}
+
+fn ttl_buckets_path() -> Option<String> {
+    TTL_BUCKETS_PATH.map(|v| v.to_string())
+}
+
+fn hashtable_path() -> Option<String> {
+    HASHTABLE_PATH.map(|v| v.to_string())
+}
+
 // definitions
 #[derive(Serialize, Deserialize, Debug)]
 pub struct Seg {
+    #[serde(default = "restore")]
+    restore: bool,
+    #[serde(default = "graceful_shutdown")]
+    graceful_shutdown: bool,
     #[serde(default = "hash_power")]
     hash_power: u8,
     #[serde(default = "overflow_factor")]
@@ -96,11 +133,19 @@ pub struct Seg {
     compact_target: usize,
     #[serde(default = "datapool_path")]
     datapool_path: Option<String>,
+    #[serde(default = "segments_fields_path")]
+    segments_fields_path: Option<String>,
+    #[serde(default = "ttl_buckets_path")]
+    ttl_buckets_path: Option<String>,
+    #[serde(default = "hashtable_path")]
+    hashtable_path: Option<String>,
 }
 
 impl Default for Seg {
     fn default() -> Self {
         Self {
+            restore: restore(),
+            graceful_shutdown: graceful_shutdown(),
             hash_power: hash_power(),
             overflow_factor: overflow_factor(),
             heap_size: heap_size(),
@@ -110,12 +155,31 @@ impl Default for Seg {
             merge_max: merge_max(),
             compact_target: compact_target(),
             datapool_path: datapool_path(),
+            segments_fields_path: segments_fields_path(),
+            ttl_buckets_path: ttl_buckets_path(),
+            hashtable_path: hashtable_path(),
         }
     }
 }
 
 // implementation
 impl Seg {
+    // Determines if the `Seg` will be restored.
+    // The restoration will be successful if `datapool_path`, `segments_fields_path`
+    // `ttl_buckets_path` and `hashtable_path` are valid paths.
+    // Otherwise, the `Seg` will be created as new.
+    pub fn restore(&self) -> bool {
+        self.restore
+    }
+
+    // Determines if the `Seg` will be gracefully shutdown.
+    // The graceful shutdown will be successful if the cache is file backed
+    // and `segments_fields_path`, `ttl_buckets_path` and `hashtable_path` are
+    // valid paths to save the relevant `Seg` fields to.
+    // Otherwise, the relevant `Seg` fields will not be saved.
+    pub fn graceful_shutdown(&self) -> bool {
+        self.graceful_shutdown
+    }
     pub fn hash_power(&self) -> u8 {
         self.hash_power
     }
@@ -151,6 +215,24 @@ impl Seg {
     pub fn datapool_path(&self) -> Option<PathBuf> {
         self.datapool_path.as_ref().map(|v| Path::new(v).to_owned())
     }
+
+    pub fn segments_fields_path(&self) -> Option<PathBuf> {
+        self.segments_fields_path
+            .as_ref()
+            .map(|v| Path::new(v).to_owned())
+    }
+
+    pub fn ttl_buckets_path(&self) -> Option<PathBuf> {
+        self.ttl_buckets_path
+            .as_ref()
+            .map(|v| Path::new(v).to_owned())
+    }
+
+    pub fn hashtable_path(&self) -> Option<PathBuf> {
+        self.hashtable_path
+            .as_ref()
+            .map(|v| Path::new(v).to_owned())
+    }
 }
 
 // trait definitions

src/rust/entrystore/src/seg/mod.rs

@@ -43,16 +43,30 @@ impl Seg {
 
         // build the datastructure from the config
         let data = ::seg::Seg::builder()
+            .restore(config.restore())
             .hash_power(config.hash_power())
             .overflow_factor(config.overflow_factor())
             .heap_size(config.heap_size())
             .segment_size(config.segment_size())
             .eviction(eviction)
             .datapool_path(config.datapool_path())
+            .segments_fields_path(config.segments_fields_path())
+            .ttl_buckets_path(config.ttl_buckets_path())
+            .hashtable_path(config.hashtable_path())
             .build();
 
         Self { data }
     }
+
+    /// Flush (gracefully shutdown) the `Seg` cache if configured to do so
+    pub fn flush<T: SegConfig>(self, config: &T) {
+        let config = config.seg();
+
+        if config.graceful_shutdown() {
+            // TODO: check if successfully shutdown and record result
+            self.data.flush();
+        };
+    }
 }
 
 impl EntryStore for Seg {

src/rust/server/segcache/src/lib.rs

@@ -71,6 +71,7 @@ impl Segcache {
     /// fully terminated. This is more likely to be used for running integration
     /// tests or other automated testing.
     pub fn shutdown(self) {
+        // TODO: flush the cache
         self.process.shutdown()
     }
 }

src/rust/storage/seg/Cargo.toml

@@ -35,6 +35,7 @@ rand_chacha = { version = "0.3.0" }
 rand_xoshiro = { version = "0.6.0" }
 storage-types = { path = "../types" }
 thiserror = "1.0.24"
+tempfile = "3.3.0"
 
 [dev-dependencies]
 criterion = "0.3.4"

src/rust/storage/seg/src/builder.rs

@@ -6,26 +6,41 @@
 
 use crate::*;
 use std::path::Path;
+use std::path::PathBuf;
 
 /// A builder that is used to construct a new [`Seg`] instance.
 pub struct Builder {
+    restore: bool,
     hash_power: u8,
     overflow_factor: f64,
     segments_builder: SegmentsBuilder,
+    ttl_buckets_path: Option<PathBuf>,
+    hashtable_path: Option<PathBuf>,
 }
 
 // Defines the default parameters
 impl Default for Builder {
     fn default() -> Self {
         Self {
+            restore: false,
             hash_power: 16,
             overflow_factor: 0.0,
             segments_builder: SegmentsBuilder::default(),
+            ttl_buckets_path: None,
+            hashtable_path: None,
         }
     }
 }
 
 impl Builder {
+    /// Specify to `Builder` and `SegmentsBuilder` whether the cache will be restored.
+    /// Otherwise, the cache will be created and treated as new.
+    pub fn restore(mut self, restore: bool) -> Self {
+        self.restore = restore;
+        self.segments_builder = self.segments_builder.restore(restore);
+        self
+    }
+
     /// Specify the hash power, which limits the size of the hashtable to 2^N
     /// entries. 1/8th of these are used for metadata storage, meaning that the
     /// total number of items which can be held in the cache is limited to
@@ -135,17 +150,35 @@ impl Builder {
         self
     }
 
-    /// Specify a backing file to be used for segment storage.
-    ///
-    /// # Panics
-    ///
-    /// This will panic if the file already exists
+    /// Specify a backing file to be used for `Segments.data` storage.
     pub fn datapool_path<T: AsRef<Path>>(mut self, path: Option<T>) -> Self {
         self.segments_builder = self.segments_builder.datapool_path(path);
         self
     }
 
+    /// Specify a backing file to be used for `Segments` fields' storage.
+    pub fn segments_fields_path<T: AsRef<Path>>(mut self, path: Option<T>) -> Self {
+        self.segments_builder = self.segments_builder.segments_fields_path(path);
+        self
+    }
+
+    /// Specify a backing file to be used for `TtlBuckets` storage.
+    pub fn ttl_buckets_path<T: AsRef<Path>>(mut self, path: Option<T>) -> Self {
+        self.ttl_buckets_path = path.map(|p| p.as_ref().to_owned());
+        self
+    }
+
+    /// Specify a backing file to be used for `HashTable` storage.
+    pub fn hashtable_path<T: AsRef<Path>>(mut self, path: Option<T>) -> Self {
+        self.hashtable_path = path.map(|p| p.as_ref().to_owned());
+        self
+    }
+
     /// Consumes the builder and returns a fully-allocated `Seg` instance.
+    /// If `restore` and valid paths to the structures are given, `Seg` will
+    /// be restored. Otherwise, create a new `Seg` instance. If valid paths are
+    /// given, the files at these paths will be used to copy the structures to
+    /// upon graceful shutdown.
     ///
     /// ```
     /// use seg::{Policy, Seg};
@@ -159,10 +192,44 @@ impl Builder {
     ///     .eviction(Policy::Random).build();
     /// ```
     pub fn build(self) -> Seg {
-        let hashtable = HashTable::new(self.hash_power, self.overflow_factor);
+        // Build `Segments`. If there is a path for the datapool set, the
+        // `Segments.data` will be file backed. If `restore` and there is a path
+        // for the `Segments` fields, restore the other relevant `Segments`
+        // fields.
         let segments = self.segments_builder.build();
-        let ttl_buckets = TtlBuckets::default();
 
+        // If `Segments` successfully restored and `restore`
+        if segments.fields_copied_back && self.restore {
+            // Attempt to restore `HashTable` and `TtlBuckets`
+            let hashtable = HashTable::restore(
+                self.hashtable_path.clone(),
+                self.hash_power,
+                self.overflow_factor,
+            );
+            let ttl_buckets = TtlBuckets::restore(self.ttl_buckets_path.clone());
+
+            // If successful, return a restored segcache
+            if hashtable.table_copied_back && ttl_buckets.buckets_copied_back {
+                return Seg {
+                    hashtable,
+                    segments,
+                    ttl_buckets,
+                };
+            }
+        }
+
+        // TODO: Should paths be checked here to see if any are None (or not
+        // valid)? Then we could take an "All or Nothing" approach. That is, if
+        // one of the paths is not valid, then all structures are created
+        // as new AND no paths are set for graceful shutdown. Otherwise, if
+        // `restore`, we restore from these paths, else, we set these paths.
+        // Currently, I am not doing this as due to the Segments having a
+        // separate builder + different control flow, it is too awkward to
+        // implement.
+
+        // If not `restore` or restoration failed, create a new cache
+        let hashtable = HashTable::new(self.hashtable_path, self.hash_power, self.overflow_factor);
+        let ttl_buckets = TtlBuckets::new(self.ttl_buckets_path);
         Seg {
             hashtable,
             segments,