make hash calculation parallel

Author: maxscherbakov

Cargo.toml

@@ -8,6 +8,9 @@ edition = "2021"
 
 [dependencies]
 chunkfs = "0.1"
+futures = "0.3"
+rayon = "1.10"
+
 
 [dev-dependencies]
 rand = "0.8.5"

runner/src/main.rs

@@ -6,15 +6,13 @@ use chunkfs::chunkers::{
     FSChunker, LeapChunker, RabinChunker, SeqChunker, SizeParams, SuperChunker, UltraChunker,
 };
 use chunkfs::hashers::Sha256Hasher;
-use chunkfs::{ChunkerRef, FileSystem};
-use chunkfs::chunkers::seq::OperationMode;
-use sbc_algorithm::decoders;
-use sbc_algorithm::encoders;
+use chunkfs::FileSystem;
+use sbc_algorithm::{decoders, encoders};
 use sbc_algorithm::{SBCMap, SBCScrubber};
 use std::collections::HashMap;
-use std::fs::OpenOptions;
-use std::io::{self, Write};
-use std::time::Instant;
+use std::fs;
+use std::io;
+
 #[allow(dead_code)]
 const MB: usize = 1024 * 1024;
 
@@ -25,72 +23,21 @@ fn generate_data(mb_size: usize) -> Vec<u8> {
 }
 
 fn main() -> io::Result<()> {
-    let chunk_size = SizeParams::new(2 * 1024, 8 * 1024, 12 * 1024);
-    let chunkers: Vec<(&str, ChunkerRef)> = vec![
-        ("super", SuperChunker::new(chunk_size.clone()).into()),
-        ("fs", FSChunker::default().into()),
-        ("leap", LeapChunker::new(chunk_size.clone()).into()),
-        ("rabin", RabinChunker::new(chunk_size.clone()).into()),
-        (
-            "seq",
-            SeqChunker::new(OperationMode::Increasing, chunk_size.clone(), Default::default()).into(),
-        ),
-        ("ultra", UltraChunker::new(chunk_size.clone()).into()),
-    ];
-    let dataset_path = "runner/files/kernels.tar";
-    let mut out_file = OpenOptions::new()
-        .write(true)
-        .create(true)
-        .truncate(false)
-        .open("out-sbc-kernels-gdelta1.csv")?;
-    writeln!(
-        out_file,
-        "Chunker\tCDCReadTime(s)\tCDCWriteTime(s)\tCDCDedupRatio\tScrubTime\tSBCReadTime\tSBCDedupRatio"
-    )?;
-
-    for (chunker_name, chunker_ref) in chunkers {
-        let data = std::fs::read(dataset_path)?;
-        let mut fs = FileSystem::new_with_scrubber(
-            HashMap::default(),
-            SBCMap::new(decoders::GdeltaDecoder),
-            Box::new(SBCScrubber::new(encoders::GdeltaEncoder)),
-            Sha256Hasher::default(),
-        );
-        let mut handle = fs.create_file("file".to_string(), chunker_ref)?;
-        let now = Instant::now();
-        fs.write_to_file(&mut handle, &data)?;
-        let cdc_write_time = now.elapsed();
-        fs.close_file(handle)?;
-        let cdc_dedup_ratio = fs.cdc_dedup_ratio();
-
-        let handle = fs.open_file_readonly("file".to_string())?;
-        let now = Instant::now();
-        fs.read_file_complete(&handle)?;
-        let cdc_read_time = now.elapsed();
-
-        let res = fs.scrub().unwrap();
-        let sbc_dedup_ratio = fs.total_dedup_ratio();
-
-        let now = Instant::now();
-        fs.read_file_complete(&handle)?;
-        let sbc_read_time = now.elapsed();
-
-        writeln!(
-            out_file,
-            "{}\t{:.10}\t{:.10}\t{:.10}\t{:.10}\t{:.10}\t{:.10}",
-            chunker_name,
-            cdc_read_time.as_secs_f64(),
-            cdc_write_time.as_secs_f64(),
-            cdc_dedup_ratio,
-            res.running_time.as_secs_f64(),
-            sbc_read_time.as_secs_f64(),
-            sbc_dedup_ratio
-        )?;
-        fs.clear_file_system()?;
-        println!("Scrubber results: {res:?}");
-        println!("CDChunking complete, dedup_ratio: {}", cdc_dedup_ratio);
-        println!("SBC dedup ratio: {}", sbc_dedup_ratio);
-        println!("delta: {}", sbc_dedup_ratio - cdc_dedup_ratio);
-    }
+    let data = fs::read("runner/files/my_data")?;
+    let chunk_size = SizeParams::new(2 * 1024, 8 * 1024, 16 * 1024);
+    let mut fs = FileSystem::new_with_scrubber(
+        HashMap::default(),
+        SBCMap::new(decoders::GdeltaDecoder),
+        Box::new(SBCScrubber::new(encoders::GdeltaEncoder)),
+        Sha256Hasher::default(),
+    );
+    let mut handle = fs.create_file("file".to_string(), SuperChunker::new(chunk_size))?;
+
+    fs.write_to_file(&mut handle, &data)?;
+    fs.close_file(handle)?;
+    let cdc_dedup_ratio = fs.cdc_dedup_ratio();
+    let res = fs.scrub().unwrap();
+    let sbc_dedup_ratio = fs.total_dedup_ratio();
+    println!("{}, {:?}, {}", cdc_dedup_ratio, res, sbc_dedup_ratio);
     Ok(())
 }

src/chunkfs_sbc.rs

@@ -1,15 +1,22 @@
 use crate::decoders::Decoder;
-use crate::encoders::{Encoder};
+use crate::encoders::Encoder;
 use crate::graph::Graph;
 use crate::{hash_functions, ChunkType, SBCHash, SBCMap};
 use chunkfs::{
     ChunkHash, Data, DataContainer, Database, IterableDatabase, Scrub, ScrubMeasurements,
 };
+use rayon::prelude::*;
+use rayon::ThreadPoolBuilder;
 use std::collections::HashMap;
 use std::io;
 use std::io::{Error, ErrorKind};
+use std::sync::{Arc, Mutex};
 use std::time::Instant;
 
+const NUM_THREADS_FOR_HASHING: usize = 4;
+
+type ClusterType<'a> = HashMap<u32, Vec<(u32, &'a mut DataContainer<SBCHash>)>>;
+
 impl<D: Decoder> Database<SBCHash, Vec<u8>> for SBCMap<D> {
     fn insert(&mut self, sbc_hash: SBCHash, chunk: Vec<u8>) -> io::Result<()> {
         self.sbc_hashmap.insert(sbc_hash, chunk);
@@ -69,29 +76,24 @@ pub struct SBCScrubber<E>
 where
     E: Encoder,
 {
-    graph: Graph,
+    graph: Arc<Mutex<Graph>>,
     encoder: E,
 }
 
 impl<E: Encoder> SBCScrubber<E> {
     pub fn new(_encoder: E) -> SBCScrubber<E> {
         SBCScrubber {
-            graph: Graph::new(),
+            graph: Arc::new(Mutex::new(Graph::new())),
             encoder: _encoder,
         }
     }
 }
 
-// impl<E: Encoder> Default for SBCScrubber<E> {
-//     fn default() -> SBCScrubber<LevenshteinEncoder> {
-//         Self::new(LevenshteinEncoder)
-//     }
-// }
-
 impl<Hash: ChunkHash, B, D: Decoder, E: Encoder> Scrub<Hash, B, SBCHash, SBCMap<D>>
     for SBCScrubber<E>
 where
-    B: IterableDatabase<Hash, DataContainer<SBCHash>>,
+    for<'data> B:
+        IterableDatabase<Hash, DataContainer<SBCHash>> + IntoParallelRefMutIterator<'data> + 'data,
 {
     fn scrub<'a>(
         &mut self,
@@ -101,28 +103,45 @@ where
     where
         Hash: 'a,
     {
-        let time_start = Instant::now();
         let mut processed_data = 0;
         let mut data_left = 0;
-        let mut clusters: HashMap<u32, Vec<(u32, &mut DataContainer<SBCHash>)>> = HashMap::new();
-        for (_, data_container) in database.iterator_mut() {
-            match data_container.extract() {
-                Data::Chunk(data) => {
-                    let sbc_hash = hash_functions::sbc_hashing(data.as_slice());
-                    let parent_hash = self.graph.add_vertex(sbc_hash);
-                    let cluster = clusters.entry(parent_hash).or_default();
-                    cluster.push((sbc_hash, data_container));
+        let pool = ThreadPoolBuilder::new()
+            .num_threads(NUM_THREADS_FOR_HASHING)
+            .build()
+            .unwrap();
+        let clusters: Arc<Mutex<ClusterType>> = Arc::new(Mutex::new(HashMap::new()));
+
+        let mut mut_refs_database: Vec<&mut DataContainer<SBCHash>> =
+            database.iterator_mut().map(|(_, b)| b).collect();
+
+        let time_start = Instant::now();
+        pool.install(|| {
+            mut_refs_database.par_iter_mut().for_each(|data_container| {
+                match data_container.extract() {
+                    Data::Chunk(data) => {
+                        let sbc_hash = hash_functions::sbc_hashing(data.as_slice());
+                        let parent_hash = self.graph.lock().unwrap().add_vertex(sbc_hash);
+                        let mut clusters_lock = clusters.lock().unwrap();
+                        let cluster = clusters_lock.entry(parent_hash).or_default();
+                        cluster.push((sbc_hash, data_container));
+                    }
+                    Data::TargetChunk(_) => {
+                        panic!()
+                    }
                 }
-                Data::TargetChunk(_) => {}
-            }
-        }
+            });
+        });
         let time_hashing = time_start.elapsed();
         println!("time for hashing: {time_hashing:?}");
-        let (clusters_data_left, clusters_processed_data) =
-            self.encoder.encode_clusters(&mut clusters, target_map);
+
+        let (clusters_data_left, clusters_processed_data) = self
+            .encoder
+            .encode_clusters(&mut clusters.lock().unwrap(), target_map);
+
         data_left += clusters_data_left;
         processed_data += clusters_processed_data;
         let running_time = time_start.elapsed();
+
         Ok(ScrubMeasurements {
             processed_data,
             running_time,