diff --git a/benches/server.rs b/benches/server.rs index 26f5572d..481493bf 100644 --- a/benches/server.rs +++ b/benches/server.rs @@ -9,20 +9,23 @@ use std::io::{Read, Write}; use std::net::{TcpListener, TcpStream}; use std::sync::mpsc; -use futures::{future, Future}; +use futures::{future, stream, Future}; use futures::sync::oneshot; use hyper::header::{ContentLength, ContentType, TransferEncoding}; use hyper::server::{self, Service}; macro_rules! bench_server { - ($b:ident, $handler:ident, $total_bytes:expr) => { + ($b:ident, $header:expr, $body:expr) => ({ let (_until_tx, until_rx) = oneshot::channel(); let addr = { let (addr_tx, addr_rx) = mpsc::channel(); ::std::thread::spawn(move || { let addr = "127.0.0.1:0".parse().unwrap(); - let srv = hyper::server::Http::new().bind(&addr, || Ok($handler)).unwrap(); + let srv = hyper::server::Http::new().bind(&addr, || Ok(BenchPayload { + header: $header, + body: $body, + })).unwrap(); let addr = srv.local_addr().unwrap(); addr_tx.send(addr).unwrap(); srv.run_until(until_rx.map_err(|_| ())).unwrap(); @@ -31,39 +34,71 @@ macro_rules! bench_server { addr_rx.recv().unwrap() }; + let total_bytes = { + let mut tcp = TcpStream::connect(addr).unwrap(); + tcp.write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n").unwrap(); + let mut buf = Vec::new(); + tcp.read_to_end(&mut buf).unwrap() + }; + let mut tcp = TcpStream::connect(addr).unwrap(); + tcp.set_read_timeout(Some(::std::time::Duration::from_secs(3))).unwrap(); let mut buf = [0u8; 8192]; - $b.bytes = 35 + $total_bytes; + $b.bytes = 35 + total_bytes as u64; $b.iter(|| { tcp.write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n").unwrap(); let mut sum = 0; - while sum < $total_bytes { + while sum < total_bytes { sum += tcp.read(&mut buf).unwrap(); } - assert_eq!(sum, $total_bytes); + assert_eq!(sum, total_bytes); }) - } + }) } - -#[bench] -fn bench_server_tcp_throughput_fixedsize_small_payload(b: &mut test::Bencher) { - bench_server!(b, FixedSizeSmallPayload, 130); +fn body(b: &'static [u8]) -> hyper::Body { + b.into() } #[bench] -fn bench_server_tcp_throughput_chunked_small_payload(b: &mut test::Bencher) { - bench_server!(b, ChunkedSmallPayload, 148); +fn throughput_fixedsize_small_payload(b: &mut test::Bencher) { + bench_server!(b, ContentLength(13), || body(b"Hello, World!")) } #[bench] -fn bench_server_tcp_throughput_chunked_large_payload(b: &mut test::Bencher) { - bench_server!(b, ChunkedLargePayload, 13_000_140); +fn throughput_fixedsize_large_payload(b: &mut test::Bencher) { + bench_server!(b, ContentLength(1_000_000), || body(&[b'x'; 1_000_000])) } #[bench] -fn bench_raw_tcp_throughput_small_payload(b: &mut test::Bencher) { +fn throughput_fixedsize_many_chunks(b: &mut test::Bencher) { + bench_server!(b, ContentLength(1_000_000), || { + static S: &'static [&'static [u8]] = &[&[b'x'; 1_000] as &[u8]; 1_000] as _; + stream::iter_ok(S.iter()) + }) +} + +#[bench] +fn throughput_chunked_small_payload(b: &mut test::Bencher) { + bench_server!(b, TransferEncoding::chunked(), || body(b"Hello, World!")) +} + +#[bench] +fn throughput_chunked_large_payload(b: &mut test::Bencher) { + bench_server!(b, TransferEncoding::chunked(), || body(&[b'x'; 1_000_000])) +} + +#[bench] +fn throughput_chunked_many_chunks(b: &mut test::Bencher) { + bench_server!(b, TransferEncoding::chunked(), || { + static S: &'static [&'static [u8]] = &[&[b'x'; 1_000] as &[u8]; 1_000] as _; + stream::iter_ok(S.iter()) + }) +} + +#[bench] +fn raw_tcp_throughput_small_payload(b: &mut test::Bencher) { let (tx, rx) = mpsc::channel(); let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let addr = listener.local_addr().unwrap(); @@ -96,54 +131,70 @@ fn bench_raw_tcp_throughput_small_payload(b: &mut test::Bencher) { tx.send(()).unwrap(); } -struct FixedSizeSmallPayload; +#[bench] +fn raw_tcp_throughput_large_payload(b: &mut test::Bencher) { + let (tx, rx) = mpsc::channel(); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); -impl Service for FixedSizeSmallPayload { + let srv_head = b"\ + HTTP/1.1 200 OK\r\n\ + Content-Length: 1000000\r\n\ + Content-Type: text/plain; charset=utf-8\r\n\ + Date: Fri, 12 May 2017 18:21:45 GMT\r\n\ + \r\n\ + "; + ::std::thread::spawn(move || { + let mut sock = listener.accept().unwrap().0; + + let mut buf = [0u8; 8192]; + while rx.try_recv().is_err() { + let r = sock.read(&mut buf).unwrap(); + if r == 0 { + break; + } + sock.write_all(srv_head).unwrap(); + sock.write_all(&[b'x'; 1_000_000]).unwrap(); + } + }); + + let mut tcp = TcpStream::connect(addr).unwrap(); + let mut buf = [0u8; 8192]; + + let expect_read = srv_head.len() + 1_000_000; + b.bytes = expect_read as u64 + 35; + + b.iter(|| { + tcp.write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n").unwrap(); + let mut sum = 0; + while sum < expect_read { + sum += tcp.read(&mut buf).unwrap(); + } + assert_eq!(sum, expect_read); + }); + tx.send(()).unwrap(); +} + +struct BenchPayload { + header: H, + body: F, +} + +impl Service for BenchPayload +where + H: hyper::header::Header + Clone, + F: Fn() -> B, +{ type Request = server::Request; - type Response = server::Response; + type Response = server::Response; type Error = hyper::Error; type Future = future::FutureResult; fn call(&self, _req: Self::Request) -> Self::Future { future::ok( server::Response::new() - .with_header(ContentLength("Hello, World!".len() as u64)) + .with_header(self.header.clone()) .with_header(ContentType::plaintext()) - .with_body("Hello, World!") + .with_body((self.body)()) ) } } - -struct ChunkedSmallPayload; - -impl Service for ChunkedSmallPayload { - type Request = server::Request; - type Response = server::Response; - type Error = hyper::Error; - type Future = future::FutureResult; - fn call(&self, _req: Self::Request) -> Self::Future { - future::ok( - server::Response::new() - .with_header(TransferEncoding::chunked()) - .with_header(ContentType::plaintext()) - .with_body("Hello, World!") - ) - } -} - -struct ChunkedLargePayload; - -impl Service for ChunkedLargePayload { - type Request = server::Request; - type Response = server::Response; - type Error = hyper::Error; - type Future = future::FutureResult; - fn call(&self, _req: Self::Request) -> Self::Future { - future::ok( - server::Response::new() - .with_header(TransferEncoding::chunked()) - .with_header(ContentType::plaintext()) - .with_body((0..1_000_000).flat_map(|_| "Hello, World!".bytes()).collect::>()) - ) - } -} -