add conn idle timeout

Author: silence

src/proto/h1/conn.rs

@@ -154,6 +154,10 @@ where
         self.state.is_write_closed()
     }
 
+    pub(crate) fn is_idle(&self) -> bool {
+        self.state.is_idle()
+    }
+
     pub(crate) fn can_read_head(&self) -> bool {
         if !matches!(self.state.reading, Reading::Init) {
             return false;

src/proto/h1/dispatch.rs

@@ -1,8 +1,10 @@
 use std::error::Error as StdError;
+use std::time::Duration;
 
 use bytes::{Buf, Bytes};
 use http::Request;
 use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::time::Sleep;
 use tracing::{debug, trace};
 
 use super::{Http1Transaction, Wants};
@@ -12,13 +14,15 @@ use crate::proto::{
     BodyLength, Conn, Dispatched, MessageHead, RequestHead,
 };
 use crate::upgrade::OnUpgrade;
-
+// TODO will be replace by config in the formal pull request
+const DEFAULT_CONN_IDLE_TIMEOUT: Duration = Duration::from_secs(65);
 pub(crate) struct Dispatcher<D, Bs: HttpBody, I, T> {
     conn: Conn<I, Bs::Data, T>,
     dispatch: D,
     body_tx: Option<crate::body::Sender>,
     body_rx: Pin<Box<Option<Bs>>>,
     is_closing: bool,
+    conn_idle_timeout: Pin<Box<Sleep>>,
 }
 
 pub(crate) trait Dispatch {
@@ -77,6 +81,7 @@ where
             body_tx: None,
             body_rx: Box::pin(None),
             is_closing: false,
+            conn_idle_timeout: Box::pin(tokio::time::sleep(DEFAULT_CONN_IDLE_TIMEOUT)),
         }
     }
 
@@ -156,11 +161,18 @@ where
         //
         // 16 was chosen arbitrarily, as that is number of pipelined requests
         // benchmarks often use. Perhaps it should be a config option instead.
+        if self.conn_idle_timeout.as_mut().poll(cx).is_ready() {
+            trace!("conn is close");
+            self.close();
+            return Poll::Ready(Ok(()));
+        }
         for _ in 0..16 {
             let _ = self.poll_read(cx)?;
             let _ = self.poll_write(cx)?;
             let _ = self.poll_flush(cx)?;
-
+            if self.conn.is_idle() {
+                self.conn_idle_timeout.as_mut().reset(tokio::time::Instant::now() + DEFAULT_CONN_IDLE_TIMEOUT);
+            }
             // This could happen if reading paused before blocking on IO,
             // such as getting to the end of a framed message, but then
             // writing/flushing set the state back to Init. In that case,
@@ -247,6 +259,9 @@ where
         // dispatch is ready for a message, try to read one
         match ready!(self.conn.poll_read_head(cx)) {
             Some(Ok((mut head, body_len, wants))) => {
+                trace!("req is comming");
+                // Avoid request failures due to timeout.
+                self.conn_idle_timeout.as_mut().reset(tokio::time::Instant::now() + Duration::from_secs(60 * 60 * 24));
                 let body = match body_len {
                     DecodedLength::ZERO => Body::empty(),
                     other => {

src/proto/h2/server.rs

@@ -1,5 +1,6 @@
 use std::error::Error as StdError;
 use std::marker::Unpin;
+use std::sync::{Arc, Mutex};
 #[cfg(feature = "runtime")]
 use std::time::Duration;
 
@@ -24,7 +25,7 @@ use crate::service::HttpService;
 
 use crate::upgrade::{OnUpgrade, Pending, Upgraded};
 use crate::{Body, Response};
-
+use tokio::time::Sleep;
 // Our defaults are chosen for the "majority" case, which usually are not
 // resource constrained, and so the spec default of 64kb can be too limiting
 // for performance.
@@ -36,6 +37,9 @@ const DEFAULT_STREAM_WINDOW: u32 = 1024 * 1024; // 1mb
 const DEFAULT_MAX_FRAME_SIZE: u32 = 1024 * 16; // 16kb
 const DEFAULT_MAX_SEND_BUF_SIZE: usize = 1024 * 400; // 400kb
 
+// TODO will be replace by config in the formal pull request
+const DEFAULT_CONN_IDLE_TIMEOUT: Duration = Duration::from_secs(65);
+
 #[derive(Clone, Debug)]
 pub(crate) struct Config {
     pub(crate) adaptive_window: bool,
@@ -100,6 +104,7 @@ where
     ping: Option<(ping::Recorder, ping::Ponger)>,
     conn: Connection<T, SendBuf<B::Data>>,
     closing: Option<crate::Error>,
+    conn_idle_timeout: Arc<Mutex<Pin<Box<Sleep>>>>
 }
 
 impl<T, S, B, E> Server<T, S, B, E>
@@ -202,6 +207,7 @@ where
                         ping,
                         conn,
                         closing: None,
+                        conn_idle_timeout: Arc::new(Mutex::new(Box::pin(tokio::time::sleep(DEFAULT_CONN_IDLE_TIMEOUT))))
                     })
                 }
                 State::Serving(ref mut srv) => {
@@ -224,6 +230,17 @@ where
     T: AsyncRead + AsyncWrite + Unpin,
     B: HttpBody + 'static,
 {
+
+    fn reset_conn_idle_timeout(&self, time: Duration) {
+        let mut sleep = self.conn_idle_timeout.lock().unwrap();
+        sleep.as_mut().reset(tokio::time::Instant::now() + time);
+    }
+
+    fn poll_conn_idle_timeout(&self, cx: &mut task::Context<'_>) -> bool  {
+        let mut sleep = self.conn_idle_timeout.lock().unwrap();
+        sleep.as_mut().poll(cx).is_ready()
+    }
+
     fn poll_server<S, E>(
         &mut self,
         cx: &mut task::Context<'_>,
@@ -269,10 +286,16 @@ where
                         break;
                     }
                 }
-
+                if self.poll_conn_idle_timeout(cx) {
+                    self.conn.graceful_shutdown();
+                    trace!("conn is close with idle");
+                    return Poll::Ready(Ok(()));
+                }
                 // When the service is ready, accepts an incoming request.
                 match ready!(self.conn.poll_accept(cx)) {
                     Some(Ok((req, mut respond))) => {
+                        // Avoid request failures due to timeout.
+                        self.reset_conn_idle_timeout(Duration::from_secs(60 * 60 * 24));
                         trace!("incoming request");
                         let content_length = headers::content_length_parse_all(req.headers());
                         let ping = self
@@ -317,7 +340,7 @@ where
                             req.extensions_mut().insert(Protocol::from_inner(protocol));
                         }
 
-                        let fut = H2Stream::new(service.call(req), connect_parts, respond);
+                        let fut = H2Stream::new(service.call(req), connect_parts, respond, self.conn_idle_timeout.clone());
                         exec.execute_h2stream(fut);
                     }
                     Some(Err(e)) => {
@@ -373,6 +396,7 @@ pin_project! {
         reply: SendResponse<SendBuf<B::Data>>,
         #[pin]
         state: H2StreamState<F, B>,
+        conn_idle_timeout: Arc<Mutex<Pin<Box<Sleep>>>>,
     }
 }
 
@@ -408,10 +432,12 @@ where
         fut: F,
         connect_parts: Option<ConnectParts>,
         respond: SendResponse<SendBuf<B::Data>>,
+        conn_idle_timeout: Arc<Mutex<Pin<Box<Sleep>>>>
     ) -> H2Stream<F, B> {
         H2Stream {
             reply: respond,
             state: H2StreamState::Service { fut, connect_parts },
+            conn_idle_timeout,
         }
     }
 }
@@ -534,7 +560,11 @@ where
     type Output = ();
 
     fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
-        self.poll2(cx).map(|res| {
+        let conn_idle_timeout = self.conn_idle_timeout.clone();
+        let poll = self.poll2(cx);
+        poll.map(|res| {
+            let mut sleep = conn_idle_timeout.lock().unwrap();
+            sleep.as_mut().reset(tokio::time::Instant::now() + DEFAULT_CONN_IDLE_TIMEOUT);
             if let Err(e) = res {
                 debug!("stream error: {}", e);
             }