add conn idle timeout

Author: silence

Cargo.toml

@@ -35,7 +35,7 @@ pin-project-lite = "0.2.4"
 tower-service = "0.3"
 tokio = { version = "1", features = ["sync"] }
 want = "0.3"
-
+log = "0.4.14"
 # Optional
 
 libc = { version = "0.2", optional = true }

src/proto/h1/conn.rs

@@ -154,6 +154,10 @@ where
         self.state.is_write_closed()
     }
 
+    pub 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;
@@ -736,11 +740,11 @@ where
     pub(crate) fn poll_shutdown(&mut self, cx: &mut task::Context<'_>) -> Poll<io::Result<()>> {
         match ready!(Pin::new(self.io.io_mut()).poll_shutdown(cx)) {
             Ok(()) => {
-                trace!("shut down IO complete");
+                log::trace!("shut down IO complete");
                 Poll::Ready(Ok(()))
             }
             Err(e) => {
-                debug!("error shutting down IO: {}", e);
+                log::debug!("error shutting down IO: {}", e);
                 Poll::Ready(Err(e))
             }
         }

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};
@@ -19,6 +21,7 @@ pub(crate) struct Dispatcher<D, Bs: HttpBody, I, T> {
     body_tx: Option<crate::body::Sender>,
     body_rx: Pin<Box<Option<Bs>>>,
     is_closing: bool,
+    idle_timeout_fut: Option<Pin<Box<Sleep>>>,
 }
 
 pub(crate) trait Dispatch {
@@ -77,6 +80,7 @@ where
             body_tx: None,
             body_rx: Box::pin(None),
             is_closing: false,
+            idle_timeout_fut: None
         }
     }
 
@@ -157,10 +161,22 @@ where
         // 16 was chosen arbitrarily, as that is number of pipelined requests
         // benchmarks often use. Perhaps it should be a config option instead.
         for _ in 0..16 {
+            if let Some(time) = &mut self.idle_timeout_fut {
+                if time.as_mut().poll(cx).is_ready() {
+                    log::info!("conn is close");
+                    self.close();
+                    return Poll::Ready(Ok(()));
+                }
+            }
             let _ = self.poll_read(cx)?;
             let _ = self.poll_write(cx)?;
             let _ = self.poll_flush(cx)?;
-
+            if self.conn.is_idle() {
+                let mut idle_timeout = Box::pin(tokio::time::sleep(Duration::from_secs(10)));
+                let _ = idle_timeout.as_mut().poll(cx);
+                self.idle_timeout_fut = Some(idle_timeout);
+                log::info!("conn is idle");
+            }
             // 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 +263,8 @@ 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))) => {
+                log::info!("req is comming");
+                self.idle_timeout_fut = None;
                 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.
@@ -100,6 +101,7 @@ where
     ping: Option<(ping::Recorder, ping::Ponger)>,
     conn: Connection<T, SendBuf<B::Data>>,
     closing: Option<crate::Error>,
+    idle_timeot: Arc<Mutex<Pin<Box<Sleep>>>>
 }
 
 impl<T, S, B, E> Server<T, S, B, E>
@@ -202,6 +204,9 @@ where
                         ping,
                         conn,
                         closing: None,
+                        idle_timeot: Arc::new(Mutex::new(Box::pin(
+                            tokio::time::sleep(Duration::from_secs(10)
+                            ))))
                     })
                 }
                 State::Serving(ref mut srv) => {
@@ -224,6 +229,16 @@ where
     T: AsyncRead + AsyncWrite + Unpin,
     B: HttpBody + 'static,
 {
+
+    fn reset_idle(&self, time: Duration) {
+        let mut sleep = self.idle_timeot.lock().unwrap();
+        sleep.as_mut().reset(tokio::time::Instant::now() + time);
+    }
+    fn poll_idle(&self,  cx: &mut task::Context<'_>)-> bool  {
+        let mut sleep = self.idle_timeot.lock().unwrap();
+        sleep.as_mut().poll(cx).is_ready()
+    }
+
     fn poll_server<S, E>(
         &mut self,
         cx: &mut task::Context<'_>,
@@ -269,10 +284,16 @@ where
                         break;
                     }
                 }
-
+                if self.poll_idle(cx) {
+                    log::info!("incoming connection complete11");
+                    self.conn.graceful_shutdown();
+                    log::info!("incoming connection complete");
+                    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))) => {
+                        self.reset_idle(Duration::from_secs(60 * 60 * 24));
                         trace!("incoming request");
                         let content_length = headers::content_length_parse_all(req.headers());
                         let ping = self
@@ -316,8 +337,9 @@ where
                         if let Some(protocol) = req.extensions_mut().remove::<h2::ext::Protocol>() {
                             req.extensions_mut().insert(Protocol::from_inner(protocol));
                         }
-
-                        let fut = H2Stream::new(service.call(req), connect_parts, respond);
+                        log::info!("{:?}",respond.stream_id());
+                        let fut = H2Stream::new(service.call(req), connect_parts, respond,
+                        self.idle_timeot.clone());
                         exec.execute_h2stream(fut);
                     }
                     Some(Err(e)) => {
@@ -373,6 +395,7 @@ pin_project! {
         reply: SendResponse<SendBuf<B::Data>>,
         #[pin]
         state: H2StreamState<F, B>,
+        idle_timeot: Arc<Mutex<Pin<Box<Sleep>>>>,
     }
 }
 
@@ -408,10 +431,12 @@ where
         fut: F,
         connect_parts: Option<ConnectParts>,
         respond: SendResponse<SendBuf<B::Data>>,
+        idle_timeot: Arc<Mutex<Pin<Box<Sleep>>>>
     ) -> H2Stream<F, B> {
         H2Stream {
             reply: respond,
             state: H2StreamState::Service { fut, connect_parts },
+            idle_timeot
         }
     }
 }
@@ -534,7 +559,12 @@ where
     type Output = ();
 
     fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
-        self.poll2(cx).map(|res| {
+        let idle_timeot = self.idle_timeot.clone();
+        let poll = self.poll2(cx);
+        poll.map(|res| {
+            log::info!("con is idle");
+            let mut sleep = idle_timeot.lock().unwrap();
+            sleep.as_mut().reset(tokio::time::Instant::now() + Duration::from_secs(10));
             if let Err(e) = res {
                 debug!("stream error: {}", e);
             }