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

[Christian Ebner <c.ebner@proxmox.com>]

nit: s/refactor:/client: pxar:/ in commit title

On 6/10/26 8:01 PM, Shan Shaji wrote:
> 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.

But that is not what happens in this patch? The progress logging is 
introduced in subsequent changes only, so maybe explicitly mention that 
this is in preparation for doing this?

> 
> 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 {

nit: non of these helpers need to be exposed as pub. Here they can be 
fully contained in the module, on subsequent patches they might the be 
relaxed to pub(crate) where needed.

> +        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)
> +    }

while at it, it would make sense to also implement the helpers for 
loading hardlink count and especially for counter increments. That makes 
the code below easier to read and improves consistency.

Further, I think the ordering constrains can be relaxed from sequential 
consistency to Acquire/Release.

>   }
>   
>   #[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());