// Computes file_table and chunk_table of a manifest using a parallel algorithm.
// All the parallel work is spawned in the specified thread pool.
Computes file_table and chunk_table of a manifest using a parallel algorithm.
All the parallel work is spawned in the specified thread pool.
Computes file_table and chunk_table of a manifest using a parallel algorithm . All the parallel work is spawned in the specified thread pool .
fn build_chunk_table_parallel(
    thread_pool: &mut scoped_threadpool::Pool,
    metrics: &ManifestMetrics,
    log: &ReplicaLogger,
    root: &Path,
    files: Vec<FileWithSize>,
    max_chunk_size: u32,
    chunk_actions: Vec<ChunkAction>,
) -> (Vec<FileInfo>, Vec<ChunkInfo>) {
    // Build a chunk table and file table filled with blank hashes.
    let mut chunk_table: Vec<ChunkInfo> = {
        let mut chunks = Vec::with_capacity(chunk_actions.len());
        for (file_index, FileWithSize(_, size_bytes)) in files.iter().enumerate() {
            let n = count_chunks(*size_bytes, max_chunk_size);
            for i in 0..n {
                let offset = i as u64 * max_chunk_size as u64;
                let size_bytes = (size_bytes - offset).min(max_chunk_size as u64) as u32;
                chunks.push(ChunkInfo {
                    file_index: file_index as u32,
                    offset,
                    size_bytes,
                    hash: [0; 32],
                });
            }
        }
        chunks
    };
    assert_eq!(chunk_table.len(), chunk_actions.len());
    let mut file_table: Vec<FileInfo> = files
        .into_iter()
        .map(|FileWithSize(relative_path, size_bytes)| FileInfo {
            relative_path,
            size_bytes,
            hash: [0; 32],
        })
        .collect();
    // We cache the files that are currently being hashed to avoid opening them
    // individually for each chunk. The values in the cache are weak references,
    // so the last thread that has a strong reference will release the value
    // and close the corresponding file.
    // This way we keep the number of files opened at the same time
    // low (it doesn't exceed the number of the threads).
    let file_cache: Arc<Mutex<HashMap<u32, Weak<ScopedMmap>>>> =
        Arc::new(Mutex::new(HashMap::new()));
    // Compute real chunk hashes in parallel.
    // NB. We must populate hashes of all the chunks in a file before we compute
    // file hashes.
    thread_pool.scoped(|scope| {
        for (chunk_idx, chunk_info) in chunk_table.iter_mut().enumerate() {
            let chunk_action = chunk_actions[chunk_idx].clone();
            let file_path = root.join(&file_table[chunk_info.file_index as usize].relative_path);
            let file_size = file_table[chunk_info.file_index as usize].size_bytes;
            let file_cache = Arc::clone(&file_cache);
            scope.execute(move || {
		let recompute_chunk_hash = || {
		    let mmap: Arc<ScopedMmap> = if file_size > max_chunk_size as u64 {
			// We only use the file cache if there is more than one chunk in the file,
			// otherwise the synchronization cost is unnecessary.
			let mut cache = file_cache.lock().unwrap();
			match cache.get(&chunk_info.file_index).and_then(Weak::upgrade) {
                            Some(mmap) => mmap,
                            None => {
				let mmap = Arc::new(
                                ScopedMmap::from_path(&file_path).expect("failed to open file"),
				);
				cache.insert(chunk_info.file_index, Arc::downgrade(&mmap));
				mmap
                            }
			}
                    } else {
			Arc::new(ScopedMmap::from_path(&file_path).expect("failed to open file"))
                    };
                    let data = mmap.as_slice();
		    let mut hasher = chunk_hasher();
                    let chunk_start = chunk_info.offset as usize;
                    let chunk_end = chunk_start + chunk_info.size_bytes as usize;
                    hasher.write(&data[chunk_start..chunk_end]);
                    hasher.finish()
                };
                chunk_info.hash = match chunk_action {
                    ChunkAction::Recompute => {
			metrics.chunk_bytes.with_label_values(&[LABEL_VALUE_HASHED]).inc_by(chunk_info.size_bytes as u64);
			recompute_chunk_hash()
		    },
                    ChunkAction::RecomputeAndCompare(precomputed_hash) => {
			metrics.chunk_bytes.with_label_values(&[LABEL_VALUE_HASHED_AND_COMPARED]).inc_by(chunk_info.size_bytes as u64);
			let recomputed_hash = recompute_chunk_hash();
			debug_assert_eq!(recomputed_hash, precomputed_hash);
                        if recomputed_hash != precomputed_hash {
                            metrics.reused_chunk_hash_error_count.inc();
                            error!(
                                log,
                                "{}: Hash mismatch in chunk with index {} in file {}, recomputed hash {:?}, reused hash {:?}",
                                CRITICAL_ERROR_REUSED_CHUNK_HASH,
                                chunk_idx,
                                file_path.display(),
                                chunk_info.hash,
                                precomputed_hash
                            );
                        }
                        recomputed_hash
                    }
		    ChunkAction::UseHash(precomputed_hash) => {
			metrics.chunk_bytes.with_label_values(&[LABEL_VALUE_REUSED]).inc_by(chunk_info.size_bytes as u64);
			precomputed_hash
		    },
                };
            });
        }
    });
    // After we computed all the chunk hashes, we can finally compute file hashes.
    for (file_index, file_info) in file_table.iter_mut().enumerate() {
        let mut hasher = file_hasher();
        let chunk_range = file_chunk_range(&chunk_table, file_index);
        (chunk_range.len() as u32).update_hash(&mut hasher);
        for chunk_idx in chunk_range {
            write_chunk_hash(&mut hasher, &chunk_table[chunk_idx])
        }
        file_info.hash = hasher.finish();
    }
    (file_table, chunk_table)
}
fn build_chunk_table_parallel ( thread_pool : & mut scoped_threadpool :: Pool , metrics : & ManifestMetrics , log : & ReplicaLogger , root : & Path , files : Vec < FileWithSize > , max_chunk_size : u32 , chunk_actions : Vec < ChunkAction > , ) -> ( Vec < FileInfo > , Vec < ChunkInfo > ) { // Build a chunk table and file table filled with blank hashes. let mut chunk_table : Vec < ChunkInfo > = { let mut chunks = Vec :: with_capacity ( chunk_actions . len ( ) ) ; for ( file_index , FileWithSize ( _ , size_bytes ) ) in files . iter ( ) . enumerate ( ) { let n = count_chunks ( * size_bytes , max_chunk_size ) ; for i in 0 .. n { let offset = i as u64 * max_chunk_size as u64 ; let size_bytes = ( size_bytes - offset ) . min ( max_chunk_size as u64 ) as u32 ; chunks . push ( ChunkInfo { file_index : file_index as u32 , offset , size_bytes , hash : [ 0 ; 32 ] , } ) ; } } chunks } ; assert_eq ! ( chunk_table len ( ) chunk_actions len ( ) ) ; let mut file_table : Vec < FileInfo > = files . into_iter ( ) . map ( | FileWithSize ( relative_path , size_bytes ) | FileInfo { relative_path , size_bytes , hash : [ 0 ; 32 ] , } ) . collect ( ) ; // We cache the files that are currently being hashed to avoid opening them // individually for each chunk. The values in the cache are weak references, // so the last thread that has a strong reference will release the value // and close the corresponding file. // This way we keep the number of files opened at the same time // low (it doesn't exceed the number of the threads). let file_cache : Arc < Mutex < HashMap < u32 , Weak < ScopedMmap > > > > = Arc :: new ( Mutex :: new ( HashMap :: new ( ) ) ) ; // Compute real chunk hashes in parallel. // NB. We must populate hashes of all the chunks in a file before we compute // file hashes. thread_pool . scoped ( | scope | { for ( chunk_idx , chunk_info ) in chunk_table . iter_mut ( ) . enumerate ( ) { let chunk_action = chunk_actions [ chunk_idx ] . clone ( ) ; let file_path = root . join ( & file_table [ chunk_info . file_index as usize ] . relative_path ) ; let file_size = file_table [ chunk_info . file_index as usize ] . size_bytes ; let file_cache = Arc :: clone ( & file_cache ) ; scope . execute ( move | | { let recompute_chunk_hash = | | { let mmap : Arc < ScopedMmap > = if file_size > max_chunk_size as u64 { // We only use the file cache if there is more than one chunk in the file, // otherwise the synchronization cost is unnecessary. let mut cache = file_cache . lock ( ) . unwrap ( ) ; match cache . get ( & chunk_info . file_index ) . and_then ( Weak :: upgrade ) { Some ( mmap ) => mmap , None => { let mmap = Arc :: new ( ScopedMmap :: from_path ( & file_path ) . expect ( " " ) , ) ; cache . insert ( chunk_info . file_index , Arc :: downgrade ( & mmap ) ) ; mmap } } } else { Arc :: new ( ScopedMmap :: from_path ( & file_path ) . expect ( " " ) ) } ; let data = mmap . as_slice ( ) ; let mut hasher = chunk_hasher ( ) ; let chunk_start = chunk_info . offset as usize ; let chunk_end = chunk_start + chunk_info . size_bytes as usize ; hasher . write ( & data [ chunk_start .. chunk_end ] ) ; hasher . finish ( ) } ; chunk_info . hash = match chunk_action { ChunkAction :: Recompute => { metrics . chunk_bytes . with_label_values ( & [ LABEL_VALUE_HASHED ] ) . inc_by ( chunk_info . size_bytes as u64 ) ; recompute_chunk_hash ( ) } , ChunkAction :: RecomputeAndCompare ( precomputed_hash ) => { metrics . chunk_bytes . with_label_values ( & [ LABEL_VALUE_HASHED_AND_COMPARED ] ) . inc_by ( chunk_info . size_bytes as u64 ) ; let recomputed_hash = recompute_chunk_hash ( ) ; debug_assert_eq ! ( recomputed_hash precomputed_hash ) ; if recomputed_hash != precomputed_hash { metrics . reused_chunk_hash_error_count . inc ( ) ; error ! ( log " " CRITICAL_ERROR_REUSED_CHUNK_HASH chunk_idx file_path display ( ) chunk_info hash precomputed_hash ) ; } recomputed_hash } ChunkAction :: UseHash ( precomputed_hash ) => { metrics . chunk_bytes . with_label_values ( & [ LABEL_VALUE_REUSED ] ) . inc_by ( chunk_info . size_bytes as u64 ) ; precomputed_hash } , } ; } ) ; } } ) ; // After we computed all the chunk hashes, we can finally compute file hashes. for ( file_index , file_info ) in file_table . iter_mut ( ) . enumerate ( ) { let mut hasher = file_hasher ( ) ; let chunk_range = file_chunk_range ( & chunk_table , file_index ) ; ( chunk_range . len ( ) as u32 ) . update_hash ( & mut hasher ) ; for chunk_idx in chunk_range { write_chunk_hash ( & mut hasher , & chunk_table [ chunk_idx ] ) } file_info . hash = hasher . finish ( ) ; } ( file_table , chunk_table ) }
Computes file_table and chunk_table of a manifest using a parallel algorithm.
Computes file_table and chunk_table of a manifest using a parallel algorithm .
thread_pool
&mut scoped_threadpool::Pool
metrics
&ManifestMetrics
log
&ReplicaLogger
root
&Path
files
Vec<FileWithSize>
max_chunk_size
u32
chunk_actions
Vec<ChunkAction>