Repository: cloudflare/pingora (language: rust)

Issue:
Raw Markdown displayed in command line help
## Describe the bug

Running a Pingora server with `--help` command line flag will produce descriptions including raw Markdown. Also, the names of command line flags unnecessarily. These are meant for online documentation and often make little sense as a command line help.

## Pingora info

Please include the following information about your environment:

**Pingora version**: https://github.com/cloudflare/pingora/commit/2501d4adb038d93613c0edbd7c1e3b3de9b415b1
**Rust version**: 1.79.0-nightly (28e7b2bc0 2024-04-05)
**Operating system version**: Fedora 39

## Steps to reproduce

Run `cargo run --example gateway -- --help` from the command line.

## Expected results

A human-readable output without any duplicated information.

## Observed results

Raw Markdown and duplicated information on command line flags, pointless references to code (the command description in particular is meaningless) and references to code (`ServerConf`) that should have really been URLs:

```
basic 0.1.0
Command-line options

Call `Opt::from_args()` to build this object from the process's command line arguments.

USAGE:
    gateway [FLAGS] [OPTIONS]

FLAGS:
    -d, --daemon       
            Whether should run this server in the background
            
            `-d` or `--daemon` can be used
    -h, --help         
            Prints help information

        --nocapture    
            Not actually used. This flag is there so that the server is not upset seeing this flag passed from `cargo
            test` sometimes
    -t, --test         
            Test the configuration and exit
            
            When this flag is set, calling `server.bootstrap()` will exit the process without errors
            
            This flag is useful for upgrading service where the user wants to make sure the new service can start before
            shutting down the old server process.
            
            `-t` or `--test` can be used
    -u, --upgrade      
            Whether this server should try to upgrade from a running old server
            
            `-u` or `--upgrade` can be used
    -V, --version      
            Prints version information


OPTIONS:
    -c, --conf <conf>    
            The path to the configuration file.
            
            See [`ServerConf`] for more details of the configuration file.
            
            `-c` or `--conf` can be used
```

## Additional context

The docstrings that this info is compiled from should really be adjusted. While an application extending the default command line options will overwrite the application description, adjusting the description of individual options is non-trivial.

Alternatively, the docstrings can be left unchanged and [structopt help messages](https://docs.rs/structopt/latest/structopt/#help-messages) can be specified separately.


Discussion / hints:

Fix patch:
diff --git a/.bleep b/.bleep
index 5b3f05fd5..fa13307f1 100644
--- a/.bleep
+++ b/.bleep
@@ -1,1 +1,1 @@
-719b4dcafd471ecfa9e4ade3abfb217929afddc6
\ No newline at end of file
+946e1733bc009bf3e73c635ad1b8bac0079dc680
\ No newline at end of file
diff --git a/pingora-core/src/apps/http_app.rs b/pingora-core/src/apps/http_app.rs
index 4dc905939..c758a1494 100644
--- a/pingora-core/src/apps/http_app.rs
+++ b/pingora-core/src/apps/http_app.rs
@@ -161,13 +161,16 @@ where
         }
         let mut module_ctx = self.modules.build_ctx();
         let req = http.req_header_mut();
-        module_ctx.request_header_filter(req).ok()?;
+        module_ctx.request_header_filter(req).await.ok()?;
         let new_response = self.app.response(&mut http).await;
         let (parts, body) = new_response.into_parts();
-        let resp_header: ResponseHeader = parts.into();
-        let mut task = HttpTask::Header(Box::new(resp_header), body.is_empty());
-        module_ctx.response_filter(&mut task).ok()?;
+        let mut resp_header: ResponseHeader = parts.into();
+        module_ctx
+            .response_header_filter(&mut resp_header, body.is_empty())
+            .await
+            .ok()?;
 
+        let task = HttpTask::Header(Box::new(resp_header), body.is_empty());
         trace!("{task:?}");
 
         match http.response_duplex_vec(vec![task]).await {
@@ -181,15 +184,13 @@ where
                 );
             }
         }
-        let mut task = if !body.is_empty() {
-            HttpTask::Body(Some(body.into()), true)
-        } else {
-            HttpTask::Body(None, true)
-        };
 
-        trace!("{task:?}");
+        let mut body = Some(body.into());
+        module_ctx.response_body_filter(&mut body, true).ok()?;
 
-        module_ctx.response_filter(&mut task).ok()?;
+        let task = HttpTask::Body(body, true);
+
+        trace!("{task:?}");
 
         // TODO: check if chunked encoding is needed
         match http.response_duplex_vec(vec![task]).await {
diff --git a/pingora-core/src/modules/http/compression.rs b/pingora-core/src/modules/http/compression.rs
index b07e1c730..134f94fe3 100644
--- a/pingora-core/src/modules/http/compression.rs
+++ b/pingora-core/src/modules/http/compression.rs
@@ -20,6 +20,7 @@ use crate::protocols::http::compression::ResponseCompressionCtx;
 /// HTTP response compression module
 pub struct ResponseCompression(ResponseCompressionCtx);
 
+#[async_trait]
 impl HttpModule for ResponseCompression {
     fn as_any(&self) -> &dyn std::any::Any {
         self
@@ -28,13 +29,30 @@ impl HttpModule for ResponseCompression {
         self
     }
 
-    fn request_header_filter(&mut self, req: &mut RequestHeader) -> Result<()> {
+    async fn request_header_filter(&mut self, req: &mut RequestHeader) -> Result<()> {
         self.0.request_filter(req);
         Ok(())
     }
 
-    fn response_filter(&mut self, t: &mut HttpTask) -> Result<()> {
-        self.0.response_filter(t);
+    async fn response_header_filter(
+        &mut self,
+        resp: &mut ResponseHeader,
+        end_of_stream: bool,
+    ) -> Result<()> {
+        self.0.response_header_filter(resp, end_of_stream);
+        Ok(())
+    }
+
+    fn response_body_filter(
+        &mut self,
+        body: &mut Option<Bytes>,
+        end_of_stream: bool,
+    ) -> Result<()> {
+        if !self.0.is_enabled() {
+            return Ok(());
+        }
+        let compressed = self.0.response_body_filter(body.as_ref(), end_of_stream);
+        *body = compressed;
         Ok(())
     }
 }
diff --git a/pingora-core/src/modules/http/mod.rs b/pingora-core/src/modules/http/mod.rs
index 0f0401b22..31c5f9a21 100644
--- a/pingora-core/src/modules/http/mod.rs
+++ b/pingora-core/src/modules/http/mod.rs
@@ -20,28 +20,44 @@
 
 pub mod compression;
 
-use crate::protocols::http::HttpTask;
+use async_trait::async_trait;
 use bytes::Bytes;
 use once_cell::sync::OnceCell;
 use pingora_error::Result;
-use pingora_http::RequestHeader;
+use pingora_http::{RequestHeader, ResponseHeader};
 use std::any::Any;
 use std::any::TypeId;
 use std::collections::HashMap;
 use std::sync::Arc;
 
 /// The trait an HTTP traffic module needs to implement
-// TODO: * async filters for, e.g., 3rd party auth server; * access the connection for, e.g., GeoIP
+#[async_trait]
 pub trait HttpModule {
-    fn request_header_filter(&mut self, _req: &mut RequestHeader) -> Result<()> {
+    async fn request_header_filter(&mut self, _req: &mut RequestHeader) -> Result<()> {
         Ok(())
     }
 
-    fn request_body_filter(&mut self, body: Option<Bytes>) -> Result<Option<Bytes>> {
-        Ok(body)
+    async fn request_body_filter(
+        &mut self,
+        _body: &mut Option<Bytes>,
+        _end_of_stream: bool,
+    ) -> Result<()> {
+        Ok(())
+    }
+
+    async fn response_header_filter(
+        &mut self,
+        _resp: &mut ResponseHeader,
+        _end_of_stream: bool,
+    ) -> Result<()> {
+        Ok(())
     }
 
-    fn response_filter(&mut self, _t: &mut HttpTask) -> Result<()> {
+    fn response_body_filter(
+        &mut self,
+        _body: &mut Option<Bytes>,
+        _end_of_stream: bool,
+    ) -> Result<()> {
         Ok(())
     }
 
@@ -168,25 +184,45 @@ impl HttpModuleCtx {
     }
 
     /// Run the `request_header_filter` for all the modules according to their orders.
-    pub fn request_header_filter(&mut self, req: &mut RequestHeader) -> Result<()> {
+    pub async fn request_header_filter(&mut self, req: &mut RequestHeader) -> Result<()> {
         for filter in self.module_ctx.iter_mut() {
-            filter.request_header_filter(req)?;
+            filter.request_header_filter(req).await?;
         }
         Ok(())
     }
 
     /// Run the `request_body_filter` for all the modules according to their orders.
-    pub fn request_body_filter(&mut self, mut body: Option<Bytes>) -> Result<Option<Bytes>> {
+    pub async fn request_body_filter(
+        &mut self,
+        body: &mut Option<Bytes>,
+        end_of_stream: bool,
+    ) -> Result<()> {
         for filter in self.module_ctx.iter_mut() {
-            body = filter.request_body_filter(body)?;
+            filter.request_body_filter(body, end_of_stream).await?;
         }
-        Ok(body)
+        Ok(())
     }
 
-    /// Run the `response_filter` for all the modules according to their orders.
-    pub fn response_filter(&mut self, t: &mut HttpTask) -> Result<()> {
+    /// Run the `response_header_filter` for all the modules according to their orders.
+    pub async fn response_header_filter(
+        &mut self,
+        req: &mut ResponseHeader,
+        end_of_stream: bool,
+    ) -> Result<()> {
         for filter in self.module_ctx.iter_mut() {
-            filter.response_filter(t)?;
+            filter.response_header_filter(req, end_of_stream).await?;
+        }
+        Ok(())
+    }
+
+    /// Run the `response_body_filter` for all the modules according to their orders.
+    pub fn response_body_filter(
+        &mut self,
+        body: &mut Option<Bytes>,
+        end_of_stream: bool,
+    ) -> Result<()> {
+        for filter in self.module_ctx.iter_mut() {
+            filter.response_body_filter(body, end_of_stream)?;
         }
         Ok(())
     }
@@ -197,6 +233,7 @@ mod tests {
     use super::*;
 
     struct MyModule;
+    #[async_trait]
     impl HttpModule for MyModule {
         fn as_any(&self) -> &dyn Any {
             self
@@ -204,7 +241,7 @@ mod tests {
         fn as_any_mut(&mut self) -> &mut dyn Any {
             self
         }
-        fn request_header_filter(&mut self, req: &mut RequestHeader) -> Result<()> {
+        async fn request_header_filter(&mut self, req: &mut RequestHeader) -> Result<()> {
             req.insert_header("my-filter", "1")
         }
     }
@@ -220,6 +257,7 @@ mod tests {
     }
 
     struct MyOtherModule;
+    #[async_trait]
     impl HttpModule for MyOtherModule {
         fn as_any(&self) -> &dyn Any {
             self
@@ -227,7 +265,7 @@ mod tests {
         fn as_any_mut(&mut self) -> &mut dyn Any {
             self
         }
-        fn request_header_filter(&mut self, req: &mut RequestHeader) -> Result<()> {
+        async fn request_header_filter(&mut self, req: &mut RequestHeader) -> Result<()> {
             if req.headers.get("my-filter").is_some() {
                 // if this MyOtherModule runs after MyModule
                 req.insert_header("my-filter", "2")
@@ -262,14 +300,14 @@ mod tests {
         assert!(ctx.get_mut::<usize>().is_none());
     }
 
-    #[test]
-    fn test_module_filter() {
+    #[tokio::test]
+    async fn test_module_filter() {
         let mut http_module = HttpModules::new();
         http_module.add_module(Box::new(MyOtherModuleBuilder));
         http_module.add_module(Box::new(MyModuleBuilder));
         let mut ctx = http_module.build_ctx();
         let mut req = RequestHeader::build("Get", b"/", None).unwrap();
-        ctx.request_header_filter(&mut req).unwrap();
+        ctx.request_header_filter(&mut req).await.unwrap();
         // MyModule runs before MyOtherModule
         assert_eq!(req.headers.get("my-filter").unwrap(), "2");
         assert!(req.headers.get("my-other-filter").is_none());
diff --git a/pingora-core/src/protocols/digest.rs b/pingora-core/src/protocols/digest.rs
index 594dbba74..e080a89ab 100644
--- a/pingora-core/src/protocols/digest.rs
+++ b/pingora-core/src/protocols/digest.rs
@@ -15,10 +15,11 @@
 //! Extra information about the connection
 
 use std::sync::Arc;
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};
 
 use once_cell::sync::OnceCell;
 
+use super::l4::ext::{get_recv_buf, get_tcp_info, TCP_INFO};
 use super::l4::socket::SocketAddr;
 use super::raw_connect::ProxyDigest;
 use super::ssl::digest::SslDigest;
@@ -88,12 +89,38 @@ impl SocketDigest {
             .get_or_init(|| SocketAddr::from_raw_fd(self.raw_fd, false))
             .as_ref()
     }
+
+    fn is_inet(&self) -> bool {
+        self.local_addr().and_then(|p| p.as_inet()).is_some()
+    }
+
+    pub fn tcp_info(&self) -> Option<TCP_INFO> {
+        if self.is_inet() {
+            get_tcp_info(self.raw_fd).ok()
+        } else {
+            None
+        }
+    }
+
+    pub fn get_recv_buf(&self) -> Option<usize> {
+        if self.is_inet() {
+            get_recv_buf(self.raw_fd).ok()
+        } else {
+            None
+        }
+    }
 }
 
 /// The interface to return timing information
 pub trait GetTimingDigest {
     /// Return the timing for each layer from the lowest layer to upper
     fn get_timing_digest(&self) -> Vec<Option<TimingDigest>>;
+    fn get_read_pending_time(&self) -> Duration {
+        Duration::ZERO
+    }
+    fn get_write_pending_time(&self) -> Duration {
+        Duration::ZERO
+    }
 }
 
 /// The interface to set or return proxy information
diff --git a/pingora-core/src/protocols/http/client.rs b/pingora-core/src/protocols/http/client.rs
index 83a7618f6..5c44b6f2d 100644
--- a/pingora-core/src/protocols/http/client.rs
+++ b/pingora-core/src/protocols/http/client.rs
@@ -19,7 +19,7 @@ use std::time::Duration;
 
 use super::v1::client::HttpSession as Http1Session;
 use super::v2::client::Http2Session;
-use crate::protocols::{Digest, SocketAddr};
+use crate::protocols::{Digest, SocketAddr, Stream};
 
 /// A type for Http client session. It can be either an Http1 connection or an Http2 stream.
 pub enum HttpSession {
@@ -174,4 +174,13 @@ impl HttpSession {
             Self::H2(s) => s.client_addr(),
         }
     }
+
+    /// Get the reference of the [Stream] that this HTTP/1 session is operating upon.
+    /// None if the HTTP session is over H2
+    pub fn stream(&self) -> Option<&Stream> {
+        match self {
+            Self::H1(s) => Some(s.stream()),
+            Self::H2(_) => None,
+        }
+    }
 }
diff --git a/pingora-core/src/protocols/http/compression/mod.rs b/pingora-core/src/protocols/http/compression/mod.rs
index 26d4fb8e0..6cbd7388d 100644
--- a/pingora-core/src/protocols/http/compression/mod.rs
+++ b/pingora-core/src/protocols/http/compression/mod.rs
@@ -159,7 +159,11 @@ impl ResponseCompressionCtx {
         }
     }
 
-    fn response_header_filter(&mut self, resp: &mut ResponseHeader, end: bool) {
+    /// Feed the response header into this ctx
+    pub fn response_header_filter(&mut self, resp: &mut ResponseHeader, end: bool) {
+        if !self.is_enabled() {
+            return;
+        }
         match &self.0 {
             CtxInner::HeaderPhase {
                 compression_level,
@@ -195,7 +199,8 @@ impl ResponseCompressionCtx {
         }
     }
 
-    fn response_body_filter(&mut self, data: Option<&Bytes>, end: bool) -> Option<Bytes> {
+    /// Stream the response body chunks into this ctx. The return value will be the compressed data
+    pub fn response_body_filter(&mut self, data: Option<&Bytes>, end: bool) -> Option<Bytes> {
         match &mut self.0 {
             CtxInner::HeaderPhase {
                 compression_level: _,
@@ -222,6 +227,7 @@ impl ResponseCompressionCtx {
         }
     }
 
+    // TODO: retire this function, replace it with the two functions above
     /// Feed the response into this ctx.
     /// This filter will mutate the response accordingly if encoding is needed.
     pub fn response_filter(&mut self, t: &mut HttpTask) {
diff --git a/pingora-core/src/protocols/http/server.rs b/pingora-core/src/protocols/http/server.rs
index a9619a88b..8dfb103d6 100644
--- a/pingora-core/src/protocols/http/server.rs
+++ b/pingora-core/src/protocols/http/server.rs
@@ -370,4 +370,13 @@ impl Session {
             Self::H2(s) => s.server_addr(),
         }
     }
+
+    /// Get the reference of the [Stream] that this HTTP/1 session is operating upon.
+    /// None if the HTTP session is over H2
+    pub fn stream(&self) -> Option<&Stream> {
+        match self {
+            Self::H1(s) => Some(s.stream()),
+            Self::H2(_) => None,
+        }
+    }
 }
diff --git a/pingora-core/src/protocols/http/v1/body.rs b/pingora-core/src/protocols/http/v1/body.rs
index c317bb728..0ddc90a28 100644
--- a/pingora-core/src/protocols/http/v1/body.rs
+++ b/pingora-core/src/protocols/http/v1/body.rs
@@ -538,20 +538,10 @@ impl BodyWriter {
 
                 let chuck_size_buf = format!("{:X}\r\n", chunk_size);
                 let mut output_buf = Bytes::from(chuck_size_buf).chain(buf).chain(&b"\r\n"[..]);
-
-                while output_buf.has_remaining() {
-                    let res = stream.write_vec(&mut output_buf).await;
-                    match res {
-                        Ok(n) => {
-                            if n == 0 {
-                                return Error::e_explain(ConnectionClosed, "while writing body");
-                            }
-                        }
-                        Err(e) => {
-                            return Error::e_because(WriteError, "while writing body", e);
-                        }
-                    }
-                }
+                stream
+                    .write_vec_all(&mut output_buf)
+                    .await
+                    .or_err(WriteError, "while writing body")?;
                 stream.flush().await.or_err(WriteError, "flushing body")?;
                 self.body_mode = BM::ChunkedEncoding(written + chunk_size);
                 Ok(Some(chunk_size))
diff --git a/pingora-core/src/protocols/http/v1/client.rs b/pingora-core/src/protocols/http/v1/client.rs
index fa7710f4d..41cf5901f 100644
--- a/pingora-core/src/protocols/http/v1/client.rs
+++ b/pingora-core/src/protocols/http/v1/client.rs
@@ -641,6 +641,11 @@ impl HttpSession {
             .as_ref()
             .map(|d| d.local_addr())?
     }
+
+    /// Get the reference of the [Stream] that this HTTP session is operating upon.
+    pub fn stream(&self) -> &Stream {
+        &self.underlying_stream
+    }
 }
 
 #[inline]
diff --git a/pingora-core/src/protocols/http/v1/server.rs b/pingora-core/src/protocols/http/v1/server.rs
index 6f8c96ce2..5168456d4 100644
--- a/pingora-core/src/protocols/http/v1/server.rs
+++ b/pingora-core/src/protocols/http/v1/server.rs
@@ -926,6 +926,11 @@ impl HttpSession {
         }
         Ok(end_stream)
     }
+
+    /// Get the reference of the [Stream] that this HTTP session is operating upon.
+    pub fn stream(&self) -> &Stream {
+        &self.underlying_stream
+    }
 }
 
 // Regex to parse request line that has illegal chars in it
diff --git a/pingora-core/src/protocols/l4/ext.rs b/pingora-core/src/protocols/l4/ext.rs
index be81f5563..aefd36c52 100644
--- a/pingora-core/src/protocols/l4/ext.rs
+++ b/pingora-core/src/protocols/l4/ext.rs
@@ -56,13 +56,12 @@ pub struct TCP_INFO {
     tcpi_rcv_ssthresh: u32,
     pub tcpi_rtt: u32,
     tcpi_rttvar: u32,
-    /* uncomment these field if needed
     tcpi_snd_ssthresh: u32,
     tcpi_snd_cwnd: u32,
     tcpi_advmss: u32,
     tcpi_reordering: u32,
     tcpi_rcv_rtt: u32,
-    tcpi_rcv_space: u32,
+    pub tcpi_rcv_space: u32,
     tcpi_total_retrans: u32,
     tcpi_pacing_rate: u64,
     tcpi_max_pacing_rate: u64,
@@ -75,8 +74,19 @@ pub struct TCP_INFO {
     tcpi_data_segs_in: u32,
     tcpi_data_segs_out: u32,
     tcpi_delivery_rate: u64,
-    */
-    /* and more, see include/linux/tcp.h */
+    tcpi_busy_time: u64,
+    tcpi_rwnd_limited: u64,
+    tcpi_sndbuf_limited: u64,
+    tcpi_delivered: u32,
+    tcpi_delivered_ce: u32,
+    tcpi_bytes_sent: u64,
+    tcpi_bytes_retrans: u64,
+    tcpi_dsack_dups: u32,
+    tcpi_reord_seen: u32,
+    tcpi_rcv_ooopack: u32,
+    tcpi_snd_wnd: u32,
+    pub tcpi_rcv_wnd: u32,
+    // and more, see include/linux/tcp.h
 }
 
 impl TCP_INFO {
@@ -223,6 +233,25 @@ pub fn set_recv_buf(_fd: RawFd, _: usize) -> Result<()> {
     Ok(())
 }
 
+#[cfg(target_os = "linux")]
+pub fn get_recv_buf(fd: RawFd) -> io::Result<usize> {
+    let mut recv_size: c_int = 0;
+    let mut size = std::mem::size_of::<c_int>() as u32;
+    get_opt(
+        fd,
+        libc::SOL_SOCKET,
+        libc::SO_RCVBUF,
+        &mut recv_size,
+        &mut size,
+    )?;
+    Ok(recv_size as usize)
+}
+
+#[cfg(not(target_os = "linux"))]
+pub fn get_recv_buf(_fd: RawFd) -> Result<usize> {
+    Ok(0)
+}
+
 /// Enable client side TCP fast open.
 #[cfg(target_os = "linux")]
 pub fn set_tcp_fastopen_connect(fd: RawFd) -> Result<()> {
@@ -358,18 +387,8 @@ mod test {
 
         #[cfg(target_os = "linux")]
         {
-            let mut recv_size: c_int = 0;
-            let mut size = std::mem::size_of::<c_int>() as u32;
-            get_opt(
-                socket.as_raw_fd(),
-                libc::SOL_SOCKET,
-                libc::SO_RCVBUF,
-                &mut recv_size,
-                &mut size,
-            )
-            .unwrap();
             // kernel doubles whatever is set
-            assert_eq!(recv_size, 102400 * 2);
+            assert_eq!(get_recv_buf(socket.as_raw_fd()).unwrap(), 102400 * 2);
         }
     }
 
diff --git a/pingora-core/src/protocols/l4/stream.rs b/pingora-core/src/protocols/l4/stream.rs
index 6169d3115..6a13c6588 100644
--- a/pingora-core/src/protocols/l4/stream.rs
+++ b/pingora-core/src/protocols/l4/stream.rs
@@ -22,7 +22,7 @@ use std::os::unix::io::AsRawFd;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
-use std::time::SystemTime;
+use std::time::{Duration, Instant, SystemTime};
 use tokio::io::{self, AsyncRead, AsyncWrite, AsyncWriteExt, BufStream, ReadBuf};
 use tokio::net::{TcpStream, UnixStream};
 
@@ -141,6 +141,8 @@ pub struct Stream {
     pub established_ts: SystemTime,
     /// The distributed tracing object for this stream
     pub tracer: Option<Tracer>,
+    read_pending_time: AccumulatedDuration,
+    write_pending_time: AccumulatedDuration,
 }
 
 impl Stream {
@@ -172,6 +174,8 @@ impl From<TcpStream> for Stream {
             proxy_digest: None,
             socket_digest: None,
             tracer: None,
+            read_pending_time: AccumulatedDuration::new(),
+            write_pending_time: AccumulatedDuration::new(),
         }
     }
 }
@@ -185,6 +189,8 @@ impl From<UnixStream> for Stream {
             proxy_digest: None,
             socket_digest: None,
             tracer: None,
+            read_pending_time: AccumulatedDuration::new(),
+            write_pending_time: AccumulatedDuration::new(),
         }
     }
 }
@@ -220,6 +226,14 @@ impl GetTimingDigest for Stream {
         }));
         digest
     }
+
+    fn get_read_pending_time(&self) -> Duration {
+        self.read_pending_time.total
+    }
+
+    fn get_write_pending_time(&self) -> Duration {
+        self.write_pending_time.total
+    }
 }
 
 impl GetProxyDigest for Stream {
@@ -282,7 +296,9 @@ impl AsyncRead for Stream {
         cx: &mut Context<'_>,
         buf: &mut ReadBuf<'_>,
     ) -> Poll<io::Result<()>> {
-        Pin::new(&mut self.stream).poll_read(cx, buf)
+        let result = Pin::new(&mut self.stream).poll_read(cx, buf);
+        self.read_pending_time.poll_time(&result);
+        result
     }
 }
 
@@ -292,15 +308,19 @@ impl AsyncWrite for Stream {
         cx: &mut Context,
         buf: &[u8],
     ) -> Poll<io::Result<usize>> {
-        if self.buffer_write {
+        let result = if self.buffer_write {
             Pin::new(&mut self.stream).poll_write(cx, buf)
         } else {
             Pin::new(&mut self.stream.get_mut()).poll_write(cx, buf)
-        }
+        };
+        self.write_pending_time.poll_write_time(&result, buf.len());
+        result
     }
 
     fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
-        Pin::new(&mut self.stream).poll_flush(cx)
+        let result = Pin::new(&mut self.stream).poll_flush(cx);
+        self.write_pending_time.poll_time(&result);
+        result
     }
 
     fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
@@ -312,11 +332,16 @@ impl AsyncWrite for Stream {
         cx: &mut Context<'_>,
         bufs: &[std::io::IoSlice<'_>],
     ) -> Poll<io::Result<usize>> {
-        if self.buffer_write {
+        let total_size = bufs.iter().fold(0, |acc, s| acc + s.len());
+
+        let result = if self.buffer_write {
             Pin::new(&mut self.stream).poll_write_vectored(cx, bufs)
         } else {
             Pin::new(&mut self.stream.get_mut()).poll_write_vectored(cx, bufs)
-        }
+        };
+
+        self.write_pending_time.poll_write_time(&result, total_size);
+        result
     }
 
     fn is_write_vectored(&self) -> bool {
@@ -350,6 +375,12 @@ pub mod async_write_vec {
         buf: &'a mut B,
     }
 
+    #[must_use = "futures do nothing unless you `.await` or poll them"]
+    pub struct WriteVecAll<'a, W, B> {
+        writer: &'a mut W,
+        buf: &'a mut B,
+    }
+
     pub trait AsyncWriteVec {
         fn poll_write_vec<B: Buf>(
             self: Pin<&mut Self>,
@@ -367,6 +398,17 @@ pub mod async_write_vec {
                 buf: src,
             }
         }
+
+        fn write_vec_all<'a, B>(&'a mut self, src: &'a mut B) -> WriteVecAll<'a, Self, B>
+        where
+            Self: Sized,
+            B: Buf,
+        {
+            WriteVecAll {
+                writer: self,
+                buf: src,
+            }
+        }
     }
 
     impl<W, B> Future for WriteVec<'_, W, B>
@@ -382,6 +424,25 @@ pub mod async_write_vec {
         }
     }
 
+    impl<W, B> Future for WriteVecAll<'_, W, B>
+    where
+        W: AsyncWriteVec + Unpin,
+        B: Buf,
+    {
+        type Output = io::Result<()>;
+
+        fn poll(mut self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            let me = &mut *self;
+            while me.buf.has_remaining() {
+                let n = ready!(Pin::new(&mut *me.writer).poll_write_vec(ctx, me.buf))?;
+                if n == 0 {
+                    return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
+                }
+            }
+            Poll::Ready(Ok(()))
+        }
+    }
+
     /* from https://github.com/tokio-rs/tokio/blob/master/tokio-util/src/lib.rs#L177 */
     impl<T> AsyncWriteVec for T
     where
@@ -414,3 +475,55 @@ pub mod async_write_vec {
 }
 
 pub use async_write_vec::AsyncWriteVec;
+
+#[derive(Debug)]
+struct AccumulatedDuration {
+    total: Duration,
+    last_start: Option<Instant>,
+}
+
+impl AccumulatedDuration {
+    fn new() -> Self {
+        AccumulatedDuration {
+            total: Duration::ZERO,
+            last_start: None,
+        }
+    }
+
+    fn start(&mut self) {
+        if self.last_start.is_none() {
+            self.last_start = Some(Instant::now());
+        }
+    }
+
+    fn stop(&mut self) {
+        if let Some(start) = self.last_start.take() {
+            self.total += start.elapsed();
+        }
+    }
+
+    fn poll_write_time(&mut self, result: &Poll<io::Result<usize>>, buf_size: usize) {
+        match result {
+            Poll::Ready(Ok(n)) => {
+                if *n == buf_size {
+                    self.stop();
+                } else {
+                    // partial write
+                    self.start();
+                }
+            }
+            // Pending: start the timer, Ready(Err()): does not matter
+            _ => self.start(),
+        }
+    }
+
+    fn poll_time(&mut self, result: &Poll<io::Result<()>>) {
+        match result {
+            Poll::Ready(Ok(())) => {
+                self.stop();
+            }
+            // Pending: start the timer, Ready(Err()): does not matter
+            _ => self.start(),
+        }
+    }
+}
diff --git a/pingora-core/src/protocols/ssl/client.rs b/pingora-core/src/protocols/ssl/client.rs
index 07811dda3..9edf29b2a 100644
--- a/pingora-core/src/protocols/ssl/client.rs
+++ b/pingora-core/src/protocols/ssl/client.rs
@@ -23,6 +23,7 @@ use crate::tls::{ssl, ssl::ConnectConfiguration, ssl_sys::X509_V_ERR_INVALID_CAL
 
 use pingora_error::{Error, ErrorType::*, OrErr, Result};
 use std::sync::Arc;
+use std::time::Duration;
 
 /// Perform the TLS handshake for the given connection with the given configuration
 pub async fn handshake<S: IO>(
@@ -82,6 +83,13 @@ where
         ts_vec.push(Some(self.timing.clone()));
         ts_vec
     }
+    fn get_read_pending_time(&self) -> Duration {
+        self.get_ref().get_read_pending_time()
+    }
+
+    fn get_write_pending_time(&self) -> Duration {
+        self.get_ref().get_write_pending_time()
+    }
 }
 
 impl<S> GetProxyDigest for SslStream<S>
diff --git a/pingora-core/src/server/configuration/mod.rs b/pingora-core/src/server/configuration/mod.rs
index 5b7609dcb..a4b5cc732 100644
--- a/pingora-core/src/server/configuration/mod.rs
+++ b/pingora-core/src/server/configuration/mod.rs
@@ -119,17 +119,18 @@ impl Default for ServerConf {
 ///
 /// Call `Opt::from_args()` to build this object from the process's command line arguments.
 #[derive(Parser, Debug)]
-#[clap(name = "basic")]
+#[clap(name = "basic", long_about = None)]
 pub struct Opt {
     /// Whether this server should try to upgrade from a running old server
     #[clap(
         short,
         long,
-        help = "This is the base set of command line arguments for a pingora-based service"
+        help = "This is the base set of command line arguments for a pingora-based service",
+        long_help = None
     )]
     pub upgrade: bool,
 
-    /// Whether should run this server in the background
+    /// Whether this server should run in the background
     #[clap(short, long)]
     pub daemon: bool,
 
@@ -149,14 +150,15 @@ pub struct Opt {
         long,
         help = "This flag is useful for upgrading service where the user wants \
                 to make sure the new service can start before shutting down \
-                the old server process."
+                the old server process.",
+        long_help = None
     )]
     pub test: bool,
 
     /// The path to the configuration file.
     ///
     /// See [`ServerConf`] for more details of the configuration file.
-    #[clap(short, long, help = "The path to the configuration file.")]
+    #[clap(short, long, help = "The path to the configuration file.", long_help = None)]
     pub conf: Option<String>,
 }
 
diff --git a/pingora-core/src/server/transfer_fd/mod.rs b/pingora-core/src/server/transfer_fd/mod.rs
index 969d9cc31..ee63ac02e 100644
--- a/pingora-core/src/server/transfer_fd/mod.rs
+++ b/pingora-core/src/server/transfer_fd/mod.rs
@@ -77,7 +77,7 @@ impl Fds {
     {
         let mut de_buf: [u8; 2048] = [0; 2048];
         let (fds, bytes) = get_fds_from(path, &mut de_buf)?;
-        let keys = deserialize_vec_string(&de_buf[..bytes]);
+        let keys = deserialize_vec_string(&de_buf[..bytes])?;
         self.deserialize(keys, fds);
         Ok(())
     }
@@ -91,13 +91,9 @@ fn serialize_vec_string(vec_string: &[String], mut buf: &mut [u8]) -> usize {
     buf.write(joined.as_bytes()).unwrap()
 }
 
-fn deserialize_vec_string(buf: &[u8]) -> Vec<String> {
-    let joined = std::str::from_utf8(buf).unwrap(); // TODO: handle error
-    let mut results: Vec<String> = Vec::new();
-    for iter in joined.split_ascii_whitespace() {
-        results.push(String::from(iter));
-    }
-    results
+fn deserialize_vec_string(buf: &[u8]) -> Result<Vec<String>, Error> {
+    let joined = std::str::from_utf8(buf).map_err(|_| Error::EINVAL)?;
+    Ok(joined.split_ascii_whitespace().map(String::from).collect())
 }
 
 #[cfg(target_os = "linux")]
@@ -369,7 +365,7 @@ mod tests {
         let vec_str: Vec<String> = vec!["aaaa".to_string(), "bbb".to_string()];
         let mut ser_buf: [u8; 1024] = [0; 1024];
         let size = serialize_vec_string(&vec_str, &mut ser_buf);
-        let de_vec_string = deserialize_vec_string(&ser_buf[..size]);
+        let de_vec_string = deserialize_vec_string(&ser_buf[..size]).unwrap();
         assert_eq!(de_vec_string.len(), 2);
         assert_eq!(de_vec_string[0], "aaaa");
         assert_eq!(de_vec_string[1], "bbb");
diff --git a/pingora-core/src/services/listening.rs b/pingora-core/src/services/listening.rs
index 54373230d..ea81799b9 100644
--- a/pingora-core/src/services/listening.rs
+++ b/pingora-core/src/services/listening.rs
@@ -35,29 +35,29 @@ use std::sync::Arc;
 pub struct Service<A> {
     name: String,
     listeners: Listeners,
-    app_logic: Arc<A>,
+    app_logic: Option<A>,
     /// The number of preferred threads. `None` to follow global setting.
     pub threads: Option<usize>,
 }
 
 impl<A> Service<A> {
     /// Create a new [`Service`] with the given application (see [`crate::apps`]).
-    pub fn new(name: String, app_logic: Arc<A>) -> Self {
+    pub fn new(name: String, app_logic: A) -> Self {
         Service {
             name,
             listeners: Listeners::new(),
-            app_logic,
+            app_logic: Some(app_logic),
             threads: None,
         }
     }
 
     /// Create a new [`Service`] with the given application (see [`crate::apps`]) and the given
     /// [`Listeners`].
-    pub fn with_listeners(name: String, listeners: Listeners, app_logic: Arc<A>) -> Self {
+    pub fn with_listeners(name: String, listeners: Listeners, app_logic: A) -> Self {
         Service {
             name,
             listeners,
-            app_logic,
+            app_logic: Some(app_logic),
             threads: None,
         }
     }
@@ -107,6 +107,16 @@ impl<A> Service<A> {
     pub fn add_address(&mut self, addr: ServerAddress) {
         self.listeners.add_address(addr);
     }
+
+    /// Get a reference to the application inside this service
+    pub fn app_logic(&self) -> Option<&A> {
+        self.app_logic.as_ref()
+    }
+
+    /// Get a mutable reference to the application inside this service
+    pub fn app_logic_mut(&mut self) -> Option<&mut A> {
+        self.app_logic.as_mut()
+    }
 }
 
 impl<A: ServerApp + Send + Sync + 'static> Service<A> {
@@ -194,18 +204,23 @@ impl<A: ServerApp + Send + Sync + 'static> ServiceTrait for Service<A> {
     async fn start_service(&mut self, fds: Option<ListenFds>, shutdown: ShutdownWatch) {
         let runtime = current_handle();
         let endpoints = self.listeners.build(fds);
+        let app_logic = self
+            .app_logic
+            .take()
+            .expect("can only start_service() once");
+        let app_logic = Arc::new(app_logic);
 
         let handlers = endpoints.into_iter().map(|endpoint| {
-            let app_logic = self.app_logic.clone();
             let shutdown = shutdown.clone();
+            let my_app_logic = app_logic.clone();
             runtime.spawn(async move {
-                Self::run_endpoint(app_logic, endpoint, shutdown).await;
+                Self::run_endpoint(my_app_logic, endpoint, shutdown).await;
             })
         });
 
         futures::future::join_all(handlers).await;
         self.listeners.cleanup();
-        self.app_logic.cleanup().await;
+        app_logic.cleanup().await;
     }
 
     fn name(&self) -> &str {
@@ -226,7 +241,7 @@ impl Service<PrometheusServer> {
     pub fn prometheus_http_service() -> Self {
         Service::new(
             "Prometheus metric HTTP".to_string(),
-            Arc::new(PrometheusServer::new()),
+            PrometheusServer::new(),
         )
     }
 }
diff --git a/pingora-proxy/src/lib.rs b/pingora-proxy/src/lib.rs
index a60e6a7fb..eaf31d52b 100644
--- a/pingora-proxy/src/lib.rs
+++ b/pingora-proxy/src/lib.rs
@@ -96,12 +96,12 @@ pub struct HttpProxy<SV> {
 }
 
 impl<SV> HttpProxy<SV> {
-    fn new(inner: SV, conf: Arc<ServerConf>) -> Arc<Self> {
-        Arc::new(HttpProxy {
+    fn new(inner: SV, conf: Arc<ServerConf>) -> Self {
+        HttpProxy {
             inner,
             client_upstream: Connector::new(Some(ConnectorOptions::from_server_conf(&conf))),
             shutdown: Notify::new(),
-        })
+        }
     }
 
     async fn handle_new_request(
diff --git a/pingora/examples/app/echo.rs b/pingora/examples/app/echo.rs
index 61f94e5c6..09725c48d 100644
--- a/pingora/examples/app/echo.rs
+++ b/pingora/examples/app/echo.rs
@@ -86,13 +86,3 @@ impl ServeHttp for HttpEchoApp {
             .unwrap()
     }
 }
-
-impl EchoApp {
-    pub fn new() -> Arc<Self> {
-        Arc::new(EchoApp {})
-    }
-}
-
-pub fn new_http_echo_app() -> Arc<HttpEchoApp> {
-    Arc::new(HttpEchoApp {})
-}
diff --git a/pingora/examples/service/echo.rs b/pingora/examples/service/echo.rs
index 8126088a7..af07da217 100644
--- a/pingora/examples/service/echo.rs
+++ b/pingora/examples/service/echo.rs
@@ -12,13 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::app::echo::{new_http_echo_app, EchoApp, HttpEchoApp};
+use crate::app::echo::{EchoApp, HttpEchoApp};
 use pingora::services::listening::Service;
 
 pub fn echo_service() -> Service<EchoApp> {
-    Service::new("Echo Service".to_string(), EchoApp::new())
+    Service::new("Echo Service".to_string(), EchoApp)
 }
 
 pub fn echo_service_http() -> Service<HttpEchoApp> {
-    Service::new("Echo Service HTTP".to_string(), new_http_echo_app())
+    Service::new("Echo Service HTTP".to_string(), HttpEchoApp)
 }
diff --git a/pingora/examples/service/proxy.rs b/pingora/examples/service/proxy.rs
index a103b4776..49c24ca02 100644
--- a/pingora/examples/service/proxy.rs
+++ b/pingora/examples/service/proxy.rs
@@ -16,7 +16,6 @@ use crate::app::proxy::ProxyApp;
 use pingora_core::listeners::Listeners;
 use pingora_core::services::listening::Service;
 use pingora_core::upstreams::peer::BasicPeer;
-use std::sync::Arc;
 
 pub fn proxy_service(addr: &str, proxy_addr: &str) -> Service<ProxyApp> {
     let proxy_to = BasicPeer::new(proxy_addr);
@@ -24,7 +23,7 @@ pub fn proxy_service(addr: &str, proxy_addr: &str) -> Service<ProxyApp> {
     Service::with_listeners(
         "Proxy Service".to_string(),
         Listeners::tcp(addr),
-        Arc::new(ProxyApp::new(proxy_to)),
+        ProxyApp::new(proxy_to),
     )
 }
 
@@ -41,6 +40,6 @@ pub fn proxy_service_tls(
     Service::with_listeners(
         "Proxy Service TLS".to_string(),
         Listeners::tls(addr, cert_path, key_path).unwrap(),
-        Arc::new(ProxyApp::new(proxy_to)),
+        ProxyApp::new(proxy_to),
     )
 }
diff --git a/tinyufo/Cargo.toml b/tinyufo/Cargo.toml
index 0326fe540..68c4236b5 100644
--- a/tinyufo/Cargo.toml
+++ b/tinyufo/Cargo.toml
@@ -29,6 +29,7 @@ zipf = "7"
 moka = { version = "0", features = ["sync"] }
 dhat = "0"
 quick_cache = "0.4"
+triomphe = "<=0.1.11" # 0.1.12 requires Rust 1.76
 
 [[bench]]
 name = "bench_perf"


Test patch:
diff --git a/pingora-core/tests/utils/mod.rs b/pingora-core/tests/utils/mod.rs
index af5de8544..d254e9469 100644
--- a/pingora-core/tests/utils/mod.rs
+++ b/pingora-core/tests/utils/mod.rs
@@ -25,7 +25,6 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use http::{Response, StatusCode};
 use pingora_timeout::timeout;
-use std::sync::Arc;
 use std::time::Duration;
 
 use pingora_core::apps::http_app::ServeHttp;
@@ -63,10 +62,6 @@ impl ServeHttp for EchoApp {
     }
 }
 
-pub fn new_http_echo_app() -> Arc<EchoApp> {
-    Arc::new(EchoApp {})
-}
-
 pub struct MyServer {
     pub handle: thread::JoinHandle<()>,
 }
@@ -88,11 +83,8 @@ fn entry_point(opt: Option<Opt>) {
     tls_settings.enable_h2();
     listeners.add_tls_with_settings("0.0.0.0:6146", None, tls_settings);
 
-    let echo_service_http = Service::with_listeners(
-        "Echo Service HTTP".to_string(),
-        listeners,
-        new_http_echo_app(),
-    );
+    let echo_service_http =
+        Service::with_listeners("Echo Service HTTP".to_string(), listeners, EchoApp);
 
     my_server.add_service(echo_service_http);
     my_server.run_forever();