From 00cdc5010a79cd74e089bcd787453ba5084c8583 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 5 Oct 2023 17:11:03 +0100 Subject: [PATCH] new generic cache and refresher --- nym-api/src/support/caching/cache.rs | 141 +++++++++++++++++++++++ nym-api/src/support/caching/mod.rs | 6 + nym-api/src/support/caching/refresher.rs | 132 +++++++++++++++++++++ 3 files changed, 279 insertions(+) create mode 100644 nym-api/src/support/caching/cache.rs create mode 100644 nym-api/src/support/caching/refresher.rs diff --git a/nym-api/src/support/caching/cache.rs b/nym-api/src/support/caching/cache.rs new file mode 100644 index 0000000000..f4a6310f51 --- /dev/null +++ b/nym-api/src/support/caching/cache.rs @@ -0,0 +1,141 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use std::ops::Deref; +use std::sync::Arc; +use std::time::Duration; +use thiserror::Error; +use time::OffsetDateTime; +use tokio::sync::{RwLock, RwLockReadGuard}; + +#[derive(Debug, Error)] +#[error("the cache item has not been initialised")] +pub struct UninitialisedCache; + +pub struct SharedCache(Arc>>); + +impl Clone for SharedCache { + fn clone(&self) -> Self { + SharedCache(Arc::clone(&self.0)) + } +} + +impl Default for SharedCache { + fn default() -> Self { + SharedCache(Arc::new(RwLock::new(CachedItem { inner: None }))) + } +} + +impl SharedCache { + pub(crate) fn new() -> Self { + SharedCache::default() + } + + pub(crate) async fn update(&self, value: T) { + let mut guard = self.0.write().await; + if let Some(ref mut existing) = guard.inner { + existing.update(value) + } else { + guard.inner = Some(Cache::new(value)) + } + } + + pub(crate) async fn get(&self) -> Result>, UninitialisedCache> { + let guard = self.0.read().await; + RwLockReadGuard::try_map(guard, |a| a.inner.as_ref()).map_err(|_| UninitialisedCache) + } + + // ignores expiration data + #[allow(dead_code)] + pub(crate) async fn unchecked_get_inner( + &self, + ) -> Result, UninitialisedCache> { + Ok(RwLockReadGuard::map(self.get().await?, |a| &a.value)) + } +} + +impl From> for SharedCache { + fn from(value: Cache) -> Self { + SharedCache(Arc::new(RwLock::new(value.into()))) + } +} + +#[derive(Default)] +pub(crate) struct CachedItem { + inner: Option>, +} + +impl CachedItem { + #[allow(dead_code)] + fn get(&self) -> Result<&Cache, UninitialisedCache> { + self.inner.as_ref().ok_or(UninitialisedCache) + } +} + +impl From> for CachedItem { + fn from(value: Cache) -> Self { + CachedItem { inner: Some(value) } + } +} + +pub struct Cache { + value: T, + as_at: i64, +} + +impl Cache { + fn new(value: T) -> Self { + Cache { + value, + as_at: current_unix_timestamp(), + } + } + + fn update(&mut self, value: T) { + self.value = value; + self.as_at = current_unix_timestamp() + } + + #[allow(dead_code)] + pub fn has_expired(&self, ttl: Duration, now: Option) -> bool { + let now = now.unwrap_or(OffsetDateTime::now_utc()).unix_timestamp(); + let diff = now - self.as_at; + + diff > (ttl.as_secs() as i64) + } + + #[allow(dead_code)] + pub fn timestamp(&self) -> i64 { + self.as_at + } + + #[allow(dead_code)] + pub fn into_inner(self) -> T { + self.value + } +} + +impl Deref for Cache { + type Target = T; + + fn deref(&self) -> &Self::Target { + &self.value + } +} + +impl Default for Cache +where + T: Default, +{ + fn default() -> Self { + Cache { + value: T::default(), + as_at: 0, + } + } +} + +fn current_unix_timestamp() -> i64 { + let now = OffsetDateTime::now_utc(); + now.unix_timestamp() +} diff --git a/nym-api/src/support/caching/mod.rs b/nym-api/src/support/caching/mod.rs index 0d2615d327..77e28714f8 100644 --- a/nym-api/src/support/caching/mod.rs +++ b/nym-api/src/support/caching/mod.rs @@ -1,7 +1,13 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + use serde::Serialize; use std::ops::Deref; use time::OffsetDateTime; +pub(crate) mod cache; +pub(crate) mod refresher; + #[derive(Serialize, Clone)] pub struct Cache { pub value: T, diff --git a/nym-api/src/support/caching/refresher.rs b/nym-api/src/support/caching/refresher.rs new file mode 100644 index 0000000000..bb8a25527a --- /dev/null +++ b/nym-api/src/support/caching/refresher.rs @@ -0,0 +1,132 @@ +// Copyright 2023 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::support::caching::cache::SharedCache; +use nym_task::TaskClient; +use std::time::Duration; +use tokio::time::interval; + +pub struct CacheRefresher { + name: String, + refreshing_interval: Duration, + + // TODO: the Send + Sync bounds are only required for the `start` method. could we maybe make it less restrictive? + provider: Box + Send + Sync>, + shared_cache: SharedCache, + // triggers: Vec>, +} + +#[async_trait] +pub trait CacheItemProvider { + type Item; + type Error: std::error::Error; + + async fn try_refresh(&self) -> Result; +} + +// pub struct TriggerFailure; +// +// #[async_trait] +// pub trait RefreshTriggerTrait { +// async fn triggerred(&mut self) -> Result<(), TriggerFailure>; +// } +// +// // TODO: how to get rid of `T: Send + Sync`? it really doesn't need to be Send + Sync +// // since it's wrapped in Shared internally anyway +// #[async_trait] +// impl RefreshTriggerTrait for watch::Receiver +// where +// T: Send + Sync, +// { +// async fn triggerred(&mut self) -> Result<(), TriggerFailure> { +// self.changed().await.map_err(|err| { +// error!("failed to process refresh trigger: {err}"); +// TriggerFailure +// }) +// } +// } + +impl CacheRefresher +where + E: std::error::Error, +{ + pub(crate) fn new( + item_provider: Box + Send + Sync>, + refreshing_interval: Duration, + ) -> Self { + CacheRefresher { + name: "GenericCacheRefresher".to_string(), + refreshing_interval, + provider: item_provider, + shared_cache: SharedCache::new(), + } + } + + pub(crate) fn new_with_initial_value( + item_provider: Box + Send + Sync>, + refreshing_interval: Duration, + shared_cache: SharedCache, + ) -> Self { + CacheRefresher { + name: "GenericCacheRefresher".to_string(), + refreshing_interval, + provider: item_provider, + shared_cache, + } + } + + #[must_use] + pub(crate) fn named(mut self, name: impl Into) -> Self { + self.name = name.into(); + self + } + + #[allow(dead_code)] + pub(crate) fn get_shared_cache(&self) -> SharedCache { + self.shared_cache.clone() + } + + async fn do_refresh_cache(&self) { + match self.provider.try_refresh().await { + Ok(updated_items) => { + self.shared_cache.update(updated_items).await; + } + Err(err) => { + error!("{}: failed to refresh the cache: {err}", self.name) + } + } + } + + pub async fn refresh(&self, task_client: &mut TaskClient) { + log::info!("{}: refreshing cache state", self.name); + + tokio::select! { + biased; + _ = task_client.recv() => { + log::trace!("{}: Received shutdown while refreshing cache", self.name) + } + _ = self.do_refresh_cache() => (), + } + } + + pub async fn run(&self, mut task_client: TaskClient) { + let mut refresh_interval = interval(self.refreshing_interval); + while !task_client.is_shutdown() { + tokio::select! { + biased; + _ = task_client.recv() => { + log::trace!("{}: Received shutdown", self.name) + } + _ = refresh_interval.tick() => self.refresh(&mut task_client).await, + } + } + } + + pub fn start(self, task_client: TaskClient) + where + T: Send + Sync + 'static, + E: Send + Sync + 'static, + { + tokio::spawn(async move { self.run(task_client).await }); + } +}