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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

use std::{fs::OpenOptions, hash::Hash, io::Read, path::Path, sync::Arc};

use crossbeam_channel::Sender;
use threadpool::ThreadPool;

use crate::{
    hasher_server::{
        data_wrapper::DataWrapper, fragment_sender::FragmentSender, hasher_wrapper::HasherWrapper,
        operation::Operation, sync_unsafe_cell::SyncUnsafeCell, HasherError, Identifier,
    },
    simple_semaphore::SimpleSemaphore,
};

#[derive(Debug)]
pub struct DataSender<Tag>
where
    Tag: Clone + Eq + Hash + Send,
{
    pub(super) block_size: usize,
    pub(super) block_count: usize,
    pub(super) operation_sender: Sender<Operation<Tag>>,
    pub(super) id_semaphore: Arc<SimpleSemaphore>,
    pub(super) reader_threads: ThreadPool,
}

impl<Tag> DataSender<Tag>
where
    Tag: Clone + Eq + Hash + Send + 'static,
{
    pub fn push_file(&self, filename: impl AsRef<Path>, hashers: Vec<HasherWrapper<Tag>>) {
        if hashers.is_empty() {
            return;
        }

        let identifier: Identifier = filename.as_ref().into();

        let operation_sender = self.operation_sender.clone();

        let mut file = unwrap_or_return!(
            OpenOptions::new().read(true).open(&filename),
            identifier,
            None,
            operation_sender
        );

        let file_size =
            unwrap_or_return!(file.metadata(), identifier, None, operation_sender).len();

        operation_sender
            .send(Operation::NewIdentifier {
                identifier: identifier.clone(),
                hashers,
            })
            .expect("Failed to send new identifier.");

        let block_count = self.block_count;
        let block_size = self.block_size;

        let id_semaphore = Arc::clone(&self.id_semaphore);
        id_semaphore.acquire();

        self.reader_threads.execute(move || {
            let block_semaphore = Arc::new(SimpleSemaphore::new(block_count));

            let mut buffers = Vec::with_capacity(block_count);
            for _ in 0..block_count {
                buffers.push(Arc::new(SyncUnsafeCell::new(vec![0u8; block_size])));
            }

            let mut sent_file_size = 0;
            let mut buffer_index = 0;

            loop {
                block_semaphore.acquire();

                // Reading data
                let size_to_read = u64::min(file_size - sent_file_size, block_size as u64);
                sent_file_size += size_to_read;

                let buffer = &buffers[buffer_index];
                buffer_index = (buffer_index + 1) % block_count;

                unwrap_or_break!(
                    file.read_exact(&mut buffer.get_mut()[..size_to_read as usize]),
                    identifier,
                    None,
                    operation_sender
                );

                // Sending data
                let block_semaphore = Arc::clone(&block_semaphore);
                let data = DataWrapper {
                    identifier: identifier.clone(),
                    data: Arc::clone(buffer),
                    semaphore: block_semaphore,
                    length: size_to_read as usize,
                    last: file_size == sent_file_size,
                    total_data_length: file_size,
                    sent_data_length: sent_file_size,
                };

                operation_sender
                    .send(Operation::Data(Arc::new(data)))
                    .expect("Failed to send data.");

                if sent_file_size == file_size {
                    break;
                }
            }

            id_semaphore.release();
        })
    }

    pub fn fragment_sender(
        &mut self,
        name: impl Into<Identifier>,
        hashers: Vec<HasherWrapper<Tag>>,
    ) -> FragmentSender<Tag> {
        self.id_semaphore.acquire();
        let id = name.into();
        self.operation_sender
            .send(Operation::NewIdentifier {
                identifier: id.clone(),
                hashers,
            })
            .expect("Failed to send new identifier.");

        FragmentSender {
            id,
            length: 0,
            sent_length: 0,
            block_size: self.block_size,
            operation_sender: self.operation_sender.clone(),
            id_semaphore: self.id_semaphore.clone(),
            block_semaphore: Arc::new(SimpleSemaphore::new(self.block_count)),
        }
    }

    pub fn end(&self) {
        self.operation_sender
            .send(Operation::EndOfNewIdentifier)
            .unwrap()
    }
}