Add per-client backpressure

upstairs/src/client.rs

@@ -10,15 +10,22 @@ use crate::{
 use crucible_common::x509::TLSContext;
 use crucible_protocol::{ReconciliationId, CRUCIBLE_MESSAGE_VERSION};
 
-use std::{collections::BTreeSet, net::SocketAddr, sync::Arc};
+use std::{
+    collections::BTreeSet,
+    net::SocketAddr,
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    },
+};
 
 use futures::StreamExt;
 use slog::{debug, error, info, o, warn, Logger};
 use tokio::{
     io::AsyncWriteExt,
     net::{TcpSocket, TcpStream},
     sync::{mpsc, oneshot},
-    time::sleep_until,
+    time::{sleep_until, Duration},
 };
 use tokio_util::codec::{Encoder, FramedRead};
 use uuid::Uuid;
@@ -216,6 +223,9 @@ pub(crate) struct DownstairsClient {
 
     /// Session ID for a clients connection to a downstairs.
     connection_id: ConnectionId,
+
+    /// Per-client delay, shared with the [`DownstairsClient`]
+    client_delay_us: Arc<AtomicU64>,
 }
 
 impl DownstairsClient {
@@ -226,6 +236,7 @@ impl DownstairsClient {
         log: Logger,
         tls_context: Option<Arc<crucible_common::x509::TLSContext>>,
     ) -> Self {
+        let client_delay_us = Arc::new(AtomicU64::new(0));
         Self {
             cfg,
             client_task: Self::new_io_task(
@@ -234,6 +245,7 @@ impl DownstairsClient {
                 false, // do not start the task until GoActive
                 client_id,
                 tls_context.clone(),
+                client_delay_us.clone(),
                 &log,
             ),
             client_id,
@@ -253,6 +265,7 @@ impl DownstairsClient {
             repair_info: None,
             io_state_count: ClientIOStateCount::new(),
             connection_id: ConnectionId(0),
+            client_delay_us,
         }
     }
 
@@ -262,6 +275,7 @@ impl DownstairsClient {
     /// client will disappear into the void.
     #[cfg(test)]
     fn test_default() -> Self {
+        let client_delay_us = Arc::new(AtomicU64::new(0));
         let cfg = Arc::new(UpstairsConfig {
             encryption_context: None,
             upstairs_id: Uuid::new_v4(),
@@ -290,6 +304,7 @@ impl DownstairsClient {
             repair_info: None,
             io_state_count: ClientIOStateCount::new(),
             connection_id: ConnectionId(0),
+            client_delay_us,
         }
     }
 
@@ -628,6 +643,7 @@ impl DownstairsClient {
         }
 
         self.connection_id.update();
+
         // Restart with a short delay
         self.start_task(true, auto_promote);
     }
@@ -652,6 +668,7 @@ impl DownstairsClient {
             connect,
             self.client_id,
             self.tls_context.clone(),
+            self.client_delay_us.clone(),
             &self.log,
         );
     }
@@ -662,6 +679,7 @@ impl DownstairsClient {
         connect: bool,
         client_id: ClientId,
         tls_context: Option<Arc<TLSContext>>,
+        client_delay_us: Arc<AtomicU64>,
         log: &Logger,
     ) -> ClientTaskHandle {
         #[cfg(test)]
@@ -672,6 +690,7 @@ impl DownstairsClient {
                 connect,
                 client_id,
                 tls_context,
+                client_delay_us,
                 log,
             )
         } else {
@@ -685,6 +704,7 @@ impl DownstairsClient {
             connect,
             client_id,
             tls_context,
+            client_delay_us,
             log,
         )
     }
@@ -695,6 +715,7 @@ impl DownstairsClient {
         connect: bool,
         client_id: ClientId,
         tls_context: Option<Arc<TLSContext>>,
+        client_delay_us: Arc<AtomicU64>,
         log: &Logger,
     ) -> ClientTaskHandle {
         // These channels must support at least MAX_ACTIVE_COUNT messages;
@@ -730,6 +751,7 @@ impl DownstairsClient {
                     log: log.clone(),
                 },
                 delay,
+                client_delay_us,
                 log,
             };
             c.run().await
@@ -1242,6 +1264,8 @@ impl DownstairsClient {
             return false;
         }
 
+        job.reply_time[self.client_id] = Some(std::time::Instant::now());
+
         let mut jobs_completed_ok = job.state_count().completed_ok();
         let mut ackable = false;
 
@@ -2220,6 +2244,16 @@ impl DownstairsClient {
             None
         }
     }
+
+    /// Sets the per-client delay
+    pub(crate) fn set_delay_us(&self, delay: u64) {
+        self.client_delay_us.store(delay, Ordering::Relaxed);
+    }
+
+    /// Looks up the per-client delay
+    pub(crate) fn get_delay_us(&self) -> u64 {
+        self.client_delay_us.load(Ordering::Relaxed)
+    }
 }
 
 /// How to handle "promote to active" requests
@@ -2420,6 +2454,9 @@ struct ClientIoTask {
     /// Handle for the rx task
     recv_task: ClientRxTask,
 
+    /// Shared handle to receive per-client backpressure delay
+    client_delay_us: Arc<AtomicU64>,
+
     log: Logger,
 }
 
@@ -2692,6 +2729,24 @@ impl ClientIoTask {
             + std::marker::Send
             + 'static,
     {
+        // Delay communication with this client based on backpressure, to keep
+        // the three clients relatively in sync with each other.
+        //
+        // We don't need to delay writes, because they're already constrained by
+        // the global backpressure system and cannot build up an unbounded
+        // queue.  This is admittedly quite subtle; see crucible#1167 for
+        // discussions and graphs.
+        if !matches!(
+            m,
+            ClientRequest::Message(Message::Write { .. })
+                | ClientRequest::RawMessage(RawMessage::Write { .. }, ..)
+        ) {
+            let d = self.client_delay_us.load(Ordering::Relaxed);
+            if d > 0 {
+                tokio::time::sleep(Duration::from_micros(d)).await;
+            }
+        }
+
         // There's some duplication between this function and `cmd_loop` above,
         // but it's not obvious whether there's a cleaner way to organize stuff.
         tokio::select! {

upstairs/src/downstairs.rs

@@ -659,6 +659,11 @@ impl Downstairs {
         // Restart the IO task for that specific client
         self.clients[client_id].reinitialize(auto_promote);
 
+        for i in ClientId::iter() {
+            // Clear per-client delay, because we're starting a new session
+            self.clients[i].set_delay_us(0);
+        }
+
         // Special-case: if a Downstairs goes away midway through initial
         // reconciliation, then we have to manually abort reconciliation.
         if self.clients.iter().any(|c| c.state() == DsState::Reconcile) {
@@ -1326,6 +1331,7 @@ impl Downstairs {
             guest_id: gw_id,
             work: noop_ioop,
             state: ClientData::new(IOState::New),
+            reply_time: ClientData::new(None),
             acked: false,
             replay: false,
             data: None,
@@ -1449,6 +1455,7 @@ impl Downstairs {
             guest_id: gw_id,
             work: repair_ioop,
             state: ClientData::new(IOState::New),
+            reply_time: ClientData::new(None),
             acked: false,
             replay: false,
             data: None,
@@ -1589,6 +1596,7 @@ impl Downstairs {
             guest_id: gw_id,
             work: reopen_ioop,
             state: ClientData::new(IOState::New),
+            reply_time: ClientData::new(None),
             acked: false,
             replay: false,
             data: None,
@@ -1662,6 +1670,7 @@ impl Downstairs {
             guest_id: gw_id,
             work: aread,
             state: ClientData::new(IOState::New),
+            reply_time: ClientData::new(None),
             acked: false,
             replay: false,
             data: None,
@@ -1732,6 +1741,7 @@ impl Downstairs {
             guest_id: gw_id,
             work: awrite,
             state: ClientData::new(IOState::New),
+            reply_time: ClientData::new(None),
             acked: false,
             replay: false,
             data: None,
@@ -1762,6 +1772,7 @@ impl Downstairs {
             guest_id: gw_id,
             work: close_ioop,
             state: ClientData::new(IOState::New),
+            reply_time: ClientData::new(None),
             acked: false,
             replay: false,
             data: None,
@@ -2087,6 +2098,7 @@ impl Downstairs {
             guest_id: gw_id,
             work: flush,
             state: ClientData::new(IOState::New),
+            reply_time: ClientData::new(None),
             acked: false,
             replay: false,
             data: None,
@@ -2211,6 +2223,7 @@ impl Downstairs {
             guest_id,
             work: aread,
             state: ClientData::new(IOState::New),
+            reply_time: ClientData::new(None),
             acked: false,
             replay: false,
             data: None,
@@ -3166,6 +3179,30 @@ impl Downstairs {
             self.ackable_work.insert(ds_id);
         }
 
+        if job.reply_time.iter().all(Option::is_some) && !job.replay {
+            let fastest_id = ClientId::iter()
+                .min_by_key(|i| job.reply_time[*i].unwrap())
+                .unwrap();
+            let slowest_time = *job.reply_time.iter().flatten().max().unwrap();
+
+            // Apply a delay to the fastest client, and clear the delay for the
+            // other two clients.
+            for i in ClientId::iter() {
+                let delay_time_us: u64 = if i == fastest_id {
+                    // 0 delay below 10ms of lead time, then linearly increasing
+                    // to a maximum of 10 ms.  These are all roughly-eyeballed
+                    // numbers!
+                    let dt = slowest_time - job.reply_time[i].unwrap();
+                    let lead_time_us: u64 = dt.as_micros().try_into().unwrap();
+                    (lead_time_us.saturating_sub(10_000) / 100).min(10_000)
+                } else {
+                    0
+                };
+
+                self.clients[i].set_delay_us(delay_time_us);
+            }
+        }
+
         /*
          * 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

upstairs/src/lib.rs

@@ -889,6 +889,9 @@ struct DownstairsIO {
     /// Map of work status, tracked on a per-client basis
     state: ClientData<IOState>,
 
+    /// At what time did we hear a reply?
+    reply_time: ClientData<Option<std::time::Instant>>,
+
     /*
      * Has this been acked to the guest yet?
      */
@@ -2024,6 +2027,8 @@ pub struct Arg {
     pub ds_extents_repaired: [usize; 3],
     /// Times we have live confirmed  an extent on this downstairs.
     pub ds_extents_confirmed: [usize; 3],
+    /// Per-client delay to keep them roughly in sync
+    pub ds_delay_us: [usize; 3],
     /// Times we skipped repairing a downstairs because we are read_only.
     pub ds_ro_lr_skipped: [usize; 3],
 }

upstairs/src/upstairs.rs

@@ -738,6 +738,8 @@ impl Upstairs {
             self.downstairs.collect_stats(|c| c.stats.extents_repaired);
         let ds_extents_confirmed =
             self.downstairs.collect_stats(|c| c.stats.extents_confirmed);
+        let ds_delay_us =
+            self.downstairs.collect_stats(|c| c.get_delay_us() as usize);
         let ds_ro_lr_skipped =
             self.downstairs.collect_stats(|c| c.stats.ro_lr_skipped);
 
@@ -762,6 +764,7 @@ impl Upstairs {
                 ds_flow_control,
                 ds_extents_repaired,
                 ds_extents_confirmed,
+                ds_delay_us,
                 ds_ro_lr_skipped,
             };
             ("stats", arg)