[pbs-devel] [RFC proxmox-backup 34/39] datastore: add local datastore cache for network attached storages
Christian Ebner
c.ebner at proxmox.com
Mon May 19 13:46:35 CEST 2025
Use a local datastore as cache using LRU cache replacement policy for
operations on a datastore backed by a network, e.g. by an S3 object
store backend. The goal is to reduce number of requests to the
backend and thereby save costs (monetary as well as time).
The cacher allows to fetch cache items on cache misses via the access
method.
Signed-off-by: Christian Ebner <c.ebner at proxmox.com>
---
pbs-datastore/src/datastore.rs | 46 ++++++-
pbs-datastore/src/lib.rs | 3 +
.../src/local_datastore_lru_cache.rs | 116 ++++++++++++++++++
3 files changed, 164 insertions(+), 1 deletion(-)
create mode 100644 pbs-datastore/src/local_datastore_lru_cache.rs
diff --git a/pbs-datastore/src/datastore.rs b/pbs-datastore/src/datastore.rs
index 22ad566ca..19d4ca02c 100644
--- a/pbs-datastore/src/datastore.rs
+++ b/pbs-datastore/src/datastore.rs
@@ -37,8 +37,9 @@ use crate::dynamic_index::{DynamicIndexReader, DynamicIndexWriter};
use crate::fixed_index::{FixedIndexReader, FixedIndexWriter};
use crate::hierarchy::{ListGroups, ListGroupsType, ListNamespaces, ListNamespacesRecursive};
use crate::index::IndexFile;
+use crate::local_datastore_lru_cache::S3Cacher;
use crate::task_tracking::{self, update_active_operations};
-use crate::DataBlob;
+use crate::{DataBlob, LocalDatastoreLruCache};
static DATASTORE_MAP: LazyLock<Mutex<HashMap<String, Arc<DataStoreImpl>>>> =
LazyLock::new(|| Mutex::new(HashMap::new()));
@@ -129,6 +130,7 @@ pub struct DataStoreImpl {
last_digest: Option<[u8; 32]>,
sync_level: DatastoreFSyncLevel,
backend_config: DatastoreBackendConfig,
+ lru_store_caching: Option<LocalDatastoreLruCache>,
}
impl DataStoreImpl {
@@ -144,6 +146,7 @@ impl DataStoreImpl {
last_digest: None,
sync_level: Default::default(),
backend_config: Default::default(),
+ lru_store_caching: None,
})
}
}
@@ -241,6 +244,37 @@ impl DataStore {
Ok(backend_type)
}
+ pub fn cache(&self) -> Option<&LocalDatastoreLruCache> {
+ self.inner.lru_store_caching.as_ref()
+ }
+
+ /// Check if the digest is present in the local datastore cache.
+ /// Always returns false if there is no cache configured for this datastore.
+ pub fn cache_contains(&self, digest: &[u8; 32]) -> bool {
+ if let Some(cache) = self.inner.lru_store_caching.as_ref() {
+ return cache.contains(digest);
+ }
+ false
+ }
+
+ /// Insert digest as most recently used on in the cache.
+ /// Returns with success if there is no cache configured for this datastore.
+ pub fn cache_insert(&self, digest: &[u8; 32], chunk: &DataBlob) -> Result<(), Error> {
+ if let Some(cache) = self.inner.lru_store_caching.as_ref() {
+ return cache.insert(digest, chunk);
+ }
+ Ok(())
+ }
+
+ pub fn cacher(&self) -> Result<Option<S3Cacher>, Error> {
+ self.backend().map(|backend| match backend {
+ DatastoreBackend::S3(s3_client) => {
+ Some(S3Cacher::new(s3_client, self.inner.chunk_store.clone()))
+ }
+ DatastoreBackend::Filesystem => None,
+ })
+ }
+
pub fn lookup_datastore(
name: &str,
operation: Option<Operation>,
@@ -423,6 +457,15 @@ impl DataStore {
None => Default::default(),
};
+ const LOCAL_DATASTORE_CACHE_SIZE: usize = 10_000_000;
+ let lru_store_caching = if let DatastoreBackendConfig::S3(_) = backend_config {
+ let cache =
+ LocalDatastoreLruCache::new(LOCAL_DATASTORE_CACHE_SIZE, chunk_store.clone());
+ Some(cache)
+ } else {
+ None
+ };
+
Ok(DataStoreImpl {
chunk_store,
gc_mutex: Mutex::new(()),
@@ -432,6 +475,7 @@ impl DataStore {
last_digest,
sync_level: tuning.sync_level.unwrap_or_default(),
backend_config,
+ lru_store_caching,
})
}
diff --git a/pbs-datastore/src/lib.rs b/pbs-datastore/src/lib.rs
index e6f65575b..f1ad3d4c2 100644
--- a/pbs-datastore/src/lib.rs
+++ b/pbs-datastore/src/lib.rs
@@ -216,3 +216,6 @@ pub use snapshot_reader::SnapshotReader;
mod local_chunk_reader;
pub use local_chunk_reader::LocalChunkReader;
+
+mod local_datastore_lru_cache;
+pub use local_datastore_lru_cache::LocalDatastoreLruCache;
diff --git a/pbs-datastore/src/local_datastore_lru_cache.rs b/pbs-datastore/src/local_datastore_lru_cache.rs
new file mode 100644
index 000000000..c711c5208
--- /dev/null
+++ b/pbs-datastore/src/local_datastore_lru_cache.rs
@@ -0,0 +1,116 @@
+//! Use a local datastore as cache for operations on a datastore attached via
+//! a network layer (e.g. via the S3 backend).
+
+use std::future::Future;
+use std::sync::Arc;
+
+use anyhow::{bail, Error};
+use hyper::body::HttpBody;
+
+use pbs_s3_client::S3Client;
+use pbs_tools::async_lru_cache::{AsyncCacher, AsyncLruCache};
+
+use crate::ChunkStore;
+use crate::DataBlob;
+
+#[derive(Clone)]
+pub struct S3Cacher {
+ client: Arc<S3Client>,
+ store: Arc<ChunkStore>,
+}
+
+impl AsyncCacher<[u8; 32], ()> for S3Cacher {
+ fn fetch(
+ &self,
+ key: [u8; 32],
+ ) -> Box<dyn Future<Output = Result<Option<()>, Error>> + Send + 'static> {
+ let client = self.client.clone();
+ let store = self.store.clone();
+ Box::new(async move {
+ match client.get_object(key.into()).await? {
+ None => bail!("could not fetch object with key {}", hex::encode(key)),
+ Some(response) => {
+ let bytes = response.content.collect().await?.to_bytes();
+ let chunk = DataBlob::from_raw(bytes.to_vec())?;
+ store.insert_chunk(&chunk, &key)?;
+ Ok(Some(()))
+ }
+ }
+ })
+ }
+}
+
+impl S3Cacher {
+ pub fn new(client: Arc<S3Client>, store: Arc<ChunkStore>) -> Self {
+ Self { client, store }
+ }
+}
+
+/// LRU cache using local datastore for caching chunks
+///
+/// Uses a LRU cache, but without storing the values in-memory but rather
+/// on the filesystem
+pub struct LocalDatastoreLruCache {
+ cache: AsyncLruCache<[u8; 32], ()>,
+ store: Arc<ChunkStore>,
+}
+
+impl LocalDatastoreLruCache {
+ pub fn new(capacity: usize, store: Arc<ChunkStore>) -> Self {
+ Self {
+ cache: AsyncLruCache::new(capacity),
+ store,
+ }
+ }
+
+ /// Insert a new chunk into the local datastore cache.
+ ///
+ /// Fails if the chunk cannot be inserted successfully.
+ pub fn insert(&self, digest: &[u8; 32], chunk: &DataBlob) -> Result<(), Error> {
+ self.store.insert_chunk(chunk, digest)?;
+ self.cache.insert(*digest, (), |digest| {
+ let (path, _digest_str) = self.store.chunk_path(&digest);
+ // Truncate to free up space but keep the inode around, since that
+ // is used as marker for chunks in use by garbage collection.
+ nix::unistd::truncate(&path, 0).map_err(Error::from)
+ })
+ }
+
+ /// Remove a chunk from the local datastore cache.
+ ///
+ /// Fails if the chunk cannot be deleted successfully.
+ pub fn remove(&self, digest: &[u8; 32]) -> Result<(), Error> {
+ self.cache.remove(*digest);
+ let (path, _digest_str) = self.store.chunk_path(digest);
+ std::fs::remove_file(path).map_err(Error::from)
+ }
+
+ pub async fn access(
+ &self,
+ digest: &[u8; 32],
+ cacher: &mut S3Cacher,
+ ) -> Result<Option<DataBlob>, Error> {
+ if self
+ .cache
+ .access(*digest, cacher, |digest| {
+ let (path, _digest_str) = self.store.chunk_path(&digest);
+ // Truncate to free up space but keep the inode around, since that
+ // is used as marker for chunks in use by garbage collection.
+ nix::unistd::truncate(&path, 0).map_err(Error::from)
+ })
+ .await?
+ .is_some()
+ {
+ let (path, _digest_str) = self.store.chunk_path(digest);
+ let mut file = std::fs::File::open(&path)?;
+ let chunk = DataBlob::load_from_reader(&mut file)?;
+ Ok(Some(chunk))
+ } else {
+ Ok(None)
+ }
+ }
+
+ pub fn contains(&self, digest: &[u8; 32]) -> bool {
+ self.cache.contains(*digest)
+ }
+}
--
2.39.5
More information about the pbs-devel
mailing list