foster/denoland-deno
0
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2024-10-29 08:58:01 -04:00
Code Issues Wiki Activity

perf(ext/http): fast path for uncompressed bodies (#14366)

Browse source
This commit is contained in:
Divy Srivastava 2022-04-25 08:13:22 +05:30 committed by GitHub
parent e2fba7b967
commit 6dcf3a447c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 70 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

65
ext/http/lib.rs
View file

@ -343,6 +343,7 @@ impl Default for HttpRequestReader {
enum HttpResponseWriter { enum HttpResponseWriter {
Headers(oneshot::Sender<Response<Body>>), Headers(oneshot::Sender<Response<Body>>),
Body(Pin<Box<dyn tokio::io::AsyncWrite>>), Body(Pin<Box<dyn tokio::io::AsyncWrite>>),
BodyUncompressed(hyper::body::Sender),
Closed, Closed,
} }
@ -638,7 +639,7 @@ async fn op_http_write_headers(
} }
None => { None => {
// If no buffer was passed, the caller will stream the response body. // If no buffer was passed, the caller will stream the response body.
if should_compress {
// Create a one way pipe that implements tokio's async io traits. To do // Create a one way pipe that implements tokio's async io traits. To do
// this we create a [tokio::io::DuplexStream], but then throw away one // this we create a [tokio::io::DuplexStream], but then throw away one
// of the directions to create a one way pipe. // of the directions to create a one way pipe.
@ -647,8 +648,6 @@ async fn op_http_write_headers(
let (_, writer) = tokio::io::split(b); let (_, writer) = tokio::io::split(b);
let writer_body: Pin<Box<dyn tokio::io::AsyncWrite>>; let writer_body: Pin<Box<dyn tokio::io::AsyncWrite>>;
if should_compress {
match *stream.accept_encoding.borrow() { match *stream.accept_encoding.borrow() {
Encoding::Brotli => { Encoding::Brotli => {
let writer = BrotliEncoder::new(writer); let writer = BrotliEncoder::new(writer);
@ -662,12 +661,14 @@ async fn op_http_write_headers(
builder = builder.header("content-encoding", "gzip"); builder = builder.header("content-encoding", "gzip");
} }
} }
} else {
writer_body = Box::pin(writer);
}
body = builder.body(Body::wrap_stream(ReaderStream::new(reader)))?; body = builder.body(Body::wrap_stream(ReaderStream::new(reader)))?;
new_wr = HttpResponseWriter::Body(writer_body); new_wr = HttpResponseWriter::Body(writer_body);
} else {
let (body_tx, body_rx) = Body::channel();
body = builder.body(body_rx)?;
new_wr = HttpResponseWriter::BodyUncompressed(body_tx);
}
} }
} }
@ -699,14 +700,14 @@ async fn op_http_write_resource(
let mut wr = RcRef::map(&http_stream, |r| &r.wr).borrow_mut().await; let mut wr = RcRef::map(&http_stream, |r| &r.wr).borrow_mut().await;
let resource = state.borrow().resource_table.get_any(stream)?; let resource = state.borrow().resource_table.get_any(stream)?;
loop { loop {
let body_writer = match &mut *wr { match *wr {
HttpResponseWriter::Body(body_writer) => body_writer,
HttpResponseWriter::Headers(_) => { HttpResponseWriter::Headers(_) => {
return Err(http_error("no response headers")) return Err(http_error("no response headers"))
} }
HttpResponseWriter::Closed => { HttpResponseWriter::Closed => {
return Err(http_error("response already completed")) return Err(http_error("response already completed"))
} }
_ => {}
}; };
let vec = vec![0u8; 64 * 1024]; // 64KB let vec = vec![0u8; 64 * 1024]; // 64KB
@ -715,9 +716,10 @@ async fn op_http_write_resource(
if nread == 0 { if nread == 0 {
break; break;
} }
match body_writer.write_all(&buf[..nread]).await {
Ok(_) => {} match &mut *wr {
Err(err) => { HttpResponseWriter::Body(body) => {
if let Err(err) = body.write_all(&buf[..nread]).await {
assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe); assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe);
// Don't return "broken pipe", that's an implementation detail. // Don't return "broken pipe", that's an implementation detail.
// Pull up the failure associated with the transport connection instead. // Pull up the failure associated with the transport connection instead.
@ -726,6 +728,17 @@ async fn op_http_write_resource(
*wr = HttpResponseWriter::Closed; *wr = HttpResponseWriter::Closed;
} }
} }
HttpResponseWriter::BodyUncompressed(body) => {
if let Err(err) = body.send_data(Bytes::from(buf.to_temp())).await {
assert!(err.is_closed());
// Pull up the failure associated with the transport connection instead.
http_stream.conn.closed().await?;
// If there was no connection error, drop body_tx.
*wr = HttpResponseWriter::Closed;
}
}
_ => unreachable!(),
};
} }
let wr = take(&mut *wr); let wr = take(&mut *wr);
@ -756,22 +769,19 @@ async fn op_http_write(
let mut wr = RcRef::map(&stream, |r| &r.wr).borrow_mut().await; let mut wr = RcRef::map(&stream, |r| &r.wr).borrow_mut().await;
loop { loop {
let body_writer = match &mut *wr { match &mut *wr {
HttpResponseWriter::Body(body_tx) => body_tx,
HttpResponseWriter::Headers(_) => { HttpResponseWriter::Headers(_) => {
break Err(http_error("no response headers")) break Err(http_error("no response headers"))
} }
HttpResponseWriter::Closed => { HttpResponseWriter::Closed => {
break Err(http_error("response already completed")) break Err(http_error("response already completed"))
} }
}; HttpResponseWriter::Body(body) => {
let mut result = body.write_all(&buf).await;
let mut res = body_writer.write_all(&buf).await; if result.is_ok() {
if res.is_ok() { result = body.flush().await;
res = body_writer.flush().await;
} }
match result {
match res {
Ok(_) => break Ok(()), Ok(_) => break Ok(()),
Err(err) => { Err(err) => {
assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe); assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe);
@ -783,6 +793,21 @@ async fn op_http_write(
} }
} }
} }
HttpResponseWriter::BodyUncompressed(body) => {
let bytes = Bytes::copy_from_slice(&buf[..]);
match body.send_data(bytes).await {
Ok(_) => break Ok(()),
Err(err) => {
assert!(err.is_closed());
// Pull up the failure associated with the transport connection instead.
stream.conn.closed().await?;
// If there was no connection error, drop body_tx.
*wr = HttpResponseWriter::Closed;
}
}
}
}
}
} }
/// Gracefully closes the write half of the HTTP stream. Note that this does not /// Gracefully closes the write half of the HTTP stream. Note that this does not