rename cache strategizer, add forget duration (untested)

src/util/jobs/cache_miss_strategizer.rs

@@ -0,0 +1,170 @@
+use anyhow::anyhow;
+use chrono::{DateTime, Utc, Duration};
+use std::collections::HashMap;
+
+/// How to handle cache miss events, as dictated by the `CacheMissStrategizer`.
+#[derive(Eq, PartialEq, Copy, Clone, Debug)]
+pub enum CacheMissStrategy {
+  /// Don't handle the cache miss right now.
+  /// Wait or skip until told otherwise.
+  WaitOrSkip,
+  /// Handle the cache miss now.
+  /// Do the download, calculation, whatever immediately.
+  Proceed,
+}
+
+struct CacheMissLog {
+  pub first_miss: DateTime<Utc>,
+  pub most_recent_miss: DateTime<Utc>,
+}
+
+/// CacheMissStrategizer
+///
+/// This is the core component of the cold cache system.
+///
+/// Keep track of a cache and tell the caller when to proceed with work.
+/// If it isn't time yet, skip or wait and let another consumer handle it.
+pub struct CacheMissStrategizer {
+  /// How long it takes after the first cache miss to do work.
+  max_cold_duration: Duration,
+
+  /// How long it takes to reset the cold cache timer.
+  /// eg. If we see a cache miss for X and then another one a day later,
+  /// that shouldn't be cause to do work now.
+  forget_duration: Duration,
+
+  /// Log of all cache misses.
+  cache_miss_log: HashMap<i64, CacheMissLog>,
+
+  /// Clock can be replaced for testing.
+  get_time_function: Box<dyn Fn() -> DateTime<Utc>>,
+}
+
+impl CacheMissLog {
+  fn new(first_miss: DateTime<Utc>) -> Self {
+    Self {
+      first_miss: first_miss.clone(),
+      most_recent_miss: first_miss,
+    }
+  }
+
+  fn with_most_recent_miss(&self, most_recent_miss: DateTime<Utc>) -> Self {
+    Self {
+      first_miss: self.first_miss.clone(),
+      most_recent_miss,
+    }
+  }
+}
+
+impl CacheMissStrategizer {
+  pub fn new(max_cold_duration: Duration, forget_duration: Duration) -> Self {
+    Self {
+      max_cold_duration,
+      forget_duration,
+      cache_miss_log: HashMap::new(),
+      get_time_function: Box::new(|| Utc::now()),
+    }
+  }
+
+  pub fn new_for_testing(
+    max_cold_duration: Duration,
+    forget_duration: Duration,
+    get_time_function: Box<dyn Fn() -> DateTime<Utc>>
+  ) -> Self {
+    Self {
+      max_cold_duration,
+      forget_duration,
+      cache_miss_log: HashMap::new(),
+      get_time_function,
+    }
+  }
+
+  // NB: Not threadsafe due to multiple operations against hashes!
+  #[must_use]
+  pub fn cache_miss(&mut self, id: i64) -> CacheMissStrategy {
+    let now : DateTime<Utc> = (self.get_time_function)();
+
+    if let Some((_id, cache_miss_log)) = self.cache_miss_log.get_key_value(&id) {
+      let admit_duration = now.signed_duration_since(cache_miss_log.first_miss.clone());
+      let forget_duration = now.signed_duration_since(cache_miss_log.most_recent_miss.clone());
+
+      if forget_duration > self.forget_duration {
+        // Too much time elapsed.
+        // Treat this as essentially seeing the miss for the first time.
+        let cache_miss_log = CacheMissLog::new(now);
+        self.cache_miss_log.insert(id, cache_miss_log);
+        CacheMissStrategy::WaitOrSkip
+      }
+      else if admit_duration > self.max_cold_duration {
+        // We're done waiting. Do the work.
+        self.cache_miss_log.remove(&id);
+        CacheMissStrategy::Proceed
+      }
+      else {
+        // Still waiting.
+        let cache_miss_log = cache_miss_log.with_most_recent_miss(now);
+        self.cache_miss_log.insert(id, cache_miss_log);
+        CacheMissStrategy::WaitOrSkip
+      }
+    } else {
+      let cache_miss_log = CacheMissLog::new(now);
+      self.cache_miss_log.insert(id, cache_miss_log);
+      CacheMissStrategy::WaitOrSkip
+    }
+  }
+
+  // NB: Private; for testing only
+  fn set_time_function(&mut self, get_time_function: Box<dyn Fn() -> DateTime<Utc>>) {
+    self.get_time_function = get_time_function;
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use chrono::{Duration, Utc, DateTime, TimeZone};
+  use crate::util::jobs::cache_miss_strategizer::CacheMissStrategizer;
+  use crate::util::jobs::cache_miss_strategizer::CacheMissStrategy;
+
+  fn get_date(datetime: &str) -> DateTime<Utc> {
+    let datetime = DateTime::parse_from_rfc3339(datetime).unwrap();
+    let utc : DateTime<Utc> = DateTime::from(datetime);
+    utc
+  }
+
+  #[test]
+  fn cold_cache_cache_miss_algorithm() {
+    let mut cache_miss_strategizer = CacheMissStrategizer::new(
+      Duration::seconds(10),
+      Duration::seconds(60),
+    );
+
+    // First invocation
+    cache_miss_strategizer.set_time_function(Box::new(|| get_date("2021-07-01T13:00:00+00:00")));
+    assert_eq!(cache_miss_strategizer.cache_miss(10), CacheMissStrategy::WaitOrSkip);
+    cache_miss_strategizer.set_time_function(Box::new(|| get_date("2021-07-01T13:00:01+00:00")));
+    assert_eq!(cache_miss_strategizer.cache_miss(10), CacheMissStrategy::WaitOrSkip);
+    cache_miss_strategizer.set_time_function(Box::new(|| get_date("2021-07-01T13:00:05+00:00")));
+    assert_eq!(cache_miss_strategizer.cache_miss(10), CacheMissStrategy::WaitOrSkip);
+    cache_miss_strategizer.set_time_function(Box::new(|| get_date("2021-07-01T13:00:10+00:00")));
+    assert_eq!(cache_miss_strategizer.cache_miss(10), CacheMissStrategy::WaitOrSkip);
+    // Final invocation after time expires. Proceed.
+    cache_miss_strategizer.set_time_function(Box::new(|| get_date("2021-07-01T13:00:11+00:00")));
+    assert_eq!(cache_miss_strategizer.cache_miss(10), CacheMissStrategy::Proceed);
+
+    // New invocation.
+    cache_miss_strategizer.set_time_function(Box::new(|| get_date("2021-07-01T13:00:15+00:00")));
+    assert_eq!(cache_miss_strategizer.cache_miss(10), CacheMissStrategy::WaitOrSkip);
+    cache_miss_strategizer.set_time_function(Box::new(|| get_date("2021-07-01T13:00:25+00:00")));
+    // New ID.
+    assert_eq!(cache_miss_strategizer.cache_miss(20), CacheMissStrategy::WaitOrSkip);
+    // Old ID (still wait)
+    assert_eq!(cache_miss_strategizer.cache_miss(10), CacheMissStrategy::WaitOrSkip);
+    // Old ID is done, new ID is still waiting.
+    cache_miss_strategizer.set_time_function(Box::new(|| get_date("2021-07-01T13:00:30+00:00")));
+    assert_eq!(cache_miss_strategizer.cache_miss(10), CacheMissStrategy::Proceed);
+    assert_eq!(cache_miss_strategizer.cache_miss(20), CacheMissStrategy::WaitOrSkip);
+    // Now the new ID is also done.
+    cache_miss_strategizer.set_time_function(Box::new(|| get_date("2021-07-01T13:00:41+00:00")));
+    assert_eq!(cache_miss_strategizer.cache_miss(20), CacheMissStrategy::Proceed);
+  }
+}

src/util/jobs/cache_miss_strategizer_multi.rs

@@ -0,0 +1,68 @@
+use anyhow::anyhow;
+use chrono::{DateTime, Utc, Duration};
+use crate::util::anyhow_result::AnyhowResult;
+use crate::util::jobs::cache_miss_strategizer::{CacheMissStrategizer, CacheMissStrategy};
+use std::sync::{RwLock, Arc};
+
+/// Keep track of multiple caches, each with different time penalties.
+/// (This isn't strictly necessary)
+pub struct MultiCacheMissStrategizer {
+  in_memory_log: CacheMissStrategizer,
+  on_disk_log: CacheMissStrategizer,
+  // ... future caches ?
+}
+
+/// And the Sync/Send threadsafe + interior mutability version.
+#[derive(Clone)]
+pub struct SyncMultiCacheMissStrategizer {
+  multi_cache_log: Arc<RwLock<MultiCacheMissStrategizer>>,
+}
+
+impl SyncMultiCacheMissStrategizer {
+  pub fn new(in_memory_log: CacheMissStrategizer, on_disk_log: CacheMissStrategizer) -> Self {
+    let multi_cache_log = MultiCacheMissStrategizer::new(
+      in_memory_log,
+      on_disk_log,
+    );
+    Self {
+      multi_cache_log: Arc::new(RwLock::new(multi_cache_log)),
+    }
+  }
+
+  #[must_use]
+  pub fn memory_cache_miss(&self, id: i64) -> AnyhowResult<CacheMissStrategy> {
+    let result = self.multi_cache_log
+        .write()
+        .map(|mut cache_logs| cache_logs.memory_cache_miss(id))
+        .map_err(|err| anyhow!("mutex error: {:?}", err))?;
+    Ok(result)
+  }
+
+  #[must_use]
+  pub fn disk_cache_miss(&self, id: i64) -> AnyhowResult<CacheMissStrategy> {
+    let result = self.multi_cache_log
+        .write()
+        .map(|mut cache_logs| cache_logs.disk_cache_miss(id))
+        .map_err(|err| anyhow!("mutex error: {:?}", err))?;
+    Ok(result)
+  }
+}
+
+impl MultiCacheMissStrategizer {
+  pub fn new(in_memory_log: CacheMissStrategizer, on_disk_log: CacheMissStrategizer) -> Self {
+    Self {
+      in_memory_log,
+      on_disk_log,
+    }
+  }
+
+  #[must_use]
+  pub fn memory_cache_miss(&mut self, id: i64) -> CacheMissStrategy {
+    self.in_memory_log.cache_miss(id)
+  }
+
+  #[must_use]
+  pub fn disk_cache_miss(&mut self, id: i64) -> CacheMissStrategy {
+    self.on_disk_log.cache_miss(id)
+  }
+}

src/util/jobs/cold_cache_strategy.rs

@@ -1,182 +0,0 @@
-use anyhow::anyhow;
-use chrono::{DateTime, Utc, Duration};
-use crate::util::anyhow_result::AnyhowResult;
-use std::collections::HashMap;
-use std::sync::{RwLock, Arc};
-
-#[derive(Eq, PartialEq, Copy, Clone, Debug)]
-pub enum ColdCacheStrategy {
-  /// Don't handle the cache miss right now.
-  /// Wait or skip until told otherwise.
-  WaitOrSkip,
-  /// Handle the cache miss now.
-  /// Do the download, calculation, whatever immediately.
-  Proceed,
-}
-
-/// Keep track of multiple caches.
-/// This tells the caller when to ignore work or when to pick it up.
-pub struct MultiColdCacheLog {
-  in_memory_log: ColdCacheLog,
-  on_disk_log: ColdCacheLog,
-  // ... future caches ?
-}
-
-/// Keep track of a cache and tell the caller when to proceed with work.
-/// If it isn't time yet, skip or wait and let another consumer handle it.
-pub struct ColdCacheLog {
-  max_cold_duration: Duration,
-  //history_expire_duration: Duration,
-  cache_miss_log: HashMap<i64, DateTime<Utc>>,
-  get_time_function: Box<dyn Fn() -> DateTime<Utc>>,
-}
-
-/// And the Sync/Send threadsafe + interior mutability version.
-#[derive(Clone)]
-pub struct SyncMultiColdCacheLog {
-  multi_cache_log: Arc<RwLock<MultiColdCacheLog>>,
-}
-
-impl SyncMultiColdCacheLog {
-  pub fn new(memory_max_cold_duration: Duration, disk_max_cold_duration: Duration) -> Self {
-    let multi_cache_log = MultiColdCacheLog::new(
-      memory_max_cold_duration,
-      disk_max_cold_duration,
-    );
-    Self {
-      multi_cache_log: Arc::new(RwLock::new(multi_cache_log)),
-    }
-  }
-
-  #[must_use]
-  pub fn memory_cache_miss(&self, id: i64) -> AnyhowResult<ColdCacheStrategy> {
-    let result = self.multi_cache_log
-        .write()
-        .map(|mut cache_logs| cache_logs.memory_cache_miss(id))
-        .map_err(|err| anyhow!("mutex error: {:?}", err))?;
-    Ok(result)
-  }
-
-  #[must_use]
-  pub fn disk_cache_miss(&self, id: i64) -> AnyhowResult<ColdCacheStrategy> {
-    let result = self.multi_cache_log
-        .write()
-        .map(|mut cache_logs| cache_logs.disk_cache_miss(id))
-        .map_err(|err| anyhow!("mutex error: {:?}", err))?;
-    Ok(result)
-  }
-}
-
-impl MultiColdCacheLog {
-  pub fn new(memory_max_cold_duration: Duration, disk_max_cold_duration: Duration) -> Self {
-    Self {
-      in_memory_log: ColdCacheLog::new(memory_max_cold_duration),
-      on_disk_log: ColdCacheLog::new(disk_max_cold_duration),
-    }
-  }
-
-  #[must_use]
-  pub fn memory_cache_miss(&mut self, id: i64) -> ColdCacheStrategy {
-    self.in_memory_log.cache_miss(id)
-  }
-
-  #[must_use]
-  pub fn disk_cache_miss(&mut self, id: i64) -> ColdCacheStrategy {
-    self.on_disk_log.cache_miss(id)
-  }
-}
-
-impl ColdCacheLog {
-  pub fn new(max_cold_duration: Duration) -> Self {
-    Self {
-      max_cold_duration,
-      cache_miss_log: HashMap::new(),
-      //get_time_function: || Utc::now(),
-      get_time_function: Box::new(|| Utc::now()),
-    }
-  }
-
-  pub fn new_for_testing(
-    max_cold_duration: Duration,
-    get_time_function: Box<dyn Fn() -> DateTime<Utc>>
-    //get_time_function: fn() -> DateTime<Utc>,
-  ) -> Self {
-    Self {
-      max_cold_duration,
-      cache_miss_log: HashMap::new(),
-      get_time_function,
-    }
-  }
-
-  // NB: Not threadsafe due to multiple operations against hashes!
-  #[must_use]
-  pub fn cache_miss(&mut self, id: i64) -> ColdCacheStrategy {
-    let now : DateTime<Utc> = (self.get_time_function)();
-
-    if let Some((_id, wait_start)) = self.cache_miss_log.get_key_value(&id) {
-      let duration = now.signed_duration_since(wait_start.clone());
-
-      if duration > self.max_cold_duration {
-        self.cache_miss_log.remove(&id);
-        return ColdCacheStrategy::Proceed;
-      }
-    } else {
-      self.cache_miss_log.insert(id, now);
-    }
-
-    ColdCacheStrategy::WaitOrSkip
-  }
-
-  // NB: For testing
-  //fn set_time_function(&mut self, get_time_function: fn() -> DateTime<Utc>) {
-  fn set_time_function(&mut self, get_time_function: Box<dyn Fn() -> DateTime<Utc>>) {
-    self.get_time_function = get_time_function;
-  }
-}
-
-#[cfg(test)]
-mod tests {
-  use crate::util::jobs::cold_cache_strategy::ColdCacheLog;
-  use crate::util::jobs::cold_cache_strategy::ColdCacheStrategy;
-  use chrono::{Duration, Utc, DateTime, TimeZone};
-
-  fn get_date(datetime: &str) -> DateTime<Utc> {
-    let datetime = DateTime::parse_from_rfc3339(datetime).unwrap();
-    let utc : DateTime<Utc> = DateTime::from(datetime);
-    utc
-  }
-
-  #[test]
-  fn cold_cache_cache_miss_algorithm() {
-    let mut cold_cache_log = ColdCacheLog::new(Duration::seconds(10));
-
-    // First invocation
-    cold_cache_log.set_time_function(Box::new(|| get_date("2021-07-01T13:00:00+00:00")));
-    assert_eq!(cold_cache_log.cache_miss(10), ColdCacheStrategy::WaitOrSkip);
-    cold_cache_log.set_time_function(Box::new(|| get_date("2021-07-01T13:00:01+00:00")));
-    assert_eq!(cold_cache_log.cache_miss(10), ColdCacheStrategy::WaitOrSkip);
-    cold_cache_log.set_time_function(Box::new(|| get_date("2021-07-01T13:00:05+00:00")));
-    assert_eq!(cold_cache_log.cache_miss(10), ColdCacheStrategy::WaitOrSkip);
-    cold_cache_log.set_time_function(Box::new(|| get_date("2021-07-01T13:00:10+00:00")));
-    assert_eq!(cold_cache_log.cache_miss(10), ColdCacheStrategy::WaitOrSkip);
-    // Final invocation after time expires. Proceed.
-    cold_cache_log.set_time_function(Box::new(|| get_date("2021-07-01T13:00:11+00:00")));
-    assert_eq!(cold_cache_log.cache_miss(10), ColdCacheStrategy::Proceed);
-
-    // New invocation.
-    cold_cache_log.set_time_function(Box::new(|| get_date("2021-07-01T13:00:15+00:00")));
-    assert_eq!(cold_cache_log.cache_miss(10), ColdCacheStrategy::WaitOrSkip);
-    cold_cache_log.set_time_function(Box::new(|| get_date("2021-07-01T13:00:25+00:00")));
-    // New ID.
-    assert_eq!(cold_cache_log.cache_miss(20), ColdCacheStrategy::WaitOrSkip);
-    // Old ID (still wait)
-    assert_eq!(cold_cache_log.cache_miss(10), ColdCacheStrategy::WaitOrSkip);
-    // Old ID is done, new ID is still waiting.
-    cold_cache_log.set_time_function(Box::new(|| get_date("2021-07-01T13:00:30+00:00")));
-    assert_eq!(cold_cache_log.cache_miss(10), ColdCacheStrategy::Proceed);
-    assert_eq!(cold_cache_log.cache_miss(20), ColdCacheStrategy::WaitOrSkip);
-    // Now the new ID is also done.
-    cold_cache_log.set_time_function(Box::new(|| get_date("2021-07-01T13:00:41+00:00")));
-    assert_eq!(cold_cache_log.cache_miss(20), ColdCacheStrategy::Proceed);
-  }
-}

src/util/jobs/mod.rs

@@ -1,2 +1,3 @@
-pub mod cold_cache_strategy;
+pub mod cache_miss_strategizer;
+pub mod cache_miss_strategizer_multi;
 pub mod virtual_lfu_cache;