[PATCH proxmox-backup v3 1/3] server: pull: run blocking file operations on the blocking pool

[Jakob Klocker <j.klocker@proxmox.com>]

The atomic renames and the unreferenced-file cleanup are blocking
operations that ran directly on the runtime's worker threads, stalling
other tasks scheduled there for their duration.

Use the tokio counterpart for the renames, the removes, and move the
cleanup onto the blocking thread pool.

Note that tokio::fs runs these on the blocking thread pool, which
carries some overhead over std::fs, but that is acceptable here as these
are one-off operations rather than tight loops.

Signed-off-by: Jakob Klocker <j.klocker@proxmox.com>
---
 src/server/pull.rs | 19 ++++++++++---------
 1 file changed, 10 insertions(+), 9 deletions(-)

diff --git a/src/server/pull.rs b/src/server/pull.rs
index 8165719c4..d4bd07d94 100644
--- a/src/server/pull.rs
+++ b/src/server/pull.rs
@@ -452,7 +452,7 @@ async fn pull_single_archive<'a>(
         .with_context(|| archive_prefix.clone())?
         .is_none()
     {
-        let _ = std::fs::remove_file(&tmp_path);
+        let _ = tokio::fs::remove_file(&tmp_path).await;
         bail!("{archive_prefix}: archive missing on source");
     };
 
@@ -593,14 +593,14 @@ async fn pull_single_archive<'a>(
         }
     }
     let source_path = if crypt_config.is_some() {
-        if let Err(err) = std::fs::remove_file(&tmp_path) {
+        if let Err(err) = tokio::fs::remove_file(&tmp_path).await {
             bail!("{archive_prefix}: Failed to remove temp. file {tmp_path:?} failed - {err}");
         }
         tmp_dec_path
     } else {
         tmp_path
     };
-    if let Err(err) = std::fs::rename(&source_path, &path) {
+    if let Err(err) = tokio::fs::rename(&source_path, &path).await {
         bail!("{archive_prefix}: Atomic rename file {path:?} failed - {err}");
     }
 
@@ -659,7 +659,7 @@ async fn pull_snapshot<'a>(
         .await
         .with_context(|| prefix.clone())?
     else {
-        let _ = std::fs::remove_file(&tmp_manifest_name);
+        let _ = tokio::fs::remove_file(&tmp_manifest_name).await;
         log_sender
             .log(
                 Level::INFO,
@@ -710,7 +710,7 @@ async fn pull_snapshot<'a>(
         log_sender
             .log(Level::INFO, format!("{prefix}: no data changes"))
             .await?;
-        let _ = std::fs::remove_file(&tmp_manifest_name);
+        let _ = tokio::fs::remove_file(&tmp_manifest_name).await;
         Ok::<(), Error>(())
     };
 
@@ -749,7 +749,7 @@ async fn pull_snapshot<'a>(
         params.verified_only,
         params.encrypted_only,
     ) {
-        let _ = std::fs::remove_file(&tmp_manifest_name);
+        let _ = tokio::fs::remove_file(&tmp_manifest_name).await;
         log_sender
             .log(
                 Level::INFO,
@@ -890,7 +890,7 @@ async fn pull_snapshot<'a>(
         nix::unistd::fsync(tmp_manifest_file.as_raw_fd())?;
     }
 
-    if let Err(err) = std::fs::rename(&tmp_manifest_name, &manifest_name) {
+    if let Err(err) = tokio::fs::rename(&tmp_manifest_name, &manifest_name).await {
         bail!("{prefix}: Atomic rename file {manifest_name:?} failed - {err}");
     }
     if let DatastoreBackend::S3(s3_client) = backend {
@@ -910,8 +910,9 @@ async fn pull_snapshot<'a>(
 
     fetch_log(crypt_config).await?;
 
-    snapshot
-        .cleanup_unreferenced_files(&manifest)
+    let snapshot = snapshot.clone();
+    tokio::task::spawn_blocking(move || snapshot.cleanup_unreferenced_files(&manifest))
+        .await?
         .map_err(|err| format_err!("{prefix}: failed to cleanup unreferenced files - {err}"))?;
 
     Ok(Some(sync_stats))
-- 
2.47.3