From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [IPv6:2a0f:8001:1:32::40]) by lore.proxmox.com (Postfix) with ESMTPS id B4B421FF0B4 for ; Tue, 10 Nov 2026 16:06:01 +0100 (CET) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id 91806216A3; Tue, 10 Nov 2026 16:05:48 +0100 (CET) Message-ID: <00aa1666-9ded-42a6-b006-da06c5ff78f5@proxmox.com> Date: Tue, 10 Nov 2026 16:05:31 +0100 MIME-Version: 1.0 User-Agent: Mozilla Thunderbird Beta Subject: Re: [RFC PATCH proxmox-backup 1/4] tape: drive: add a pool to use several drives for one task To: Jakob-Tobias Winter , pbs-devel@lists.proxmox.com References: <20261106133826.1692620-2-tobias@stuttgart.de> <20261106133826.1692620-3-tobias@stuttgart.de> Content-Language: en-US From: Dominik Csapak In-Reply-To: <20261106133826.1692620-3-tobias@stuttgart.de> Content-Type: text/plain; charset=UTF-8; format=flowed Content-Transfer-Encoding: 7bit X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1794323131618 X-SPAM-LEVEL: Spam detection results: 0 AWL 0.313 Adjusted score from AWL reputation of From: address DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment (newer systems) RCVD_IN_DNSWL_MED -2.3 Sender listed at https://www.dnswl.org/, medium trust SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: 4YBXOA2KK5BHRMDE4CV34NGEUKT2VYH7 X-Message-ID-Hash: 4YBXOA2KK5BHRMDE4CV34NGEUKT2VYH7 X-MailFrom: d.csapak@proxmox.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; loop; banned-address; emergency; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header X-Mailman-Version: 3.3.10 Precedence: list List-Id: Proxmox Backup Server development discussion List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: some comments inline On 11/6/26 2:39 PM, Jakob-Tobias Winter wrote: > Restoring a media set loads one tape after the other into a single > drive, even if the library has more drives that are idle. In > preparation for using them in parallel, add a pool of drives that runs > a list of jobs, each needing one media, on all of its drives. > > With a single drive, the jobs run in the given order in the calling > thread, exactly like the existing loops that load one media after the > other. > > With several drives, each drive runs in its own thread. A drive takes > the job for the media already loaded in it, else the next one whose > media is not loaded in another drive of the pool, as loading such a > media fails until that drive moves on. Changer operations are > serialized, so two drives do not pick the same free slot when > unloading a media. The first error stops all drives and is returned > once all of them finished. not sure i understand completely here: normally we load the tapes simply in order, we don't 'take what's in the drive'. What happens if this is an unrelated tape? this would also make the loop a bit easier, see below > > As the lines of all drives are interleaved in the task log, their > messages start with the name of the drive, and the jobs get this > prefix for their own messages. With a single drive, the prefix is > empty, so the log stays as it is. > > Signed-off-by: Jakob-Tobias Winter > --- > src/tape/drive/mod.rs | 41 ++++- > src/tape/drive/pool.rs | 358 +++++++++++++++++++++++++++++++++++++++++ > 2 files changed, 393 insertions(+), 6 deletions(-) > create mode 100644 src/tape/drive/pool.rs > > diff --git a/src/tape/drive/mod.rs b/src/tape/drive/mod.rs > index fa84fd54a..4df30be72 100644 > --- a/src/tape/drive/mod.rs > +++ b/src/tape/drive/mod.rs > @@ -1,6 +1,7 @@ > //! Tape drivers > > use std::path::PathBuf; > +use std::sync::{Mutex, PoisonError}; > > use anyhow::{Error, bail, format_err}; > use nix::fcntl::OFlag; > @@ -39,6 +40,9 @@ mod virtual_tape; > mod lto; > pub use lto::*; > > +mod pool; > +pub use pool::*; > + > /// Tape driver interface > pub trait TapeDriver { > /// Flush all data to the tape > @@ -378,11 +382,28 @@ pub fn request_and_load_media( > drive: &str, > label: &MediaLabel, > notification_mode: &TapeNotificationMode, > +) -> Result<(Box, MediaId), Error> { > + request_and_load_media_locked(worker, config, drive, label, notification_mode, None, "") > +} > + > +/// Like [`request_and_load_media`], but holds `changer_lock` while the changer moves media and > +/// starts each log message with `log_prefix`. > +/// > +/// This serializes the changer operations of several drives used by the same task, and tells > +/// their messages apart in the task log. > +pub fn request_and_load_media_locked( > + worker: &dyn WorkerTaskContext, > + config: &SectionConfigData, > + drive: &str, > + label: &MediaLabel, > + notification_mode: &TapeNotificationMode, > + changer_lock: Option<&Mutex<()>>, > + log_prefix: &str, > ) -> Result<(Box, MediaId), Error> { > let check_label = |handle: &mut dyn TapeDriver, uuid: &proxmox_uuid::Uuid| { > if let Ok((Some(media_id), _)) = handle.read_label() { > info!( > - "found media label {} ({})", > + "{log_prefix}found media label {} ({})", > media_id.label.label_text, media_id.label.uuid, > ); > > @@ -422,7 +443,7 @@ pub fn request_and_load_media( > new: TapeRequestError| > -> Result<(), Error> { > if new != *old { > - info!("{new}"); > + info!("{log_prefix}{new}"); > let (device_type, device) = if let Some(changer) = changer { > ("changer", changer.as_str()) > } else { > @@ -430,7 +451,8 @@ pub fn request_and_load_media( > }; > > info!( > - "Please insert media '{label_text}' into {device_type} '{device}'" > + "{log_prefix}Please insert media '{label_text}' into {device_type} \ > + '{device}'" > ); > send_load_media_notification( > notification_mode, > @@ -455,13 +477,20 @@ pub fn request_and_load_media( > std::thread::sleep(std::time::Duration::from_millis(100)); > } > } else if drive_config.changer.is_none() { > - info!("Checking for media '{label_text}' in drive '{drive}'"); > + info!( > + "{log_prefix}Checking for media '{label_text}' in drive '{drive}'" > + ); > } else { > - info!("trying to load media '{label_text}' into drive '{drive}'"); > + info!( > + "{log_prefix}trying to load media '{label_text}' into drive \ > + '{drive}'" > + ); > } > > if drive_config.changer.is_some() { > let mut changer = MtxMediaChanger::with_drive_config(&drive_config)?; > + let _changer_guard = changer_lock > + .map(|lock| lock.lock().unwrap_or_else(PoisonError::into_inner)); i don't really like that we more or less ignore the poisonerror here, but see below > if let Err(err) = changer.load_media(&label_text) { > update_and_log_request_error( > &mut last_error, > @@ -485,7 +514,7 @@ pub fn request_and_load_media( > let request_error = match handle.read_label() { > Ok((Some(media_id), _)) if media_id.label.uuid == label.uuid => { > info!( > - "found media label {} ({})", > + "{log_prefix}found media label {} ({})", > media_id.label.label_text, > media_id.label.uuid.to_string(), > ); > diff --git a/src/tape/drive/pool.rs b/src/tape/drive/pool.rs > new file mode 100644 > index 000000000..3ee7a2984 > --- /dev/null > +++ b/src/tape/drive/pool.rs > @@ -0,0 +1,358 @@ > +//! Use several tape drives for one task > + > +use std::collections::{HashMap, HashSet, VecDeque}; > +use std::sync::atomic::{AtomicBool, Ordering}; > +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; > + > +use anyhow::{Error, bail}; > +use tracing::{info, warn}; > + > +use proxmox_log::LogContext; > +use proxmox_section_config::SectionConfigData; > +use proxmox_worker_task::WorkerTaskContext; > + > +use pbs_api_types::LtoTapeDrive; > +use pbs_tape::{ElementStatus, MtxStatus}; > + > +use crate::tape::{ > + MediaId, TapeNotificationMode, > + changer::{MediaChange, MtxMediaChanger}, > + file_formats::MediaLabel, > +}; > + > +use super::{TapeDriver, request_and_load_media, request_and_load_media_locked}; > + > +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { > + mutex.lock().unwrap_or_else(PoisonError::into_inner) > +} > + > +/// Task context for the jobs of one drive, also stops when a job on another drive failed. > +struct DriveJobContext { > + worker: Arc, > + stop: Arc, > +} > + > +impl WorkerTaskContext for DriveJobContext { > + fn abort_requested(&self) -> bool { > + self.stop.load(Ordering::SeqCst) || self.worker.abort_requested() > + } > + > + fn shutdown_requested(&self) -> bool { > + self.worker.shutdown_requested() > + } > +} > + > +/// Stops the other drives if a drive thread panics. > +struct StopOnPanic<'a>(&'a AtomicBool); > + > +impl Drop for StopOnPanic<'_> { > + fn drop(&mut self) { > + if std::thread::panicking() { > + self.0.store(true, Ordering::SeqCst); > + } > + } > +} > + instead of doing this whole dance with creating a new workertask context, handling panic on mutex, etc. i think we could simply: - panic on a poisonerror: this is only triggered in cases we want to stop, so this is fine IMO - check the normal worker task - call abort_task_nowait (or whatever it's called in proxmox_rest_server) to abort the whole task, this will then abort all other threads the next time check_abort will be checked or am i missing something here? > +/// A media loaded into a drive of a [`DrivePool`]. > +pub struct LoadedMedia { > + /// The task context the job has to use for abort checks. > + pub worker: Arc, > + pub drive: Box, > + pub media_id: MediaId, > + /// The start of the job's log messages, names the drive if the pool has several. > + pub log_prefix: String, > +} > + > +/// A set of drives that work on the media of one task in parallel. > +pub struct DrivePool<'a> { > + worker: Arc, > + config: &'a SectionConfigData, > + drives: &'a [String], > + notification_mode: &'a TapeNotificationMode, > + /// Serializes the changer operations of all drives. > + changer_lock: Mutex<()>, > +} > + > +impl<'a> DrivePool<'a> { > + /// Create a pool of `drives`, which the caller has to lock. > + pub fn new( > + worker: Arc, > + config: &'a SectionConfigData, > + drives: &'a [String], > + notification_mode: &'a TapeNotificationMode, > + ) -> Self { > + Self { > + worker, > + config, > + drives, > + notification_mode, > + changer_lock: Mutex::new(()), > + } > + } > + > + /// Load the media of each job into a drive and run `func` for it. > + /// > + /// With a single drive, the jobs run in the given order in the calling thread. With several > + /// drives, each drive runs in its own thread and takes the next job whose media is not loaded > + /// in another drive of the pool, preferring the media loaded in itself. The first error stops > + /// the jobs on all drives and is returned once all of them finished. > + /// > + /// The results are returned in the order the jobs finished. > + pub fn run(&self, jobs: Vec<(MediaLabel, T)>, func: F) -> Result, Error> > + where > + T: Send, > + R: Send, > + F: Fn(LoadedMedia, T) -> Result + Sync, > + { > + if let [drive_name] = self.drives { > + let mut results = Vec::with_capacity(jobs.len()); > + for (label, job) in jobs { > + let (drive, media_id) = request_and_load_media( > + &*self.worker, > + self.config, > + drive_name, > + &label, > + self.notification_mode, > + )?; > + let media = LoadedMedia { > + worker: Arc::clone(&self.worker), > + drive, > + media_id, > + log_prefix: String::new(), > + }; > + results.push(func(media, job)?); > + } > + return Ok(results); > + } > + > + let queue = Mutex::new(VecDeque::from(jobs)); this here is a vecdeque, but we only call remove(pos) on it. why not simply do a pop_front each time a drive wants the next job and simply loads the tape? (or use a vec and reverse the order when pushing in, so a `pop()` returns the first) i think this would make the code much shorter > + let results = Mutex::new(Vec::new()); > + let first_error = Mutex::new(None); > + let stop = Arc::new(AtomicBool::new(false)); > + let log_context = LogContext::current(); > + > + std::thread::scope(|scope| { > + for drive_name in self.drives { > + let worker: Arc = Arc::new(DriveJobContext { > + worker: Arc::clone(&self.worker), > + stop: Arc::clone(&stop), > + }); > + let (queue, results, first_error, stop, func) = > + (&queue, &results, &first_error, &stop, &func); > + let log_context = log_context.clone(); > + > + scope.spawn(move || { > + let run = || { > + let _stop_on_panic = StopOnPanic(stop); > + let res = self.drive_loop(drive_name, worker, queue, results, func); > + if let Err(err) = res { > + let mut first_error = lock(first_error); > + if first_error.is_none() { > + if !stop.load(Ordering::SeqCst) && !self.worker.abort_requested() { > + warn!( > + "drive '{drive_name}': failed, stopping all drives - {err}" > + ); > + } > + *first_error = Some(err); > + } > + stop.store(true, Ordering::SeqCst); > + } > + }; > + match log_context { > + Some(log_context) => log_context.sync_scope(run), > + None => run(), > + } > + }); > + } > + }); > + > + if let Some(err) = lock(&first_error).take() { > + return Err(err); > + } > + > + let left = lock(&queue); > + if !left.is_empty() { > + // only possible without drives, a drive always takes the media loaded in itself > + let labels: Vec<&str> = left > + .iter() > + .map(|(label, _)| label.label_text.as_str()) > + .collect(); > + bail!("unable to process media {}", labels.join(", ")); > + } > + > + Ok(results.into_inner().unwrap_or_else(PoisonError::into_inner)) > + } > + > + fn drive_loop( > + &self, > + drive_name: &str, > + worker: Arc, > + queue: &Mutex>, > + results: &Mutex>, > + func: &F, > + ) -> Result<(), Error> > + where > + F: Fn(LoadedMedia, T) -> Result, > + { > + // the lines of all drives are interleaved in the task log > + let log_prefix = format!("drive '{drive_name}': "); > + > + loop { > + worker.check_abort()?; > + > + let (label, job) = { > + let mut queue = lock(queue); > + if queue.is_empty() { > + return Ok(()); > + } > + > + let (own, busy) = self.loaded_media(drive_name, &log_prefix); > + match next_job(&queue, own.as_deref(), &busy) { > + Some(pos) => queue.remove(pos).expect("position is valid"), > + None => return Ok(()), > + } IMO this reads very strangely. this assumes that only valid tapes are loaded in the relevant drives, and that relevant tapes that are loaded, are loaded in drives that belong to this job (or did i misread here?) i don't think either must be true, especially when users manually move tapes around (for whatever reason) I'd do: build a list of necessary tapes each drive unloads it's current media so every participating drive is empty (takes a few minutes probably but it's not a large part compared with the actual drive restore for modern lto versions, also the changer must have capacity to hold all drives anyway...) if there is any tape loaded in a unrelated drive, abort right then. then every drive: - takes and removes the first media from the list - loads and restores that media - unloads it again this makes the whole procedure much more straight forward and easier to reason about. what do you think? > + }; > + > + info!("{log_prefix}using media '{}'", label.label_text); > + > + let (drive, media_id) = request_and_load_media_locked( > + &*worker, > + self.config, > + drive_name, > + &label, > + self.notification_mode, > + Some(&self.changer_lock), > + &log_prefix, > + )?; > + let media = LoadedMedia { > + worker: Arc::clone(&worker), > + drive, > + media_id, > + log_prefix: log_prefix.clone(), > + }; > + let result = func(media, job)?; > + lock(results).push(result); > + > + info!("{log_prefix}done with media '{}'", label.label_text); > + } > + } > + > + /// The label texts of the media loaded in `drive_name` and in the other drives of this pool. > + /// > + /// Only drives in a changer can report this, errors are ignored (loading a media that is in > + /// another drive then fails and is retried by [`request_and_load_media_locked`]). > + fn loaded_media( > + &self, > + drive_name: &str, > + log_prefix: &str, > + ) -> (Option, HashSet) { > + let mut own = None; > + let mut others = HashSet::new(); > + let mut status_list: HashMap> = HashMap::new(); > + > + let _changer_guard = lock(&self.changer_lock); > + > + for name in self.drives { > + let Ok(drive) = self.config.lookup::("lto", name) else { > + continue; > + }; > + let Some(changer) = drive.changer.clone() else { > + continue; > + }; > + > + let status = status_list.entry(changer).or_insert_with_key(|changer| { > + match MtxMediaChanger::with_drive_config(&drive).and_then(|mut c| c.status()) { > + Ok(status) => Some(status), > + Err(err) => { > + warn!("{log_prefix}unable to get status of changer '{changer}' - {err}"); > + None > + } > + } > + }); > + > + let drive_status = status.as_ref().and_then(|status| { > + status > + .drives > + .get(drive.changer_drivenum.unwrap_or(0) as usize) > + }); > + if let Some(drive_status) = drive_status { > + if let ElementStatus::VolumeTag(ref tag) = drive_status.status { > + if name == drive_name { > + own = Some(tag.clone()); > + } else { > + others.insert(tag.clone()); > + } > + } > + } > + } > + > + (own, others) > + } > +} > + > +/// Choose the next job for a drive: the media loaded in it, else the first one in order that is > +/// not loaded in another drive (loading that would fail, the other drive will take it). > +fn next_job( > + queue: &VecDeque<(MediaLabel, T)>, > + own: Option<&str>, > + busy: &HashSet, > +) -> Option { > + if let Some(own) = own { > + if let Some(pos) = queue.iter().position(|(label, _)| label.label_text == own) { > + return Some(pos); > + } > + } > + queue > + .iter() > + .position(|(label, _)| !busy.contains(&label.label_text)) > +} > + > +#[cfg(test)] > +mod test { > + use std::collections::{HashSet, VecDeque}; > + > + use proxmox_uuid::Uuid; > + > + use super::next_job; > + use crate::tape::file_formats::MediaLabel; > + > + fn queue(labels: &[&str]) -> VecDeque<(MediaLabel, ())> { > + labels > + .iter() > + .map(|label_text| { > + let label = MediaLabel { > + label_text: label_text.to_string(), > + uuid: Uuid::generate(), > + ctime: 0, > + pool: None, > + }; > + (label, ()) > + }) > + .collect() > + } > + > + #[test] > + fn next_job_in_order() { > + let queue = queue(&["A", "B", "C"]); > + assert_eq!(next_job(&queue, None, &HashSet::new()), Some(0)); > + // the media in this drive is not in the queue > + assert_eq!(next_job(&queue, Some("X"), &HashSet::new()), Some(0)); > + } > + > + #[test] > + fn next_job_prefers_own_media() { > + let queue = queue(&["A", "B", "C"]); > + assert_eq!(next_job(&queue, Some("C"), &HashSet::new()), Some(2)); > + } > + > + #[test] > + fn next_job_skips_media_in_other_drives() { > + let queue = queue(&["A", "B", "C"]); > + let busy = HashSet::from(["A".to_string(), "B".to_string()]); > + assert_eq!(next_job(&queue, None, &busy), Some(2)); > + > + let busy = HashSet::from(["A".to_string(), "B".to_string(), "C".to_string()]); > + assert_eq!(next_job(&queue, None, &busy), None); > + assert_eq!(next_job(&queue, Some("B"), &busy), Some(1)); > + } > +}