Closed sanjams2 closed 1 year ago
seanmonstar
commented
1 year ago The connection needs to be constantly polled to allow all connection-based and streams to make progress. It's best to spawn a new task for each request, since otherwise you awaiting data will mean it isn't awaiting the connection. (The h2.accept().await line).
sanjams2
commented
1 year ago Thanks for the quick response @seanmonstar. That does seem to work. Thanks for the pointer.
For posterity, the updated code is:
use std::net::SocketAddr;
use std::time::Duration;
use bytes::Bytes;
use h2::{client, server};
use http::{Method, Request, Response, StatusCode};
use tokio::net::{TcpListener, TcpStream};
use tokio::time::sleep;
async fn run_server(addr: SocketAddr) {
tracing::info!("[server] starting server on {}", addr);
let listener = TcpListener::bind(addr).await.unwrap();
while let Ok((socket, remote_addr)) = listener.accept().await {
tracing::debug!("[server] received connection from {}", remote_addr);
tokio::task::spawn(async move {
let mut h2 = server::handshake(socket).await.unwrap();
tracing::debug!("[server] h2 stream established with {}", remote_addr);
while let Some(request) = h2.accept().await {
tokio::spawn(async move {
let (mut request, mut respond) = request.unwrap();
tracing::debug!("[server] received h2 request from {}: {:?} {} {}", remote_addr, request.version(), request.method(), request.uri());
let body = request.body_mut();
tracing::trace!("[server] (flow control) avail cap: {}, used cap: {}", body.flow_control().available_capacity(), body.flow_control().used_capacity());
while let Some(body_chunk) = body.data().await {
tracing::debug!("[server] received body chunk from {}", remote_addr);
match body_chunk {
Err(e) => {
tracing::error!("[server] error getting body chunk: {:?}", e);
break
},
Ok(body_bytes) => {
let req_body = String::from_utf8_lossy(&body_bytes);
tracing::debug!("[server] request body chunk:\n{}", req_body);
let _ = body.flow_control().release_capacity(body_bytes.len());
}
}
}
tracing::debug!("[server] sending response to {}", remote_addr);
let response = Response::builder().status(StatusCode::OK).body(()).unwrap();
respond.send_response(response, true).unwrap();
tracing::info!("[server] successfully responded to {}", remote_addr);
});
}
tracing::debug!("[server] connection with {} ended", remote_addr);
});
}
tracing::info!("[server] shutting down");
}
#[tokio::main]
async fn main() {
tracing_subscriber::fmt::init();
// Start server
let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
tokio::spawn(async move {
run_server(addr).await
});
// Let server spin up
sleep(Duration::from_secs(2)).await;
// Client
// Establish connection
tracing::debug!("[client] starting up...");
let addr = TcpStream::connect(addr).await.unwrap();
let (h2, connection) = client::handshake(addr).await.unwrap();
tokio::spawn(async move {
if let Err(e) = connection.await {
tracing::error!("[client] error on connection: {:?}", e);
}
});
let mut h2 = h2.ready().await.unwrap();
// Start request
tracing::info!("[client] sending request");
let request = Request::builder()
.method(Method::POST)
.uri("http://127.0.0.1:3000")
.body(())
.unwrap();
let (response, mut stream) = h2.send_request(request, false).unwrap();
// Simulate delay
sleep(Duration::from_millis(2)).await;
// Send body
tracing::debug!("[client] sending body");
let req_body = Bytes::from("hello, server ".to_string().repeat(10));
stream.reserve_capacity(req_body.len());
tracing::trace!("[client] (flow control) stream capacity: {}", stream.capacity());
stream.send_data(req_body, true).unwrap();
// Process response
tracing::debug!("[client] waiting for response...");
let response = response.await.unwrap();
let (head, _body) = response.into_parts();
tracing::info!("[client] received response: {:?}", head);
}I will close this issue.
Im not sure this is a bug but the behavior feels buggy so I thought I would cut an issue.
Here is the code in question:
Here is what the code prints out:
and the process just hangs here
What I notice is that when there is any sort of delay between the
send_requestandsend_datacalls, the server will hang attempting to read the body. This is simulated in the above code via a sleep (see// Simulate delay). When the sleep is removed, the server is able to read the body of the request and continue on. The piece of the server code that hangs is thebody.data().await(see// <-- server hangs here)To ensure that the server was actually receiving the request data, I also performed a packet capture using tcpdump:
Perhaps there is something about this is how the server should be acting based on the HTTP/2 RFC (maybe due to receiving a HEADER packet with no data?), but this feels a bit strange. It seems like there could be a delay between the
send_requestandsend_datacalls for any number of reasons and therefore the case is one that the server should be able to handle?I can include more trace logs, but have left it at this for brevity.