bench.rs

use parking_lot::Mutex;
use std::{
    sync::{atomic, Arc},
    thread, time,
};

#[tokio::main]
async fn main() -> web3::Result {
    let _ = env_logger::try_init();
    let requests = 200_000;

    let http = web3::transports::Http::new("http://localhost:8545/")?;
    bench("http", http, requests);

    let ipc = web3::transports::WebSocket::new("./jsonrpc.ipc").await?;
    bench(" ipc", ipc, requests);

    Ok(())
}

fn bench<T: web3::Transport>(id: &str, transport: T, max: usize)
where
    T::Out: Send + 'static,
{
    use futures::FutureExt;

    let web3 = web3::Web3::new(transport);
    let ticker = Arc::new(Ticker::new(id));

    for _ in 0..max {
        let ticker = ticker.clone();
        ticker.start();
        let accounts = web3.eth().block_number(None).then(move |res| {
            if let Err(e) = res {
                println!("Error: {:?}", e);
            }
            ticker.tick();
            futures::future::ready(())
        });
        tokio::spawn(accounts);
    }

    ticker.wait();
}

struct Ticker {
    id: String,
    started: atomic::AtomicUsize,
    reqs: atomic::AtomicUsize,
    time: Mutex<time::Instant>,
}

impl Ticker {
    pub fn new(id: &str) -> Self {
        Ticker {
            id: id.to_owned(),
            time: Mutex::new(time::Instant::now()),
            started: Default::default(),
            reqs: Default::default(),
        }
    }

    pub fn start(&self) {
        self.started.fetch_add(1, atomic::Ordering::AcqRel);
    }

    pub fn tick(&self) {
        let reqs = self.reqs.fetch_add(1, atomic::Ordering::AcqRel) as u64;
        self.started.fetch_sub(1, atomic::Ordering::AcqRel);

        if reqs >= 100_000 {
            self.print_result(reqs);
        }
    }

    pub fn print_result(&self, reqs: u64) {
        fn as_millis(dur: time::Duration) -> u64 {
            dur.as_secs() * 1_000 + dur.subsec_nanos() as u64 / 1_000_000
        }

        let mut time = self.time.lock();
        let elapsed = as_millis(time.elapsed());
        let result = reqs * 1_000 / elapsed;

        println!("[{}] {} reqs/s ({} reqs in {} ms)", self.id, result, reqs, elapsed);

        self.reqs.store(0, atomic::Ordering::Release);
        *time = time::Instant::now();
    }

    pub fn wait(&self) {
        while self.started.load(atomic::Ordering::Relaxed) > 0 {
            thread::sleep(time::Duration::from_millis(100));
        }
        self.print_result(self.reqs.load(atomic::Ordering::Acquire) as u64);
    }
}