test(benches): add http1 pipeline bench target

Cargo.toml

@@ -143,6 +143,11 @@ name = "end_to_end"
 path = "benches/end_to_end.rs"
 required-features = ["runtime"]
 
+[[bench]]
+name = "pipeline"
+path = "benches/pipeline.rs"
+required-features = ["runtime"]
+
 [[bench]]
 name = "server"
 path = "benches/server.rs"

benches/pipeline.rs

@@ -0,0 +1,76 @@
+#![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);
+    });
+}
+