refactor: move datapool out into separate crate (twitter#426)

Moves the datapool out from the seg storage crate and into its own
crate.

Adds a basic header to file-backed datapools and introduces a new
datapool for keeping content in memory with a file available to
save and restore state.

Minor refactoring to accomodate returning a result which allows us
to better surface any issues creating the datapool.

Adds a header to the datapools to allow versioning, validation, and
staleness checks.

Cargo.toml

@@ -17,6 +17,7 @@ members = [
     "src/rust/server/pingserver",
     "src/rust/server/segcache",
     "src/rust/session",
+    "src/rust/storage/datapool",
     "src/rust/storage/seg",
     "src/rust/storage/types",
 ]

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

@@ -23,7 +23,7 @@ pub struct Seg {
 impl Seg {
     /// Create `Seg` storage based on the config and the `TimeType` which is
     /// used to interpret various expiry time formats.
-    pub fn new<T: SegConfig>(config: &T) -> Self {
+    pub fn new<T: SegConfig>(config: &T) -> Result<Self, std::io::Error> {
         let config = config.seg();
 
         // build up the eviction policy from the config
@@ -49,9 +49,9 @@ impl Seg {
             .segment_size(config.segment_size())
             .eviction(eviction)
             .datapool_path(config.datapool_path())
-            .build();
+            .build()?;
 
-        Self { data }
+        Ok(Self { data })
     }
 }
 

src/rust/server/segcache/benches/benchmark.rs

@@ -25,7 +25,7 @@ fn get_benchmark(c: &mut Criterion) {
     let config = SegcacheConfig::default();
 
     // launch the server
-    let server = Segcache::new(config);
+    let server = Segcache::new(config).expect("failed to launch segcache");
 
     // wait for server to startup. duration is chosen to be longer than we'd
     // expect startup to take in a slow ci environment.

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

@@ -25,15 +25,15 @@ pub struct Segcache {
 
 impl Segcache {
     /// Creates a new `Segcache` process from the given `SegcacheConfig`.
-    pub fn new(config: SegcacheConfig) -> Self {
+    pub fn new(config: SegcacheConfig) -> Result<Self, std::io::Error> {
         // initialize logging
         let log_drain = configure_logging(&config);
 
         // initialize metrics
         common::metrics::init();
 
         // initialize storage
-        let storage = Storage::new(&config);
+        let storage = Storage::new(&config)?;
 
         let max_buffer_size = std::cmp::max(
             server::DEFAULT_BUFFER_SIZE,
@@ -59,7 +59,7 @@ impl Segcache {
         // spawn threads
         let process = process_builder.spawn();
 
-        Self { process }
+        Ok(Self { process })
     }
 
     /// Wait for all threads to complete. Blocks until the process has fully

src/rust/server/segcache/src/main.rs

@@ -117,5 +117,10 @@ fn main() {
     }
 
     // launch segcache
-    Segcache::new(config).wait()
+    match Segcache::new(config) {
+        Ok(segcache) => segcache.wait(),
+        Err(e) => {
+            error!("error launching segcache: {}", e);
+        }
+    }
 }

src/rust/server/segcache/tests/integration.rs

@@ -19,7 +19,7 @@ use std::time::Duration;
 
 fn main() {
     debug!("launching server");
-    let server = Segcache::new(SegcacheConfig::default());
+    let server = Segcache::new(SegcacheConfig::default()).expect("failed to launch segcache");
 
     // wait for server to startup. duration is chosen to be longer than we'd
     // expect startup to take in a slow ci environment.

src/rust/server/segcache/tests/integration_multi.rs

@@ -21,7 +21,7 @@ fn main() {
     debug!("launching multi-worker server");
     let mut config = SegcacheConfig::default();
     config.worker_mut().set_threads(2);
-    let server = Segcache::new(config);
+    let server = Segcache::new(config).expect("failed to launch segcache");
 
     // wait for server to startup. duration is chosen to be longer than we'd
     // expect startup to take in a slow ci environment.

src/rust/storage/datapool/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "datapool"
+version = "0.1.0"
+edition = "2021"
+authors = ["Brian Martin <[email protected]>"]
+description = "abstractions for byte storage pools"
+homepage = "https://pelikan.io"
+repository = "https://github.com/twitter/pelikan"
+license = "Apache-2.0"
+
+[dependencies]
+blake3 = "1.3.1"
+common = { path = "../../common" }
+libc = "0.2.125"
+memmap2 = "0.5.3"
+
+[dev-dependencies]
+tempfile = "3.3.0"