diff --git a/src/backup/verify.rs b/src/backup/verify.rs index fd48d907..0c55305f 100644 --- a/src/backup/verify.rs +++ b/src/backup/verify.rs @@ -84,7 +84,7 @@ fn verify_index_chunks( worker: Arc, ) -> Result<(), Error> { - let errors = AtomicUsize::new(0); + let errors = Arc::new(AtomicUsize::new(0)); let start_time = Instant::now(); @@ -95,7 +95,7 @@ fn verify_index_chunks( let datastore2 = Arc::clone(&datastore); let corrupt_chunks2 = Arc::clone(&corrupt_chunks); let verified_chunks2 = Arc::clone(&verified_chunks); - let errors2 = &errors; + let errors2 = Arc::clone(&errors); let decoder_pool = ParallelHandler::new( "verify chunk decoder", 4, diff --git a/src/client/pull.rs b/src/client/pull.rs index d88d64f9..d69cf485 100644 --- a/src/client/pull.rs +++ b/src/client/pull.rs @@ -50,12 +50,13 @@ async fn pull_index_chunks( }) ); + let target2 = target.clone(); let verify_pool = ParallelHandler::new( "sync chunk writer", 4, - |(chunk, digest, size): (DataBlob, [u8;32], u64)| { + move |(chunk, digest, size): (DataBlob, [u8;32], u64)| { // println!("verify and write {}", proxmox::tools::digest_to_hex(&digest)); chunk.verify_unencrypted(size as usize, &digest)?; - target.insert_chunk(&chunk, &digest)?; + target2.insert_chunk(&chunk, &digest)?; Ok(()) } ); diff --git a/src/tools/parallel_handler.rs b/src/tools/parallel_handler.rs index 36a7b859..5482ccb3 100644 --- a/src/tools/parallel_handler.rs +++ b/src/tools/parallel_handler.rs @@ -38,11 +38,10 @@ impl SendHandle { /// /// When done, the 'complete()' method needs to be called to check for /// outstanding errors. -pub struct ParallelHandler<'a, I> { +pub struct ParallelHandler { handles: Vec>, name: String, input: Option>, - _marker: std::marker::PhantomData<&'a ()>, } impl Clone for SendHandle { @@ -54,11 +53,11 @@ impl Clone for SendHandle { } } -impl<'a, I: Send + 'static> ParallelHandler<'a, I> { +impl ParallelHandler { /// Create a new thread pool, each thread processing incoming data /// with 'handler_fn'. pub fn new(name: &str, threads: usize, handler_fn: F) -> Self - where F: Fn(I) -> Result<(), Error> + Send + Clone + 'a, + where F: Fn(I) -> Result<(), Error> + Send + Clone + 'static, { let mut handles = Vec::new(); let (input_tx, input_rx) = bounded::(threads); @@ -68,13 +67,7 @@ impl<'a, I: Send + 'static> ParallelHandler<'a, I> { for i in 0..threads { let input_rx = input_rx.clone(); let abort = Arc::clone(&abort); - - // Erase the 'a lifetime bound. This is safe because we - // join all thread in the drop handler. - let handler_fn: Box Result<(), Error> + Send + 'a> = - Box::new(handler_fn.clone()); - let handler_fn: Box Result<(), Error> + Send + 'static> = - unsafe { std::mem::transmute(handler_fn) }; + let handler_fn = handler_fn.clone(); handles.push( std::thread::Builder::new() @@ -104,7 +97,6 @@ impl<'a, I: Send + 'static> ParallelHandler<'a, I> { input: input_tx, abort, }), - _marker: std::marker::PhantomData, } } @@ -164,7 +156,7 @@ impl<'a, I: Send + 'static> ParallelHandler<'a, I> { } // Note: We make sure that all threads will be joined -impl<'a, I> Drop for ParallelHandler<'a, I> { +impl Drop for ParallelHandler { fn drop(&mut self) { drop(self.input.take()); while let Some(handle) = self.handles.pop() {