From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from firstgate.proxmox.com (firstgate.proxmox.com [IPv6:2a01:7e0:0:424::9]) by lore.proxmox.com (Postfix) with ESMTPS id 54FDD1FF183 for ; Wed, 5 Nov 2025 16:51:25 +0100 (CET) Received: from firstgate.proxmox.com (localhost [127.0.0.1]) by firstgate.proxmox.com (Proxmox) with ESMTP id D73F72222; Wed, 5 Nov 2025 16:52:05 +0100 (CET) From: Nicolas Frey To: pbs-devel@lists.proxmox.com Date: Wed, 5 Nov 2025 16:51:26 +0100 Message-ID: <20251105155129.517430-3-n.frey@proxmox.com> X-Mailer: git-send-email 2.47.3 In-Reply-To: <20251105155129.517430-1-n.frey@proxmox.com> References: <20251105155129.517430-1-n.frey@proxmox.com> MIME-Version: 1.0 X-SPAM-LEVEL: Spam detection results: 0 AWL -0.104 Adjusted score from AWL reputation of From: address BAYES_00 -1.9 Bayes spam probability is 0 to 1% DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment KAM_LAZY_DOMAIN_SECURITY 1 Sending domain does not have any anti-forgery methods RDNS_NONE 0.793 Delivered to internal network by a host with no rDNS SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_NONE 0.001 SPF: sender does not publish an SPF Record Subject: [pbs-devel] [PATCH proxmox-backup 1/4] api: verify: move chunk loading into parallel handler X-BeenThere: pbs-devel@lists.proxmox.com X-Mailman-Version: 2.1.29 Precedence: list List-Id: Proxmox Backup Server development discussion List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , Reply-To: Proxmox Backup Server development discussion Content-Type: text/plain; charset="us-ascii" Content-Transfer-Encoding: 7bit Errors-To: pbs-devel-bounces@lists.proxmox.com Sender: "pbs-devel" This way, the chunks will be loaded in parallel in addition to being checked in parallel. Depending on the underlying storage, this can speed up reading chunks from disk, especially when the underlying storage is IO depth dependent, and the CPU is faster than the storage. Originally-by: Dominik Csapak Signed-off-by: Nicolas Frey --- src/backup/verify.rs | 120 +++++++++++++++++++++++++++---------------- 1 file changed, 75 insertions(+), 45 deletions(-) diff --git a/src/backup/verify.rs b/src/backup/verify.rs index e33fdf50..7f91f38c 100644 --- a/src/backup/verify.rs +++ b/src/backup/verify.rs @@ -1,6 +1,6 @@ use pbs_config::BackupLockGuard; use std::collections::HashSet; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Instant; @@ -20,7 +20,7 @@ use pbs_datastore::index::{ChunkReadInfo, IndexFile}; use pbs_datastore::manifest::{BackupManifest, FileInfo}; use pbs_datastore::{DataBlob, DataStore, DatastoreBackend, StoreProgress}; -use crate::tools::parallel_handler::ParallelHandler; +use crate::tools::parallel_handler::{ParallelHandler, SendHandle}; use crate::backup::hierarchy::ListAccessibleBackupGroups; @@ -156,23 +156,20 @@ impl VerifyWorker { let start_time = Instant::now(); - let mut read_bytes = 0; - let mut decoded_bytes = 0; + let read_bytes = Arc::new(AtomicU64::new(0)); + let decoded_bytes = Arc::new(AtomicU64::new(0)); - let datastore2 = Arc::clone(&self.datastore); - let corrupt_chunks2 = Arc::clone(&self.corrupt_chunks); - let verified_chunks2 = Arc::clone(&self.verified_chunks); - let errors2 = Arc::clone(&errors); - - let decoder_pool = ParallelHandler::new( - "verify chunk decoder", - 4, + let decoder_pool = ParallelHandler::new("verify chunk decoder", 4, { + let datastore = Arc::clone(&self.datastore); + let corrupt_chunks = Arc::clone(&self.corrupt_chunks); + let verified_chunks = Arc::clone(&self.verified_chunks); + let errors = Arc::clone(&errors); move |(chunk, digest, size): (DataBlob, [u8; 32], u64)| { let chunk_crypt_mode = match chunk.crypt_mode() { Err(err) => { - corrupt_chunks2.lock().unwrap().insert(digest); + corrupt_chunks.lock().unwrap().insert(digest); info!("can't verify chunk, unknown CryptMode - {err}"); - errors2.fetch_add(1, Ordering::SeqCst); + errors.fetch_add(1, Ordering::SeqCst); return Ok(()); } Ok(mode) => mode, @@ -182,21 +179,21 @@ impl VerifyWorker { info!( "chunk CryptMode {chunk_crypt_mode:?} does not match index CryptMode {crypt_mode:?}" ); - errors2.fetch_add(1, Ordering::SeqCst); + errors.fetch_add(1, Ordering::SeqCst); } if let Err(err) = chunk.verify_unencrypted(size as usize, &digest) { - corrupt_chunks2.lock().unwrap().insert(digest); + corrupt_chunks.lock().unwrap().insert(digest); info!("{err}"); - errors2.fetch_add(1, Ordering::SeqCst); - Self::rename_corrupted_chunk(datastore2.clone(), &digest); + errors.fetch_add(1, Ordering::SeqCst); + Self::rename_corrupted_chunk(datastore.clone(), &digest); } else { - verified_chunks2.lock().unwrap().insert(digest); + verified_chunks.lock().unwrap().insert(digest); } Ok(()) - }, - ); + } + }); let skip_chunk = |digest: &[u8; 32]| -> bool { if self.verified_chunks.lock().unwrap().contains(digest) { @@ -223,6 +220,29 @@ impl VerifyWorker { .datastore .get_chunks_in_order(&*index, skip_chunk, check_abort)?; + let reader_pool = ParallelHandler::new("read chunks", 4, { + let decoder_pool = decoder_pool.channel(); + let datastore = Arc::clone(&self.datastore); + let corrupt_chunks = Arc::clone(&self.corrupt_chunks); + let read_bytes = Arc::clone(&read_bytes); + let decoded_bytes = Arc::clone(&decoded_bytes); + let errors = Arc::clone(&errors); + let backend = self.backend.clone(); + + move |info: ChunkReadInfo| { + Self::verify_chunk_by_backend( + &backend, + Arc::clone(&datastore), + Arc::clone(&corrupt_chunks), + Arc::clone(&read_bytes), + Arc::clone(&decoded_bytes), + Arc::clone(&errors), + &decoder_pool, + &info, + ) + } + }); + for (pos, _) in chunk_list { self.worker.check_abort()?; self.worker.fail_on_shutdown()?; @@ -234,19 +254,16 @@ impl VerifyWorker { continue; // already verified or marked corrupt } - self.verify_chunk_by_backend( - &info, - &mut read_bytes, - &mut decoded_bytes, - Arc::clone(&errors), - &decoder_pool, - )?; + reader_pool.send(info)?; } - decoder_pool.complete()?; + reader_pool.complete()?; let elapsed = start_time.elapsed().as_secs_f64(); + let read_bytes = read_bytes.load(Ordering::SeqCst); + let decoded_bytes = decoded_bytes.load(Ordering::SeqCst); + let read_bytes_mib = (read_bytes as f64) / (1024.0 * 1024.0); let decoded_bytes_mib = (decoded_bytes as f64) / (1024.0 * 1024.0); @@ -266,26 +283,31 @@ impl VerifyWorker { Ok(()) } + #[allow(clippy::too_many_arguments)] fn verify_chunk_by_backend( - &self, - info: &ChunkReadInfo, - read_bytes: &mut u64, - decoded_bytes: &mut u64, + backend: &DatastoreBackend, + datastore: Arc, + corrupt_chunks: Arc>>, + read_bytes: Arc, + decoded_bytes: Arc, errors: Arc, - decoder_pool: &ParallelHandler<(DataBlob, [u8; 32], u64)>, + decoder_pool: &SendHandle<(DataBlob, [u8; 32], u64)>, + info: &ChunkReadInfo, ) -> Result<(), Error> { - match &self.backend { - DatastoreBackend::Filesystem => match self.datastore.load_chunk(&info.digest) { - Err(err) => self.add_corrupt_chunk( + match backend { + DatastoreBackend::Filesystem => match datastore.load_chunk(&info.digest) { + Err(err) => Self::add_corrupt_chunk( + datastore, + corrupt_chunks, info.digest, errors, &format!("can't verify chunk, load failed - {err}"), ), Ok(chunk) => { let size = info.size(); - *read_bytes += chunk.raw_size(); + read_bytes.fetch_add(chunk.raw_size(), Ordering::SeqCst); decoder_pool.send((chunk, info.digest, size))?; - *decoded_bytes += size; + decoded_bytes.fetch_add(size, Ordering::SeqCst); } }, DatastoreBackend::S3(s3_client) => { @@ -302,9 +324,9 @@ impl VerifyWorker { match chunk_result { Ok(chunk) => { let size = info.size(); - *read_bytes += chunk.raw_size(); + read_bytes.fetch_add(chunk.raw_size(), Ordering::SeqCst); decoder_pool.send((chunk, info.digest, size))?; - *decoded_bytes += size; + decoded_bytes.fetch_add(size, Ordering::SeqCst); } Err(err) => { errors.fetch_add(1, Ordering::SeqCst); @@ -312,7 +334,9 @@ impl VerifyWorker { } } } - Ok(None) => self.add_corrupt_chunk( + Ok(None) => Self::add_corrupt_chunk( + datastore, + corrupt_chunks, info.digest, errors, &format!( @@ -330,13 +354,19 @@ impl VerifyWorker { Ok(()) } - fn add_corrupt_chunk(&self, digest: [u8; 32], errors: Arc, message: &str) { + fn add_corrupt_chunk( + datastore: Arc, + corrupt_chunks: Arc>>, + digest: [u8; 32], + errors: Arc, + message: &str, + ) { // Panic on poisoned mutex - let mut corrupt_chunks = self.corrupt_chunks.lock().unwrap(); + let mut corrupt_chunks = corrupt_chunks.lock().unwrap(); corrupt_chunks.insert(digest); error!(message); errors.fetch_add(1, Ordering::SeqCst); - Self::rename_corrupted_chunk(self.datastore.clone(), &digest); + Self::rename_corrupted_chunk(datastore.clone(), &digest); } fn verify_fixed_index(&self, backup_dir: &BackupDir, info: &FileInfo) -> Result<(), Error> { -- 2.47.3 _______________________________________________ pbs-devel mailing list pbs-devel@lists.proxmox.com https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel