Skip to content

Commit

Permalink
feat(h1): re-enable writev support
Browse files Browse the repository at this point in the history 
Tokio's `AsyncWrite` trait once again has support for vectored writes in
Tokio 0.3.4 (see tokio-rs/tokio#3149.

This branch re-enables vectored writes in Hyper for HTTP/1. Using
vectored writes in HTTP/2 will require an upstream change in the `h2`
crate as well.

I've removed the adaptive write buffer implementation
that attempts to detect whether vectored IO is or is not available,
since the Tokio 0.3.4 `AsyncWrite` trait exposes this directly via the
`is_write_vectored` method. Now, we just ask the IO whether or not it
supports vectored writes, and configure the buffer accordingly. This
makes the implementation somewhat simpler.

Signed-off-by: Eliza Weisman <eliza@buoyant.io>
  • Loading branch information
@hawkw
hawkw committed Nov 19, 2020
1 parent abb6471 commit 4ddae75
Show file tree
Hide file tree
Showing 2 changed files with 58 additions and 93 deletions.
2 changes: 1 addition & 1 deletion Cargo.toml
View file
Expand Up @@ -249,4 +249,4 @@ required-features = ["runtime", "stream"]
[[test]]
name = "server"
path = "tests/server.rs"
required-features = ["runtime"]
required-features = ["runtime"]
149 changes: 57 additions & 92 deletions src/proto/h1/io.rs
View file
@@ -1,4 +1,3 @@
use std::cell::Cell;
use std::cmp;
use std::fmt;
use std::io::{self, IoSlice};
Expand Down Expand Up @@ -56,13 +55,14 @@ where
B: Buf,
{
pub fn new(io: T) -> Buffered<T, B> {
let write_buf = WriteBuf::new(&io);
Buffered {
flush_pipeline: false,
io,
read_blocked: false,
read_buf: BytesMut::with_capacity(0),
read_buf_strategy: ReadStrategy::default(),
write_buf: WriteBuf::new(),
write_buf,
}
}

Expand Down Expand Up @@ -233,13 +233,36 @@ where
if let WriteStrategy::Flatten = self.write_buf.strategy {
return self.poll_flush_flattened(cx);
}

// The `IoSlice` type can't have a zero-length size, so create
// a dummy version from a 1-length slice which we'll overwrite with
// the `bytes_vectored` method.
static S: &[u8] = &[0];

loop {
// TODO(eliza): this basically ignores all of `WriteBuf`...put
// back vectored IO and `poll_write_buf` when the appropriate Tokio
// changes land...
let n = ready!(Pin::new(&mut self.io)
// .poll_write_buf(cx, &mut self.write_buf.auto()))?;
.poll_write(cx, self.write_buf.auto().bytes()))?;
debug_assert!(self.io.is_write_vectored(), "using vectored writes on an IO that does not provide fast vectored write support, this is a bug");
let n = {
let mut iovs: [IoSlice<'_>; MAX_BUF_LIST_BUFFERS] = [
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
IoSlice::new(S),
];
let len = self.write_buf.bytes_vectored(&mut iovs);
ready!(Pin::new(&mut self.io).poll_write_vectored(cx, &iovs[..len]))?
};
// TODO(eliza): we have to do this manually because
// `poll_write_buf` doesn't exist in Tokio 0.3 yet...when
// `poll_write_buf` comes back, the manual advance will need to leave!
Expand Down Expand Up @@ -458,12 +481,17 @@ pub(super) struct WriteBuf<B> {
}

impl<B: Buf> WriteBuf<B> {
fn new() -> WriteBuf<B> {
fn new(io: &impl AsyncWrite) -> WriteBuf<B> {
let strategy = if io.is_write_vectored() {
WriteStrategy::Queue
} else {
WriteStrategy::Flatten
};
WriteBuf {
headers: Cursor::new(Vec::with_capacity(INIT_BUFFER_SIZE)),
max_buf_size: DEFAULT_MAX_BUFFER_SIZE,
queue: BufList::new(),
strategy: WriteStrategy::Auto,
strategy,
}
}
}
Expand All @@ -476,12 +504,6 @@ where
self.strategy = strategy;
}

// TODO(eliza): put back writev!
#[inline]
fn auto(&mut self) -> WriteBufAuto<'_, B> {
WriteBufAuto::new(self)
}

pub(super) fn buffer<BB: Buf + Into<B>>(&mut self, mut buf: BB) {
debug_assert!(buf.has_remaining());
match self.strategy {
Expand All @@ -501,7 +523,7 @@ where
buf.advance(adv);
}
}
WriteStrategy::Auto | WriteStrategy::Queue => {
WriteStrategy::Queue => {
self.queue.push(buf.into());
}
}
Expand All @@ -510,7 +532,7 @@ where
fn can_buffer(&self) -> bool {
match self.strategy {
WriteStrategy::Flatten => self.remaining() < self.max_buf_size,
WriteStrategy::Auto | WriteStrategy::Queue => {
WriteStrategy::Queue => {
self.queue.bufs_cnt() < MAX_BUF_LIST_BUFFERS && self.remaining() < self.max_buf_size
}
}
Expand Down Expand Up @@ -569,65 +591,8 @@ impl<B: Buf> Buf for WriteBuf<B> {
}
}

/// Detects when wrapped `WriteBuf` is used for vectored IO, and
/// adjusts the `WriteBuf` strategy if not.
struct WriteBufAuto<'a, B: Buf> {
bytes_called: Cell<bool>,
bytes_vec_called: Cell<bool>,
inner: &'a mut WriteBuf<B>,
}

impl<'a, B: Buf> WriteBufAuto<'a, B> {
fn new(inner: &'a mut WriteBuf<B>) -> WriteBufAuto<'a, B> {
WriteBufAuto {
bytes_called: Cell::new(false),
bytes_vec_called: Cell::new(false),
inner,
}
}
}

impl<'a, B: Buf> Buf for WriteBufAuto<'a, B> {
#[inline]
fn remaining(&self) -> usize {
self.inner.remaining()
}

#[inline]
fn bytes(&self) -> &[u8] {
self.bytes_called.set(true);
self.inner.bytes()
}

#[inline]
fn advance(&mut self, cnt: usize) {
self.inner.advance(cnt)
}

#[inline]
fn bytes_vectored<'t>(&'t self, dst: &mut [IoSlice<'t>]) -> usize {
self.bytes_vec_called.set(true);
self.inner.bytes_vectored(dst)
}
}

impl<'a, B: Buf + 'a> Drop for WriteBufAuto<'a, B> {
fn drop(&mut self) {
if let WriteStrategy::Auto = self.inner.strategy {
if self.bytes_vec_called.get() {
self.inner.strategy = WriteStrategy::Queue;
} else if self.bytes_called.get() {
trace!("detected no usage of vectored write, flattening");
self.inner.strategy = WriteStrategy::Flatten;
self.inner.headers.bytes.put(&mut self.inner.queue);
}
}
}
}

#[derive(Debug)]
enum WriteStrategy {
Auto,
Flatten,
Queue,
}
Expand All @@ -639,8 +604,8 @@ mod tests {

use tokio_test::io::Builder as Mock;

#[cfg(feature = "nightly")]
use test::Bencher;
// #[cfg(feature = "nightly")]
// use test::Bencher;

/*
impl<T: Read> MemRead for AsyncIo<T> {
Expand Down Expand Up @@ -924,19 +889,19 @@ mod tests {
assert_eq!(buffered.write_buf.queue.bufs_cnt(), 0);
}

#[cfg(feature = "nightly")]
#[bench]
fn bench_write_buf_flatten_buffer_chunk(b: &mut Bencher) {
let s = "Hello, World!";
b.bytes = s.len() as u64;

let mut write_buf = WriteBuf::<bytes::Bytes>::new();
write_buf.set_strategy(WriteStrategy::Flatten);
b.iter(|| {
let chunk = bytes::Bytes::from(s);
write_buf.buffer(chunk);
::test::black_box(&write_buf);
write_buf.headers.bytes.clear();
})
}
// #[cfg(feature = "nightly")]
// #[bench]
// fn bench_write_buf_flatten_buffer_chunk(b: &mut Bencher) {
// let s = "Hello, World!";
// b.bytes = s.len() as u64;

// let mut write_buf = WriteBuf::<bytes::Bytes>::new();
// write_buf.set_strategy(WriteStrategy::Flatten);
// b.iter(|| {
// let chunk = bytes::Bytes::from(s);
// write_buf.buffer(chunk);
// ::test::black_box(&write_buf);
// write_buf.headers.bytes.clear();
// })
// }
}

0 comments on commit 4ddae75

Please sign in to comment.