From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [45.144.208.40]) by lore.proxmox.com (Postfix) with ESMTPS id 3A6D31FF0AF for ; Fri, 06 Nov 2026 14:38:56 +0100 (CET) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id DFE42216F7; Fri, 06 Nov 2026 14:38:50 +0100 (CET) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=stuttgart.de; s=s1022; t=1793972325; bh=LtrW//VYKejCkx4gifPjAmcUmX9cw5vDQerVisUtqvA=; h=From:To:Cc:Subject:Date:In-Reply-To:References:From; b=qcdiQlAV1xg0ssSf3JnS5aAlwday8whDged9KCw48uhZN+Xh9XX3nP3W84ptgwMG4 RHXdkoEYy2MrtCL4rcJ3a3NTdfYYyg6QgYUyo5DgOaYnbCYw0CNfW6Koy48SoeGn4L BlGK7OsiWtJwgW+2DD4xl+eMS7CrQUbTYpKXcAYI3QMFp3dBTQReZVD0GEwlAcfW/6 Ca23olzwJfPrtDz7+r227NMleEapaByP01Oaey9RcFZB5j3jHoDYM+WpYCE8FBG/vb TzziOKKruNvcMF5aWJVx+6/604p/6g0dJRiynt4e286bR9o2CyWYoXgUHnghf5PnuX ptyI7l7yFkzMA== From: Jakob-Tobias Winter To: pbs-devel@lists.proxmox.com Subject: [RFC PATCH proxmox-backup 1/4] tape: drive: add a pool to use several drives for one task Date: Fri, 6 Nov 2026 14:38:24 +0100 Message-ID: <20261106133826.1692620-3-tobias@stuttgart.de> X-Mailer: git-send-email 2.53.0 In-Reply-To: <20261106133826.1692620-2-tobias@stuttgart.de> References: <20261106133826.1692620-2-tobias@stuttgart.de> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit X-SPAM-LEVEL: Spam detection results: 0 AWL 0.021 Adjusted score from AWL reputation of From: address DKIM_SIGNED 0.1 Message has a DKIM or DK signature, not necessarily valid DKIM_VALID -0.1 Message has at least one valid DKIM or DK signature DKIM_VALID_AU -0.1 Message has a valid DKIM or DK signature from author's domain DKIM_VALID_EF -0.1 Message has a valid DKIM or DK signature from envelope-from domain DMARC_PASS -0.1 DMARC pass policy RCVD_IN_DNSWL_MED -2.3 Sender listed at https://www.dnswl.org/, medium trust SPF_HELO_PASS -0.001 SPF: HELO matches SPF record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: JKV43DJPHDVQWJCRJRMSLIX5DJ52CORY X-Message-ID-Hash: JKV43DJPHDVQWJCRJRMSLIX5DJ52CORY X-MailFrom: prvs=957be33b11=tobias@stuttgart.de 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 CC: Jakob-Tobias Winter 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: 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. 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)); 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); + } + } +} + +/// 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)); + 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(()), + } + }; + + 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)); + } +} -- 2.53.0