[PATCH proxmox-backup 2/6] client: pxar: use atomic values in `PxarArchiverProgressStats`

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

Make all fields of the PxarArchiverProgressStats struct atomic.
This is in preparation to enable lock free access of the archiver stats
from within the progress logger callback.

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

diff --git a/pbs-client/src/pxar/create.rs b/pbs-client/src/pxar/create.rs
index ab2b9301c..bde66ac2d 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,89 @@ 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 {
+    #[inline(always)]
+    pub(crate) fn files_reencoded_count(&self) -> u64 {
+        self.files_reencoded_count.load(Ordering::Acquire)
+    }
+
+    #[inline(always)]
+    pub(crate) fn total_reencoded_size(&self) -> u64 {
+        self.total_reencoded_size.load(Ordering::Acquire)
+    }
+
+    #[inline(always)]
+    pub(crate) fn files_reused_count(&self) -> u64 {
+        self.files_reused_count.load(Ordering::Acquire)
+    }
+
+    #[inline(always)]
+    pub(crate) fn total_reused_payload_size(&self) -> u64 {
+        self.total_reused_payload_size.load(Ordering::Acquire)
+    }
+
+    fn padding(&self) -> u64 {
+        self.total_injected_size.load(Ordering::Acquire) - self.total_reused_payload_size()
+    }
+
+    #[inline(always)]
+    fn total_hardlink_count(&self) -> u64 {
+        self.files_hardlink_count.load(Ordering::Acquire)
+    }
+
+    fn partial_chunks_count(&self) -> u64 {
+        self.partial_chunks_count.load(Ordering::Acquire)
+    }
+
+    #[inline(always)]
+    pub(crate) fn total_files_processed(&self) -> u64 {
+        self.files_reused_count() + self.total_hardlink_count() + self.files_reencoded_count()
+    }
+
+    #[inline(always)]
+    fn increment_reused_payload_size(&self, file_size: u64) {
+        self.total_reused_payload_size.fetch_add(
+            file_size + size_of::<pxar::format::Header>() as u64,
+            Ordering::Release,
+        );
+        self.files_reused_count.fetch_add(1, Ordering::Release);
+    }
+
+    #[inline(always)]
+    fn increment_reencoded_size(&self, file_size: u64) {
+        self.total_reencoded_size.fetch_add(
+            file_size + size_of::<pxar::format::Header>() as u64,
+            Ordering::Release,
+        );
+        self.files_reencoded_count.fetch_add(1, Ordering::Release);
+    }
+
+    #[inline(always)]
+    fn increment_hardlink_count(&self) {
+        self.files_hardlink_count.fetch_add(1, Ordering::Release);
+    }
+
+    #[inline(always)]
+    fn increment_partial_chunk_count(&self) {
+        self.partial_chunks_count.fetch_add(1, Ordering::Release);
+    }
+
+    #[inline(always)]
+    fn increment_injected_size(&self, chunk_size: u64) {
+        self.total_injected_size
+            .fetch_add(chunk_size, Ordering::Release);
+        self.total_injected_count.fetch_add(1, Ordering::Release);
+    }
 }
 
 #[derive(Serialize, Deserialize)]
@@ -325,30 +401,26 @@ 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_processed(),
+            archiver.progress_stats.total_hardlink_count(),
         );
         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(),
         );
     }
+
     Ok(())
 }
 
@@ -968,24 +1040,18 @@ 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.increment_reused_payload_size(file_size);
                     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.increment_reencoded_size(file_size);
                     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.increment_hardlink_count();
                     self.hardlinks
                         .insert(link_info, (self.path.clone(), offset));
                 }
@@ -1212,11 +1278,9 @@ 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.increment_injected_size(chunk.size());
                 if chunk.padding > 0 {
-                    self.progress_stats.partial_chunks_count += 1;
+                    self.progress_stats.increment_partial_chunk_count();
                 }
 
                 size = size.add(chunk.size());
-- 
2.47.3