berkus
/
actix-web
mirror of https://github.com/fafhrd91/actix-web
1
0
Fork 0
Code Issues Releases Wiki Activity

actix-http: h1: stop pipelining when not reading full requests 

The existing pipelining behavior of the h1 dispatcher can cause client timeouts
if the entire request body isn't read. It puts the dispatcher into a state where
it refuses to read more (payload dropped) but there are still bytes in the buffer
from the request body.

This solution adds the SHUTDOWN flag in addition to the FINISHED flag
when completing a response when both the following are true:

1. There are no messages in `this.messages`
2. There is still a payload in `this.payload`

This combination implies two things. First, that we have not parsed a
pipelined request after the request we have just responded to. Second,
that the current request payload has not been fed an EOF. Because there
are no pipelined requests, we know that the current request payload
belongs to the request we have just responded to, and because the
request payload has not been fed an EOF, we know we never finished
reading it.

When this occurs, adding the SHUTDOWN flag to the dispatcher triggers a
`flush` and a `poll_shutdown` on the IO resource on the next poll.
This commit is contained in:
asonix 2025-08-07 17:56:44 -05:00
parent 351f7a17ed
commit af30b7e679
1 changed files with 49 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
actix-http/src/h1/dispatcher.rs
View File

@ -386,7 +386,14 @@ where
let mut this = self.project();
this.state.set(match size {
BodySize::None | BodySize::Sized(0) => {
this.flags.insert(Flags::FINISHED);
let payload_unfinished = this.payload.is_some();
if payload_unfinished {
this.flags.insert(Flags::SHUTDOWN | Flags::FINISHED);
} else {
this.flags.insert(Flags::FINISHED);
}
State::None
}
_ => State::SendPayload { body },
@ -404,7 +411,14 @@ where
let mut this = self.project();
this.state.set(match size {
BodySize::None | BodySize::Sized(0) => {
this.flags.insert(Flags::FINISHED);
let payload_unfinished = this.payload.is_some();
if payload_unfinished {
this.flags.insert(Flags::SHUTDOWN | Flags::FINISHED);
} else {
this.flags.insert(Flags::FINISHED);
}
State::None
}
_ => State::SendErrorPayload { body },
@ -503,10 +517,25 @@ where
Poll::Ready(None) => {
this.codec.encode(Message::Chunk(None), this.write_buf)?;
// if we have not yet pipelined to the next request, then
// this.payload was the payload for the request we just finished
// responding to. We can check to see if we finished reading it
// yet, and if not, shutdown the connection.
let payload_unfinished = this.payload.is_some();
let not_pipelined = this.messages.is_empty();
println!("not pipelined: {not_pipelined}");
println!("payload unfinished: {payload_unfinished}");
// payload stream finished.
// set state to None and handle next message
this.state.set(State::None);
this.flags.insert(Flags::FINISHED);
if not_pipelined && payload_unfinished {
this.flags.insert(Flags::SHUTDOWN | Flags::FINISHED);
} else {
this.flags.insert(Flags::FINISHED);
}
continue 'res;
}
@ -542,10 +571,25 @@ where
Poll::Ready(None) => {
this.codec.encode(Message::Chunk(None), this.write_buf)?;
// payload stream finished
// if we have not yet pipelined to the next request, then
// this.payload was the payload for the request we just finished
// responding to. We can check to see if we finished reading it
// yet, and if not, shutdown the connection.
let payload_unfinished = this.payload.is_some();
let not_pipelined = this.messages.is_empty();
println!("not pipelined: {not_pipelined}");
println!("payload unfinished: {payload_unfinished}");
// payload stream finished.
// set state to None and handle next message
this.state.set(State::None);
this.flags.insert(Flags::FINISHED);
if not_pipelined && payload_unfinished {
this.flags.insert(Flags::SHUTDOWN | Flags::FINISHED);
} else {
this.flags.insert(Flags::FINISHED);
}
continue 'res;
}