From fded1f31547ec9a23c981d84670634b0ea145644 Mon Sep 17 00:00:00 2001
From: Dietmar Maurer <dietmar@proxmox.com>
Date: Sat, 29 Jun 2019 10:08:12 +0200
Subject: [PATCH] add code to test H2 speed

I currently get about 1MB/s, which is ways too slow (must be a bug)?
---
 src/bin/h2client.rs | 106 ++++++++++++++++++++++++++++++++++++++++++++
 src/bin/h2server.rs |  65 +++++++++++++++++++++++++++
 2 files changed, 171 insertions(+)
 create mode 100644 src/bin/h2client.rs
 create mode 100644 src/bin/h2server.rs

diff --git a/src/bin/h2client.rs b/src/bin/h2client.rs
new file mode 100644
index 00000000..ca129cc3
--- /dev/null
+++ b/src/bin/h2client.rs
@@ -0,0 +1,106 @@
+use failure::*;
+use futures::*;
+
+// Simple H2 client to test H2 download speed using h2server.rs
+
+use tokio::net::TcpStream;
+
+struct Process {
+    body: h2::RecvStream,
+    trailers: bool,
+    bytes: usize,
+}
+
+impl Future for Process {
+    type Item = usize;
+    type Error = Error;
+
+    fn poll(&mut self) -> Poll<usize, Error> {
+        loop {
+            if self.trailers {
+                let trailers = try_ready!(self.body.poll_trailers());
+                if let Some(trailers) = trailers {
+                    println!("trailers: {:?}", trailers);
+                }
+                println!("Received {} bytes", self.bytes);
+
+                return Ok(Async::Ready(self.bytes));
+            } else {
+                match try_ready!(self.body.poll()) {
+                    Some(chunk) => {
+                        self.body.release_capacity().release_capacity(chunk.len())?;
+                        self.bytes += chunk.len();
+                        // println!("GOT FRAME {}", chunk.len());
+                    },
+                    None => {
+                        self.trailers = true;
+                    },
+                }
+            }
+        }
+    }
+}
+
+fn send_request(mut client: h2::client::SendRequest<bytes::Bytes>) -> impl Future<Item=usize, Error=Error> {
+
+    println!("sending request");
+
+    let request = http::Request::builder()
+        .uri("http://localhost/")
+        .body(())
+        .unwrap();
+
+    let (response, _stream) = client.send_request(request, true).unwrap();
+
+    response
+        .map_err(Error::from)
+        .and_then(|response| {
+            Process { body: response.into_body(), trailers: false, bytes: 0 }
+        })
+}
+
+pub fn main() -> Result<(), Error> {
+
+    let tcp_stream = TcpStream::connect(&"127.0.0.1:8008".parse().unwrap());
+
+    let start = std::time::SystemTime::now();
+
+    let tcp = tcp_stream
+        .map_err(Error::from)
+        .and_then(|c| {
+            h2::client::handshake(c)
+                .map_err(Error::from)
+        })
+        .and_then(|(client, h2)| {
+
+            // Spawn a task to run the conn...
+            tokio::spawn(h2.map_err(|e| println!("GOT ERR={:?}", e)));
+
+            futures::stream::repeat(())
+                .take(10)
+                .and_then(move |_| send_request(client.clone()))
+                .fold(0, move |mut acc, size| {
+                    acc += size;
+                    Ok::<_, Error>(acc)
+                })
+        })
+        .then(move |result| {
+            match result {
+                Err(err) => {
+                    println!("ERROR {}", err);
+                }
+                Ok(bytes) => {
+                    let elapsed = start.elapsed().unwrap();
+                    let elapsed = (elapsed.as_secs() as f64) +
+                        (elapsed.subsec_millis() as f64)/1000.0;
+
+                    println!("Downloaded {} bytes, {} MB/s", bytes, (bytes as f64)/(elapsed*1024.0*1024.0));
+                }
+            }
+            Ok(())
+        });
+
+    tokio::run(tcp);
+
+    Ok(())
+}
diff --git a/src/bin/h2server.rs b/src/bin/h2server.rs
new file mode 100644
index 00000000..dc8f2729
--- /dev/null
+++ b/src/bin/h2server.rs
@@ -0,0 +1,65 @@
+use failure::*;
+use futures::*;
+
+// Simple H2 server to test H2 speed with h2client.rs
+
+use tokio::net::TcpListener;
+
+use proxmox_backup::client::pipe_to_stream::*;
+
+pub fn main() -> Result<(), Error> {
+
+    start_h2_server()?;
+
+    Ok(())
+}
+
+pub fn start_h2_server() -> Result<(), Error> {
+
+    let listener = TcpListener::bind(&"127.0.0.1:8008".parse().unwrap()).unwrap();
+
+    println!("listening on {:?}", listener.local_addr());
+
+    let server = listener.incoming().for_each(move |socket| {
+
+        let connection = h2::server::handshake(socket)
+            .map_err(Error::from)
+            .and_then(|conn| {
+                println!("H2 connection bound");
+
+                conn
+                    .map_err(Error::from)
+                    .for_each(|(request, mut respond)| {
+                        println!("GOT request: {:?}", request);
+
+                        let response = http::Response::builder().status(http::StatusCode::OK).body(()).unwrap();
+
+                        let send = respond.send_response(response, false).unwrap();
+                        let data = vec![65u8; 1024*1024];
+                        PipeToSendStream::new(bytes::Bytes::from(data), send)
+                            .and_then(|_| {
+                                println!("DATA SENT");
+                                Ok(())
+                            })
+                    })
+            })
+            .and_then(|_| {
+                println!("H2 connection CLOSE !");
+                Ok(())
+            })
+            .then(|res| {
+                if let Err(e) = res {
+                    println!("  -> err={:?}", e);
+                }
+                Ok(())
+            });
+
+        tokio::spawn(Box::new(connection));
+        Ok(())
+    })
+    .map_err(|e| eprintln!("accept error: {}", e));
+
+    tokio::run(server);
+
+    Ok(())
+}