[PATCH proxmox-backup v2 08/13] task tracking: use parameter for initial count and refactor updates

[Robert Obkircher <r.obkircher@proxmox.com>]

Don't hard-code the initial count to 1 and bail out if it is negative.
Replace the ActiveOperationStats initializers with the update logic
because they would be reformatted to multiple lines after adding a
field and an operation.

Signed-off-by: Robert Obkircher <r.obkircher@proxmox.com>
---
 pbs-datastore/src/task_tracking.rs | 49 ++++++++++++++++++------------
 1 file changed, 29 insertions(+), 20 deletions(-)

diff --git a/pbs-datastore/src/task_tracking.rs b/pbs-datastore/src/task_tracking.rs
index 528ac8c18..a67b77221 100644
--- a/pbs-datastore/src/task_tracking.rs
+++ b/pbs-datastore/src/task_tracking.rs
@@ -1,4 +1,4 @@
-use anyhow::Error;
+use anyhow::{Error, bail};
 use libc::pid_t;
 use nix::unistd::Pid;
 use std::iter::Sum;
@@ -15,6 +15,16 @@ pub struct ActiveOperationStats {
     pub write: i64,
 }
 
+impl ActiveOperationStats {
+    fn add(&mut self, count: i64, operation: Operation) {
+        match operation {
+            Operation::Read => self.read += count,
+            Operation::Write => self.write += count,
+            Operation::Lookup => (), // no IO must happen there
+        }
+    }
+}
+
 impl Sum<Self> for ActiveOperationStats {
     fn sum<I>(iter: I) -> Self
     where
@@ -102,12 +112,7 @@ pub fn update_active_operations(
 
     let pid = std::process::id();
 
-    let mut updated_active_operations = match operation {
-        Operation::Read => ActiveOperationStats { read: 1, write: 0 },
-        Operation::Write => ActiveOperationStats { read: 0, write: 1 },
-        Operation::Lookup => ActiveOperationStats { read: 0, write: 0 },
-    };
-    let mut found_entry = false;
+    let mut updated = None;
     let mut updated_tasks: Vec<TaskOperations> = match file_read_optional_string(&path)? {
         Some(data) => serde_json::from_str::<Vec<TaskOperations>>(&data)?
             .iter_mut()
@@ -116,13 +121,8 @@ pub fn update_active_operations(
                     Some(stat) if pid == task.pid && stat.starttime != task.starttime => None,
                     Some(_) => {
                         if pid == task.pid {
-                            found_entry = true;
-                            match operation {
-                                Operation::Read => task.active_operations.read += count,
-                                Operation::Write => task.active_operations.write += count,
-                                Operation::Lookup => (), // no IO must happen there
-                            };
-                            updated_active_operations = task.active_operations;
+                            task.active_operations.add(count, operation);
+                            updated = Some(task.active_operations);
                         }
                         Some(task.clone())
                     }
@@ -133,18 +133,27 @@ pub fn update_active_operations(
         None => Vec::new(),
     };
 
-    if !found_entry {
+    let result = if let Some(updated) = updated {
+        updated
+    } else {
+        if count < 0 {
+            bail!("unexpected initial active operation count: {count} {operation:?}")
+        }
+        let mut new = ActiveOperationStats::default();
+        new.add(count, operation);
         updated_tasks.push(TaskOperations {
             pid,
             starttime: procfs::PidStat::read_from_pid(Pid::from_raw(pid as pid_t))?.starttime,
-            active_operations: updated_active_operations,
-        })
-    }
+            active_operations: new,
+        });
+        new
+    };
+
     replace_file(
         &path,
         serde_json::to_string(&updated_tasks)?.as_bytes(),
         options,
         false,
-    )
-    .map(|_| updated_active_operations)
+    )?;
+    Ok(result)
 }
-- 
2.47.3