tyler
/
proxmox-backup
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

src/tools/parallel_handler.rs: remove static lifetime bound from handler_fn

This commit is contained in:
Dietmar Maurer 2020-09-26 09:22:50 +02:00
parent 1c13afa8f9
commit b02b374b46
1 changed files with 14 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
src/tools/parallel_handler.rs
View File

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