Make backpressure ignore Offline downstairs (#1431)

One of my complaints in #1258 is that the `Offline` → `Faulted`
transition depends on reaching a certain number of jobs in the queue,
but accumulating those jobs is subject to backpressure.

This has a few awkward consequences:
- The time it takes to transition from `Offline` → `Faulted` depends on
backpressure tuning, and we have to pick backpressure parameters that
make this take a reasonable amount of time (tested in
`check_offline_timeout`)
- The offline Downstairs slows down the other Downstairs based on the
number of jobs in its queue, meaning the system as a whole gets slower
and slower until that Downstairs is marked as faulted

This PR tweaks the backpressure implementation so that only `Active`
Downstairs are counted. To do so, we move the `write_bytes_outstanding`
to a per-client field. The backpressure implementation is also moved to
a separate module, since we'll be adding more to it in the future.

The `check_offline_timeout` test is no longer relevant and has been
removed.

Author: mkeeter

upstairs/src/backpressure.rs

@@ -0,0 +1,45 @@
+// Copyright 2024 Oxide Computer Company
+
+use crate::{ClientId, DownstairsIO, IOop};
+
+/// Helper struct to contain a count of backpressure bytes
+#[derive(Debug)]
+pub struct BackpressureBytes(u64);
+
+impl BackpressureBytes {
+    pub fn new() -> Self {
+        BackpressureBytes(0)
+    }
+
+    pub fn get(&self) -> u64 {
+        self.0
+    }
+
+    /// Ensures that the given `DownstairsIO` is counted for backpressure
+    ///
+    /// This is idempotent: if the job has already been counted, indicated by
+    /// the `DownstairsIO::backpressure_bytes[c]` member being `Some(..)`, it
+    /// will not be counted again.
+    pub fn increment(&mut self, io: &mut DownstairsIO, c: ClientId) {
+        match &io.work {
+            IOop::Write { data, .. } | IOop::WriteUnwritten { data, .. } => {
+                if !io.backpressure_bytes.contains(&c) {
+                    let n = data.len() as u64;
+                    io.backpressure_bytes.insert(c, n);
+                    self.0 += n;
+                }
+            }
+            _ => (),
+        };
+    }
+
+    /// Remove the given job's contribution to backpressure
+    ///
+    /// This is idempotent: `DownstairsIO::backpressure_bytes[c]` is set to
+    /// `None` by this function call, so it's harmless to call repeatedly.
+    pub fn decrement(&mut self, io: &mut DownstairsIO, c: ClientId) {
+        if let Some(n) = io.backpressure_bytes.take(&c) {
+            self.0 = self.0.checked_sub(n).unwrap();
+        }
+    }
+}

upstairs/src/client.rs

@@ -1,10 +1,10 @@
 // Copyright 2023 Oxide Computer Company
 use crate::{
-    cdt, integrity_hash, live_repair::ExtentInfo, upstairs::UpstairsConfig,
-    upstairs::UpstairsState, ClientIOStateCount, ClientId, CrucibleDecoder,
-    CrucibleError, DownstairsIO, DsState, EncryptionContext, IOState, IOop,
-    JobId, Message, RawReadResponse, ReconcileIO, RegionDefinitionStatus,
-    RegionMetadata, Validation,
+    backpressure::BackpressureBytes, cdt, integrity_hash,
+    live_repair::ExtentInfo, upstairs::UpstairsConfig, upstairs::UpstairsState,
+    ClientIOStateCount, ClientId, CrucibleDecoder, CrucibleError, DownstairsIO,
+    DsState, EncryptionContext, IOState, IOop, JobId, Message, RawReadResponse,
+    ReconcileIO, RegionDefinitionStatus, RegionMetadata, Validation,
 };
 use crucible_common::{
     deadline_secs, verbose_timeout, x509::TLSContext, ExtentId,
@@ -126,6 +126,9 @@ pub(crate) struct DownstairsClient {
     /// estimate per-client backpressure to keep the 3x downstairs in sync.
     pub(crate) bytes_outstanding: u64,
 
+    /// Write bytes in this queue, used for global backpressure
+    pub(crate) write_bytes_outstanding: BackpressureBytes,
+
     /// UUID for this downstairs region
     ///
     /// Unpopulated until provided by `Message::RegionInfo`
@@ -230,6 +233,7 @@ impl DownstairsClient {
             repair_info: None,
             io_state_count: ClientIOStateCount::new(),
             bytes_outstanding: 0,
+            write_bytes_outstanding: BackpressureBytes::new(),
             connection_id: ConnectionId(0),
             client_delay_us,
         }
@@ -270,6 +274,7 @@ impl DownstairsClient {
             repair_info: None,
             io_state_count: ClientIOStateCount::new(),
             bytes_outstanding: 0,
+            write_bytes_outstanding: BackpressureBytes::new(),
             connection_id: ConnectionId(0),
             client_delay_us,
         }
@@ -379,10 +384,12 @@ impl DownstairsClient {
                 .bytes_outstanding
                 .checked_sub(job.work.job_bytes())
                 .unwrap();
+            self.write_bytes_outstanding.decrement(job, self.client_id);
         } else if is_running && !was_running {
             // This should only happen if a job is replayed, but that still
             // counts!
             self.bytes_outstanding += job.work.job_bytes();
+            self.write_bytes_outstanding.increment(job, self.client_id);
         }
 
         old_state
@@ -934,6 +941,7 @@ impl DownstairsClient {
         };
         if r == IOState::New {
             self.bytes_outstanding += io.work.job_bytes();
+            self.write_bytes_outstanding.increment(io, self.client_id);
         }
         self.io_state_count.incr(&r);
         r

upstairs/src/downstairs.rs

@@ -73,17 +73,6 @@ pub(crate) struct Downstairs {
     /// The active list of IO for the downstairs.
     pub(crate) ds_active: ActiveJobs,
 
-    /// The number of write bytes that haven't finished yet
-    ///
-    /// This is used to configure backpressure to the host, because writes
-    /// (uniquely) will return before actually being completed by a Downstairs
-    /// and can clog up the queues.
-    ///
-    /// It is stored in the Downstairs because from the perspective of the
-    /// Upstairs, writes complete immediately; only the Downstairs is actually
-    /// tracking the pending jobs.
-    write_bytes_outstanding: BackpressureBytes,
-
     /// The next Job ID this Upstairs should use for downstairs work.
     next_id: JobId,
 
@@ -139,44 +128,6 @@ pub(crate) struct Downstairs {
     reqwest_client: reqwest::Client,
 }
 
-/// Helper struct to contain a count of backpressure bytes
-#[derive(Debug)]
-struct BackpressureBytes(u64);
-
-impl BackpressureBytes {
-    fn new() -> Self {
-        BackpressureBytes(0)
-    }
-
-    /// Ensures that the given `DownstairsIO` is counted for backpressure
-    ///
-    /// This is idempotent: if the job has already been counted, indicated by
-    /// the `DownstairsIO::backpressure_bytes` member being `Some(..)`, it will
-    /// not be counted again.
-    fn increment(&mut self, io: &mut DownstairsIO) {
-        match &io.work {
-            IOop::Write { data, .. } | IOop::WriteUnwritten { data, .. } => {
-                if io.backpressure_bytes.is_none() {
-                    let n = data.len() as u64;
-                    io.backpressure_bytes = Some(n);
-                    self.0 += n;
-                }
-            }
-            _ => (),
-        };
-    }
-
-    /// Remove the given job's contribution to backpressure
-    ///
-    /// This is idempotent: `DownstairsIO::backpressure_bytes` is set to `None`
-    /// by this function call, so it's harmless to call repeatedly.
-    fn decrement(&mut self, io: &mut DownstairsIO) {
-        if let Some(n) = io.backpressure_bytes.take() {
-            self.0 = self.0.checked_sub(n).unwrap();
-        }
-    }
-}
-
 /// State machine for a live-repair operation
 ///
 /// We pass through all states (except `FinalFlush`) in order for each extent,
@@ -331,7 +282,6 @@ impl Downstairs {
             cfg,
             next_flush: 0,
             ds_active: ActiveJobs::new(),
-            write_bytes_outstanding: BackpressureBytes::new(),
             completed: AllocRingBuffer::new(2048),
             completed_jobs: AllocRingBuffer::new(8),
             next_id: JobId(1000),
@@ -936,10 +886,6 @@ impl Downstairs {
             if self.clients[client_id].replay_job(job) {
                 count += 1;
             }
-
-            // Make sure this job counts for backpressure (this is a no-op if
-            // the job is already counted).
-            self.write_bytes_outstanding.increment(job);
         });
         info!(
             self.log,
@@ -1519,7 +1465,7 @@ impl Downstairs {
             replay: false,
             data: None,
             read_validations: Vec::new(),
-            backpressure_bytes: None,
+            backpressure_bytes: ClientMap::new(),
         }
     }
 
@@ -1643,7 +1589,7 @@ impl Downstairs {
             replay: false,
             data: None,
             read_validations: Vec::new(),
-            backpressure_bytes: None,
+            backpressure_bytes: ClientMap::new(),
         }
     }
 
@@ -1784,7 +1730,7 @@ impl Downstairs {
             replay: false,
             data: None,
             read_validations: Vec::new(),
-            backpressure_bytes: None,
+            backpressure_bytes: ClientMap::new(),
         }
     }
 
@@ -1842,7 +1788,7 @@ impl Downstairs {
             replay: false,
             data: None,
             read_validations: Vec::new(),
-            backpressure_bytes: None,
+            backpressure_bytes: ClientMap::new(),
         };
         self.enqueue(io);
         ds_id
@@ -1926,7 +1872,7 @@ impl Downstairs {
             replay: false,
             data: None,
             read_validations: Vec::new(),
-            backpressure_bytes: None,
+            backpressure_bytes: ClientMap::new(),
         };
         self.enqueue(io);
         ds_id
@@ -1957,7 +1903,7 @@ impl Downstairs {
             replay: false,
             data: None,
             read_validations: Vec::new(),
-            backpressure_bytes: None,
+            backpressure_bytes: ClientMap::new(),
         }
     }
 
@@ -2366,7 +2312,7 @@ impl Downstairs {
             replay: false,
             data: None,
             read_validations: Vec::new(),
-            backpressure_bytes: None,
+            backpressure_bytes: ClientMap::new(),
         };
 
         self.enqueue(fl);
@@ -2493,7 +2439,7 @@ impl Downstairs {
             replay: false,
             data: None,
             read_validations: Vec::new(),
-            backpressure_bytes: None,
+            backpressure_bytes: ClientMap::new(),
         };
 
         self.enqueue(io);
@@ -2580,7 +2526,6 @@ impl Downstairs {
         }
 
         // Make sure this job is counted for backpressure
-        self.write_bytes_outstanding.increment(&mut io);
         let is_write = matches!(io.work, IOop::Write { .. });
 
         // Puts the IO onto the downstairs work queue.
@@ -2963,12 +2908,17 @@ impl Downstairs {
                 // they are completed (in `process_io_completion_inner`),
                 // **not** when they are retired.  We'll do a sanity check here
                 // and print a warning if that's not the case.
-                if job.backpressure_bytes.is_some() {
-                    warn!(
-                        self.log,
-                        "job {ds_id} had pending backpressure bytes"
-                    );
-                    self.write_bytes_outstanding.decrement(&mut job);
+                for c in ClientId::iter() {
+                    if job.backpressure_bytes.contains(&c) {
+                        warn!(
+                            self.log,
+                            "job {ds_id} had pending backpressure bytes \
+                             for client {c}"
+                        );
+                        self.clients[c]
+                            .write_bytes_outstanding
+                            .decrement(&mut job, c);
+                    }
                 }
             }
 
@@ -3467,14 +3417,6 @@ impl Downstairs {
             self.ackable_work.insert(ds_id);
         }
 
-        // Write bytes no longer count for backpressure once all 3x downstairs
-        // have returned (although they'll continue to be stored until they are
-        // retired by the next flush).
-        let wc = job.state_count();
-        if (wc.error + wc.skipped + wc.done) == 3 {
-            self.write_bytes_outstanding.decrement(job);
-        }
-
         /*
          * If all 3 jobs are done, we can check here to see if we can
          * remove this job from the DS list. If we have completed the ack
@@ -3492,6 +3434,7 @@ impl Downstairs {
             // If we are a write or a flush with one success, then
             // we must switch our state to failed.  This condition is
             // handled when we check the job result.
+            let wc = job.state_count();
             if (wc.error + wc.skipped + wc.done) == 3 {
                 self.ackable_work.insert(ds_id);
                 debug!(self.log, "[{}] Set AckReady {}", client_id, job.ds_id);
@@ -3589,7 +3532,12 @@ impl Downstairs {
     }
 
     pub(crate) fn write_bytes_outstanding(&self) -> u64 {
-        self.write_bytes_outstanding.0
+        self.clients
+            .iter()
+            .filter(|c| matches!(c.state(), DsState::Active))
+            .map(|c| c.write_bytes_outstanding.get())
+            .max()
+            .unwrap_or(0)
     }
 
     /// Marks a single job as acked

upstairs/src/guest.rs

@@ -1461,55 +1461,6 @@ mod test {
         Ok(())
     }
 
-    /// Confirm that the offline timeout is reasonable
-    #[test]
-    fn check_offline_timeout() {
-        for job_size in
-            [512, 4 * 1024, 16 * 1024, 64 * 1024, 256 * 1024, 1024 * 1024]
-        {
-            let mut bytes_in_flight = 0;
-            let mut jobs_in_flight = 0;
-            let mut time_usec: u64 = 0;
-            let cfg = Guest::default_backpressure_config();
-
-            let (t, desc) = loop {
-                let bp_usec =
-                    cfg.get_backpressure_us(bytes_in_flight, jobs_in_flight);
-                time_usec = time_usec.saturating_add(bp_usec);
-
-                if bytes_in_flight >= IO_OUTSTANDING_MAX_BYTES {
-                    break (time_usec, "bytes");
-                }
-
-                if jobs_in_flight >= IO_OUTSTANDING_MAX_JOBS as u64 {
-                    break (time_usec, "jobs");
-                }
-
-                bytes_in_flight += job_size;
-                jobs_in_flight += 1;
-            };
-
-            let timeout = Duration::from_micros(t);
-            assert!(
-                timeout > Duration::from_secs(1),
-                "offline -> faulted transition happens too quickly \
-                 with job size {job_size};  expected > 1 sec, got {}",
-                humantime::format_duration(timeout)
-            );
-            assert!(
-                timeout < Duration::from_secs(180),
-                "offline -> faulted transition happens too slowly \
-                 with job size {job_size};  expected < 3 mins, got {}",
-                humantime::format_duration(timeout)
-            );
-
-            println!(
-                "job size {job_size:>8}:\n    Timeout in {} ({desc})\n",
-                humantime::format_duration(timeout)
-            );
-        }
-    }
-
     #[test]
     fn check_max_backpressure() {
         let cfg = Guest::default_backpressure_config();