[pbs-devel] [PATCH proxmox-backup v4 5/9] verify: determine the number of threads to use with {read, verify}-threads

[Nicolas Frey <n.frey@proxmox.com>]

use previously introduced {read,verify}-threads in API, where default
values match the ones of the schema definition.

Tested-by: Christian Ebner <c.ebner@proxmox.com>
Signed-off-by: Nicolas Frey <n.frey@proxmox.com>
---
 src/api2/admin/datastore.rs    | 14 +++++++++++++-
 src/api2/backup/environment.rs |  4 +++-
 src/backup/verify.rs           | 16 ++++++++++++++--
 src/server/verify_job.rs       |  7 ++++++-
 4 files changed, 36 insertions(+), 5 deletions(-)

diff --git a/src/api2/admin/datastore.rs b/src/api2/admin/datastore.rs
index 2d70e997..044dfb57 100644
--- a/src/api2/admin/datastore.rs
+++ b/src/api2/admin/datastore.rs
@@ -46,6 +46,7 @@ use pbs_api_types::{
     IGNORE_VERIFIED_BACKUPS_SCHEMA, MAX_NAMESPACE_DEPTH, NS_MAX_DEPTH_SCHEMA, PRIV_DATASTORE_AUDIT,
     PRIV_DATASTORE_BACKUP, PRIV_DATASTORE_MODIFY, PRIV_DATASTORE_PRUNE, PRIV_DATASTORE_READ,
     PRIV_DATASTORE_VERIFY, PRIV_SYS_MODIFY, UPID, UPID_SCHEMA, VERIFICATION_OUTDATED_AFTER_SCHEMA,
+    VERIFY_JOB_READ_THREADS_SCHEMA, VERIFY_JOB_VERIFY_THREADS_SCHEMA,
 };
 use pbs_client::pxar::{create_tar, create_zip};
 use pbs_config::CachedUserInfo;
@@ -675,6 +676,14 @@ pub async fn status(
                 schema: NS_MAX_DEPTH_SCHEMA,
                 optional: true,
             },
+            "read-threads": {
+                schema: VERIFY_JOB_READ_THREADS_SCHEMA,
+                optional: true,
+            },
+            "verify-threads": {
+                schema: VERIFY_JOB_VERIFY_THREADS_SCHEMA,
+                optional: true,
+            },
         },
     },
     returns: {
@@ -700,6 +709,8 @@ pub fn verify(
     ignore_verified: Option<bool>,
     outdated_after: Option<i64>,
     max_depth: Option<usize>,
+    read_threads: Option<usize>,
+    verify_threads: Option<usize>,
     rpcenv: &mut dyn RpcEnvironment,
 ) -> Result<Value, Error> {
     let auth_id: Authid = rpcenv.get_auth_id().unwrap().parse()?;
@@ -779,7 +790,8 @@ pub fn verify(
         auth_id.to_string(),
         to_stdout,
         move |worker| {
-            let verify_worker = VerifyWorker::new(worker.clone(), datastore)?;
+            let verify_worker =
+                VerifyWorker::new(worker.clone(), datastore, read_threads, verify_threads)?;
             let failed_dirs = if let Some(backup_dir) = backup_dir {
                 let mut res = Vec::new();
                 if !verify_worker.verify_backup_dir(
diff --git a/src/api2/backup/environment.rs b/src/api2/backup/environment.rs
index 1b8e0e1d..bd9c5211 100644
--- a/src/api2/backup/environment.rs
+++ b/src/api2/backup/environment.rs
@@ -800,7 +800,9 @@ impl BackupEnvironment {
             move |worker| {
                 worker.log_message("Automatically verifying newly added snapshot");
 
-                let verify_worker = VerifyWorker::new(worker.clone(), datastore)?;
+                // FIXME: update once per-datastore read/verify settings
+                // are available to not use default amount of threads here
+                let verify_worker = VerifyWorker::new(worker.clone(), datastore, None, None)?;
                 if !verify_worker.verify_backup_dir_with_lock(
                     &backup_dir,
                     worker.upid().clone(),
diff --git a/src/backup/verify.rs b/src/backup/verify.rs
index 07d51dcf..734c7b30 100644
--- a/src/backup/verify.rs
+++ b/src/backup/verify.rs
@@ -32,6 +32,8 @@ pub struct VerifyWorker {
     verified_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
     corrupt_chunks: Arc<Mutex<HashSet<[u8; 32]>>>,
     backend: DatastoreBackend,
+    read_threads: usize,
+    verify_threads: usize,
 }
 
 struct IndexVerifyState {
@@ -67,6 +69,8 @@ impl VerifyWorker {
     pub fn new(
         worker: Arc<dyn WorkerTaskContext>,
         datastore: Arc<DataStore>,
+        read_threads: Option<usize>,
+        verify_threads: Option<usize>,
     ) -> Result<Self, Error> {
         let backend = datastore.backend()?;
         Ok(Self {
@@ -77,6 +81,8 @@ impl VerifyWorker {
             // start with 64 chunks since we assume there are few corrupt ones
             corrupt_chunks: Arc::new(Mutex::new(HashSet::with_capacity(64))),
             backend,
+            read_threads: read_threads.unwrap_or(1),
+            verify_threads: verify_threads.unwrap_or(4),
         })
     }
 
@@ -115,7 +121,7 @@ impl VerifyWorker {
             &self.verified_chunks,
         ));
 
-        let decoder_pool = ParallelHandler::new("verify chunk decoder", 4, {
+        let decoder_pool = ParallelHandler::new("verify chunk decoder", self.verify_threads, {
             let verify_state = Arc::clone(&verify_state);
             move |(chunk, digest, size): (DataBlob, [u8; 32], u64)| {
                 let chunk_crypt_mode = match chunk.crypt_mode() {
@@ -177,7 +183,7 @@ impl VerifyWorker {
             .datastore
             .get_chunks_in_order(&*index, skip_chunk, check_abort)?;
 
-        let reader_pool = ParallelHandler::new("read chunks", 1, {
+        let reader_pool = ParallelHandler::new("read chunks", self.read_threads, {
             let decoder_pool = decoder_pool.channel();
             let verify_state = Arc::clone(&verify_state);
             let backend = self.backend.clone();
@@ -585,6 +591,12 @@ impl VerifyWorker {
         let group_count = list.len();
         info!("found {group_count} groups");
 
+        log::info!(
+            "using {} read and {} verify thread(s)",
+            self.read_threads,
+            self.verify_threads,
+        );
+
         let mut progress = StoreProgress::new(group_count as u64);
 
         for (pos, group) in list.into_iter().enumerate() {
diff --git a/src/server/verify_job.rs b/src/server/verify_job.rs
index c8792174..e0b03155 100644
--- a/src/server/verify_job.rs
+++ b/src/server/verify_job.rs
@@ -41,7 +41,12 @@ pub fn do_verification_job(
                 None => Default::default(),
             };
 
-            let verify_worker = VerifyWorker::new(worker.clone(), datastore)?;
+            let verify_worker = VerifyWorker::new(
+                worker.clone(),
+                datastore,
+                verification_job.read_threads,
+                verification_job.verify_threads,
+            )?;
             let result = verify_worker.verify_all_backups(
                 worker.upid(),
                 ns,
-- 
2.47.3


_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel