diff --git a/src/util/jobs/cache_miss_strategizer.rs b/src/util/jobs/cache_miss_strategizer.rs new file mode 100644 index 0000000000..7b965fcc0d --- /dev/null +++ b/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, + pub most_recent_miss: DateTime, +} + +/// 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, + + /// Clock can be replaced for testing. + get_time_function: Box DateTime>, +} + +impl CacheMissLog { + fn new(first_miss: DateTime) -> Self { + Self { + first_miss: first_miss.clone(), + most_recent_miss: first_miss, + } + } + + fn with_most_recent_miss(&self, most_recent_miss: DateTime) -> 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 DateTime> + ) -> 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 = (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 DateTime>) { + 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 { + let datetime = DateTime::parse_from_rfc3339(datetime).unwrap(); + let utc : DateTime = 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); + } +} \ No newline at end of file diff --git a/src/util/jobs/cache_miss_strategizer_multi.rs b/src/util/jobs/cache_miss_strategizer_multi.rs new file mode 100644 index 0000000000..4c91f051a7 --- /dev/null +++ b/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>, +} + +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 { + 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 { + 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) + } +} diff --git a/src/util/jobs/cold_cache_strategy.rs b/src/util/jobs/cold_cache_strategy.rs deleted file mode 100644 index 06611d4c91..0000000000 --- a/src/util/jobs/cold_cache_strategy.rs +++ /dev/null @@ -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>, - get_time_function: Box DateTime>, -} - -/// And the Sync/Send threadsafe + interior mutability version. -#[derive(Clone)] -pub struct SyncMultiColdCacheLog { - multi_cache_log: Arc>, -} - -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 { - 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 { - 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 DateTime> - //get_time_function: fn() -> DateTime, - ) -> 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 = (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) { - fn set_time_function(&mut self, get_time_function: Box DateTime>) { - 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 { - let datetime = DateTime::parse_from_rfc3339(datetime).unwrap(); - let utc : DateTime = 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); - } -} \ No newline at end of file diff --git a/src/util/jobs/mod.rs b/src/util/jobs/mod.rs index e67cfe8380..b871837a3e 100644 --- a/src/util/jobs/mod.rs +++ b/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;