logs for tests

Author: seanmonstar

.github/workflows/CI.yml

@@ -7,6 +7,7 @@ on:
 
 env:
   RUST_BACKTRACE: 1
+  RUST_LOG: hyper=trace,tokio=trace,mio=trace,h2=trace
 
 jobs:
   style:
@@ -72,7 +73,7 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: test
-          args: ${{ matrix.features }}
+          args: ${{ matrix.features }} --test server upgrades_ignored
 
       - name: Test all benches
         if: matrix.benches

src/proto/h1/conn.rs

@@ -276,6 +276,7 @@ where
 
     pub fn poll_read_keep_alive(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
         debug_assert!(!self.can_read_head() && !self.can_read_body());
+        trace!("poll_read_keep_alive");
 
         if self.is_read_closed() {
             Poll::Pending
@@ -334,7 +335,13 @@ where
         debug_assert!(!self.can_read_head() && !self.can_read_body() && !self.is_read_closed());
         debug_assert!(self.is_mid_message());
 
+        trace!("mid_message_detect_eof >");
         if self.state.allow_half_close || !self.io.read_buf().is_empty() {
+            trace!(
+                "allow_half_close: {}, read buf is empty: {}",
+                self.state.allow_half_close,
+                !self.io.read_buf().is_empty()
+            );
             return Poll::Pending;
         }
 
@@ -345,14 +352,18 @@ where
             self.state.close_read();
             Poll::Ready(Err(crate::Error::new_incomplete()))
         } else {
+            trace!("mid_message_detect_eof < read {} bytes", num_read);
             Poll::Ready(Ok(()))
         }
     }
 
     fn force_io_read(&mut self, cx: &mut task::Context<'_>) -> Poll<io::Result<usize>> {
         debug_assert!(!self.state.is_read_closed());
 
+        trace!("force_io_read >");
         let result = ready!(self.io.poll_read_from_io(cx));
+
+        trace!("force_io_read < {:?}", result);
         Poll::Ready(result.map_err(|e| {
             trace!("force_io_read; io error = {:?}", e);
             self.state.close();

src/proto/h1/dispatch.rs

@@ -148,9 +148,15 @@ where
         //
         // 16 was chosen arbitrarily, as that is number of pipelined requests
         // benchmarks often use. Perhaps it should be a config option instead.
+        trace!("poll_loop");
         for _ in 0..16 {
-            let _ = self.poll_read(cx)?;
+            trace!("poll_read >");
+            let ret = self.poll_read(cx);
+            trace!("poll_read < {:?}", ret);
+            let _ = ret?;
+            trace!("poll_write >");
             let _ = self.poll_write(cx)?;
+            trace!("poll_write <");
             let _ = self.poll_flush(cx)?;
 
             // This could happen if reading paused before blocking on IO,
@@ -163,6 +169,7 @@ where
             // the Conn is noticeably faster in pipelined benchmarks.
             if !self.conn.wants_read_again() {
                 //break;
+                trace!("! wants_read_again()");
                 return Poll::Ready(Ok(()));
             }
         }
@@ -173,7 +180,9 @@ where
     }
 
     fn poll_read(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
+        trace!("poll_read >>");
         loop {
+            trace!("poll_read; loop");
             if self.is_closing {
                 return Poll::Ready(Ok(()));
             } else if self.conn.can_read_head() {
@@ -276,6 +285,7 @@ where
     }
 
     fn poll_write(&mut self, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
+        trace!("poll_write");
         loop {
             if self.is_closing {
                 return Poll::Ready(Ok(()));
@@ -475,8 +485,11 @@ where
         cx: &mut task::Context<'_>,
     ) -> Poll<Option<Result<(Self::PollItem, Self::PollBody), Self::PollError>>> {
         let mut this = self.as_mut();
+        trace!("Server::poll_msg ->");
         let ret = if let Some(ref mut fut) = this.in_flight.as_mut().as_pin_mut() {
+            trace!("Server::poll_msg -> polling fut");
             let resp = ready!(fut.as_mut().poll(cx)?);
+            trace!("Server::poll_msg -> response!");
             let (parts, body) = resp.into_parts();
             let head = MessageHead {
                 version: parts.version,
@@ -495,12 +508,14 @@ where
 
     fn recv_msg(&mut self, msg: crate::Result<(Self::RecvItem, Body)>) -> crate::Result<()> {
         let (msg, body) = msg?;
+        trace!("recv_msg; req!");
         let mut req = Request::new(body);
         *req.method_mut() = msg.subject.0;
         *req.uri_mut() = msg.subject.1;
         *req.headers_mut() = msg.headers;
         *req.version_mut() = msg.version;
         let fut = self.service.call(req);
+        trace!("recv_msg; service called");
         self.in_flight.set(Some(fut));
         Ok(())
     }

src/proto/h1/io.rs

@@ -183,6 +183,7 @@ where
     }
 
     pub fn poll_read_from_io(&mut self, cx: &mut task::Context<'_>) -> Poll<io::Result<usize>> {
+        trace!("poll_read_from_io >");
         self.read_blocked = false;
         let next = self.read_buf_strategy.next();
         if self.read_buf_remaining_mut() < next {
@@ -192,6 +193,7 @@ where
         match Pin::new(&mut self.io).poll_read(cx, &mut buf) {
             Poll::Ready(Ok(_)) => {
                 let n = buf.filled().len();
+                trace!("poll_read_from_io < {} bytes", n);
                 unsafe {
                     // Safety: we just read that many bytes into the
                     // uninitialized part of the buffer, so this is okay.
@@ -202,6 +204,7 @@ where
                 Poll::Ready(Ok(n))
             }
             Poll::Pending => {
+                trace!("poll_read_from_io < blocked");
                 self.read_blocked = true;
                 Poll::Pending
             }

tests/server.rs

@@ -1141,6 +1141,7 @@ async fn disconnect_after_reading_request_before_responding() {
 
 #[tokio::test]
 async fn returning_1xx_response_is_error() {
+    let _ = pretty_env_logger::try_init();
     let listener = tcp_bind(&"127.0.0.1:0".parse().unwrap()).unwrap();
     let addr = listener.local_addr().unwrap();
 
@@ -1155,7 +1156,7 @@ async fn returning_1xx_response_is_error() {
     });
 
     let (socket, _) = listener.accept().await.unwrap();
-    Http::new()
+    let err = Http::new()
         .serve_connection(
             socket,
             service_fn(|_| async move {
@@ -1169,6 +1170,7 @@ async fn returning_1xx_response_is_error() {
         )
         .await
         .expect_err("1xx status code should error");
+    assert!(err.is_user(), "should be a user error: {}", err);
 }
 
 #[test]
@@ -1379,9 +1381,10 @@ async fn upgrades_ignored() {
     let listener = tcp_bind(&"127.0.0.1:0".parse().unwrap()).unwrap();
     let addr = listener.local_addr().unwrap();
 
-    tokio::spawn(async move {
+    let t1 = tokio::spawn(async move {
         let svc = service_fn(move |req: Request<Body>| {
             assert_eq!(req.headers()["upgrade"], "yolo");
+            println!("server got request");
             future::ok::<_, hyper::Error>(Response::new(hyper::Body::empty()))
         });
 
@@ -1411,6 +1414,11 @@ async fn upgrades_ignored() {
 
     let res2 = client.request(make_req()).await.expect("req 2");
     assert_eq!(res2.status(), 200);
+    drop(res2);
+
+    drop(client);
+
+    t1.await.unwrap();
 }
 
 #[tokio::test]
@@ -2348,6 +2356,7 @@ fn has_header(msg: &str, name: &str) -> bool {
 
 fn tcp_bind(addr: &SocketAddr) -> ::tokio::io::Result<TcpListener> {
     let std_listener = StdTcpListener::bind(addr).unwrap();
+    std_listener.set_nonblocking(true).unwrap();
     TcpListener::from_std(std_listener)
 }