dsc 0.1.3

dsc is a cli tool for finding and removing duplicate files on one or multiple file systems, while respecting your gitignore rules. 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

use std::path::PathBuf;
use std::thread::JoinHandle;

use anyhow::{anyhow, Result};
use crossbeam::channel;
use crossbeam::channel::{Receiver, Sender};

use crate::candidate_selection::message_handler::MessageHandler;
use crate::candidate_selection::messages::{AnalysisMessage, CoordinatorMessage, WorkerMessage};
use crate::candidate_selection::path_walker;
use crate::candidate_selection::worker::Worker;
use crate::options::Options;
use crate::types::Duplicate;

pub struct Coordinator {
    options: Options,
}

impl Coordinator {
    pub fn new(options: Options) -> Self {
        Coordinator { options }
    }

    pub fn coordinate(
        &self,
        path_vec: &[PathBuf],
        update_sender: Sender<AnalysisMessage>,
    ) -> Result<Vec<Duplicate>> {
        info!("Coordinator starting");

        let (coordinator_out, worker_in) = channel::unbounded();
        let (worker_out, coordinator_work_in) = channel::unbounded();
        let (file_walker_out, coordinator_files_in) = channel::bounded(2048);

        info!("Spawning workers");

        self.spawn_workers(worker_in, worker_out);

        info!("Finished spawning workers");

        let mut message_handler = MessageHandler::new(
            self.options,
            coordinator_out,
            coordinator_files_in,
            coordinator_work_in,
        );

        let handle = std::thread::spawn(move || {
            info!("Asking MessageHandler to start handling messages");

            let res = message_handler.handle_messages(update_sender);

            info!("Finished handling messages {:?}", res);

            let result = message_handler.into_duplicates();

            res?;

            Ok(result)
        });

        info!("Walking paths");

        path_walker::walk_paths(path_vec, file_walker_out)?;

        info!("Walk finished, waiting for message handler to complete");

        handle
            .join()
            .map_err(|_| anyhow!("Analysis failed unexpectedly"))?
    }

    fn spawn_workers(
        &self,
        receiver: Receiver<CoordinatorMessage>,
        sender: Sender<WorkerMessage>,
    ) -> Vec<JoinHandle<Result<Worker>>> {
        (0..self.options.thread_count)
            .into_iter()
            .map(|_| {
                let worker = Worker::new(
                    receiver.clone(),
                    sender.clone(),
                    self.options.block_size as usize,
                );
                worker.work()
            })
            .collect()
    }
}