[RFC PATCH proxmox-backup 3/6] refactor: use atomic values in `PxarArchiverProgressStats`

[Shan Shaji <s.shaji@proxmox.com>]

Make all fields of the PxarArchiverProgressStats struct atomic.
This enables safe, lock free access to the archiver stats from
within the progress logs while uploading the stream.

Signed-off-by: Shan Shaji <s.shaji@proxmox.com>
---
 pbs-client/src/pxar/create.rs | 121 ++++++++++++++++++++++++----------
 1 file changed, 85 insertions(+), 36 deletions(-)

diff --git a/pbs-client/src/pxar/create.rs b/pbs-client/src/pxar/create.rs
index 9100ac1a3..1246f67d9 100644
--- a/pbs-client/src/pxar/create.rs
+++ b/pbs-client/src/pxar/create.rs
@@ -7,6 +7,7 @@ use std::ops::Range;
 use std::os::unix::ffi::OsStrExt;
 use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd, OwnedFd, RawFd};
 use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex, mpsc};
 
 use anyhow::{Context, Error, bail};
@@ -151,14 +152,43 @@ pub(crate) struct HardLinkInfo {
 
 #[derive(Default)]
 struct PxarArchiverProgressStats {
-    files_reused_count: u64,
-    files_hardlink_count: u64,
-    files_reencoded_count: u64,
-    total_injected_count: u64,
-    partial_chunks_count: u64,
-    total_injected_size: u64,
-    total_reused_payload_size: u64,
-    total_reencoded_size: u64,
+    files_reused_count: AtomicU64,
+    files_hardlink_count: AtomicU64,
+    files_reencoded_count: AtomicU64,
+    total_injected_count: AtomicU64,
+    partial_chunks_count: AtomicU64,
+    total_injected_size: AtomicU64,
+    total_reused_payload_size: AtomicU64,
+    total_reencoded_size: AtomicU64,
+}
+
+impl PxarArchiverProgressStats {
+    pub fn total_files_count(&self) -> u64 {
+        self.files_reused_count.load(Ordering::SeqCst)
+            + self.files_hardlink_count.load(Ordering::SeqCst)
+            + self.files_reencoded_count.load(Ordering::SeqCst)
+    }
+
+    pub fn files_reencoded_count(&self) -> u64 {
+        self.files_reencoded_count.load(Ordering::SeqCst)
+    }
+
+    pub fn total_reencoded_size(&self) -> u64 {
+        self.total_reencoded_size.load(Ordering::SeqCst)
+    }
+
+    pub fn files_reused_count(&self) -> u64 {
+        self.files_reused_count.load(Ordering::SeqCst)
+    }
+
+    pub fn total_reused_payload_size(&self) -> u64 {
+        self.total_reused_payload_size.load(Ordering::SeqCst)
+    }
+
+    pub fn padding(&self) -> u64 {
+        self.total_injected_size.load(Ordering::SeqCst)
+            - self.total_reused_payload_size.load(Ordering::SeqCst)
+    }
 }
 
 #[derive(Serialize, Deserialize)]
@@ -325,37 +355,40 @@ where
         info!("Change detection summary:");
         info!(
             " - {} total files ({} hardlinks)",
-            archiver.progress_stats.files_reused_count
-                + archiver.progress_stats.files_reencoded_count
-                + archiver.progress_stats.files_hardlink_count,
-            archiver.progress_stats.files_hardlink_count,
+            archiver.progress_stats.total_files_count(),
+            archiver
+                .progress_stats
+                .files_hardlink_count
+                .load(Ordering::SeqCst),
         );
         info!(
             " - {} unchanged, reusable files with {} data",
-            archiver.progress_stats.files_reused_count,
-            HumanByte::from(archiver.progress_stats.total_reused_payload_size),
+            archiver.progress_stats.files_reused_count(),
+            HumanByte::from(archiver.progress_stats.total_reused_payload_size()),
         );
         info!(
             " - {} changed or non-reusable files with {} data",
-            archiver.progress_stats.files_reencoded_count,
-            HumanByte::from(archiver.progress_stats.total_reencoded_size),
+            archiver.progress_stats.files_reencoded_count(),
+            HumanByte::from(archiver.progress_stats.total_reencoded_size()),
         );
         info!(
             " - {} padding in {} partially reused chunks",
-            HumanByte::from(
-                archiver.progress_stats.total_injected_size
-                    - archiver.progress_stats.total_reused_payload_size
-            ),
-            archiver.progress_stats.partial_chunks_count,
+            HumanByte::from(archiver.progress_stats.padding()),
+            archiver
+                .progress_stats
+                .partial_chunks_count
+                .load(Ordering::SeqCst),
         );
     } else {
         info!("Processing summary:");
         info!(
-            "- {} total files ({} hardlinks) with {} data",
-            archiver.progress_stats.files_reencoded_count
-                + archiver.progress_stats.files_hardlink_count,
-            archiver.progress_stats.files_hardlink_count,
-            HumanByte::from(archiver.progress_stats.total_reencoded_size),
+            " - {} total files ({} hardlinks) with {} data",
+            archiver.progress_stats.total_files_count(),
+            archiver
+                .progress_stats
+                .files_hardlink_count
+                .load(Ordering::SeqCst),
+            HumanByte::from(archiver.progress_stats.total_reencoded_size()),
         );
     }
     Ok(())
@@ -980,24 +1013,34 @@ impl Archiver {
                 }
 
                 let offset: LinkOffset = if let Some(payload_offset) = payload_offset {
-                    self.progress_stats.total_reused_payload_size +=
-                        file_size + size_of::<pxar::format::Header>() as u64;
-                    self.progress_stats.files_reused_count += 1;
+                    self.progress_stats.total_reused_payload_size.fetch_add(
+                        file_size + size_of::<pxar::format::Header>() as u64,
+                        Ordering::SeqCst,
+                    );
+                    self.progress_stats
+                        .files_reused_count
+                        .fetch_add(1, Ordering::SeqCst);
 
                     encoder
                         .add_payload_ref(metadata, file_name, file_size, payload_offset)
                         .await?
                 } else {
-                    self.progress_stats.total_reencoded_size +=
-                        file_size + size_of::<pxar::format::Header>() as u64;
-                    self.progress_stats.files_reencoded_count += 1;
+                    self.progress_stats.total_reencoded_size.fetch_add(
+                        file_size + size_of::<pxar::format::Header>() as u64,
+                        Ordering::SeqCst,
+                    );
+                    self.progress_stats
+                        .files_reencoded_count
+                        .fetch_add(1, Ordering::SeqCst);
 
                     self.add_regular_file(encoder, fd, file_name, metadata, file_size)
                         .await?
                 };
 
                 if stat.st_nlink > 1 {
-                    self.progress_stats.files_hardlink_count += 1;
+                    self.progress_stats
+                        .files_hardlink_count
+                        .fetch_add(1, Ordering::SeqCst);
                     self.hardlinks
                         .insert(link_info, (self.path.clone(), offset));
                 }
@@ -1224,11 +1267,17 @@ impl Archiver {
                     HumanByte::from(chunk.padding),
                     HumanByte::from(chunk.size()),
                 );
-                self.progress_stats.total_injected_size += chunk.size();
-                self.progress_stats.total_injected_count += 1;
+                self.progress_stats
+                    .total_injected_size
+                    .fetch_add(chunk.size(), Ordering::SeqCst);
+                self.progress_stats
+                    .total_injected_count
+                    .fetch_add(1, Ordering::SeqCst);
 
                 if chunk.padding > 0 {
-                    self.progress_stats.partial_chunks_count += 1;
+                    self.progress_stats
+                        .partial_chunks_count
+                        .fetch_add(1, Ordering::SeqCst);
                 }
 
                 size = size.add(chunk.size());
-- 
2.47.3