Add per-client backpressure

mkeeter · mkeeter · commit 9abd890a7fe9 · 2024-02-26T11:13:32.000-05:00
diff --git a/upstairs/src/client.rs b/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,
         }
     }
 
@@ -652,6 +667,7 @@ impl DownstairsClient {
             connect,
             self.client_id,
             self.tls_context.clone(),
+            self.client_delay_us.clone(),
             &self.log,
         );
     }
@@ -662,6 +678,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 +689,7 @@ impl DownstairsClient {
                 connect,
                 client_id,
                 tls_context,
+                client_delay_us,
                 log,
             )
         } else {
@@ -685,6 +703,7 @@ impl DownstairsClient {
             connect,
             client_id,
             tls_context,
+            client_delay_us,
             log,
         )
     }
@@ -695,6 +714,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 +750,7 @@ impl DownstairsClient {
                     log: log.clone(),
                 },
                 delay,
+                client_delay_us,
                 log,
             };
             c.run().await
@@ -1242,6 +1263,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 +2243,11 @@ 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);
+    }
 }
 
 /// How to handle "promote to active" requests
@@ -2420,6 +2448,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 +2723,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! {
diff --git a/upstairs/src/downstairs.rs b/upstairs/src/downstairs.rs
@@ -1326,6 +1326,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 +1450,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 +1591,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 +1665,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 +1736,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 +1767,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 +2093,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 +2218,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 +3174,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
diff --git a/upstairs/src/lib.rs b/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?
      */
