MemoryStore now can be configured to evict entries

cas/grpc_service/tests/ac_server_test.rs

@@ -34,11 +34,11 @@ fn make_store_manager() -> Result<StoreManager, Error> {
     let mut store_manager = StoreManager::new();
     store_manager.make_store(
         "main_cas",
-        &config::backends::StoreConfig::memory(config::backends::MemoryStore {}),
+        &config::backends::StoreConfig::memory(config::backends::MemoryStore::default()),
     )?;
     store_manager.make_store(
         "main_ac",
-        &config::backends::StoreConfig::memory(config::backends::MemoryStore {}),
+        &config::backends::StoreConfig::memory(config::backends::MemoryStore::default()),
     )?;
     Ok(store_manager)
 }

cas/grpc_service/tests/bytestream_server_test.rs

@@ -20,7 +20,7 @@ fn make_store_manager() -> Result<StoreManager, Error> {
     let mut store_manager = StoreManager::new();
     store_manager.make_store(
         "main_cas",
-        &config::backends::StoreConfig::memory(config::backends::MemoryStore {}),
+        &config::backends::StoreConfig::memory(config::backends::MemoryStore::default()),
     )?;
     Ok(store_manager)
 }

cas/grpc_service/tests/cas_server_test.rs

@@ -26,7 +26,7 @@ fn make_store_manager() -> Result<StoreManager, Error> {
     let mut store_manager = StoreManager::new();
     store_manager.make_store(
         "main_cas",
-        &config::backends::StoreConfig::memory(config::backends::MemoryStore {}),
+        &config::backends::StoreConfig::memory(config::backends::MemoryStore::default()),
     )?;
     Ok(store_manager)
 }

cas/store/BUILD

@@ -37,6 +37,7 @@ rust_library(
         "//third_party:tokio",
         "//util:common",
         "//util:error",
+        "//util:evicting_map",
         ":traits",
     ],
     proc_macro_deps = ["//third_party:async_trait"],

cas/store/lib.rs

@@ -4,7 +4,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use config::{self, backends::StoreConfig};
-use error::{Error, ResultExt};
+use error::Error;
 use memory_store::MemoryStore;
 use verify_store::VerifyStore;
 
@@ -19,12 +19,7 @@ fn private_make_store(backend: &StoreConfig) -> Result<Arc<dyn Store>, Error> {
         StoreConfig::memory(config) => Ok(Arc::new(MemoryStore::new(&config))),
         StoreConfig::verify(config) => Ok(Arc::new(VerifyStore::new(
             &config,
-            private_make_store(
-                config
-                    .backend
-                    .as_ref()
-                    .err_tip(|| "Expected verify store to have 'backend'")?,
-            )?,
+            private_make_store(&config.backend)?,
         ))),
     }
 }

cas/store/memory_store.rs

@@ -1,8 +1,8 @@
 // Copyright 2020-2021 Nathan (Blaise) Bruer.  All rights reserved.
 
-use std::collections::HashMap;
 use std::marker::Send;
 use std::sync::Arc;
+use std::time::Instant;
 
 use async_mutex::Mutex;
 use async_trait::async_trait;
@@ -11,17 +11,19 @@ use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use common::DigestInfo;
 use config;
 use error::{Code, ResultExt};
+use evicting_map::EvictingMap;
 use traits::{ResultFuture, StoreTrait};
 
-#[derive(Debug)]
 pub struct MemoryStore {
-    map: Mutex<HashMap<[u8; 32], Arc<Vec<u8>>>>,
+    map: Mutex<EvictingMap<Instant>>,
 }
 
 impl MemoryStore {
-    pub fn new(_config: &config::backends::MemoryStore) -> Self {
+    pub fn new(config: &config::backends::MemoryStore) -> Self {
+        let empty_policy = config::backends::EvictionPolicy::default();
+        let eviction_policy = config.eviction_policy.as_ref().unwrap_or(&empty_policy);
         MemoryStore {
-            map: Mutex::new(HashMap::new()),
+            map: Mutex::new(EvictingMap::new(eviction_policy, Instant::now())),
         }
     }
 }
@@ -30,7 +32,7 @@ impl MemoryStore {
 impl StoreTrait for MemoryStore {
     fn has<'a>(self: std::pin::Pin<&'a Self>, digest: DigestInfo) -> ResultFuture<'a, bool> {
         Box::pin(async move {
-            let map = self.map.lock().await;
+            let mut map = self.map.lock().await;
             Ok(map.contains_key(&digest.packed_hash))
         })
     }
@@ -57,7 +59,7 @@ impl StoreTrait for MemoryStore {
         length: Option<usize>,
     ) -> ResultFuture<'a, ()> {
         Box::pin(async move {
-            let map = self.map.lock().await;
+            let mut map = self.map.lock().await;
             let value = map
                 .get(&digest.packed_hash)
                 .err_tip_with_code(|_| (Code::NotFound, format!("Hash {} not found", digest.str())))?

cas/store/tests/memory_store_test.rs

@@ -1,4 +1,4 @@
-// Copyright 2020 Nathan (Blaise) Bruer.  All rights reserved.
+// Copyright 2020-2021 Nathan (Blaise) Bruer.  All rights reserved.
 
 use std::pin::Pin;
 
@@ -19,7 +19,7 @@ mod memory_store_tests {
 
     #[tokio::test]
     async fn insert_one_item_then_update() -> Result<(), Error> {
-        let store_owned = MemoryStore::new(&config::backends::MemoryStore {});
+        let store_owned = MemoryStore::new(&config::backends::MemoryStore::default());
         let store = Pin::new(&store_owned);
 
         {
@@ -73,7 +73,7 @@ mod memory_store_tests {
 
     #[tokio::test]
     async fn read_partial() -> Result<(), Error> {
-        let store_owned = MemoryStore::new(&config::backends::MemoryStore {});
+        let store_owned = MemoryStore::new(&config::backends::MemoryStore::default());
         let store = Pin::new(&store_owned);
 
         const VALUE1: &str = "1234";
@@ -97,7 +97,7 @@ mod memory_store_tests {
 
     #[tokio::test]
     async fn errors_with_invalid_inputs() -> Result<(), Error> {
-        let store_owned = MemoryStore::new(&config::backends::MemoryStore {});
+        let store_owned = MemoryStore::new(&config::backends::MemoryStore::default());
         let store = Pin::new(&store_owned);
         const VALUE1: &str = "123";
         {

cas/store/tests/verify_store_test.rs

@@ -21,10 +21,10 @@ mod verify_store_tests {
 
     #[tokio::test]
     async fn verify_size_false_passes_on_update() -> Result<(), Error> {
-        let inner_store = Arc::new(MemoryStore::new(&config::backends::MemoryStore {}));
+        let inner_store = Arc::new(MemoryStore::new(&config::backends::MemoryStore::default()));
         let store_owned = VerifyStore::new(
             &config::backends::VerifyStore {
-                backend: None,
+                backend: config::backends::StoreConfig::memory(config::backends::MemoryStore::default()),
                 verify_size: false,
             },
             inner_store.clone(),
@@ -50,10 +50,10 @@ mod verify_store_tests {
 
     #[tokio::test]
     async fn verify_size_true_fails_on_update() -> Result<(), Error> {
-        let inner_store = Arc::new(MemoryStore::new(&config::backends::MemoryStore {}));
+        let inner_store = Arc::new(MemoryStore::new(&config::backends::MemoryStore::default()));
         let store_owned = VerifyStore::new(
             &config::backends::VerifyStore {
-                backend: None,
+                backend: config::backends::StoreConfig::memory(config::backends::MemoryStore::default()),
                 verify_size: true,
             },
             inner_store.clone(),
@@ -82,10 +82,10 @@ mod verify_store_tests {
 
     #[tokio::test]
     async fn verify_size_true_suceeds_on_update() -> Result<(), Error> {
-        let inner_store = Arc::new(MemoryStore::new(&config::backends::MemoryStore {}));
+        let inner_store = Arc::new(MemoryStore::new(&config::backends::MemoryStore::default()));
         let store_owned = VerifyStore::new(
             &config::backends::VerifyStore {
-                backend: None,
+                backend: config::backends::StoreConfig::memory(config::backends::MemoryStore::default()),
                 verify_size: true,
             },
             inner_store.clone(),

config/backends.rs

@@ -19,20 +19,22 @@ pub enum StoreConfig {
     verify(Box<VerifyStore>),
 }
 
-#[derive(Deserialize, Debug)]
+#[derive(Deserialize, Debug, Default)]
 pub struct MemoryStore {
-    // TODO(allada) MemoryStore needs to implement deleting deleting old objects when they are deleted.
+    /// Policy used to evict items out of the store. Failure to set this
+    /// value will cause items to never be removed from the store causing
+    /// infinite memory usage.
+    pub eviction_policy: Option<EvictionPolicy>,
 }
 
 #[derive(Deserialize, Debug)]
 pub struct VerifyStore {
-    #[serde(default)]
     /// The underlying store wrap around. All content will first flow
     /// through self before forwarding to backend. In the event there
     /// is an error detected in self, the connection to the backend
     /// will be terminated, and early termination should always cause
     /// updates to fail on the backend.
-    pub backend: Option<StoreConfig>,
+    pub backend: StoreConfig,
 
     /// If set the store will verify the size of the data before accepting
     /// an upload of data.
@@ -41,3 +43,25 @@ pub struct VerifyStore {
     #[serde(default)]
     pub verify_size: bool,
 }
+
+/// Eviction policy always works on LRU (Least Recently Used). Any time an entry
+/// is touched it updates the timestamp. Inserts and updates will execute the
+/// eviction policy removing any expired entries and/or the oldest entries
+/// until the store size becomes smaller than max_bytes.
+#[derive(Deserialize, Debug, Default)]
+pub struct EvictionPolicy {
+    /// Maximum number of bytes before eviction takes place.
+    /// Default: 0. Zero means never evict based on size.
+    #[serde(default)]
+    pub max_bytes: usize,
+
+    /// Maximum number of seconds for an entry to live before an eviction.
+    /// Default: 0. Zero means never evict based on time.
+    #[serde(default)]
+    pub max_seconds: u32,
+
+    /// Maximum size of the store before an eviction takes place.
+    /// Default: 0. Zero means never evict based on count.
+    #[serde(default)]
+    pub max_count: u64,
+}

config/examples/basic_cas.json

@@ -9,7 +9,12 @@
       }
     },
     "AC_MAIN_STORE": {
-      "memory": {}
+      "memory": {
+        "eviction_policy": {
+          // 100mb.
+          "max_bytes": 100000000,
+        }
+      }
     }
   },
   "servers": [{

util/BUILD

@@ -28,6 +28,16 @@ rust_library(
     visibility = ["//visibility:public"],
 )
 
+rust_library(
+    name = "evicting_map",
+    srcs = ["evicting_map.rs"],
+    deps = [
+        "//config",
+        "//third_party:lru",
+    ],
+    visibility = ["//visibility:public"],
+)
+
 rust_library(
     name = "async_fixed_buffer",
     srcs = ["async_fixed_buffer.rs"],
@@ -39,8 +49,8 @@ rust_library(
 )
 
 rust_test(
-    name = "utils_tests",
-    srcs = ["tests/async_fixed_buffer_tests.rs"],
+    name = "async_fixed_buffer_test",
+    srcs = ["tests/async_fixed_buffer_test.rs"],
     deps = [
         "//third_party:futures",
         "//third_party:pretty_assertions",
@@ -49,3 +59,16 @@ rust_test(
         ":error",
     ],
 )
+
+rust_test(
+    name = "evicting_map_test",
+    srcs = ["tests/evicting_map_test.rs"],
+    deps = [
+        "//config",
+        "//third_party:hex",
+        "//third_party:mock_instant",
+        "//third_party:tokio",
+        ":error",
+        ":evicting_map",
+    ],
+)