From 0c792d06b681e8dfd966831f29e7436bf7410076 Mon Sep 17 00:00:00 2001 From: Alex Zenla Date: Tue, 16 Apr 2024 14:04:37 +0000 Subject: [PATCH] oci: utilize rw-lock for increased cache performance --- crates/oci/src/packer/cache.rs | 17 +++++++++++------ crates/oci/src/packer/service.rs | 11 ++++++++--- 2 files changed, 19 insertions(+), 9 deletions(-) diff --git a/crates/oci/src/packer/cache.rs b/crates/oci/src/packer/cache.rs index cd53d81..0155659 100644 --- a/crates/oci/src/packer/cache.rs +++ b/crates/oci/src/packer/cache.rs @@ -14,12 +14,12 @@ use std::{ path::{Path, PathBuf}, sync::Arc, }; -use tokio::{fs, sync::Mutex}; +use tokio::{fs, sync::RwLock}; #[derive(Clone)] pub struct OciPackerCache { cache_dir: PathBuf, - index: Arc>, + index: Arc>, } const ANNOTATION_IMAGE_NAME: &str = "io.containerd.image.name"; @@ -34,24 +34,29 @@ impl OciPackerCache { .build()?; let cache = OciPackerCache { cache_dir: cache_dir.to_path_buf(), - index: Arc::new(Mutex::new(index)), + index: Arc::new(RwLock::new(index)), }; { - let mut mutex = cache.index.lock().await; + let mut mutex = cache.index.write().await; *mutex = cache.load_index().await?; } Ok(cache) } + pub async fn list(&self) -> Result> { + let index = self.index.read().await; + Ok(index.manifests().clone()) + } + pub async fn recall( &self, name: ImageName, digest: &str, format: OciPackedFormat, ) -> Result> { - let index = self.index.lock().await; + let index = self.index.read().await; let mut descriptor: Option = None; for manifest in index.manifests() { @@ -109,7 +114,7 @@ impl OciPackerCache { } pub async fn store(&self, packed: OciPackedImage) -> Result { - let mut index = self.index.lock().await; + let mut index = self.index.write().await; let mut manifests = index.manifests().clone(); debug!("cache store digest={}", packed.digest); let mut fs_path = self.cache_dir.clone(); diff --git a/crates/oci/src/packer/service.rs b/crates/oci/src/packer/service.rs index db35012..9fff54b 100644 --- a/crates/oci/src/packer/service.rs +++ b/crates/oci/src/packer/service.rs @@ -6,6 +6,7 @@ use std::{ }; use anyhow::{anyhow, Result}; +use oci_spec::image::Descriptor; use tokio::{ sync::{watch, Mutex}, task::JoinHandle, @@ -51,6 +52,10 @@ impl OciPackerService { }) } + pub async fn list(&self) -> Result> { + self.cache.list().await + } + pub async fn recall( &self, digest: &str, @@ -144,7 +149,7 @@ impl OciPackerService { fetcher: OciImageFetcher, progress: OciBoundProgress, ) -> JoinHandle<()> { - info!("packer task {} started", key); + info!("started packer task {}", key); tokio::task::spawn(async move { let _task_drop_guard = scopeguard::guard((key.clone(), self.clone()), |(key, service)| { @@ -223,7 +228,7 @@ impl OciPackerService { match result.as_ref() { Ok(_) => { - info!("packer task {} completed", key); + info!("completed packer task {}", key); } Err(err) => { @@ -249,7 +254,7 @@ impl OciPackerService { tokio::task::spawn(async move { let mut tasks = self.tasks.lock().await; if let Some(task) = tasks.remove(&key) { - warn!("packer task {} aborted", key); + warn!("aborted packer task {}", key); task.watch.send_replace(Some(Err(anyhow!("task aborted")))); } });