[RFC PATCH proxmox-backup 0/4] fix 7333 use multiple tape drives for restoring

[RFC PATCH proxmox-backup 0/4] fix 7333 use multiple tape drives for restoring

[Jakob-Tobias Winter]

Restoring a media set reads one tape after the other with a single
drive, even if the library has more drives that are idle. This series
lets a restore use up to three additional drives, so several tapes are
read at the same time [0]. Even when the host cannot keep up with the
combined data rate, loading, positioning and changer moves on one
drive then overlap with reading on the others.

I am sending this as RFC to discuss the approach.

The change is kept small and reuses the existing selective restore:

Patch 1 adds a pool of drives that runs one job per media. With a
single drive, this is the existing loop, so a restore with one drive
behaves as before. With several drives, each drive runs in its own
thread and takes the media already loaded in it, else the next one
that is not loaded in another drive of the pool. Changer operations
are serialized, so two drives do not pick the same free slot when
unloading. The first error, or an abort, stops all drives. As the lines
of all drives are interleaved in the task log, every message of a drive
starts with "drive 'NAME': ".

Patch 2 loads the media of the first two phases of the selective
restore through the pool, without functional change.

Patch 3 adds the optional 'additional-drives' parameter to the API and
to 'proxmox-tape restore'. Every drive is checked for Tape.Read, locked
and marked as in use for the whole task. A selective restore then reads
the snapshot archives (phase 1) and the chunks (phase 2) on all drives.

A full restore writes the snapshot archives straight into the
datastore. That is safe with a single drive, because a tape backup
writes the chunks of a snapshot before its snapshot archive, to the
same or an earlier media. With several drives, a snapshot archive can
be read while an earlier media is still being restored, and an aborted
restore would leave snapshots with missing chunks behind. So with more
than one drive, a full restore runs the phases of the selective restore
for all snapshots, which only add the snapshots once all chunks are
restored. The restore window already takes that path whenever it sends
a namespace mapping. The trade-offs are that the media set catalog is
needed, every media is loaded twice (once for the snapshot archives,
once for the chunks), and the snapshots only show up at the end.

Patch 4 adds a multi-select field for the additional drives to the
restore window.

On abort or error, the existing cleanup of the selective restore
removes the temporary directory, so an abort before phase 3 does not
add any snapshot. Chunks that were already written stay until garbage
collection removes them, as today.

Open questions:

- Is 'drive' plus 'additional-drives' fine, or would you prefer
  something else, e.g. a 'drives' list, or the changer plus "use all
  free drives"?
- Is a maximum of four drives in total OK or can we drop the limit?
- Standalone drives (without changer) work, but the pool cannot know
  which tape is in which drive, so the operator has to follow the
  per-drive "insert media" notifications. Is that OK?
- Should the "drive 'NAME': " prefix also be used with a single drive,
  for consistency?

Testing: I built the series on top of 4.2.8, and tested it on a tape
library with four drives (LTO-7 media), with one and with several
drives. With four drives, each drive read at about 260-290 MB/s.

[0] https://bugzilla.proxmox.com/show_bug.cgi?id=7333

Jakob-Tobias Winter (4):
  tape: drive: add a pool to use several drives for one task
  tape: restore: use a drive pool for the selective restore
  close #7333: tape: restore: allow using several drives in parallel
  ui: tape restore: add selector for additional drives

 src/api2/tape/restore.rs       | 179 +++++++++++------
 src/bin/proxmox-tape.rs        |  10 +
 src/tape/drive/mod.rs          |  41 +++-
 src/tape/drive/pool.rs         | 358 +++++++++++++++++++++++++++++++++
 www/tape/window/TapeRestore.js |  15 ++
 5 files changed, 540 insertions(+), 63 deletions(-)
 create mode 100644 src/tape/drive/pool.rs


base-commit: 1df6dff9c0acdd11386245502b2c7aeed2f440dd
-- 
2.53.0

[RFC PATCH proxmox-backup 1/4] tape: drive: add a pool to use several drives for one task

[Jakob-Tobias Winter]

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 <tobias@stuttgart.de>
---
 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<dyn TapeDriver>, 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<dyn TapeDriver>, 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<T>(mutex: &Mutex<T>) -> 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<dyn WorkerTaskContext>,
+    stop: Arc<AtomicBool>,
+}
+
+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<dyn WorkerTaskContext>,
+    pub drive: Box<dyn TapeDriver>,
+    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<dyn WorkerTaskContext>,
+    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<dyn WorkerTaskContext>,
+        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<T, R, F>(&self, jobs: Vec<(MediaLabel, T)>, func: F) -> Result<Vec<R>, Error>
+    where
+        T: Send,
+        R: Send,
+        F: Fn(LoadedMedia, T) -> Result<R, Error> + 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<dyn WorkerTaskContext> = 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<T, R, F>(
+        &self,
+        drive_name: &str,
+        worker: Arc<dyn WorkerTaskContext>,
+        queue: &Mutex<VecDeque<(MediaLabel, T)>>,
+        results: &Mutex<Vec<R>>,
+        func: &F,
+    ) -> Result<(), Error>
+    where
+        F: Fn(LoadedMedia, T) -> Result<R, Error>,
+    {
+        // 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<String>, HashSet<String>) {
+        let mut own = None;
+        let mut others = HashSet::new();
+        let mut status_list: HashMap<String, Option<MtxStatus>> = HashMap::new();
+
+        let _changer_guard = lock(&self.changer_lock);
+
+        for name in self.drives {
+            let Ok(drive) = self.config.lookup::<LtoTapeDrive>("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<T>(
+    queue: &VecDeque<(MediaLabel, T)>,
+    own: Option<&str>,
+    busy: &HashSet<String>,
+) -> Option<usize> {
+    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

[RFC PATCH proxmox-backup 2/4] tape: restore: use a drive pool for the selective restore

[Jakob-Tobias Winter]

In preparation for restoring from several drives in parallel, load the
media of the first two phases of a selective restore through a drive
pool, and start the messages logged while reading a media with the
prefix the pool provides. The pool only contains the given drive for
now, so the media are loaded in the same order as before and the
prefix is empty.

No functional change intended.

Signed-off-by: Jakob-Tobias Winter <tobias@stuttgart.de>
---
 src/api2/tape/restore.rs | 116 ++++++++++++++++++++++++---------------
 1 file changed, 72 insertions(+), 44 deletions(-)

diff --git a/src/api2/tape/restore.rs b/src/api2/tape/restore.rs
index 6544aa726..ee8906f85 100644
--- a/src/api2/tape/restore.rs
+++ b/src/api2/tape/restore.rs
@@ -40,7 +40,9 @@ use pbs_tape::{
 use crate::backup::check_ns_modification_privs;
 use crate::tape::{
     Inventory, MediaCatalog, MediaId, MediaSet, MediaSetCatalog, TAPE_STATUS_DIR,
-    drive::{TapeDriver, lock_tape_device, request_and_load_media, set_tape_device_state},
+    drive::{
+        DrivePool, TapeDriver, lock_tape_device, request_and_load_media, set_tape_device_state,
+    },
     file_formats::{
         CatalogArchiveHeader, ChunkArchiveDecoder, ChunkArchiveHeader,
         PROXMOX_BACKUP_CATALOG_ARCHIVE_MAGIC_1_0, PROXMOX_BACKUP_CATALOG_ARCHIVE_MAGIC_1_1,
@@ -418,7 +420,7 @@ pub fn restore(
                     inventory,
                     media_set_uuid,
                     drive_config,
-                    &drive,
+                    std::slice::from_ref(&drive),
                     store_map,
                     restore_owner,
                     &notification_mode,
@@ -628,7 +630,7 @@ fn restore_list_worker(
     inventory: Inventory,
     media_set_uuid: Uuid,
     drive_config: SectionConfigData,
-    drive_name: &str,
+    drives: &[String],
     store_map: DataStoreMap,
     restore_owner: &Authid,
     notification_mode: &TapeNotificationMode,
@@ -636,6 +638,7 @@ fn restore_list_worker(
     auth_id: &Authid,
 ) -> Result<(), Error> {
     let catalog = get_media_set_catalog(&inventory, &media_set_uuid)?;
+    let drive_pool = DrivePool::new(worker, &drive_config, drives, notification_mode);
 
     let mut datastore_locks = Vec::new();
     let mut snapshot_file_hash: BTreeMap<Uuid, Vec<(u64, String)>> = BTreeMap::new();
@@ -760,30 +763,39 @@ fn restore_list_worker(
 
         info!("Phase 1: temporarily restore snapshots to temp dir");
         log_required_tapes(&inventory, snapshot_file_hash.keys());
-        let mut datastore_chunk_map: HashMap<String, HashSet<[u8; 32]>> = HashMap::new();
-        let mut tmp_paths = Vec::new();
-        for (media_uuid, file_list) in snapshot_file_hash.iter_mut() {
-            let media_id = inventory.lookup_media(media_uuid).unwrap();
-            let (drive, info) = request_and_load_media(
-                &worker,
-                &drive_config,
-                drive_name,
-                &media_id.label,
-                notification_mode,
-            )?;
+        let jobs = snapshot_file_hash
+            .into_iter()
+            .map(|(media_uuid, file_list)| {
+                let media_id = inventory.lookup_media(&media_uuid).unwrap();
+                (media_id.label.clone(), file_list)
+            })
+            .collect();
+        let results = drive_pool.run(jobs, |media, mut file_list| {
             file_list.sort_unstable();
-
+            let mut chunk_map = HashMap::new();
             let tmp_path = restore_snapshots_to_tmpdir(
-                worker.clone(),
+                media.worker,
                 &store_map,
-                file_list,
-                drive,
-                &info,
+                &file_list,
+                media.drive,
+                &media.media_id,
                 &media_set_uuid,
-                &mut datastore_chunk_map,
+                &mut chunk_map,
+                &media.log_prefix,
             )
             .map_err(|err| format_err!("could not restore snapshots to tmpdir: {}", err))?;
+            Ok((tmp_path, chunk_map))
+        })?;
+        let mut datastore_chunk_map: HashMap<String, HashSet<[u8; 32]>> = HashMap::new();
+        let mut tmp_paths = Vec::new();
+        for (tmp_path, chunk_map) in results {
             tmp_paths.extend(tmp_path);
+            for (source_datastore, chunks) in chunk_map {
+                datastore_chunk_map
+                    .entry(source_datastore)
+                    .or_default()
+                    .extend(chunks);
+            }
         }
 
         // sorted media_uuid => (sorted file_num => (set of digests)))
@@ -816,17 +828,23 @@ fn restore_list_worker(
             info!("All chunks are already present, skip phase 2...");
         }
 
-        for (media_uuid, file_chunk_map) in media_file_chunk_map.iter_mut() {
-            let media_id = inventory.lookup_media(media_uuid).unwrap();
-            let (mut drive, _info) = request_and_load_media(
-                &worker,
-                &drive_config,
-                drive_name,
-                &media_id.label,
-                notification_mode,
-            )?;
-            restore_file_chunk_map(worker.clone(), &mut drive, &store_map, file_chunk_map)?;
-        }
+        let jobs = media_file_chunk_map
+            .into_iter()
+            .map(|(media_uuid, file_chunk_map)| {
+                let media_id = inventory.lookup_media(&media_uuid).unwrap();
+                (media_id.label.clone(), file_chunk_map)
+            })
+            .collect();
+        drive_pool.run(jobs, |media, mut file_chunk_map| {
+            let mut drive = media.drive;
+            restore_file_chunk_map(
+                media.worker,
+                &mut drive,
+                &store_map,
+                &mut file_chunk_map,
+                &media.log_prefix,
+            )
+        })?;
 
         info!("Phase 3: copy snapshots from temp dir to datastores");
         let mut errors = false;
@@ -997,14 +1015,16 @@ fn snapshot_tmpdir(
     Ok(path)
 }
 
+#[allow(clippy::too_many_arguments)]
 fn restore_snapshots_to_tmpdir(
-    worker: Arc<WorkerTask>,
+    worker: Arc<dyn WorkerTaskContext>,
     store_map: &DataStoreMap,
     file_list: &[(u64, String)],
     mut drive: Box<dyn TapeDriver>,
     media_id: &MediaId,
     media_set_uuid: &Uuid,
     chunks_list: &mut HashMap<String, HashSet<[u8; 32]>>,
+    log_prefix: &str,
 ) -> Result<Vec<PathBuf>, Error> {
     let mut tmp_paths = Vec::new();
     match media_id.media_set_label {
@@ -1030,10 +1050,10 @@ fn restore_snapshots_to_tmpdir(
     for (file_num, snapshot_name) in file_list {
         let current_file_number = drive.current_file_number()?;
         if current_file_number != *file_num {
-            info!("was at file {current_file_number}, moving to {file_num}");
+            info!("{log_prefix}was at file {current_file_number}, moving to {file_num}");
             drive.move_to_file(*file_num)?;
             let current_file_number = drive.current_file_number()?;
-            info!("now at file {current_file_number}");
+            info!("{log_prefix}now at file {current_file_number}");
         }
         let mut reader = drive.read_next_file()?;
 
@@ -1062,7 +1082,9 @@ fn restore_snapshots_to_tmpdir(
                     bail!("unexpected snapshot name (expected {snapshot_name}, got {snapshot})");
                 }
 
-                info!("File {file_num}: snapshot archive {source_datastore}:{snapshot}",);
+                info!(
+                    "{log_prefix}File {file_num}: snapshot archive {source_datastore}:{snapshot}",
+                );
 
                 let mut decoder =
                     pxar::decoder::sync::Decoder::from_std(pxar::PxarVariant::Unified(reader))?;
@@ -1070,7 +1092,10 @@ fn restore_snapshots_to_tmpdir(
                 let target_datastore = match store_map.target_store(&source_datastore) {
                     Some(datastore) => datastore,
                     None => {
-                        warn!("could not find target datastore for {source_datastore}:{snapshot}",);
+                        warn!(
+                            "{log_prefix}could not find target datastore for \
+                            {source_datastore}:{snapshot}",
+                        );
                         continue;
                     }
                 };
@@ -1115,18 +1140,19 @@ fn restore_snapshots_to_tmpdir(
 }
 
 fn restore_file_chunk_map(
-    worker: Arc<WorkerTask>,
+    worker: Arc<dyn WorkerTaskContext>,
     drive: &mut Box<dyn TapeDriver>,
     store_map: &DataStoreMap,
     file_chunk_map: &mut BTreeMap<u64, HashSet<[u8; 32]>>,
+    log_prefix: &str,
 ) -> Result<(), Error> {
     for (nr, chunk_map) in file_chunk_map.iter_mut() {
         let current_file_number = drive.current_file_number()?;
         if current_file_number != *nr {
-            info!("was at file {current_file_number}, moving to {nr}");
+            info!("{log_prefix}was at file {current_file_number}, moving to {nr}");
             drive.move_to_file(*nr)?;
             let current_file_number = drive.current_file_number()?;
-            info!("now at file {current_file_number}");
+            info!("{log_prefix}now at file {current_file_number}");
         }
         let mut reader = drive.read_next_file()?;
         let header: MediaContentHeader = unsafe { reader.read_le_value()? };
@@ -1143,7 +1169,7 @@ fn restore_file_chunk_map(
 
                 let source_datastore = archive_header.store;
 
-                info!("File {nr}: chunk archive for datastore '{source_datastore}'",);
+                info!("{log_prefix}File {nr}: chunk archive for datastore '{source_datastore}'",);
 
                 let datastore = store_map.target_store(&source_datastore).ok_or_else(|| {
                     format_err!("unexpected chunk archive for store: {source_datastore}")
@@ -1154,8 +1180,9 @@ fn restore_file_chunk_map(
                     reader,
                     datastore.clone(),
                     chunk_map,
+                    log_prefix,
                 )?;
-                info!("restored {count} chunks");
+                info!("{log_prefix}restored {count} chunks");
             }
             _ => bail!("unexpected content magic {:?}", header.content_magic),
         }
@@ -1165,10 +1192,11 @@ fn restore_file_chunk_map(
 }
 
 fn restore_partial_chunk_archive<'a>(
-    worker: Arc<WorkerTask>,
+    worker: Arc<dyn WorkerTaskContext>,
     reader: Box<dyn 'a + TapeRead>,
     datastore: Arc<DataStore>,
     chunk_list: &mut HashSet<[u8; 32]>,
+    log_prefix: &str,
 ) -> Result<usize, Error> {
     let mut decoder = ChunkArchiveDecoder::new(reader);
 
@@ -1217,7 +1245,7 @@ fn restore_partial_chunk_archive<'a>(
     let elapsed = start_time.elapsed()?.as_secs_f64();
     let bytes = bytes.load(std::sync::atomic::Ordering::SeqCst) as f64;
     info!(
-        "restored {} ({:.2}/s)",
+        "{log_prefix}restored {} ({:.2}/s)",
         HumanByte::new_decimal(bytes),
         HumanByte::new_decimal(bytes / elapsed),
     );
@@ -1675,7 +1703,7 @@ fn restore_snapshot_archive<'a>(
 }
 
 fn try_restore_snapshot_archive<R: pxar::decoder::SeqRead>(
-    worker: Arc<WorkerTask>,
+    worker: Arc<dyn WorkerTaskContext>,
     decoder: &mut pxar::decoder::sync::Decoder<R>,
     snapshot_path: &Path,
 ) -> Result<BackupManifest, Error> {
-- 
2.53.0

[RFC PATCH proxmox-backup 3/4] close #7333: tape: restore: allow using several drives in parallel

[Jakob-Tobias Winter]

Restoring a media set that spans several tapes from a single drive
takes long, even if other drives of the library are idle. Add the
optional 'additional-drives' parameter to the restore API and the
proxmox-tape CLI, for up to three drives in addition to the given one.
All drives are checked for the Tape.Read privilege, locked and marked
as in use for the whole task.

A selective restore then reads the media of its first two phases in
parallel. A full restore cannot simply do the same: a tape backup
writes the chunks of a snapshot before its snapshot archive, possibly
to an earlier media, so with several drives a snapshot could be added
before all of its chunks are restored, and an aborted restore would
leave it behind with chunks missing. So with several drives, a full
restore uses the phases of the selective restore for all snapshots,
which only add the snapshots once all chunks are restored. This needs
the media set catalog, the snapshots only show up at the end of the
restore, and each media is loaded twice, once for the small snapshot
archives and once for the chunks. With a single drive, a full restore
works as before.

Signed-off-by: Jakob-Tobias Winter <tobias@stuttgart.de>
---
 src/api2/tape/restore.rs | 65 +++++++++++++++++++++++++++++++---------
 src/bin/proxmox-tape.rs  | 10 +++++++
 2 files changed, 61 insertions(+), 14 deletions(-)

diff --git a/src/api2/tape/restore.rs b/src/api2/tape/restore.rs
index ee8906f85..8040a6c21 100644
--- a/src/api2/tape/restore.rs
+++ b/src/api2/tape/restore.rs
@@ -286,6 +286,16 @@ pub const ROUTER: Router = Router::new().post(&API_METHOD_RESTORE);
             drive: {
                 schema: DRIVE_NAME_SCHEMA,
             },
+            "additional-drives": {
+                description: "Use these drives in addition to 'drive', to restore from several \
+                    media in parallel.",
+                type: Array,
+                optional: true,
+                max_length: 3,
+                items: {
+                    schema: DRIVE_NAME_SCHEMA,
+                },
+            },
             "media-set": {
                 description: "Media set UUID.",
                 type: String,
@@ -318,8 +328,9 @@ pub const ROUTER: Router = Router::new().post(&API_METHOD_RESTORE);
     access: {
         // Note: parameters are no uri parameter, so we need to test inside function body
         description: "The user needs Tape.Read privilege on /tape/pool/{pool} and \
-            /tape/device/{drive}, Datastore.Backup privilege on /datastore/{store}/[{namespace}], \
-            Datastore.Modify privileges to create namespaces (if they don't exist).",
+            /tape/device/{drive} (and each additional drive), Datastore.Backup privilege on \
+            /datastore/{store}/[{namespace}], Datastore.Modify privileges to create namespaces \
+            (if they don't exist).",
         permission: &Permission::Anybody,
     },
 )]
@@ -328,6 +339,7 @@ pub const ROUTER: Router = Router::new().post(&API_METHOD_RESTORE);
 pub fn restore(
     store: String,
     drive: String,
+    additional_drives: Option<Vec<String>>,
     namespaces: Option<Vec<String>>,
     media_set: String,
     notify_user: Option<Userid>,
@@ -374,7 +386,16 @@ pub fn restore(
             }
         }
     }
-    user_info.check_privs(&auth_id, &["tape", "device", &drive], PRIV_TAPE_READ, false)?;
+    let mut drives = vec![drive];
+    for drive in additional_drives.unwrap_or_default() {
+        if drives.contains(&drive) {
+            bail!("drive '{drive}' given more than once");
+        }
+        drives.push(drive);
+    }
+    for drive in drives.iter() {
+        user_info.check_privs(&auth_id, &["tape", "device", drive], PRIV_TAPE_READ, false)?;
+    }
 
     let media_set_uuid = media_set.parse()?;
 
@@ -388,7 +409,10 @@ pub fn restore(
     let (drive_config, _digest) = pbs_config::drive::config()?;
 
     // early check/lock before starting worker
-    let drive_lock = lock_tape_device(&drive_config, &drive)?;
+    let drive_locks = drives
+        .iter()
+        .map(|drive| lock_tape_device(&drive_config, drive))
+        .collect::<Result<Vec<_>, _>>()?;
 
     let to_stdout = rpcenv.env_type() == RpcEnvironmentType::CLI;
 
@@ -404,23 +428,28 @@ pub fn restore(
         auth_id.to_string(),
         to_stdout,
         move |worker| {
-            let _drive_lock = drive_lock; // keep lock guard
+            let _drive_locks = drive_locks; // keep lock guards
 
-            set_tape_device_state(&drive, &worker.upid().to_string())?;
+            for drive in drives.iter() {
+                set_tape_device_state(drive, &worker.upid().to_string())?;
+            }
 
             let restore_owner = owner.as_ref().unwrap_or(&auth_id);
 
             info!("Mediaset '{media_set}'");
             info!("Pool: {pool}");
 
-            let res = if snapshots.is_some() || namespaces {
+            // the chunks of a snapshot can be on an earlier media that another drive did not
+            // restore yet, so with several drives a full restore uses the phases of the selective
+            // restore, which only add the snapshots after all chunks are restored
+            let res = if snapshots.is_some() || namespaces || drives.len() > 1 {
                 restore_list_worker(
                     worker,
-                    snapshots.unwrap_or_default(),
+                    snapshots,
                     inventory,
                     media_set_uuid,
                     drive_config,
-                    std::slice::from_ref(&drive),
+                    &drives,
                     store_map,
                     restore_owner,
                     &notification_mode,
@@ -433,7 +462,7 @@ pub fn restore(
                     inventory,
                     media_set_uuid,
                     drive_config,
-                    &drive,
+                    &drives[0],
                     store_map,
                     restore_owner,
                     &notification_mode,
@@ -443,8 +472,10 @@ pub fn restore(
             if res.is_ok() {
                 info!("Restore mediaset '{media_set}' done");
             }
-            if let Err(err) = set_tape_device_state(&drive, "") {
-                info!("could not unset drive state for {drive}: {err}");
+            for drive in drives.iter() {
+                if let Err(err) = set_tape_device_state(drive, "") {
+                    info!("could not unset drive state for {drive}: {err}");
+                }
             }
 
             res
@@ -626,7 +657,7 @@ fn log_required_tapes<'a>(inventory: &Inventory, list: impl Iterator<Item = &'a
 #[allow(clippy::too_many_arguments)]
 fn restore_list_worker(
     worker: Arc<WorkerTask>,
-    snapshots: Vec<String>,
+    snapshots: Option<Vec<String>>,
     inventory: Inventory,
     media_set_uuid: Uuid,
     drive_config: SectionConfigData,
@@ -646,11 +677,17 @@ fn restore_list_worker(
 
     let res = proxmox_lang::try_block!({
         // phase 0
+        // without snapshots and namespace mapping, this is a full restore using several drives
+        let restore_all = snapshots.is_none() && store_map.ns_map.is_none();
+        let snapshots = snapshots.unwrap_or_default();
         let snapshots = if snapshots.is_empty() {
             let mut restorable = Vec::new();
             // restore source namespaces
             for (store, snapshot) in catalog.list_snapshots() {
                 let (ns, dir) = match parse_ns_and_snapshot(snapshot) {
+                    Ok((ns, dir)) if restore_all && store_map.target_store(store).is_some() => {
+                        (ns, dir)
+                    }
                     Ok((ns, dir)) if store_map.has_full_mapping(store, &ns) => (ns, dir),
                     Err(err) => {
                         warn!("couldn't parse snapshot {snapshot} - {err}");
@@ -665,7 +702,7 @@ fn restore_list_worker(
                     &snapshot,
                     &ns,
                     &dir,
-                    false,
+                    restore_all,
                     &user_info,
                     auth_id,
                     restore_owner,
diff --git a/src/bin/proxmox-tape.rs b/src/bin/proxmox-tape.rs
index dcbf2d0fb..860ed38fc 100644
--- a/src/bin/proxmox-tape.rs
+++ b/src/bin/proxmox-tape.rs
@@ -906,6 +906,16 @@ async fn backup(mut param: Value) -> Result<(), Error> {
                 schema: DRIVE_NAME_SCHEMA,
                 optional: true,
             },
+            "additional-drives": {
+                description: "Use these drives in addition to 'drive', to restore from several \
+                    media in parallel.",
+                type: Array,
+                optional: true,
+                max_length: 3,
+                items: {
+                    schema: DRIVE_NAME_SCHEMA,
+                },
+            },
             "media-set": {
                 description: "Media set UUID.",
                 type: String,
-- 
2.53.0

[RFC PATCH proxmox-backup 4/4] ui: tape restore: add selector for additional drives

[Jakob-Tobias Winter]

Allow choosing up to three more drives for a tape restore, to read
several media in parallel.

Signed-off-by: Jakob-Tobias Winter <tobias@stuttgart.de>
---
 www/tape/window/TapeRestore.js | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/www/tape/window/TapeRestore.js b/www/tape/window/TapeRestore.js
index 8ee4f17a9..d28d2f242 100644
--- a/www/tape/window/TapeRestore.js
+++ b/www/tape/window/TapeRestore.js
@@ -374,6 +374,11 @@ Ext.define('PBS.TapeManagement.TapeRestoreWindow', {
                         let controller = me.up('window').getController();
                         let vm = controller.getViewModel();
                         let datastores = [];
+
+                        let additionalDrives = values['additional-drives'];
+                        if (!additionalDrives || additionalDrives.length === 0) {
+                            delete values['additional-drives'];
+                        }
                         if (values.store.toString() !== '') {
                             let target = values.store.toString();
                             delete values.store;
@@ -479,6 +484,16 @@ Ext.define('PBS.TapeManagement.TapeRestoreWindow', {
                             fieldLabel: gettext('Drive'),
                             labelWidth: 120,
                         },
+                        {
+                            xtype: 'pbsDriveSelector',
+                            name: 'additional-drives',
+                            fieldLabel: gettext('Additional Drives'),
+                            labelWidth: 120,
+                            multiSelect: true,
+                            allowBlank: true,
+                            skipEmptyText: true,
+                            emptyText: gettext('None'),
+                        },
                         {
                             xtype: 'pbsDataStoreSelector',
                             name: 'store',
-- 
2.53.0