perf(http/prom): avoid oneshot channel for request duration start time (#4448)

RecordRequestDuration knows the start time at call() time, whereas
RecordResponseDuration defers it until the request body flushes. The
latter needs async delivery from the request body wrapper, but the
former is known synchronously.

Replace the shared oneshot::Receiver<Instant> in ResponseState with an
enum supporting both immediate and deferred start times, removing one
heap allocation per request.

Signed-off-by: Alejandro Martinez Ruiz <amr@buoyant.io>

Author: unleashed

linkerd/http/prom/src/record_response.rs

@@ -81,7 +81,27 @@ where
 struct ResponseState<L: StreamLabel> {
     labeler: L,
     duration: DurationFamily<L::DurationLabels>,
-    start: oneshot::Receiver<time::Instant>,
+    start: StartTime,
+}
+
+/// Start time for a duration measurement.
+///
+/// Request duration knows the start time immediately whereas response duration
+/// needs to wait until the request body is fully flushed.
+pub(crate) enum StartTime {
+    /// Start time is already known.
+    Known(Option<time::Instant>),
+    /// Start time will be sent when the request body finishes streaming.
+    Pending(oneshot::Receiver<time::Instant>),
+}
+
+impl StartTime {
+    fn recv(&mut self) -> Option<time::Instant> {
+        match self {
+            Self::Known(t) => t.take(),
+            Self::Pending(rx) => rx.try_recv().ok(),
+        }
+    }
 }
 
 type DurationFamily<L> = Family<L, Histogram, MkDurationHistogram>;
@@ -269,7 +289,7 @@ fn end_stream<L>(
 
     labeler.end_response(res);
 
-    let elapsed = if let Ok(start) = start.try_recv() {
+    let elapsed = if let Some(start) = start.recv() {
         time::Instant::now().saturating_duration_since(start)
     } else {
         time::Duration::ZERO

linkerd/http/prom/src/record_response/request.rs

@@ -8,7 +8,7 @@ use std::{
     sync::Arc,
     task::{Context, Poll},
 };
-use tokio::{sync::oneshot, time};
+use tokio::time;
 
 /// Metrics type that tracks completed requests.
 #[derive(Debug)]
@@ -81,8 +81,7 @@ where
 
     fn call(&mut self, req: http::Request<ReqB>) -> Self::Future {
         let state = self.labeler.mk_stream_labeler(&req).map(|labeler| {
-            let (tx, start) = oneshot::channel();
-            tx.send(time::Instant::now()).unwrap();
+            let start = super::StartTime::Known(Some(time::Instant::now()));
             let RequestMetrics { duration } = self.metric.clone();
             super::ResponseState {
                 labeler,

linkerd/http/prom/src/record_response/response.rs

@@ -95,7 +95,7 @@ where
             let ResponseMetrics { duration } = self.metric.clone();
             Some(super::ResponseState {
                 labeler,
-                start,
+                start: super::StartTime::Pending(start),
                 duration,
             })
         } else {