Repository: pkolaczk/fclones (language: rust)

Issue:
High CPU-usage while hardlinking files
Hello,

while fclones was one of the rare duplicate-finders that managed to find all duplicates in a reasonable amount of time (below 2 days) in my backup, hardlinking is very slow.

```
# fclones --version
fclones 0.25.0
```
`top -H -c` shows 3 threads with high CPU-usage:

```
# top -H -c

top - 15:43:21 up 5 days, 30 min,  5 users,  load average: 6,45, 7,32, 7,28
Threads: 346 total,   5 running, 341 sleeping,   0 stopped,   0 zombie
%CPU(s): 32,8 us, 43,6 sy,  0,0 ni,  0,0 id, 23,5 wa,  0,0 hi,  0,1 si,  0,0 st
MiB Spch:  32055,4 total,    372,4 free,   6103,3 used,  25579,7 buff/cache
MiB Swap:  10240,0 total,   5483,6 free,   4756,4 used.  25498,2 avail Spch

    PID USER      PR  NI    VIRT    RES    SHR S  %CPU  %MEM     ZEIT+ BEFEHL                                                                                                                                                                   
  39209 root      20   0  346956  24516   4856 R  99,9   0,1 102:39.10 fclones link                                                                                                                                                                 
  39210 root      20   0  346956  24516   4856 R  99,9   0,1 106:14.93 fclones link                                                                                                                                                                 
  39211 root      20   0  346956  24516   4856 R  99,9   0,1  95:29.37 fclones link      
[...]
```
Connecting `strace -p <PID>`to the threads shows that one thread is calling 
```
statx(AT_FDCWD, "/filename/...", AT_STATX_SYNC_AS_STAT, STATX_ALL, {stx_mask=STATX_ALL|STATX_MNT_ID, stx_attributes=0, stx_mode=S_IFREG|0600, stx_size=47320, ...}) = 0
```
for different versions of a file, while all other threads are calling 
```
sched_yield()                           = 0
sched_yield()                           = 0
sched_yield()                           = 0

```
in a loop. So there seems to be some active waiting, while deleting files

Environment
- PC has an old 4-Core CPU
- Disk is a slow 2.5 in HDD, probably with SMR (Shingled Magnetic Recording), not CMR, connected via USB 3 / SATA.
- The disk contains incremental backups in different versions, created by [`cp -al` and then rsynced](http://www.mikerubel.org/computers/rsync_snapshots/).
- `dupes.txt` is about 4 GB
- Most files are already hardlinked, between 10 and 100 links
The idea is to fix missing hardlinks caused by incomplete backups.



Discussion / hints:
This is quite likely caused by the fact the internal sequence of commands is generated in a single thread, but the commands are processed in parallel. So generating the stream of commands became a bottleneck and deduplication threads are simply fighting for work, actively spinning (this is how rayon works). 

Generating commands was made single-threaded due to other feature request about making the stream of commands match the order of the input files. I need to find a different way. 

Two ideas here:
1. use rayon ParallelIterator to generate commands, but record the order in each item and then put the commands in correct order when printing them
2. switch from rayon to async which would allow a lot more flexibility
This seems to be the matching [rayon issue](https://github.com/rayon-rs/rayon/issues/795).

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index baeeffc..7934acc 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -355,6 +355,7 @@ dependencies = [
  "nix",
  "nom",
  "num_cpus",
+ "priority-queue",
  "rand",
  "rayon",
  "reflink",
@@ -842,6 +843,16 @@ version = "0.2.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "eb9f9e6e233e5c4a35559a617bf40a4ec447db2e84c20b55a6f83167b7e57872"
 
+[[package]]
+name = "priority-queue"
+version = "1.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "de9cde7493f5f5d2d163b174be9f9a72d756b79b0f6ed85654128d238c347c1e"
+dependencies = [
+ "autocfg",
+ "indexmap",
+]
+
 [[package]]
 name = "proc-macro-error"
 version = "1.0.4"
diff --git a/Cargo.toml b/Cargo.toml
index 71bf383..2d06c70 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -44,6 +44,7 @@ maplit = "1.0"
 metrohash = "1.0"
 nom = { version = "5.1", features = ["regexp"] }
 num_cpus = "1.13"
+priority-queue = "1.2"
 rand = "0.8"
 rayon = "1.5"
 regex = "1.5"
diff --git a/src/dedupe.rs b/src/dedupe.rs
index 67646eb..d3e2175 100644
--- a/src/dedupe.rs
+++ b/src/dedupe.rs
@@ -3,18 +3,19 @@
 use std::cmp::{max, min, Reverse};
 use std::collections::HashMap;
 use std::fmt::{Display, Formatter};
+use std::hash::{Hash, Hasher};
 use std::io::{ErrorKind, Write};
 use std::ops::{Add, AddAssign};
+use std::sync::mpsc::channel;
 use std::sync::Arc;
 use std::time::SystemTime;
 use std::{fmt, fs, io};
 
 use chrono::{DateTime, FixedOffset, Local};
-use crossbeam_utils::atomic::AtomicCell;
+use priority_queue::PriorityQueue;
 use rand::distributions::Alphanumeric;
 use rand::Rng;
-use rayon::iter::ParallelBridge;
-use rayon::iter::ParallelIterator;
+use rayon::iter::{IntoParallelIterator, ParallelBridge, ParallelIterator};
 
 use crate::config::{DedupeConfig, Priority};
 use crate::device::DiskDevices;
@@ -79,6 +80,7 @@ impl Display for PathAndMetadata {
 }
 
 /// Portable abstraction for commands used to remove duplicates
+#[derive(Debug)]
 pub enum FsCommand {
     Remove {
         file: PathAndMetadata,
@@ -864,6 +866,12 @@ where
 /// - was matched by any of the `retain_path` or `retain_name` patterns
 /// - was not matched by all `drop_path` and `drop_name` patterns
 ///
+/// The commands in the list are grouped into vectors where each
+/// vector has its sequential id. This id allows to convert the parallel iterator into a
+/// sequential iterator with the same order as the groups in the input file.
+/// Unfortunately Rayon does not allow to convert a parallel iterator
+/// to a sequential iterator easily, so we need this hack with prepending ids of each group.
+///
 /// # Parameters
 /// - `groups`: iterator over groups of identical files
 /// - `op`: what to do with duplicates
@@ -874,20 +882,22 @@ pub fn dedupe<'a, I, P>(
     op: DedupeOp,
     config: &'a DedupeConfig,
     log: &'a Log,
-) -> impl Iterator<Item = FsCommand> + Send + 'a
+) -> impl ParallelIterator<Item = (usize, Vec<FsCommand>)> + Send + 'a
 where
     I: IntoIterator<Item = FileGroup<P>> + 'a,
     I::IntoIter: Send,
-    P: Into<Path>,
+    P: Into<Path> + Send,
 {
     let devices = DiskDevices::new(&HashMap::new());
     groups
         .into_iter()
-        .flat_map(move |group| match partition(group, config, log) {
-            Ok(group) => group.dedupe_script(&op, &devices),
+        .enumerate()
+        .par_bridge()
+        .map(move |(i, group)| match partition(group, config, log) {
+            Ok(group) => (i, group.dedupe_script(&op, &devices)),
             Err(e) => {
                 log.warn(e);
-                Vec::new()
+                (i, vec![])
             }
         })
 }
@@ -902,12 +912,11 @@ where
 /// Returns the number of files processed and the amount of disk space reclaimed.
 pub fn run_script<I>(script: I, should_lock: bool, log: &Log) -> DedupeResult
 where
-    I: IntoIterator<Item = FsCommand>,
-    I::IntoIter: Send,
+    I: IntoParallelIterator<Item = (usize, Vec<FsCommand>)>,
 {
     script
-        .into_iter()
-        .par_bridge()
+        .into_par_iter()
+        .flat_map(|(_, cmd_vec)| cmd_vec)
         .map(|cmd| cmd.execute(should_lock, log))
         .inspect(|res| {
             if let Err(e) = res {
@@ -922,50 +931,116 @@ where
         .reduce(DedupeResult::default, |a, b| a + b)
 }
 
+/// We need this so we can put command vectors in a priority queue.
+struct FsCommandGroup {
+    index: usize,
+    commands: Vec<FsCommand>,
+}
+
+impl FsCommandGroup {
+    pub fn new(index: usize, commands: Vec<FsCommand>) -> FsCommandGroup {
+        FsCommandGroup { index, commands }
+    }
+}
+
+impl PartialEq<Self> for FsCommandGroup {
+    fn eq(&self, other: &Self) -> bool {
+        self.index == other.index
+    }
+}
+
+impl Eq for FsCommandGroup {}
+
+impl Hash for FsCommandGroup {
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.index.hash(state)
+    }
+}
+
 /// Prints a script generated by [`dedupe`] to stdout.
 ///
 /// Does not perform any filesystem changes.
 /// Returns the number of files processed and the amount of disk space that would be
 /// reclaimed if all commands of the script were executed with no error.
 pub fn log_script(
-    script: impl IntoIterator<Item = FsCommand>,
+    script: impl IntoParallelIterator<Item = (usize, Vec<FsCommand>)> + Send,
     mut out: impl Write + Send,
 ) -> io::Result<DedupeResult> {
-    let err = AtomicCell::new(None);
-    let result = script
-        .into_iter()
-        .map(|cmd| {
-            for line in cmd.to_shell_str() {
-                if let Err(e) = writeln!(out, "{}", line) {
-                    err.store(Some(e));
-                    return None;
+    // Unfortunately the items may come in any order from the ParallelIterator,
+    // and that order may change with each run, because multiple threads race to produce
+    // the next item. However, we want to print the commands
+    // in the same deterministic order as the groups in the input file.
+    // That's why we first send the commands to a PriorityQueue through a channel in order to
+    // get them all on a single thread. The queue sorts them by their sequential identifiers.
+    // Because the identifiers are consecutive integers, we know which item should be printed next.
+
+    let (tx, rx) = channel();
+
+    // Scope needed so the script iterator doesn't need to be 'static.
+    // This way we tell the compiler the background thread we start to process the iterator
+    // terminates before exiting this function.
+    crossbeam_utils::thread::scope(move |s| {
+        // Process items in parallel in a background thread, so we can read as soon as they
+        // are produced:
+        s.spawn(move |_| {
+            script
+                .into_par_iter()
+                .for_each_with(tx, |tx, item| tx.send(item).unwrap())
+        });
+
+        let mut queue = PriorityQueue::new();
+        let mut next_group_index = 0;
+        let mut processed_count = 0;
+        let mut reclaimed_space = FileLen(0);
+
+        while let Ok((group_index, commands)) = rx.recv() {
+            // Push the command group we received from the iterator.
+            // We may receive them in an incorrect order, so we push them to a PriorityQueue.
+            queue.push(
+                FsCommandGroup::new(group_index, commands),
+                Reverse(group_index), // we want to get items with lowest-index first
+            );
+
+            // Process items in the queue as soon as possible to save memory.
+            while let Some((group, _)) = queue.peek() {
+                // Only process the item if it is the next one we expect.
+                // If we see an out-of-order item, we simply need to wait for more items
+                // to be pushed.
+                if group.index != next_group_index {
+                    break;
+                }
+                // We got the right item, let's pop it from the queue and log it:
+                next_group_index += 1;
+                let cmd_vec = queue.pop().unwrap().0.commands;
+                for cmd in cmd_vec {
+                    processed_count += 1;
+                    reclaimed_space += cmd.space_to_reclaim();
+                    for line in cmd.to_shell_str() {
+                        writeln!(out, "{}", line)?;
+                    }
                 }
             }
-            Some(DedupeResult {
-                processed_count: 1,
-                reclaimed_space: cmd.space_to_reclaim(),
-            })
-        })
-        .take_while(|r| r.is_some())
-        .map(|r| r.unwrap())
-        .fold(DedupeResult::default(), |a, b| a + b);
+        }
 
-    match err.take() {
-        None => Ok(result),
-        Some(e) => Err(e),
-    }
+        Ok(DedupeResult {
+            processed_count,
+            reclaimed_space,
+        })
+    })
+    .unwrap()
 }
 
 #[cfg(test)]
 mod test {
     use std::collections::HashSet;
-    use std::fs::create_dir;
+    use std::fs::{create_dir, create_dir_all};
     use std::path::PathBuf;
     use std::{thread, time};
 
-    use crate::config::GroupConfig;
     use chrono::Duration;
+    use itertools::Itertools;
 
+    use crate::config::GroupConfig;
     use crate::file::FileHash;
     use crate::group_files;
     use crate::pattern::Pattern;
@@ -1106,7 +1181,8 @@ mod test {
     }
 
     /// Creates 3 empty files with different creation time and returns a FileGroup describing them
-    fn make_group(root: &PathBuf) -> FileGroup<Path> {
+    fn make_group(root: &PathBuf, file_hash: FileHash) -> FileGroup<Path> {
+        create_dir_all(root).unwrap();
         let file_1 = root.join("file_1");
         let file_2 = root.join("file_2");
         let file_3 = root.join("file_3");
@@ -1117,7 +1193,7 @@ mod test {
 
         let group = FileGroup {
             file_len: FileLen(0),
-            file_hash: FileHash(0),
+            file_hash,
             files: vec![
                 Path::from(&file_1),
                 Path::from(&file_2),
@@ -1130,7 +1206,7 @@ mod test {
     #[test]
     fn test_partition_selects_files_for_removal() {
         with_dir("dedupe/partition/basic", |root| {
-            let group = make_group(root);
+            let group = make_group(root, FileHash(0));
             let config = DedupeConfig::default();
             let partitioned = partition(group, &config, &Log::new()).unwrap();
             assert_eq!(partitioned.to_keep.len(), 1);
@@ -1141,7 +1217,7 @@ mod test {
     #[test]
     fn test_partition_bails_out_if_file_modified_too_late() {
         with_dir("dedupe/partition/modification", |root| {
-            let group = make_group(root);
+            let group = make_group(root, FileHash(0));
             let mut config = DedupeConfig::default();
             config.modified_before = Some(DateTime::from(Local::now() - Duration::days(1)));
             let partitioned = partition(group, &config, &Log::new());
@@ -1152,7 +1228,7 @@ mod test {
     #[test]
     fn test_partition_skips_file_with_different_len() {
         with_dir("dedupe/partition/file_len", |root| {
-            let group = make_group(root);
+            let group = make_group(root, FileHash(0));
             let path = group.files[0].clone();
             write_file(&path.to_path_buf(), "foo");
 
@@ -1184,7 +1260,7 @@ mod test {
                 // file creation time
                 return;
             }
-            let group = make_group(root);
+            let group = make_group(root, FileHash(0));
             let mut config = DedupeConfig::default();
             config.priority = vec![Priority::Newest];
             let partitioned_1 = partition(group.clone(), &config, &Log::new()).unwrap();
@@ -1205,7 +1281,7 @@ mod test {
     #[test]
     fn test_partition_respects_modification_time_priority() {
         with_dir("dedupe/partition/mtime_priority", |root| {
-            let group = make_group(root);
+            let group = make_group(root, FileHash(0));
 
             thread::sleep(time::Duration::from_millis(10));
             let path = group.files[0].clone();
@@ -1231,7 +1307,7 @@ mod test {
     #[test]
     fn test_partition_respects_keep_patterns() {
         with_dir("dedupe/partition/keep", |root| {
-            let group = make_group(root);
+            let group = make_group(root, FileHash(0));
             let mut config = DedupeConfig::default();
             config.priority = vec![Priority::LeastRecentlyModified];
             config.keep_name_patterns = vec![Pattern::glob("*_1").unwrap()];
@@ -1250,7 +1326,7 @@ mod test {
     #[test]
     fn test_partition_respects_drop_patterns() {
         with_dir("dedupe/partition/drop", |root| {
-            let group = make_group(root);
+            let group = make_group(root, FileHash(0));
             let mut config = DedupeConfig::default();
             config.priority = vec![Priority::LeastRecentlyModified];
             config.name_patterns = vec![Pattern::glob("*_3").unwrap()];
@@ -1274,8 +1350,8 @@ mod test {
             create_dir(&root1).unwrap();
             create_dir(&root2).unwrap();
 
-            let group1 = make_group(&root1);
-            let group2 = make_group(&root2);
+            let group1 = make_group(&root1, FileHash(0));
+            let group2 = make_group(&root2, FileHash(0));
             let group = FileGroup {
                 file_len: group1.file_len,
                 file_hash: group1.file_hash,
@@ -1347,12 +1423,12 @@ mod test {
 
     #[test]
     fn test_run_dedupe_script() {
-        with_dir("dedupe/partition/dedupe_script", |root| {
+        with_dir("dedupe/partition/run_dedupe_script", |root| {
             let mut log = Log::new();
             log.no_progress = true;
             log.log_stderr_to_stdout = true;
 
-            let group = make_group(root);
+            let group = make_group(root, FileHash(0));
             let mut config = DedupeConfig::default();
             config.priority = vec![Priority::LeastRecentlyModified];
             let script = dedupe(vec![group], DedupeOp::Remove, &config, &log);
@@ -1364,6 +1440,39 @@ mod test {
         });
     }
 
+    #[test]
+    fn test_log_dedupe_script() {
+        with_dir("dedupe/partition/log_dedupe_script", |root| {
+            let mut log = Log::new();
+            log.no_progress = true;
+            log.log_stderr_to_stdout = true;
+
+            let group_1 = make_group(&root.join("group_1"), FileHash(0));
+            let group_2 = make_group(&root.join("group_2"), FileHash(1));
+            let group_3 = make_group(&root.join("group_3"), FileHash(2));
+            let groups = vec![group_1, group_2, group_3];
+
+            let mut config = DedupeConfig::default();
+            config.priority = vec![Priority::LeastRecentlyModified];
+
+            let script = dedupe(groups, DedupeOp::Remove, &config, &log);
+
+            let mut out = Vec::new();
+            let dedupe_result = log_script(script, &mut out).unwrap();
+            assert_eq!(dedupe_result.processed_count, 6);
+
+            let out = String::from_utf8(out).unwrap();
+            let out_lines = out.lines().collect_vec();
+            assert_eq!(out_lines.len(), 6);
+            assert!(out_lines[0].contains("group_1"));
+            assert!(out_lines[1].contains("group_1"));
+            assert!(out_lines[2].contains("group_2"));
+            assert!(out_lines[3].contains("group_2"));
+            assert!(out_lines[4].contains("group_3"));
+            assert!(out_lines[5].contains("group_3"));
+        });
+    }
+
     #[test]
     fn test_hard_link_merges_subgroups_of_hard_links() {
         with_dir("dedupe/merge_subgroups_of_hardlinks", |root| {


Test patch: