forked/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(h1): improve parsing and encoding of http1 messages

Browse Source
This commit is contained in:
Sean McArthur
2018-05-11 15:40:42 -07:00
parent c3c35e866c
commit 26417fc24a
13 changed files with 1006 additions and 442 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
src/proto/h1/conn.rs
View File

@@ -4,13 +4,13 @@ use std::marker::PhantomData;
use bytes::{Buf, Bytes};
use futures::{Async, Poll};
use http::{Method, Version};
use http::{HeaderMap, Method, Version};
use tokio_io::{AsyncRead, AsyncWrite};
use ::Chunk;
use proto::{BodyLength, Decode, Http1Transaction, MessageHead};
use proto::{BodyLength, MessageHead};
use super::io::{Buffered};
use super::{EncodedBuf, Encoder, Decoder};
use super::{EncodedBuf, Encode, Encoder, Decode, Decoder, Http1Transaction, ParseContext};
const H2_PREFACE: &'static [u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
@@ -36,6 +36,7 @@ where I: AsyncRead + AsyncWrite,
Conn {
io: Buffered::new(io),
state: State {
cached_headers: None,
error: None,
keep_alive: KA::Busy,
method: None,
@@ -118,8 +119,11 @@ where I: AsyncRead + AsyncWrite,
trace!("Conn::read_head");
loop {
let (version, head) = match self.io.parse::<T>() {
Ok(Async::Ready(head)) => (head.version, head),
let msg = match self.io.parse::<T>(ParseContext {
cached_headers: &mut self.state.cached_headers,
req_method: &mut self.state.method,
}) {
Ok(Async::Ready(msg)) => msg,
Ok(Async::NotReady) => return Ok(Async::NotReady),
Err(e) => {
// If we are currently waiting on a message, then an empty
@@ -141,48 +145,32 @@ where I: AsyncRead + AsyncWrite,
}
};
match version {
Version::HTTP_10 |
Version::HTTP_11 => {},
_ => {
error!("unimplemented HTTP Version = {:?}", version);
self.state.close_read();
//TODO: replace this with a more descriptive error
return Err(::Error::new_version());
}
};
self.state.version = version;
let decoder = match T::decoder(&head, &mut self.state.method) {
Ok(Decode::Normal(d)) => {
self.state.version = msg.head.version;
let head = msg.head;
let decoder = match msg.decode {
Decode::Normal(d) => {
d
},
Ok(Decode::Final(d)) => {
Decode::Final(d) => {
trace!("final decoder, HTTP ending");
debug_assert!(d.is_eof());
self.state.close_read();
d
},
Ok(Decode::Ignore) => {
Decode::Ignore => {
// likely a 1xx message that we can ignore
continue;
}
Err(e) => {
debug!("decoder error = {:?}", e);
self.state.close_read();
return self.on_parse_error(e)
.map(|()| Async::NotReady);
}
};
debug!("incoming body is {}", decoder);
self.state.busy();
if head.expecting_continue() {
let msg = b"HTTP/1.1 100 Continue\r\n\r\n";
self.io.write_buf_mut().extend_from_slice(msg);
if msg.expect_continue {
let cont = b"HTTP/1.1 100 Continue\r\n\r\n";
self.io.write_buf_mut().extend_from_slice(cont);
}
let wants_keep_alive = head.should_keep_alive();
let wants_keep_alive = msg.keep_alive;
self.state.keep_alive &= wants_keep_alive;
let (body, reading) = if decoder.is_eof() {
(false, Reading::KeepAlive)
@@ -410,8 +398,17 @@ where I: AsyncRead + AsyncWrite,
self.enforce_version(&mut head);
let buf = self.io.write_buf_mut();
self.state.writing = match T::encode(head, body, &mut self.state.method, self.state.title_case_headers, buf) {
self.state.writing = match T::encode(Encode {
head: &mut head,
body,
keep_alive: self.state.wants_keep_alive(),
req_method: &mut self.state.method,
title_case_headers: self.state.title_case_headers,
}, buf) {
Ok(encoder) => {
debug_assert!(self.state.cached_headers.is_none());
debug_assert!(head.headers.is_empty());
self.state.cached_headers = Some(head.headers);
if !encoder.is_eof() {
Writing::Body(encoder)
} else if encoder.is_last() {
@@ -430,24 +427,12 @@ where I: AsyncRead + AsyncWrite,
// If we know the remote speaks an older version, we try to fix up any messages
// to work with our older peer.
fn enforce_version(&mut self, head: &mut MessageHead<T::Outgoing>) {
//use header::Connection;
let wants_keep_alive = if self.state.wants_keep_alive() {
let ka = head.should_keep_alive();
self.state.keep_alive &= ka;
ka
} else {
false
};
match self.state.version {
Version::HTTP_10 => {
// If the remote only knows HTTP/1.0, we should force ourselves
// to do only speak HTTP/1.0 as well.
head.version = Version::HTTP_10;
if wants_keep_alive {
//TODO: head.headers.set(Connection::keep_alive());
}
},
_ => {
// If the remote speaks HTTP/1.1, then it *should* be fine with
@@ -617,13 +602,27 @@ impl<I, B: Buf, T> fmt::Debug for Conn<I, B, T> {
}
struct State {
/// Re-usable HeaderMap to reduce allocating new ones.
cached_headers: Option<HeaderMap>,
/// If an error occurs when there wasn't a direct way to return it
/// back to the user, this is set.
error: Option<::Error>,
/// Current keep-alive status.
keep_alive: KA,
/// If mid-message, the HTTP Method that started it.
///
/// This is used to know things such as if the message can include
/// a body or not.
method: Option<Method>,
title_case_headers: bool,
/// Set to true when the Dispatcher should poll read operations
/// again. See the `maybe_notify` method for more.
notify_read: bool,
/// State of allowed reads
reading: Reading,
/// State of allowed writes
writing: Writing,
/// Either HTTP/1.0 or 1.1 connection
version: Version,
}

3
src/proto/h1/dispatch.rs
View File

@@ -4,7 +4,8 @@ use http::{Request, Response, StatusCode};
use tokio_io::{AsyncRead, AsyncWrite};
use body::{Body, Payload};
use proto::{BodyLength, Conn, Http1Transaction, MessageHead, RequestHead, RequestLine, ResponseHead};
use proto::{BodyLength, Conn, MessageHead, RequestHead, RequestLine, ResponseHead};
use super::Http1Transaction;
use service::Service;
pub(crate) struct Dispatcher<D, Bs: Payload, I, T> {

7
src/proto/h1/encode.rs
View File

@@ -7,7 +7,7 @@ use iovec::IoVec;
use common::StaticBuf;
/// Encoders to handle different Transfer-Encodings.
#[derive(Debug, Clone)]
#[derive(Debug, Clone, PartialEq)]
pub struct Encoder {
kind: Kind,
is_last: bool,
@@ -70,8 +70,9 @@ impl Encoder {
}
}
pub fn set_last(&mut self) {
self.is_last = true;
pub fn set_last(mut self, is_last: bool) -> Self {
self.is_last = is_last;
self
}
pub fn is_last(&self) -> bool {

22
src/proto/h1/io.rs
View File

@@ -8,7 +8,7 @@ use futures::{Async, Poll};
use iovec::IoVec;
use tokio_io::{AsyncRead, AsyncWrite};
use proto::{Http1Transaction, MessageHead};
use super::{Http1Transaction, ParseContext, ParsedMessage};
/// The initial buffer size allocated before trying to read from IO.
pub(crate) const INIT_BUFFER_SIZE: usize = 8192;
@@ -126,12 +126,16 @@ where
}
}
pub(super) fn parse<S: Http1Transaction>(&mut self) -> Poll<MessageHead<S::Incoming>, ::Error> {
pub(super) fn parse<S>(&mut self, ctx: ParseContext)
-> Poll<ParsedMessage<S::Incoming>, ::Error>
where
S: Http1Transaction,
{
loop {
match try!(S::parse(&mut self.read_buf)) {
Some((head, len)) => {
debug!("parsed {} headers ({} bytes)", head.headers.len(), len);
return Ok(Async::Ready(head))
match try!(S::parse(&mut self.read_buf, ParseContext { cached_headers: ctx.cached_headers, req_method: ctx.req_method, })) {
Some(msg) => {
debug!("parsed {} headers", msg.head.headers.len());
return Ok(Async::Ready(msg))
},
None => {
if self.read_buf.capacity() >= self.max_buf_size {
@@ -617,7 +621,11 @@ mod tests {
let mock = AsyncIo::new_buf(raw, raw.len());
let mut buffered = Buffered::<_, Cursor<Vec<u8>>>::new(mock);
assert_eq!(buffered.parse::<::proto::ClientTransaction>().unwrap(), Async::NotReady);
let ctx = ParseContext {
cached_headers: &mut None,
req_method: &mut None,
};
assert!(buffered.parse::<::proto::ClientTransaction>(ctx).unwrap().is_not_ready());
assert!(buffered.io.blocked());
}

60
src/proto/h1/mod.rs
View File

@@ -1,3 +1,8 @@
use bytes::BytesMut;
use http::{HeaderMap, Method};
use proto::{MessageHead, BodyLength};
pub(crate) use self::conn::Conn;
pub(crate) use self::dispatch::Dispatcher;
pub use self::decode::Decoder;
@@ -11,5 +16,58 @@ mod decode;
pub(crate) mod dispatch;
mod encode;
mod io;
pub mod role;
mod role;
pub(crate) type ServerTransaction = self::role::Server<self::role::YesUpgrades>;
//pub type ServerTransaction = self::role::Server<self::role::NoUpgrades>;
//pub type ServerUpgradeTransaction = self::role::Server<self::role::YesUpgrades>;
pub(crate) type ClientTransaction = self::role::Client<self::role::NoUpgrades>;
pub(crate) type ClientUpgradeTransaction = self::role::Client<self::role::YesUpgrades>;
pub(crate) trait Http1Transaction {
type Incoming;
type Outgoing: Default;
fn parse(bytes: &mut BytesMut, ctx: ParseContext) -> ParseResult<Self::Incoming>;
fn encode(enc: Encode<Self::Outgoing>, dst: &mut Vec<u8>) -> ::Result<Encoder>;
fn on_error(err: &::Error) -> Option<MessageHead<Self::Outgoing>>;
fn should_error_on_parse_eof() -> bool;
fn should_read_first() -> bool;
}
pub(crate) type ParseResult<T> = Result<Option<ParsedMessage<T>>, ::error::Parse>;
#[derive(Debug)]
pub(crate) struct ParsedMessage<T> {
head: MessageHead<T>,
decode: Decode,
expect_continue: bool,
keep_alive: bool,
}
pub(crate) struct ParseContext<'a> {
cached_headers: &'a mut Option<HeaderMap>,
req_method: &'a mut Option<Method>,
}
/// Passed to Http1Transaction::encode
pub(crate) struct Encode<'a, T: 'a> {
head: &'a mut MessageHead<T>,
body: Option<BodyLength>,
keep_alive: bool,
req_method: &'a mut Option<Method>,
title_case_headers: bool,
}
#[derive(Debug, PartialEq)]
pub enum Decode {
/// Decode normally.
Normal(Decoder),
/// After this decoder is done, HTTP is done.
Final(Decoder),
/// A header block that should be ignored, like unknown 1xx responses.
Ignore,
}

1011
src/proto/h1/role.rs
View File

File diff suppressed because it is too large Load Diff

47
src/proto/mod.rs
View File

@@ -1,10 +1,7 @@
//! Pieces pertaining to the HTTP message protocol.
use bytes::BytesMut;
use http::{HeaderMap, Method, StatusCode, Uri, Version};
use headers;
pub(crate) use self::h1::{dispatch, Conn};
pub(crate) use self::h1::{dispatch, Conn, ClientTransaction, ClientUpgradeTransaction, ServerTransaction};
pub(crate) mod h1;
pub(crate) mod h2;
@@ -30,6 +27,7 @@ pub struct RequestLine(pub Method, pub Uri);
/// An incoming response message.
pub type ResponseHead = MessageHead<StatusCode>;
/*
impl<S> MessageHead<S> {
pub fn should_keep_alive(&self) -> bool {
should_keep_alive(self.version, &self.headers)
@@ -55,33 +53,7 @@ pub fn should_keep_alive(version: Version, headers: &HeaderMap) -> bool {
pub fn expecting_continue(version: Version, headers: &HeaderMap) -> bool {
version == Version::HTTP_11 && headers::expect_continue(headers)
}
pub(crate) type ServerTransaction = h1::role::Server<h1::role::YesUpgrades>;
//pub type ServerTransaction = h1::role::Server<h1::role::NoUpgrades>;
//pub type ServerUpgradeTransaction = h1::role::Server<h1::role::YesUpgrades>;
pub(crate) type ClientTransaction = h1::role::Client<h1::role::NoUpgrades>;
pub(crate) type ClientUpgradeTransaction = h1::role::Client<h1::role::YesUpgrades>;
pub(crate) trait Http1Transaction {
type Incoming;
type Outgoing: Default;
fn parse(bytes: &mut BytesMut) -> ParseResult<Self::Incoming>;
fn decoder(head: &MessageHead<Self::Incoming>, method: &mut Option<Method>) -> ::Result<Decode>;
fn encode(
head: MessageHead<Self::Outgoing>,
body: Option<BodyLength>,
method: &mut Option<Method>,
title_case_headers: bool,
dst: &mut Vec<u8>,
) -> ::Result<h1::Encoder>;
fn on_error(err: &::Error) -> Option<MessageHead<Self::Outgoing>>;
fn should_error_on_parse_eof() -> bool;
fn should_read_first() -> bool;
}
pub(crate) type ParseResult<T> = Result<Option<(MessageHead<T>, usize)>, ::error::Parse>;
*/
#[derive(Debug)]
pub enum BodyLength {
@@ -91,17 +63,7 @@ pub enum BodyLength {
Unknown,
}
#[derive(Debug)]
pub enum Decode {
/// Decode normally.
Normal(h1::Decoder),
/// After this decoder is done, HTTP is done.
Final(h1::Decoder),
/// A header block that should be ignored, like unknown 1xx responses.
Ignore,
}
/*
#[test]
fn test_should_keep_alive() {
let mut headers = HeaderMap::new();
@@ -129,3 +91,4 @@ fn test_expecting_continue() {
assert!(!expecting_continue(Version::HTTP_10, &headers));
assert!(expecting_continue(Version::HTTP_11, &headers));
}
*/