libmhash 0.2.1

A file hashing library that can do multiple hashes for multile files at the same time.
Documentation 
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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

use std::{collections::HashMap, hash::Hash, thread};

enum Operation<K>
where
    K: Eq + Hash + Send + Clone + 'static,
{
    Add(Job<K>),
    Done(K),
    DoneBy(Box<dyn Fn(&K) -> bool + Send + 'static>),
}

enum ExecutorOperation {
    Job(Box<dyn FnOnce() + Send>),
    Done,
}

struct Job<K>
where
    K: Eq + Hash + Send + Clone + 'static,
{
    pub tag: K,
    pub job: Box<dyn FnOnce() + Send>,
}

impl<K> Job<K>
where
    K: Eq + Hash + Send + Clone + 'static,
{
    fn new(tag: K, job: Box<dyn FnOnce() + Send>) -> Self {
        Self { tag, job }
    }
}

pub struct TagThreadPool<K>
where
    K: Eq + Hash + Send + Clone + 'static,
{
    dispatcher: crossbeam_channel::Sender<Operation<K>>,
}

impl<K> TagThreadPool<K>
where
    K: Eq + Hash + Send + Clone + 'static,
{
    pub fn new() -> Self {
        let (dispatcher, receiver) = crossbeam_channel::unbounded::<Operation<K>>();
        let thread_pool = threadpool::Builder::new().build();
        let dispatcher_for_finish = dispatcher.clone();
        thread::spawn(move || {
            let mut job_senders = HashMap::new();

            loop {
                let operation = match receiver.recv() {
                    Ok(operation) => operation,
                    Err(_) => break,
                };

                let _ = match operation {
                    Operation::Add(job) => job_senders
                        .entry(job.tag.clone())
                        .or_insert_with(|| {
                            let (sender, receiver) =
                                crossbeam_channel::unbounded::<ExecutorOperation>();

                            thread_pool.execute(move || loop {
                                if let Ok(operation) = receiver.recv() {
                                    match operation {
                                        ExecutorOperation::Job(job) => (job)(),
                                        ExecutorOperation::Done => break,
                                    }
                                }
                            });

                            sender
                        })
                        .send(ExecutorOperation::Job(job.job)),
                    Operation::Done(tag) => match job_senders.remove(&tag) {
                        Some(sender) => sender.send(ExecutorOperation::Done),
                        None => Ok(()),
                    },
                    Operation::DoneBy(filter) => {
                        for key in job_senders.keys().filter(|k| (filter)(k)) {
                            let _ = dispatcher_for_finish.send(Operation::Done(key.clone()));
                        }
                        Ok(())
                    }
                };
            }
        });

        TagThreadPool { dispatcher }
    }

    pub fn dispatch<F>(&self, tag: K, job: F)
    where
        F: FnOnce() + Send + 'static,
    {
        self.dispatcher
            .send(Operation::Add(Job::new(tag, Box::new(job))))
            .unwrap();
    }

    pub fn finish(&self, tag: K) {
        self.dispatcher.send(Operation::Done(tag)).unwrap();
    }

    pub fn finish_by<F>(&self, filter: F)
    where
        F: Fn(&K) -> bool + Send + 'static,
    {
        self.dispatcher
            .send(Operation::DoneBy(Box::new(filter)))
            .unwrap();
    }
}