Merge pull request stratum-mining#558 from darricksee/scale-test

Scale test

Cargo.toml

@@ -34,6 +34,7 @@ members = [
     "roles/v2/pool",
     "roles/v2/test-utils/mining-device",
     "roles/translator",
+    "test/scale",
     "utils/network-helpers",
     "utils/buffer",
     "utils/error-handling",

test/scale/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "scale"
+version = "0.1.0"
+edition = "2021"
+
+[profile.release]
+lto = true
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+clap = "2.33.3"
+serde = { version = "1.0.89", features = ["derive", "alloc"], default-features = false, optional = true}
+async-channel = "1.5.1"
+async-std="1.8.0"
+bytes = "1.0.1"
+binary_sv2 = { path = "../../protocols/v2/binary-sv2/binary-sv2" }
+codec_sv2 = { path = "../../protocols/v2/codec-sv2", features=["noise_sv2"] }
+network_helpers = { version = "0.1", path = "../../utils/network-helpers", features = ["with_tokio"] }
+roles_logic_sv2 = { path = "../../protocols/v2/roles-logic-sv2" }
+tokio = { version = "1", features = ["full"] }
+

test/scale/README.md

@@ -0,0 +1,22 @@
+# Scale Test
+
+This test simply outputs the time spent sending 1,000,000 SubmitSharesStandard 
+through the system. When you start the test you specify -h <num of hops> -e (for encryption). 
+The test spawns <num of hops> "proxies" (ports 19000->19000+<num of hops>) which simply decrypt/encrypt each 
+SubmitSharesStandard message coming in (if encryption is on). Then it sends 
+1,000,000 share messages to the first proxy and then times the whole system to see 
+how long it takes for the last proxy to receive all 1M messages. It uses the same
+network_helpers that the pool, and proxies use so it should be a good approximation
+of the work they do. 
+
+The test is run with the following command:
+NOTE: running without `--release` dramatically slows down the test.
+
+```cargo run --release -- -h 4 -e```
+This runs the test with 4 hops and encryption on.
+
+```cargo run --release -- -h 4```
+This runs the test with 4 hops and encryption off.
+
+
+

test/scale/src/main.rs

@@ -0,0 +1,235 @@
+use std::thread;
+use tokio::{
+    net::{TcpListener, TcpStream},
+    task,
+};
+
+use async_channel::{bounded, Receiver, Sender};
+
+use binary_sv2::{Deserialize, GetSize, Serialize};
+use clap::{App, Arg};
+use codec_sv2::{HandshakeRole, Initiator, Responder, StandardEitherFrame, StandardSv2Frame};
+use std::time::Duration;
+
+use network_helpers::{
+    noise_connection_tokio::Connection, plain_connection_tokio::PlainConnection,
+};
+
+use roles_logic_sv2::{
+    mining_sv2::*,
+    parsers::{Mining, MiningDeviceMessages},
+};
+
+pub type EitherFrame = StandardEitherFrame<Message>;
+
+pub const AUTHORITY_PUBLIC_K: [u8; 32] = [
+    215, 11, 47, 78, 34, 232, 25, 192, 195, 168, 170, 209, 95, 181, 40, 114, 154, 226, 176, 190,
+    90, 169, 238, 89, 191, 183, 97, 63, 194, 119, 11, 31,
+];
+
+pub const AUTHORITY_PRIVATE_K: [u8; 32] = [
+    204, 93, 167, 220, 169, 204, 172, 35, 9, 84, 174, 208, 171, 89, 25, 53, 196, 209, 161, 148, 4,
+    5, 173, 0, 234, 59, 15, 127, 31, 160, 136, 131,
+];
+
+static HOST: &str = "127.0.0.1";
+
+#[tokio::main]
+async fn main() {
+    let matches = App::new("ScaleTest")
+        .arg(Arg::with_name("encrypt").short("e").help("Use encryption"))
+        .arg(
+            Arg::with_name("hops")
+                .short("h")
+                .takes_value(true)
+                .help("Number of hops"),
+        )
+        .get_matches();
+
+    let total_messages = 1_000_000;
+    let encrypt = matches.is_present("encrypt");
+    let hops: u16 = matches.value_of("hops").unwrap_or("0").parse().unwrap_or(0);
+    let mut orig_port: u16 = 19000;
+
+    // create channel to tell final server number of messages
+    let (tx, rx) = bounded(1);
+
+    if hops > 0 {
+        orig_port = spawn_proxies(encrypt, hops, tx, total_messages).await;
+    } else {
+        println!("Usage: ./program -h <hops> -e");
+    }
+    println!("Connecting to localhost:{}", orig_port);
+    setup_driver(orig_port, encrypt, rx, total_messages, hops).await;
+}
+
+async fn setup_driver(
+    server_port: u16,
+    encrypt: bool,
+    rx: Receiver<String>,
+    total_messages: i32,
+    hops: u16,
+) {
+    let server_stream = TcpStream::connect(format!("{}:{}", HOST, server_port))
+        .await
+        .unwrap();
+    let (_server_receiver, server_sender): (Receiver<EitherFrame>, Sender<EitherFrame>);
+
+    if encrypt {
+        let initiator = Initiator::from_raw_k(AUTHORITY_PUBLIC_K).unwrap();
+
+        (_server_receiver, server_sender) =
+            Connection::new(server_stream, HandshakeRole::Initiator(initiator)).await;
+    } else {
+        (_server_receiver, server_sender) = PlainConnection::new(server_stream).await;
+    }
+    // Create timer to see how long this method takes
+    let start = std::time::Instant::now();
+
+    send_messages(server_sender, total_messages).await;
+
+    //listen for message on rx
+    let msg = rx.recv().await.unwrap();
+
+    let end = std::time::Instant::now();
+
+    println!(
+        "client: {} - Took {}s hops: {} encryption: {}",
+        msg,
+        (end - start).as_secs(),
+        hops,
+        encrypt
+    );
+}
+
+pub type Message = MiningDeviceMessages<'static>;
+pub type StdFrame = StandardSv2Frame<Message>;
+
+async fn send_messages(stream: Sender<EitherFrame>, total_messages: i32) {
+    let mut number: i32 = 0;
+    println!("Creating share");
+    let share = MiningDeviceMessages::Mining(Mining::SubmitSharesStandard(SubmitSharesStandard {
+        channel_id: 1,
+        sequence_number: number as u32,
+        job_id: 2,
+        nonce: 3,
+        ntime: 4,
+        version: 5,
+    }));
+
+    while number <= total_messages {
+        println!("client: sending msg-{}", number);
+        let frame: StdFrame = share.clone().try_into().unwrap();
+        let binary: EitherFrame = frame.into();
+
+        stream.send(binary).await.unwrap();
+        number += 1;
+    }
+}
+
+async fn handle_messages<Mining: Serialize + Deserialize<'static> + GetSize + Send + 'static>(
+    name: String,
+    client: Receiver<EitherFrame>,
+    server: Option<Sender<EitherFrame>>,
+    total_messages: i32,
+    tx: Sender<String>,
+) {
+    let mut messages_received = 0;
+
+    while messages_received <= total_messages {
+        let frame: StdFrame = client.recv().await.unwrap().try_into().unwrap();
+
+        let binary: EitherFrame = frame.into();
+
+        if server.is_some() {
+            server.as_ref().unwrap().send(binary).await.unwrap();
+        } else {
+            messages_received += 1;
+            println!("last server: {} got msg {}", name, messages_received);
+        }
+    }
+    tx.send("got all messages".to_string()).await.unwrap();
+}
+
+async fn create_proxy(
+    name: String,
+    listen_port: u16,
+    server_port: u16,
+    encrypt: bool,
+    total_messages: i32,
+    tx: Sender<String>,
+) {
+    println!(
+        "Creating proxy listener {}: {} connecting to: {}",
+        name, listen_port, server_port
+    );
+    let listener = TcpListener::bind(format!("0.0.0.0:{}", listen_port))
+        .await
+        .unwrap();
+    println!("Bound - now waiting for connection...");
+    let cli_stream = listener.accept().await.unwrap().0;
+    let (cli_receiver, _cli_sender): (Receiver<EitherFrame>, Sender<EitherFrame>);
+
+    if encrypt {
+        let responder = Responder::from_authority_kp(
+            &AUTHORITY_PUBLIC_K[..],
+            &AUTHORITY_PRIVATE_K[..],
+            Duration::from_secs(3600),
+        )
+        .unwrap();
+        (cli_receiver, _cli_sender) =
+            Connection::new(cli_stream, HandshakeRole::Responder(responder)).await;
+    } else {
+        (cli_receiver, _cli_sender) = PlainConnection::new(cli_stream).await;
+    }
+
+    let mut server = None;
+    if server_port > 0 {
+        println!("Proxy {} Connecting to server: {}", name, server_port);
+        let server_stream = TcpStream::connect(format!("{}:{}", HOST, server_port))
+            .await
+            .unwrap();
+        let (_server_receiver, server_sender): (Receiver<EitherFrame>, Sender<EitherFrame>);
+
+        if encrypt {
+            let initiator = Initiator::from_raw_k(AUTHORITY_PUBLIC_K).unwrap();
+            (_server_receiver, server_sender) =
+                Connection::new(server_stream, HandshakeRole::Initiator(initiator)).await;
+        } else {
+            (_server_receiver, server_sender) = PlainConnection::new(server_stream).await;
+        }
+        server = Some(server_sender);
+    }
+
+    println!("Proxy {} has a client", name);
+    handle_messages::<Mining>(name, cli_receiver, server, total_messages, tx).await;
+}
+
+async fn spawn_proxies(encrypt: bool, hops: u16, tx: Sender<String>, total_messages: i32) -> u16 {
+    let orig_port: u16 = 19000;
+    let final_server_port = orig_port + (hops - 1);
+    let mut listen_port = final_server_port;
+    let mut server_port: u16 = 0;
+
+    for name in (0..hops).rev() {
+        let tx_clone = tx.clone();
+        let name_clone = name.to_string();
+
+        task::spawn(async move {
+            create_proxy(
+                name_clone,
+                listen_port,
+                server_port,
+                encrypt,
+                total_messages,
+                tx_clone,
+            )
+            .await;
+        });
+
+        thread::sleep(std::time::Duration::from_secs(1));
+        server_port = listen_port;
+        listen_port -= 1;
+    }
+    orig_port
+}