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

chore(lib): individually disable tests and examples that aren't updated

Browse Source
This commit is contained in:
Sean McArthur
2019-07-12 13:29:57 -07:00
parent f93463b3d9
commit 5da17df97f
29 changed files with 120 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

258
benches/end_to_end.rs
View File

@@ -1,258 +0,0 @@
#![feature(test)]
#![deny(warnings)]
extern crate futures;
extern crate hyper;
extern crate pretty_env_logger;
extern crate test;
extern crate tokio;
use std::net::SocketAddr;
use futures::{Future, Stream};
use tokio::runtime::current_thread::Runtime;
use hyper::{Body, Method, Request, Response, Server};
use hyper::client::HttpConnector;
#[bench]
fn http1_get(b: &mut test::Bencher) {
opts()
.bench(b)
}
#[bench]
fn http1_post(b: &mut test::Bencher) {
opts()
.method(Method::POST)
.request_body(b"foo bar baz quux")
.bench(b)
}
#[bench]
fn http1_body_both_100kb(b: &mut test::Bencher) {
let body = &[b'x'; 1024 * 100];
opts()
.method(Method::POST)
.request_body(body)
.response_body(body)
.bench(b)
}
#[bench]
fn http1_body_both_10mb(b: &mut test::Bencher) {
let body = &[b'x'; 1024 * 1024 * 10];
opts()
.method(Method::POST)
.request_body(body)
.response_body(body)
.bench(b)
}
#[bench]
fn http1_parallel_x10_empty(b: &mut test::Bencher) {
opts()
.parallel(10)
.bench(b)
}
#[bench]
fn http1_parallel_x10_req_10mb(b: &mut test::Bencher) {
let body = &[b'x'; 1024 * 1024 * 10];
opts()
.parallel(10)
.method(Method::POST)
.request_body(body)
.bench(b)
}
#[bench]
fn http2_get(b: &mut test::Bencher) {
opts()
.http2()
.bench(b)
}
#[bench]
fn http2_post(b: &mut test::Bencher) {
opts()
.http2()
.method(Method::POST)
.request_body(b"foo bar baz quux")
.bench(b)
}
#[bench]
fn http2_req_100kb(b: &mut test::Bencher) {
let body = &[b'x'; 1024 * 100];
opts()
.http2()
.method(Method::POST)
.request_body(body)
.bench(b)
}
#[bench]
fn http2_parallel_x10_empty(b: &mut test::Bencher) {
opts()
.http2()
.parallel(10)
.bench(b)
}
#[bench]
fn http2_parallel_x10_req_10mb(b: &mut test::Bencher) {
let body = &[b'x'; 1024 * 1024 * 10];
opts()
.http2()
.parallel(10)
.method(Method::POST)
.request_body(body)
.http2_stream_window(std::u32::MAX >> 1)
.http2_conn_window(std::u32::MAX >> 1)
.bench(b)
}
// ==== Benchmark Options =====
struct Opts {
http2: bool,
http2_stream_window: Option<u32>,
http2_conn_window: Option<u32>,
parallel_cnt: u32,
request_method: Method,
request_body: Option<&'static [u8]>,
response_body: &'static [u8],
}
fn opts() -> Opts {
Opts {
http2: false,
http2_stream_window: None,
http2_conn_window: None,
parallel_cnt: 1,
request_method: Method::GET,
request_body: None,
response_body: b"Hello",
}
}
impl Opts {
fn http2(mut self) -> Self {
self.http2 = true;
self
}
fn http2_stream_window(mut self, sz: impl Into<Option<u32>>) -> Self {
self.http2_stream_window = sz.into();
self
}
fn http2_conn_window(mut self, sz: impl Into<Option<u32>>) -> Self {
self.http2_conn_window = sz.into();
self
}
fn method(mut self, m: Method) -> Self {
self.request_method = m;
self
}
fn request_body(mut self, body: &'static [u8]) -> Self {
self.request_body = Some(body);
self
}
fn response_body(mut self, body: &'static [u8]) -> Self {
self.response_body = body;
self
}
fn parallel(mut self, cnt: u32) -> Self {
assert!(cnt > 0, "parallel count must be larger than 0");
self.parallel_cnt = cnt;
self
}
fn bench(self, b: &mut test::Bencher) {
let _ = pretty_env_logger::try_init();
let mut rt = Runtime::new().unwrap();
b.bytes = self.response_body.len() as u64 + self.request_body.map(|b| b.len()).unwrap_or(0) as u64;
let addr = spawn_hello(&mut rt, &self);
let connector = HttpConnector::new(1);
let client = hyper::Client::builder()
.http2_only(self.http2)
.http2_initial_stream_window_size(self.http2_stream_window)
.http2_initial_connection_window_size(self.http2_conn_window)
.build::<_, Body>(connector);
let url: hyper::Uri = format!("http://{}/hello", addr).parse().unwrap();
let make_request = || {
let body = self
.request_body
.map(Body::from)
.unwrap_or_else(|| Body::empty());
let mut req = Request::new(body);
*req.method_mut() = self.request_method.clone();
req
};
if self.parallel_cnt == 1 {
b.iter(move || {
let mut req = make_request();
*req.uri_mut() = url.clone();
rt.block_on(client.request(req).and_then(|res| {
res.into_body().for_each(|_chunk| {
Ok(())
})
})).expect("client wait");
});
} else {
b.iter(|| {
let futs = (0..self.parallel_cnt)
.into_iter()
.map(|_| {
let mut req = make_request();
*req.uri_mut() = url.clone();
client.request(req).and_then(|res| {
res.into_body().for_each(|_chunk| {
Ok(())
})
}).map_err(|e| panic!("client error: {}", e))
});
let _ = rt.block_on(::futures::future::join_all(futs));
});
}
}
}
fn spawn_hello(rt: &mut Runtime, opts: &Opts) -> SocketAddr {
use hyper::service::{service_fn};
let addr = "127.0.0.1:0".parse().unwrap();
let body = opts.response_body;
let srv = Server::bind(&addr)
.http2_only(opts.http2)
.http2_initial_stream_window_size_(opts.http2_stream_window)
.http2_initial_connection_window_size_(opts.http2_conn_window)
.serve(move || {
service_fn(move |req: Request<Body>| {
req
.into_body()
.for_each(|_chunk| {
Ok(())
})
.map(move |_| {
Response::new(Body::from(body))
})
})
});
let addr = srv.local_addr();
let fut = srv.map_err(|err| panic!("server error: {}", err));
rt.spawn(fut);
return addr
}

76
benches/pipeline.rs
View File

@@ -1,76 +0,0 @@
#![feature(test)]
#![deny(warnings)]
extern crate futures;
extern crate hyper;
extern crate pretty_env_logger;
extern crate test;
extern crate tokio;
use std::io::{Read, Write};
use std::net::{TcpStream};
use std::sync::mpsc;
use futures::Future;
use futures::sync::oneshot;
use hyper::{Body, Response, Server};
use hyper::service::service_fn_ok;
const PIPELINED_REQUESTS: usize = 16;
#[bench]
fn hello_world(b: &mut test::Bencher) {
let _ = pretty_env_logger::try_init();
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 = Server::bind(&addr)
.http1_pipeline_flush(true)
.serve(|| {
service_fn_ok(move |_| {
Response::new(Body::from("Hello, World!"))
})
});
addr_tx.send(srv.local_addr()).unwrap();
let fut = srv
.map_err(|e| panic!("server error: {}", e))
.select(until_rx.then(|_| Ok(())))
.then(|_| Ok(()));
let mut rt = tokio::runtime::current_thread::Runtime::new().unwrap();
rt.spawn(fut);
rt.run().unwrap();
});
addr_rx.recv().unwrap()
};
let mut pipelined_reqs = Vec::new();
for _ in 0..PIPELINED_REQUESTS {
pipelined_reqs.extend_from_slice(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n");
}
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()
} * PIPELINED_REQUESTS;
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 = (pipelined_reqs.len() + total_bytes) as u64;
b.iter(|| {
tcp.write_all(&pipelined_reqs).unwrap();
let mut sum = 0;
while sum < total_bytes {
sum += tcp.read(&mut buf).unwrap();
}
assert_eq!(sum, total_bytes);
});
}

193
benches/server.rs
View File

@@ -1,193 +0,0 @@
#![feature(test)]
#![deny(warnings)]
extern crate futures;
extern crate hyper;
extern crate pretty_env_logger;
extern crate test;
extern crate tokio;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::mpsc;
use futures::{stream, Future, Stream};
use futures::sync::oneshot;
use hyper::{Body, Response, Server};
use hyper::service::service_fn_ok;
macro_rules! bench_server {
($b:ident, $header:expr, $body:expr) => ({
let _ = pretty_env_logger::try_init();
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 = Server::bind(&addr)
.serve(|| {
let header = $header;
let body = $body;
service_fn_ok(move |_| {
Response::builder()
.header(header.0, header.1)
.header("content-type", "text/plain")
.body(body())
.unwrap()
})
});
addr_tx.send(srv.local_addr()).unwrap();
let fut = srv
.map_err(|e| panic!("server error: {}", e))
.select(until_rx.then(|_| Ok(())))
.then(|_| Ok(()));
let mut rt = tokio::runtime::current_thread::Runtime::new().unwrap();
rt.spawn(fut);
rt.run().unwrap();
});
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 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 {
sum += tcp.read(&mut buf).unwrap();
}
assert_eq!(sum, total_bytes);
});
})
}
fn body(b: &'static [u8]) -> hyper::Body {
b.into()
}
#[bench]
fn throughput_fixedsize_small_payload(b: &mut test::Bencher) {
bench_server!(b, ("content-length", "13"), || body(b"Hello, World!"))
}
#[bench]
fn throughput_fixedsize_large_payload(b: &mut test::Bencher) {
bench_server!(b, ("content-length", "1000000"), || body(&[b'x'; 1_000_000]))
}
#[bench]
fn throughput_fixedsize_many_chunks(b: &mut test::Bencher) {
bench_server!(b, ("content-length", "1000000"), || {
static S: &'static [&'static [u8]] = &[&[b'x'; 1_000] as &[u8]; 1_000] as _;
Body::wrap_stream(stream::iter_ok::<_, String>(S.iter()).map(|&s| s))
})
}
#[bench]
fn throughput_chunked_small_payload(b: &mut test::Bencher) {
bench_server!(b, ("transfer-encoding", "chunked"), || body(b"Hello, World!"))
}
#[bench]
fn throughput_chunked_large_payload(b: &mut test::Bencher) {
bench_server!(b, ("transfer-encoding", "chunked"), || body(&[b'x'; 1_000_000]))
}
#[bench]
fn throughput_chunked_many_chunks(b: &mut test::Bencher) {
bench_server!(b, ("transfer-encoding", "chunked"), || {
static S: &'static [&'static [u8]] = &[&[b'x'; 1_000] as &[u8]; 1_000] as _;
Body::wrap_stream(stream::iter_ok::<_, String>(S.iter()).map(|&s| s))
})
}
#[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();
::std::thread::spawn(move || {
let mut sock = listener.accept().unwrap().0;
let mut buf = [0u8; 8192];
while rx.try_recv().is_err() {
sock.read(&mut buf).unwrap();
sock.write_all(b"\
HTTP/1.1 200 OK\r\n\
Content-Length: 13\r\n\
Content-Type: text/plain; charset=utf-8\r\n\
Date: Fri, 12 May 2017 18:21:45 GMT\r\n\
\r\n\
Hello, World!\
").unwrap();
}
});
let mut tcp = TcpStream::connect(addr).unwrap();
let mut buf = [0u8; 4096];
b.bytes = 130 + 35;
b.iter(|| {
tcp.write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n").unwrap();
let n = tcp.read(&mut buf).unwrap();
assert_eq!(n, 130);
});
tx.send(()).unwrap();
}
#[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();
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();
}