Use to-infinity backpressure with per-client backpressure as well

Author: mkeeter

upstairs/src/backpressure.rs

@@ -0,0 +1,74 @@
+// Copyright 2024 Oxide Computer Company
+
+use tokio::time::Duration;
+
+/// Configuration for host-side backpressure
+///
+/// Backpressure adds an artificial delay to host write messages (which are
+/// otherwise acked immediately, before actually being complete).  The delay is
+/// varied based on two metrics:
+///
+/// - number of write bytes outstanding
+/// - number of jobs in the queue
+///
+/// The backpressure delay is the max of byte-based delay and queue-based delay.
+#[derive(Copy, Clone, Debug)]
+pub(crate) struct BackpressureConfig {
+    /// Configuration for byte-based backpressure
+    pub bytes: BackpressureScale,
+    /// Configuration for job-based backpressure
+    pub jobs: BackpressureScale,
+}
+
+/// Single backpressure curve configuration
+#[derive(Copy, Clone, Debug)]
+pub(crate) struct BackpressureScale {
+    pub start: u64,
+    pub max: u64,
+    pub scale: Duration,
+}
+
+impl BackpressureScale {
+    // Our chosen backpressure curve is quadratic for 1/2 of its range, then
+    // goes to infinity in the second half.  This gives C0 + C1 continuity.
+    pub fn curve(&self, n: u64) -> Duration {
+        // Calculate a value from 0-2
+        let frac = n.saturating_sub(self.start) as f64
+            / (self.max - self.start) as f64
+            * 2.0;
+
+        let v = if frac < 1.0 {
+            frac
+        } else {
+            1.0 / (1.0 - (frac - 1.0))
+        };
+        self.scale.mul_f64(v.powi(2))
+    }
+}
+
+impl BackpressureConfig {
+    #[cfg(test)]
+    pub fn disable(&mut self) {
+        self.bytes.scale = Duration::ZERO;
+        self.jobs.scale = Duration::ZERO;
+    }
+
+    pub fn get_backpressure_us(&self, bytes: u64, jobs: u64) -> u64 {
+        // Special case if backpressure is disabled
+        if self.bytes.scale == Duration::ZERO
+            && self.jobs.scale == Duration::ZERO
+        {
+            return 0;
+        }
+
+        // Saturate at 1 hour per job, which is basically infinite
+        if bytes >= self.bytes.max || jobs >= self.jobs.max {
+            return Duration::from_secs(60 * 60).as_micros() as u64;
+        }
+
+        let delay_bytes = self.bytes.curve(bytes).as_micros() as u64;
+        let delay_jobs = self.jobs.curve(jobs).as_micros() as u64;
+
+        delay_bytes.max(delay_jobs)
+    }
+}

upstairs/src/downstairs.rs

@@ -13,12 +13,12 @@ use crate::{
     live_repair::ExtentInfo,
     stats::UpStatOuter,
     upstairs::{UpstairsConfig, UpstairsState},
-    AckStatus, ActiveJobs, AllocRingBuffer, ClientData, ClientIOStateCount,
-    ClientId, ClientMap, CrucibleError, DownstairsIO, DownstairsMend, DsState,
-    ExtentFix, ExtentRepairIDs, GuestWorkId, IOState, IOStateCount, IOop,
-    ImpactedBlocks, JobId, Message, ReadRequest, ReadResponse, ReconcileIO,
-    ReconciliationId, RegionDefinition, ReplaceResult, SnapshotDetails,
-    WorkSummary,
+    AckStatus, ActiveJobs, AllocRingBuffer, BackpressureConfig,
+    BackpressureScale, ClientData, ClientIOStateCount, ClientId, ClientMap,
+    CrucibleError, DownstairsIO, DownstairsMend, DsState, ExtentFix,
+    ExtentRepairIDs, GuestWorkId, IOState, IOStateCount, IOop, ImpactedBlocks,
+    JobId, Message, ReadRequest, ReadResponse, ReconcileIO, ReconciliationId,
+    RegionDefinition, ReplaceResult, SnapshotDetails, WorkSummary,
 };
 use crucible_protocol::{RawWrite, WriteHeader};
 
@@ -65,7 +65,17 @@ pub(crate) struct Downstairs {
     pub(crate) clients: ClientData<DownstairsClient>,
 
     /// Per-client backpressure configuration
-    backpressure_config: DownstairsBackpressureConfig,
+    ///
+    /// This backpressure adds an artificial delay to the client queues, to keep
+    /// the three clients relatively in sync.  The delay is varied based on two
+    /// metrics:
+    ///
+    /// - number of write bytes outstanding
+    /// - queue length
+    ///
+    /// These metrics are _relative_ to the slowest downstairs; the goal is to
+    /// slow down the faster Downstairs to keep the gap bounded.
+    backpressure_config: BackpressureConfig,
 
     /// The active list of IO for the downstairs.
     pub(crate) ds_active: ActiveJobs,
@@ -304,22 +314,20 @@ impl Downstairs {
         let clients = clients.map(Option::unwrap);
         Self {
             clients: ClientData(clients),
-            backpressure_config: DownstairsBackpressureConfig {
+            backpressure_config: BackpressureConfig {
                 // start byte-based backpressure at 25 MiB of difference
-                bytes_start: 25 * 1024 * 1024,
-                // bytes_scale is chosen to have 1.5 ms of delay at 100 MiB of
-                // discrepancy
-                bytes_scale: 5e-7,
+                bytes: BackpressureScale {
+                    start: 25 * 1024 * 1024,
+                    max: 1024 * 1024 * 200, // max difference is 200 MiB
+                    scale: Duration::from_micros(1500), // 1.5 ms at 100 MiB
+                },
 
                 // start job-based backpressure at 100 jobs of discrepancy
-                jobs_start: 100,
-                // job_scale is chosen to have 1 ms of of per-job delay at 900
-                // jobs of discrepancy
-                jobs_scale: 0.04,
-
-                // max delay is 100 ms, chosen experimentally to keep downstairs
-                // in sync even in heavily loaded systems
-                max_delay: Duration::from_millis(100),
+                jobs: BackpressureScale {
+                    start: 100,
+                    max: 2000,
+                    scale: Duration::from_millis(1), // 1 ms at 1000 jobs
+                },
             },
             cfg,
             next_flush: 0,
@@ -3801,7 +3809,7 @@ impl Downstairs {
 
     #[cfg(test)]
     pub(crate) fn disable_client_backpressure(&mut self) {
-        self.backpressure_config.max_delay = Duration::ZERO;
+        self.backpressure_config.disable();
     }
 
     pub(crate) fn set_client_backpressure(&self) {
@@ -3839,22 +3847,10 @@ impl Downstairs {
                 let bytes_gap =
                     (max_bytes - c.total_bytes_outstanding()) as u64;
 
-                let job_delay_us = (job_gap
-                    .saturating_sub(self.backpressure_config.jobs_start)
-                    as f64
-                    * self.backpressure_config.jobs_scale)
-                    .powf(2.0);
-
-                let bytes_delay_us = (bytes_gap
-                    .saturating_sub(self.backpressure_config.bytes_start)
-                    as f64
-                    * self.backpressure_config.bytes_scale)
-                    .powf(2.0);
-
-                let delay_us = (job_delay_us.max(bytes_delay_us) as u64)
-                    .min(self.backpressure_config.max_delay.as_micros() as u64);
-
-                delays[i] = Some(delay_us);
+                delays[i] = Some(
+                    self.backpressure_config
+                        .get_backpressure_us(bytes_gap, job_gap),
+                );
             }
         }
         // Cancel out any common delay, because it wouldn't be useful.
@@ -4427,33 +4423,6 @@ impl Downstairs {
     }
 }
 
-/// Configuration for per-client backpressure
-///
-/// Per-client backpressure adds an artificial delay to the client queues, to
-/// keep the three clients relatively in sync.  The delay is varied based on two
-/// metrics:
-///
-/// - number of write bytes outstanding
-/// - queue length
-///
-/// These metrics are _relative_ to the slowest downstairs; the goal is to slow
-/// down the faster Downstairs to keep the gap bounded.
-#[derive(Copy, Clone, Debug)]
-struct DownstairsBackpressureConfig {
-    /// When should backpressure start (in bytes)?
-    bytes_start: u64,
-    /// Scale for byte-based quadratic backpressure
-    bytes_scale: f64,
-
-    /// When should job-count-based backpressure start?
-    jobs_start: u64,
-    /// Scale for job-count-based quadratic backpressure
-    jobs_scale: f64,
-
-    /// Maximum delay
-    max_delay: Duration,
-}
-
 #[cfg(test)]
 pub(crate) mod test {
     use super::Downstairs;

upstairs/src/dummy_downstairs_tests.rs

@@ -518,8 +518,7 @@ impl TestHarness {
         // Configure our guest without backpressure, to speed up tests which
         // require triggering a timeout
         let (g, mut io) = Guest::new(Some(log.clone()));
-        io.disable_queue_backpressure();
-        io.disable_byte_backpressure();
+        io.disable_backpressure();
         let guest = Arc::new(g);
 
         let crucible_opts = CrucibleOpts {

upstairs/src/guest.rs

@@ -10,6 +10,7 @@ use std::{
 };
 
 use crate::{
+    backpressure::{BackpressureConfig, BackpressureScale},
     BlockIO, BlockOp, BlockOpWaiter, BlockRes, Buffer, JobId, ReplaceResult,
     UpstairsAction, IO_OUTSTANDING_MAX_BYTES, IO_OUTSTANDING_MAX_JOBS,
 };
@@ -304,71 +305,6 @@ pub struct Guest {
     log: Logger,
 }
 
-/// Configuration for host-side backpressure
-///
-/// Backpressure adds an artificial delay to host write messages (which are
-/// otherwise acked immediately, before actually being complete).  The delay is
-/// varied based on two metrics:
-///
-/// - number of write bytes outstanding (as a fraction of max)
-/// - queue length as a fraction (where 1.0 is full)
-///
-/// These two metrics are used for quadratic backpressure, picking the larger of
-/// the two delays.
-#[derive(Copy, Clone, Debug)]
-struct BackpressureConfig {
-    /// When should backpressure start, in units of bytes
-    bytes_start: u64,
-    /// Maximum number of bytes (i.e. backpressure goes to infinity)
-    bytes_max: u64,
-    /// Scale of bytes-based backpressure
-    bytes_scale: Duration,
-
-    /// When should backpressure start, in units of jobs
-    queue_start: u64,
-    /// Maximum number of jobs (i.e. backpressure goes to infinity)
-    queue_max: u64,
-    /// Scale of queue-based delay
-    queue_scale: Duration,
-}
-
-impl BackpressureConfig {
-    // Our chosen backpressure curve is quadratic for 1/2 of its range, then
-    // goes to infinity in the second half.  This gives C0 + C1 continuity.
-    fn curve(frac: f64, scale: Duration) -> Duration {
-        // Remap from 0-1 to 0-1.5 for ease of calculation
-        let frac = frac * 2.0;
-        let v = if frac < 1.0 {
-            frac
-        } else {
-            1.0 / (1.0 - (frac - 1.0))
-        };
-        scale.mul_f64(v.powi(2))
-    }
-
-    fn get_backpressure_us(&self, bytes: u64, jobs: u64) -> u64 {
-        // Saturate at 1 hour per job, which is basically infinite
-        if bytes >= self.bytes_max || jobs >= self.queue_max {
-            return Duration::from_secs(60 * 60).as_micros() as u64;
-        }
-
-        // These ratios start at 0 (at *_start) and hit 1 when backpressure
-        // should be infinite.
-        let jobs_frac = jobs.saturating_sub(self.queue_start) as f64
-            / (self.queue_max - self.queue_start) as f64;
-        let bytes_frac = bytes.saturating_sub(self.bytes_start) as f64
-            / (self.bytes_max - self.bytes_start) as f64;
-
-        // Delay should be 0 at frac = 0, and infinite at frac = 1
-        let delay_bytes =
-            Self::curve(bytes_frac, self.bytes_scale).as_micros() as u64;
-        let delay_jobs =
-            Self::curve(jobs_frac, self.queue_scale).as_micros() as u64;
-
-        delay_bytes.max(delay_jobs)
-    }
-}
-
 /*
  * These methods are how to add or checking for new work on the Guest struct
  */
@@ -429,14 +365,18 @@ impl Guest {
     fn default_backpressure_config() -> BackpressureConfig {
         BackpressureConfig {
             // Byte-based backpressure
-            bytes_start: 50 * 1024u64.pow(2), // 50 MiB
-            bytes_max: IO_OUTSTANDING_MAX_BYTES * 2,
-            bytes_scale: Duration::from_millis(100),
+            bytes: BackpressureScale {
+                start: 50 * 1024u64.pow(2), // 50 MiB
+                max: IO_OUTSTANDING_MAX_BYTES * 2,
+                scale: Duration::from_millis(100),
+            },
 
             // Queue-based backpressure
-            queue_start: 500,
-            queue_max: IO_OUTSTANDING_MAX_JOBS as u64 * 2,
-            queue_scale: Duration::from_millis(5),
+            jobs: BackpressureScale {
+                start: 500,
+                max: IO_OUTSTANDING_MAX_JOBS as u64 * 2,
+                scale: Duration::from_millis(5),
+            },
         }
     }
 
@@ -959,18 +899,13 @@ impl GuestIoHandle {
     }
 
     #[cfg(test)]
-    pub fn disable_queue_backpressure(&mut self) {
-        self.backpressure_config.queue_scale = Duration::ZERO;
-    }
-
-    #[cfg(test)]
-    pub fn disable_byte_backpressure(&mut self) {
-        self.backpressure_config.bytes_scale = Duration::ZERO;
+    pub fn disable_backpressure(&mut self) {
+        self.backpressure_config.disable();
     }
 
     #[cfg(test)]
-    pub fn is_queue_backpressure_disabled(&self) -> bool {
-        self.backpressure_config.queue_scale == Duration::ZERO
+    pub fn is_backpressure_disabled(&self) -> bool {
+        self.backpressure_config.jobs.scale == Duration::ZERO
     }
 
     /// Set `self.backpressure_us` based on outstanding IO ratio

upstairs/src/lib.rs

@@ -55,6 +55,9 @@ mod buffer;
 pub use buffer::Buffer; // used in BlockIO::Read, so it must be public
 pub(crate) use buffer::UninitializedBuffer; // only used in pub(crate) functions
 
+mod backpressure;
+pub(crate) use backpressure::{BackpressureConfig, BackpressureScale};
+
 mod mend;
 pub use mend::{DownstairsMend, ExtentFix, RegionMetadata};
 pub use pseudo_file::CruciblePseudoFile;
@@ -1700,7 +1703,7 @@ pub fn up_main(
     };
 
     #[cfg(test)]
-    let disable_backpressure = guest.is_queue_backpressure_disabled();
+    let disable_backpressure = guest.is_backpressure_disabled();
 
     /*
      * Build the Upstairs struct that we use to share data between