Fix benchmark compilation

.editorconfig

@@ -7,3 +7,6 @@ insert_final_newline = true
 
 [*.{yaml,yml}]
 indent_size = 2
+
+[Makefile]
+indent_style = tab

Makefile

@@ -0,0 +1,12 @@
+.PHONY: build
+build:
+	cargo build
+
+.PHONY: test
+test:
+	cargo test
+
+.PHONY: bench
+bench:
+	cargo bench -p isahc-benchmarks
+

benchmarks/Cargo.toml

@@ -4,16 +4,16 @@ version = "0.0.0"
 edition = "2018"
 publish = false
 
-[dependencies.isahc]
-path = ".."
-
-[dev-dependencies]
+[dependencies]
 criterion = "0.2"
-env_logger = "0.6"
+curl = "0.4"
 rayon = "1"
 reqwest = "0.9"
 rouille = "3"
 
+[dependencies.isahc]
+path = ".."
+
 [[bench]]
 name = "download"
 harness = false

benchmarks/benches/download.rs

@@ -1,13 +1,14 @@
 //! Benchmark for downloading files over localhost.
 
 use criterion::*;
-use utilities::server;
+use isahc_benchmarks::TestServer;
+use std::io::{Write, sink};
 
 static DATA: [u8; 0x10000] = [1; 0x10000]; // 64K
 
 fn benchmark(c: &mut Criterion) {
     c.bench_function("download 64K: curl", move |b| {
-        let server = server::spawn(|_| server::static_response(&DATA));
+        let server = TestServer::static_response(&DATA);
         let endpoint = server.endpoint();
 
         b.iter_batched(
@@ -17,12 +18,12 @@ fn benchmark(c: &mut Criterion) {
                 easy
             },
             |mut easy| {
-                let mut body = Vec::new();
+                let mut sink = sink();
                 let mut transfer = easy.transfer();
 
                 transfer
                     .write_function(|bytes| {
-                        body.extend_from_slice(bytes);
+                        sink.write_all(bytes).unwrap();
                         Ok(bytes.len())
                     })
                     .unwrap();
@@ -34,37 +35,35 @@ fn benchmark(c: &mut Criterion) {
     });
 
     c.bench_function("download 64K: isahc", move |b| {
-        use std::io::Read;
+        use isahc::prelude::*;
 
-        let server = server::spawn(|_| server::static_response(&DATA));
+        let server = TestServer::static_response(&DATA);
         let endpoint = server.endpoint();
 
         b.iter_batched(
-            || isahc::Client::new(),
+            || isahc::HttpClient::new().unwrap(),
             |client| {
-                let mut body = Vec::new();
-
-                let mut response = client.get(&endpoint).unwrap();
-                response.body_mut().read_to_end(&mut body).unwrap();
+                client.get(&endpoint)
+                    .unwrap()
+                    .copy_to(sink())
+                    .unwrap();
             },
             BatchSize::SmallInput,
         )
     });
 
     c.bench_function("download 64K: reqwest", move |b| {
-        let server = server::spawn(|_| server::static_response(&DATA));
+        let server = TestServer::static_response(&DATA);
         let endpoint = server.endpoint();
 
         b.iter_batched(
             || reqwest::Client::new(),
             |client| {
-                let mut body = Vec::new();
-
                 client
                     .get(&endpoint)
                     .send()
                     .unwrap()
-                    .copy_to(&mut body)
+                    .copy_to(&mut sink())
                     .unwrap();
             },
             BatchSize::SmallInput,

benchmarks/src/lib.rs

@@ -0,0 +1,55 @@
+use rouille::{Request, Response};
+use std::net::SocketAddr;
+use std::thread;
+use std::sync::Arc;
+
+pub struct TestServer {
+    addr: SocketAddr,
+    counter: Option<Arc<()>>,
+    handle: Option<thread::JoinHandle<()>>,
+}
+
+impl TestServer {
+    pub fn static_response(body: &'static [u8]) -> Self {
+        Self::new(move |_| {
+            use std::io::Cursor;
+
+            rouille::Response {
+                status_code: 200,
+                headers: vec![],
+                data: rouille::ResponseBody::from_reader(Cursor::new(body)),
+                upgrade: None,
+            }
+        })
+    }
+
+    pub fn new(handler: impl Send + Sync + 'static + Fn(&Request) -> Response) -> Self {
+        let server = rouille::Server::new("localhost:0", handler).unwrap();
+        let addr = server.server_addr();
+
+        let counter_outer = Arc::new(());
+        let counter_inner = counter_outer.clone();
+        let handle = thread::spawn(move || {
+            while Arc::strong_count(&counter_inner) > 1 {
+                server.poll();
+            }
+        });
+
+        Self {
+            addr: addr,
+            counter: Some(counter_outer),
+            handle: Some(handle),
+        }
+    }
+
+    pub fn endpoint(&self) -> String {
+        format!("http://{}", self.addr)
+    }
+}
+
+impl Drop for TestServer {
+    fn drop(&mut self) {
+        self.counter.take();
+        self.handle.take().unwrap().join().unwrap();
+    }
+}

src/agent.rs

@@ -11,6 +11,7 @@ use crate::handler::RequestHandler;
 use crate::task::{UdpWaker, WakerExt};
 use crate::Error;
 use crossbeam_channel::{Receiver, Sender};
+use crossbeam_utils::sync::WaitGroup;
 use curl::multi::WaitFd;
 use futures_util::task::ArcWake;
 use slab::Slab;
@@ -107,7 +108,10 @@ pub(crate) fn new() -> Result<Handle, Error> {
 
     let (message_tx, message_rx) = crossbeam_channel::unbounded();
 
-    Ok(Handle {
+    let wait_group = WaitGroup::new();
+    let wait_group_thread = wait_group.clone();
+
+    let handle = Handle {
         message_tx: message_tx.clone(),
         waker: waker.clone(),
         join_handle: Some(
@@ -125,12 +129,17 @@ pub(crate) fn new() -> Result<Handle, Error> {
                         waker,
                     };
 
+                    drop(wait_group_thread);
                     log::debug!("agent took {:?} to start up", create_start.elapsed());
 
                     agent.run()
                 })?,
         ),
-    })
+    };
+
+    wait_group.wait();
+
+    Ok(handle)
 }
 
 impl Handle {