stroage: garbage collect unused blob cache mgr

storage/src/cache/dummycache.rs

@@ -19,6 +19,7 @@
 //!   The [is_chunk_cached()](../trait.BlobCache.html#tymethod.is_chunk_cached) method always
 //!   return true to enable data prefetching.
 use std::io::Result;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 
 use fuse_backend_rs::transport::FileVolatileSlice;
@@ -161,6 +162,7 @@ pub struct DummyCacheMgr {
     cached: bool,
     prefetch: bool,
     validate: bool,
+    closed: AtomicBool,
 }
 
 impl DummyCacheMgr {
@@ -176,6 +178,7 @@ impl DummyCacheMgr {
             cached,
             validate: config.cache_validate,
             prefetch: enable_prefetch,
+            closed: AtomicBool::new(false),
         })
     }
 }
@@ -186,7 +189,14 @@ impl BlobCacheMgr for DummyCacheMgr {
     }
 
     fn destroy(&self) {
-        self.backend().shutdown()
+        if !self.closed.load(Ordering::Acquire) {
+            self.closed.store(true, Ordering::Release);
+            self.backend().shutdown();
+        }
+    }
+
+    fn gc(&self, _id: Option<&str>) -> bool {
+        false
     }
 
     fn backend(&self) -> &(dyn BlobBackend) {
@@ -209,3 +219,9 @@ impl BlobCacheMgr for DummyCacheMgr {
         }))
     }
 }
+
+impl Drop for DummyCacheMgr {
+    fn drop(&mut self) {
+        self.destroy();
+    }
+}

storage/src/cache/filecache/mod.rs

@@ -6,7 +6,7 @@
 use std::collections::HashMap;
 use std::fs::OpenOptions;
 use std::io::Result;
-use std::sync::atomic::AtomicU32;
+use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::sync::{Arc, RwLock};
 
 use tokio::runtime::Runtime;
@@ -37,6 +37,7 @@ pub struct FileCacheMgr {
     validate: bool,
     disable_indexed_map: bool,
     is_compressed: bool,
+    closed: Arc<AtomicBool>,
 }
 
 impl FileCacheMgr {
@@ -65,6 +66,7 @@ impl FileCacheMgr {
             disable_indexed_map: blob_config.disable_indexed_map,
             validate: config.cache_validate,
             is_compressed: config.cache_compressed,
+            closed: Arc::new(AtomicBool::new(false)),
         })
     }
 
@@ -109,12 +111,15 @@ impl BlobCacheMgr for FileCacheMgr {
     }
 
     fn destroy(&self) {
-        self.worker_mgr.stop();
-        self.backend().shutdown();
-        self.metrics.release().unwrap_or_else(|e| error!("{:?}", e));
+        if !self.closed.load(Ordering::Acquire) {
+            self.closed.store(true, Ordering::Release);
+            self.worker_mgr.stop();
+            self.backend().shutdown();
+            self.metrics.release().unwrap_or_else(|e| error!("{:?}", e));
+        }
     }
 
-    fn gc(&self, id: Option<&str>) {
+    fn gc(&self, id: Option<&str>) -> bool {
         let mut reclaim = Vec::new();
 
         if let Some(blob_id) = id {
@@ -137,6 +142,8 @@ impl BlobCacheMgr for FileCacheMgr {
             }
             guard.remove(key);
         }
+
+        self.blobs.read().unwrap().len() == 0
     }
 
     fn backend(&self) -> &(dyn BlobBackend) {
@@ -149,6 +156,12 @@ impl BlobCacheMgr for FileCacheMgr {
     }
 }
 
+impl Drop for FileCacheMgr {
+    fn drop(&mut self) {
+        self.destroy();
+    }
+}
+
 impl FileCacheEntry {
     fn new_file_cache(
         mgr: &FileCacheMgr,

storage/src/cache/fscache/mod.rs

@@ -4,7 +4,7 @@
 
 use std::collections::HashMap;
 use std::io::Result;
-use std::sync::atomic::AtomicU32;
+use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 use std::sync::{Arc, RwLock};
 
 use tokio::runtime::Runtime;
@@ -33,6 +33,7 @@ pub struct FsCacheMgr {
     worker_mgr: Arc<AsyncWorkerMgr>,
     work_dir: String,
     validate: bool,
+    closed: Arc<AtomicBool>,
 }
 
 impl FsCacheMgr {
@@ -59,6 +60,7 @@ impl FsCacheMgr {
             worker_mgr: Arc::new(worker_mgr),
             work_dir: work_dir.to_owned(),
             validate: config.cache_validate,
+            closed: Arc::new(AtomicBool::new(false)),
         })
     }
 
@@ -103,12 +105,15 @@ impl BlobCacheMgr for FsCacheMgr {
     }
 
     fn destroy(&self) {
-        self.worker_mgr.stop();
-        self.backend().shutdown();
-        self.metrics.release().unwrap_or_else(|e| error!("{:?}", e));
+        if !self.closed.load(Ordering::Acquire) {
+            self.closed.store(true, Ordering::Release);
+            self.worker_mgr.stop();
+            self.backend().shutdown();
+            self.metrics.release().unwrap_or_else(|e| error!("{:?}", e));
+        }
     }
 
-    fn gc(&self, id: Option<&str>) {
+    fn gc(&self, id: Option<&str>) -> bool {
         if let Some(blob_id) = id {
             self.blobs.write().unwrap().remove(blob_id);
         } else {
@@ -130,6 +135,8 @@ impl BlobCacheMgr for FsCacheMgr {
                 }
             }
         }
+
+        self.blobs.read().unwrap().len() == 0
     }
 
     fn backend(&self) -> &(dyn BlobBackend) {
@@ -142,6 +149,12 @@ impl BlobCacheMgr for FsCacheMgr {
     }
 }
 
+impl Drop for FsCacheMgr {
+    fn drop(&mut self) {
+        self.destroy();
+    }
+}
+
 impl FileCacheEntry {
     pub fn new_fs_cache(
         mgr: &FsCacheMgr,

storage/src/cache/mod.rs

@@ -330,17 +330,17 @@ pub trait BlobCache: Send + Sync {
 ///
 /// The main responsibility of the blob cache manager is to create blob cache objects for blobs,
 /// all IO requests should be issued to the blob cache object directly.
-pub trait BlobCacheMgr: Send + Sync {
+pub(crate) trait BlobCacheMgr: Send + Sync {
     /// Initialize the blob cache manager.
     fn init(&self) -> Result<()>;
 
     /// Tear down the blob cache manager.
     fn destroy(&self);
 
     /// Garbage-collect unused resources.
-    fn gc(&self, _id: Option<&str>) {
-        todo!()
-    }
+    ///
+    /// Return true if the blob cache manager itself should be garbage-collected.
+    fn gc(&self, _id: Option<&str>) -> bool;
 
     /// Get the underlying `BlobBackend` object of the blob cache object.
     fn backend(&self) -> &(dyn BlobBackend);

storage/src/factory.rs

@@ -213,16 +213,36 @@ impl BlobFactory {
 
     /// Garbage-collect unused blob cache managers and blob caches.
     pub fn gc(&self, victim: Option<(&Arc<FactoryConfig>, &str)>) {
+        let mut mgrs = Vec::new();
+
         if let Some((config, id)) = victim {
             let key = BlobCacheMgrKey {
                 config: config.clone(),
             };
             let mgr = self.mgrs.lock().unwrap().get(&key).cloned();
             if let Some(mgr) = mgr {
-                mgr.gc(Some(id));
+                if mgr.gc(Some(id)) {
+                    mgrs.push((key, mgr.clone()));
+                }
             }
         } else {
-            unimplemented!("TODO")
+            for (key, mgr) in self.mgrs.lock().unwrap().iter() {
+                if mgr.gc(None) {
+                    mgrs.push((
+                        BlobCacheMgrKey {
+                            config: key.config.clone(),
+                        },
+                        mgr.clone(),
+                    ));
+                }
+            }
+        }
+
+        for (key, mgr) in mgrs {
+            let mut guard = self.mgrs.lock().unwrap();
+            if mgr.gc(None) {
+                guard.remove(&key);
+            }
         }
     }