Thin client1

Cargo.toml

@@ -65,3 +65,4 @@ getopts = "^0.2"
 isatty = "0.1"
 rand = "0.4.2"
 pnet = "^0.21.0"
+mio = "0.6.14"

src/bin/client-demo.rs

@@ -19,7 +19,7 @@ use solana::transaction::Transaction;
 use std::env;
 use std::fs::File;
 use std::io::{stdin, Read};
-use std::net::{IpAddr, SocketAddr, UdpSocket};
+use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket};
 use std::process::exit;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, RwLock};
@@ -50,13 +50,13 @@ fn get_ip_addr() -> Option<IpAddr> {
 
 fn main() {
     let mut threads = 4usize;
-    let mut num_nodes = 10usize;
-    let mut leader = "leader.json".to_string();
+    let mut num_nodes = 1usize;
 
     let mut opts = Options::new();
     opts.optopt("l", "", "leader", "leader.json");
     opts.optopt("c", "", "client port", "port");
     opts.optopt("t", "", "number of threads", &format!("{}", threads));
+    opts.optflag("d", "dyn", "detect network address dynamically");
     opts.optopt(
         "n",
         "",
@@ -78,15 +78,14 @@ fn main() {
         print_usage(&program, opts);
         return;
     }
-    if matches.opt_present("l") {
-        leader = matches.opt_str("l").unwrap();
-    }
-    let mut addr: SocketAddr = "127.0.0.1:8010".parse().unwrap();
+    let mut addr: SocketAddr = "0.0.0.0:8100".parse().unwrap();
     if matches.opt_present("c") {
         let port = matches.opt_str("c").unwrap().parse().unwrap();
         addr.set_port(port);
     }
-    addr.set_ip(get_ip_addr().unwrap());
+    if matches.opt_present("d") {
+        addr.set_ip(get_ip_addr().unwrap());
+    }
     let client_addr: Arc<RwLock<SocketAddr>> = Arc::new(RwLock::new(addr));
     if matches.opt_present("t") {
         threads = matches.opt_str("t").unwrap().parse().expect("integer");
@@ -95,7 +94,13 @@ fn main() {
         num_nodes = matches.opt_str("n").unwrap().parse().expect("integer");
     }
 
-    let leader: ReplicatedData = read_leader(leader);
+    let leader = if matches.opt_present("l") {
+        read_leader(matches.opt_str("l").unwrap())
+    } else {
+        let server_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 8000);
+        ReplicatedData::new_leader(&server_addr)
+    };
+
     let signal = Arc::new(AtomicBool::new(false));
     let mut c_threads = vec![];
     let validators = converge(
@@ -152,14 +157,18 @@ fn main() {
         nsps / 1_000_f64
     );
 
-    let first_count = client.transaction_count();
+    let first_count = client.get_transaction_count();
     println!("initial count {}", first_count);
 
     println!("Transfering {} transactions in {} batches", txs, threads);
     let sz = transactions.len() / threads;
     let chunks: Vec<_> = transactions.chunks(sz).collect();
     chunks.into_par_iter().for_each(|txs| {
-        println!("Transferring 1 unit {} times... to", txs.len());
+        println!(
+            "Transferring 1 unit {} times... to {:?}",
+            txs.len(),
+            leader.transactions_addr
+        );
         let client = mk_client(&client_addr, &leader);
         for tx in txs {
             client.transfer_signed(tx.clone()).unwrap();
@@ -170,9 +179,9 @@ fn main() {
     validators.into_par_iter().for_each(|val| {
         let mut client = mk_client(&client_addr, &val);
         let mut now = Instant::now();
-        let mut initial_tx_count = client.transaction_count();
+        let mut initial_tx_count = client.get_transaction_count();
         for i in 0..100 {
-            let tx_count = client.transaction_count();
+            let tx_count = client.get_transaction_count();
             let duration = now.elapsed();
             now = Instant::now();
             let sample = tx_count - initial_tx_count;
@@ -207,15 +216,15 @@ fn main() {
 fn mk_client(locked_addr: &Arc<RwLock<SocketAddr>>, r: &ReplicatedData) -> ThinClient {
     let mut addr = locked_addr.write().unwrap();
     let port = addr.port();
-    let transactions_socket = UdpSocket::bind(addr.clone()).unwrap();
+    let transactions_addr = addr.clone();
     addr.set_port(port + 1);
-    let requests_socket = UdpSocket::bind(addr.clone()).unwrap();
+    let requests_addr = addr.clone();
     addr.set_port(port + 2);
     ThinClient::new(
         r.requests_addr,
-        requests_socket,
+        requests_addr,
         r.transactions_addr,
-        transactions_socket,
+        transactions_addr,
     )
 }
 

src/lib.rs

@@ -49,6 +49,7 @@ extern crate serde_derive;
 extern crate serde_json;
 extern crate sha2;
 extern crate untrusted;
+extern crate mio;
 
 #[cfg(test)]
 #[macro_use]