src/tools/parallel_handler.rs: remove lifetime hacks, require 'static

In theory, one can do std::mem::forget, and ignore the drop handler. With
the lifetime hack, this could result in a crash.

So we simply require 'static lifetime now (futures also needs that).
This commit is contained in:
Dietmar Maurer 2020-10-01 14:48:49 +02:00
parent df766e668f
commit a71bc08ff4
3 changed files with 10 additions and 17 deletions

View File

@ -84,7 +84,7 @@ fn verify_index_chunks(
worker: Arc<WorkerTask>, worker: Arc<WorkerTask>,
) -> Result<(), Error> { ) -> Result<(), Error> {
let errors = AtomicUsize::new(0); let errors = Arc::new(AtomicUsize::new(0));
let start_time = Instant::now(); let start_time = Instant::now();
@ -95,7 +95,7 @@ fn verify_index_chunks(
let datastore2 = Arc::clone(&datastore); let datastore2 = Arc::clone(&datastore);
let corrupt_chunks2 = Arc::clone(&corrupt_chunks); let corrupt_chunks2 = Arc::clone(&corrupt_chunks);
let verified_chunks2 = Arc::clone(&verified_chunks); let verified_chunks2 = Arc::clone(&verified_chunks);
let errors2 = &errors; let errors2 = Arc::clone(&errors);
let decoder_pool = ParallelHandler::new( let decoder_pool = ParallelHandler::new(
"verify chunk decoder", 4, "verify chunk decoder", 4,

View File

@ -50,12 +50,13 @@ async fn pull_index_chunks<I: IndexFile>(
}) })
); );
let target2 = target.clone();
let verify_pool = ParallelHandler::new( let verify_pool = ParallelHandler::new(
"sync chunk writer", 4, "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)); // println!("verify and write {}", proxmox::tools::digest_to_hex(&digest));
chunk.verify_unencrypted(size as usize, &digest)?; chunk.verify_unencrypted(size as usize, &digest)?;
target.insert_chunk(&chunk, &digest)?; target2.insert_chunk(&chunk, &digest)?;
Ok(()) Ok(())
} }
); );

View File

@ -38,11 +38,10 @@ impl<I: Send> 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<'a, I> { pub struct ParallelHandler<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> {
@ -54,11 +53,11 @@ impl<I> Clone for SendHandle<I> {
} }
} }
impl<'a, I: Send + 'static> ParallelHandler<'a, I> { impl<I: Send + 'static> ParallelHandler<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'.
pub fn new<F>(name: &str, threads: usize, handler_fn: F) -> Self pub fn new<F>(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 mut handles = Vec::new();
let (input_tx, input_rx) = bounded::<I>(threads); let (input_tx, input_rx) = bounded::<I>(threads);
@ -68,13 +67,7 @@ impl<'a, I: Send + 'static> ParallelHandler<'a, 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 = Arc::clone(&abort); let abort = Arc::clone(&abort);
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()
@ -104,7 +97,6 @@ impl<'a, I: Send + 'static> ParallelHandler<'a, I> {
input: input_tx, input: input_tx,
abort, 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 // Note: We make sure that all threads will be joined
impl<'a, I> Drop for ParallelHandler<'a, I> { impl<I> Drop for ParallelHandler<I> {
fn drop(&mut self) { fn drop(&mut self) {
drop(self.input.take()); drop(self.input.take());
while let Some(handle) = self.handles.pop() { while let Some(handle) = self.handles.pop() {