diff --git a/src/tools/parallel_handler.rs b/src/tools/parallel_handler.rs index cfa32199..f0274afe 100644 --- a/src/tools/parallel_handler.rs +++ b/src/tools/parallel_handler.rs @@ -36,10 +36,11 @@ impl SendHandle { /// /// When done, the 'complete()' method needs to be called to check for /// outstanding errors. -pub struct ParallelHandler { +pub struct ParallelHandler<'a, I> { handles: Vec>, name: String, input: Option>, + _marker: std::marker::PhantomData<&'a ()>, } impl Clone for SendHandle { @@ -48,7 +49,7 @@ impl Clone for SendHandle { } } -impl ParallelHandler { +impl <'a, I: Send + Sync + 'static> ParallelHandler<'a, I> { /// Create a new thread pool, each thread processing incoming data /// with 'handler_fn'. @@ -57,7 +58,7 @@ impl ParallelHandler { threads: usize, handler_fn: F, ) -> Self - where F: Fn(I) -> Result<(), Error> + Send + Clone + 'static, + where F: Fn(I) -> Result<(), Error> + Send + Clone + 'a, { let mut handles = Vec::new(); let (input_tx, input_rx) = bounded::(threads); @@ -67,7 +68,14 @@ impl ParallelHandler { for i in 0..threads { let input_rx = input_rx.clone(); let abort = abort.clone(); - let handler_fn = handler_fn.clone(); + + // 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) }; + handles.push( std::thread::Builder::new() .name(format!("{} ({})", name, i)) @@ -98,6 +106,7 @@ impl ParallelHandler { input: input_tx, abort, }), + _marker: std::marker::PhantomData, } } @@ -152,7 +161,7 @@ impl ParallelHandler { } // Note: We make sure that all threads will be joined -impl Drop for ParallelHandler { +impl <'a, I> Drop for ParallelHandler<'a, I> { fn drop(&mut self) { drop(self.input.take()); loop {