Repository: pkolaczk/fclones (language: rust)

Issue:
Don't lose the cache if interrupted
The cache from a user perspective only is committed after a complete successful run - losing 30TB of progress. I'd like to incrementally populate the cache and age out old entries to have tighter runs.


Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index f3d74b8..8ae927a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -471,7 +471,7 @@ checksum = "25cbce373ec4653f1a01a31e8a5e5ec0c622dc27ff9c4e6606eefef5cbbed4a5"
 
 [[package]]
 name = "fclones"
-version = "0.32.2"
+version = "0.33.0"
 dependencies = [
  "assert_matches",
  "atomic-counter",
diff --git a/fclones/Cargo.toml b/fclones/Cargo.toml
index b8d6449..25fc311 100644
--- a/fclones/Cargo.toml
+++ b/fclones/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "fclones"
-version = "0.32.2"
+version = "0.33.0"
 description = "Finds and removes duplicate files"
 authors = ["Piotr Kołaczkowski <pkolaczk@gmail.com>"]
 homepage = "https://github.com/pkolaczk/fclones"
diff --git a/fclones/src/cache.rs b/fclones/src/cache.rs
index f083f4b..daf701b 100644
--- a/fclones/src/cache.rs
+++ b/fclones/src/cache.rs
@@ -2,6 +2,10 @@
 
 use std::fmt::{Display, Formatter};
 use std::fs::create_dir_all;
+use std::sync::mpsc::{channel, RecvTimeoutError, Sender};
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::thread::JoinHandle;
 use std::time::{Duration, UNIX_EPOCH};
 
 use serde::{Deserialize, Serialize};
@@ -32,13 +36,18 @@ struct CachedFileInfo {
     hash: FileHash,
 }
 
+type InnerCache = typed_sled::Tree<Key, CachedFileInfo>;
+
+const FLUSH_INTERVAL: Duration = Duration::from_millis(1000);
+
 /// Caches file hashes to avoid repeated computations in subsequent runs of fclones.
 ///
 /// Most files don't change very frequently so their hashes don't change.
 /// Usually it is a lot faster to retrieve the hash from an embedded database that to compute
 /// them from file data.
 pub struct HashCache {
-    cache: typed_sled::Tree<Key, CachedFileInfo>,
+    cache: Arc<InnerCache>,
+    _flusher: HashCacheFlusher,
 }
 
 impl HashCache {
@@ -65,8 +74,12 @@ impl HashCache {
         })?;
 
         let tree_id = format!("hash_db:{:?}:{}", algorithm, transform.unwrap_or("<none>"));
-        let cache = typed_sled::Tree::open(&db, tree_id);
-        Ok(HashCache { cache })
+        let cache = Arc::new(typed_sled::Tree::open(&db, tree_id));
+        let flusher = HashCacheFlusher::start(&cache);
+        Ok(HashCache {
+            cache,
+            _flusher: flusher,
+        })
     }
 
     /// Opens the file hash database located in `fclones` subdir of user cache directory.
@@ -101,6 +114,7 @@ impl HashCache {
         self.cache
             .insert(key, &value)
             .map_err(|e| format!("Failed to write entry to cache: {e}"))?;
+
         Ok(())
     }
 
@@ -151,6 +165,44 @@ impl HashCache {
     }
 }
 
+/// Periodically flushes the cache in a background thread
+struct HashCacheFlusher {
+    thread_handle: Option<JoinHandle<()>>,
+    control_channel: Option<Mutex<Sender<()>>>, // wrapped in Mutex because Sender is not Send in older versions of Rust
+}
+
+impl HashCacheFlusher {
+    fn start(cache: &Arc<InnerCache>) -> HashCacheFlusher {
+        let cache = Arc::downgrade(cache);
+        let (tx, rx) = channel::<()>();
+
+        let thread_handle = thread::spawn(move || {
+            while let Err(RecvTimeoutError::Timeout) = rx.recv_timeout(FLUSH_INTERVAL) {
+                if let Some(cache) = cache.upgrade() {
+                    if let Err(e) = cache.flush() {
+                        eprintln!("Failed to flush hash cache: {e}");
+                        return;
+                    }
+                }
+            }
+        });
+
+        HashCacheFlusher {
+            thread_handle: Some(thread_handle),
+            control_channel: Some(Mutex::new(tx)),
+        }
+    }
+}
+
+impl Drop for HashCacheFlusher {
+    fn drop(&mut self) {
+        // Signal the flusher thread to exit:
+        drop(self.control_channel.take());
+        // Wait for the flusher thread to exit:
+        self.thread_handle.take().unwrap().join().unwrap();
+    }
+}
+
 #[cfg(test)]
 mod test {
     use std::fs::OpenOptions;


Test patch: