[PATCH proxmox v2 02/25] parallel-handler: introduce custom error type

Lukas Wagner
2026-03-19T09:46:41Z

Derive a custom error type using `thiserror`. For the handler functions,
we still use anyhow::Error, since that would involve bigger changes in
the callers.

Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
Reviewed-by: Arthur Bied-Charreton <a.bied-charreton@proxmox.com>
Tested-by: Arthur Bied-Charreton <a.bied-charreton@proxmox.com>
---
proxmox-parallel-handler/Cargo.toml | 1 +
proxmox-parallel-handler/src/lib.rs | 67 +++++++++++++++++++----------
2 files changed, 46 insertions(+), 22 deletions(-)

diff --git a/proxmox-parallel-handler/Cargo.toml b/proxmox-parallel-handler/Cargo.toml
index e55e7c63..5fe67889 100644
--- a/proxmox-parallel-handler/Cargo.toml
+++ b/proxmox-parallel-handler/Cargo.toml
@@ -13,3 +13,4 @@ repository.workspace = true
[dependencies]
anyhow.workspace = true
crossbeam-channel.workspace = true
+thiserror.workspace = true
diff --git a/proxmox-parallel-handler/src/lib.rs b/proxmox-parallel-handler/src/lib.rs
index 75eab184..4c2ac118 100644
--- a/proxmox-parallel-handler/src/lib.rs
+++ b/proxmox-parallel-handler/src/lib.rs
@@ -3,9 +3,25 @@
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;

-use anyhow::{bail, format_err, Error};
use crossbeam_channel::{bounded, Sender};

+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+ #[error("send failed - channel closed")]
+ ChannelClosed,
+
+ #[error("handler failed: {0}")]
+ HandlerFailed(String),
+
+ #[error("thread {name} panicked")]
+ ThreadPanicked {
+ /// The name of the thread.
+ name: String,
+ /// The panic message extracted from the panic payload.
+ message: Option<String>,
+ },
+}
+
/// A handle to send data to the worker thread (implements clone)
pub struct SendHandle {
input: Sender,
@@ -16,7 +32,7 @@ pub struct SendHandle {
fn check_abort(abort: &Mutex<Option<String>>) -> Result<(), Error> {
let guard = abort.lock().unwrap();
if let Some(err_msg) = &*guard {
- return Err(format_err!("{}", err_msg));
+ return Err(Error::HandlerFailed(err_msg.clone()));
}
Ok(())
}
@@ -25,10 +41,7 @@ impl<I: Send> SendHandle {
/// Send data to the worker threads
pub fn send(&self, input: I) -> Result<(), Error> {
check_abort(&self.abort)?;
- match self.input.send(input) {
- Ok(()) => Ok(()),
- Err(_) => bail!("send failed - channel closed"),
- }
+ self.input.send(input).map_err(|_| Error::ChannelClosed)
}
}

@@ -42,7 +55,6 @@ impl<I: Send> SendHandle {
/// outstanding errors.
pub struct ParallelHandler {
handles: Vec<JoinHandle<()>>,
- name: String,
input: Option<SendHandle>,
}

@@ -60,7 +72,7 @@ impl<I: Send + 'static> ParallelHandler {
/// with 'handler_fn'.
pub fn new<F>(name: &str, threads: usize, handler_fn: F) -> Self
where
- F: Fn(I) -> Result<(), Error> + Send + Clone + 'static,
+ F: Fn(I) -> Result<(), anyhow::Error> + Send + Clone + 'static,
{
let mut handles = Vec::new();
let (input_tx, input_rx) = bounded::(threads);
@@ -83,7 +95,7 @@ impl<I: Send + 'static> ParallelHandler {
if let Err(err) = (handler_fn)(data) {
let mut guard = abort.lock().unwrap();
if guard.is_none() {
- *guard = Some(err.to_string());
+ *guard = Some(format!("{err:#}"));
}
}
})
@@ -92,7 +104,6 @@ impl<I: Send + 'static> ParallelHandler {
}
Self {
handles,
- name: name.to_string(),
input: Some(SendHandle {
input: input_tx,
abort,
@@ -118,32 +129,44 @@ impl<I: Send + 'static> ParallelHandler {
check_abort(&abort)?;
drop(input);

- let msg_list = self.join_threads();
+ let mut msg_list = self.join_threads();

// an error might be encountered while waiting for the join
check_abort(&abort)?;

- if msg_list.is_empty() {
- return Ok(());
+ if let Some(e) = msg_list.pop() {
+ // Any error here is due to a thread panicking - let's just report that
+ // last panic that occurred.
+ Err(e)
+ } else {
+ Ok(())
}
- Err(format_err!("{}", msg_list.join("\n")))
}

- fn join_threads(&mut self) -> Vec<String> {
+ fn join_threads(&mut self) -> Vec<Error> {
let mut msg_list = Vec::new();

- let mut i = 0;
while let Some(handle) = self.handles.pop() {
+ let thread_name = handle.thread().name().unwrap_or("<unknown>").to_string();
+
if let Err(panic) = handle.join() {
- if let Some(panic_msg) = panic.downcast_ref::<&str>() {
- msg_list.push(format!("thread {} ({i}) panicked: {panic_msg}", self.name));
- } else if let Some(panic_msg) = panic.downcast_ref::<String>() {
- msg_list.push(format!("thread {} ({i}) panicked: {panic_msg}", self.name));
+ if let Some(message) = panic.downcast_ref::<&str>() {
+ msg_list.push(Error::ThreadPanicked {
+ name: thread_name,
+ message: Some(message.to_string()),
+ });
+ } else if let Some(message) = panic.downcast_ref::<String>() {
+ msg_list.push(Error::ThreadPanicked {
+ name: thread_name,
+ message: Some(message.to_string()),
+ });
} else {
- msg_list.push(format!("thread {} ({i}) panicked", self.name));
+ msg_list.push(Error::ThreadPanicked {
+ name: thread_name,
+ message: None,
+ });
}
}
- i += 1;
}
msg_list
}
--
2.47.3