public inbox for pbs-devel@lists.proxmox.com
 help / color / mirror / Atom feed
From: Christian Ebner <c.ebner@proxmox.com>
To: pbs-devel@lists.proxmox.com
Subject: [pbs-devel] [PATCH v4 proxmox-backup] client: pxar: fix race in pxar backup stream
Date: Sat,  7 Dec 2024 12:07:20 +0100	[thread overview]
Message-ID: <20241207110720.205989-1-c.ebner@proxmox.com> (raw)

Fixes a race condition where the backup upload stream can miss an
error returned by pxar::create_archive, because the error state is
only set after the backup stream was already polled.

On instantiation, `PxarBackupStream` spawns a future handling the
pxar archive creation, which sends the encoded pxar archive stream
(or streams in case of split archives) through a channel, received
by the pxar backup stream on polling.

In case this channel is closed as signaled by returning an error, the
poll logic will propagate an eventual error occurred during pxar
creation by taking it from the `PxarBackupStream`.

As this error might not have been set just yet, this can lead to
incorrectly terminating a backup snapshot with success, eventhough an
error occurred.

To fix this, introduce `ArchiverState` to hold a finish flag as well
as the error and add a notification channel, allowing the archiver
future to signal the waiting stream. As the notification waiter will
block on subsequent polls even if it has already been notified about
the archive creation finish, or it might not have been registered
just yet when the notification was send out, only block and wait for
notifications if the finished flag in the `ArchiverState` is not set.
If it is set, there is no need to wait for a notification, as the
archiver is finished for sure.

In case of premature termination of the pxar backup stream, no
additional measures have to been taken, as the abort handle already
terminates the archive creation.

Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
---
changes since version 3:
- fix a possible deadlock encountered during further testing by
  strictly limiting the archiver state's mutex lock scope.

 pbs-client/src/pxar_backup_stream.rs | 61 +++++++++++++++++++++-------
 1 file changed, 47 insertions(+), 14 deletions(-)

diff --git a/pbs-client/src/pxar_backup_stream.rs b/pbs-client/src/pxar_backup_stream.rs
index 2bfb5cf29..3fb1927d0 100644
--- a/pbs-client/src/pxar_backup_stream.rs
+++ b/pbs-client/src/pxar_backup_stream.rs
@@ -11,6 +11,7 @@ use futures::stream::Stream;
 use nix::dir::Dir;
 use nix::fcntl::OFlag;
 use nix::sys::stat::Mode;
+use tokio::sync::Notify;
 
 use proxmox_async::blocking::TokioWriterAdapter;
 use proxmox_io::StdChannelWriter;
@@ -30,7 +31,13 @@ pub struct PxarBackupStream {
     rx: Option<std::sync::mpsc::Receiver<Result<Vec<u8>, Error>>>,
     pub suggested_boundaries: Option<std::sync::mpsc::Receiver<u64>>,
     handle: Option<AbortHandle>,
-    error: Arc<Mutex<Option<Error>>>,
+    archiver_state: Arc<Mutex<ArchiverState>>,
+    archiver_finished_notification: Arc<Notify>,
+}
+
+struct ArchiverState {
+    finished: bool,
+    error: Option<Error>,
 }
 
 impl Drop for PxarBackupStream {
@@ -78,10 +85,16 @@ impl PxarBackupStream {
                 (pxar::PxarVariant::Unified(writer), None, None, None)
             };
 
-        let error = Arc::new(Mutex::new(None));
-        let error2 = Arc::clone(&error);
+        let archiver_state = ArchiverState {
+            finished: false,
+            error: None,
+        };
+        let archiver_state = Arc::new(Mutex::new(archiver_state));
+        let archiver_state2 = Arc::clone(&archiver_state);
+        let pxar_backup_stream_notifier = Arc::new(Notify::new());
+        let archiver_finished_notification = pxar_backup_stream_notifier.clone();
         let handler = async move {
-            if let Err(err) = crate::pxar::create_archive(
+            let result = crate::pxar::create_archive(
                 dir,
                 PxarWriters::new(
                     writer,
@@ -96,10 +109,19 @@ impl PxarBackupStream {
                 boundaries,
                 suggested_boundaries_tx,
             )
-            .await
-            {
-                let mut error = error2.lock().unwrap();
-                *error = Some(err);
+            .await;
+
+            let mut state = archiver_state2.lock().unwrap();
+            state.finished = true;
+            if let Err(err) = result {
+                state.error = Some(err);
+            }
+            drop(state);
+
+            // Notify upload streams that archiver is finished (with or without error)
+            pxar_backup_stream_notifier.notify_one();
+            if separate_payload_stream {
+                pxar_backup_stream_notifier.notify_one();
             }
         };
 
@@ -111,14 +133,16 @@ impl PxarBackupStream {
             rx: Some(rx),
             suggested_boundaries: None,
             handle: Some(handle.clone()),
-            error: Arc::clone(&error),
+            archiver_state: archiver_state.clone(),
+            archiver_finished_notification: archiver_finished_notification.clone(),
         };
 
         let backup_payload_stream = payload_rx.map(|rx| Self {
             rx: Some(rx),
             suggested_boundaries: suggested_boundaries_rx,
             handle: Some(handle),
-            error,
+            archiver_state,
+            archiver_finished_notification,
         });
 
         Ok((backup_stream, backup_payload_stream))
@@ -143,8 +167,8 @@ impl Stream for PxarBackupStream {
     fn poll_next(self: Pin<&mut Self>, _cx: &mut Context) -> Poll<Option<Self::Item>> {
         {
             // limit lock scope
-            let mut error = self.error.lock().unwrap();
-            if let Some(err) = error.take() {
+            let mut state = self.archiver_state.lock().unwrap();
+            if let Some(err) = state.error.take() {
                 return Poll::Ready(Some(Err(err)));
             }
         }
@@ -152,8 +176,17 @@ impl Stream for PxarBackupStream {
         match proxmox_async::runtime::block_in_place(|| self.rx.as_ref().unwrap().recv()) {
             Ok(data) => Poll::Ready(Some(data)),
             Err(_) => {
-                let mut error = self.error.lock().unwrap();
-                if let Some(err) = error.take() {
+                // If the archiver did not signal it is finished, wait for finished completion
+                // to avoid potentially miss errors
+                let finished = { self.archiver_state.lock().unwrap().finished };
+                if !finished {
+                    proxmox_async::runtime::block_on(
+                        self.archiver_finished_notification.notified(),
+                    );
+                }
+
+                let error = { self.archiver_state.lock().unwrap().error.take() };
+                if let Some(err) = error {
                     return Poll::Ready(Some(Err(err)));
                 }
                 Poll::Ready(None) // channel closed, no error
-- 
2.39.5



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


                 reply	other threads:[~2024-12-07 11:08 UTC|newest]

Thread overview: [no followups] expand[flat|nested]  mbox.gz  Atom feed

Reply instructions:

You may reply publicly to this message via plain-text email
using any one of the following methods:

* Save the following mbox file, import it into your mail client,
  and reply-to-all from there: mbox

  Avoid top-posting and favor interleaved quoting:
  https://en.wikipedia.org/wiki/Posting_style#Interleaved_style

* Reply using the --to, --cc, and --in-reply-to
  switches of git-send-email(1):

  git send-email \
    --in-reply-to=20241207110720.205989-1-c.ebner@proxmox.com \
    --to=c.ebner@proxmox.com \
    --cc=pbs-devel@lists.proxmox.com \
    /path/to/YOUR_REPLY

  https://kernel.org/pub/software/scm/git/docs/git-send-email.html

* If your mail client supports setting the In-Reply-To header
  via mailto: links, try the mailto: link
Be sure your reply has a Subject: header at the top and a blank line before the message body.
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox
Service provided by Proxmox Server Solutions GmbH | Privacy | Legal