reactors 0.1.4

Performance-focused cross-platform asynchronous IO implementation
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

use std::{thread::spawn, time::Duration};

use criterion::{async_executor::FuturesExecutor, *};
use futures::executor::block_on;
use reactors::Reactor;

async fn setup_tokio_server() -> anyhow::Result<()> {
    use tokio::io::AsyncWriteExt;
    use tokio::{io::AsyncReadExt, net::TcpListener};

    let listener = TcpListener::bind("127.0.0.1:1812").await?;

    loop {
        let (mut conn, _) = listener.accept().await?;

        let mut buff = [0u8; 11];

        conn.read_exact(&mut buff).await?;

        assert_eq!(&buff, b"hello world");

        conn.write_all(&buff).await?;
    }
}

async fn tokio_client() -> anyhow::Result<()> {
    use tokio::io::AsyncWriteExt;
    use tokio::{io::AsyncReadExt, net::TcpStream};

    let mut conn = TcpStream::connect("127.0.0.1:1812").await?;

    let mut buff = [0u8; 11];

    conn.write_all(&b"hello world"[..]).await?;

    conn.read_exact(&mut buff).await?;

    assert_eq!(&buff, b"hello world");

    Ok(())
}

fn bench_tokio(c: &mut Criterion) {
    spawn(|| {
        use tokio::runtime::Runtime;

        let rt = Runtime::new().unwrap();

        rt.block_on(setup_tokio_server()).unwrap();
    });

    let rt = tokio::runtime::Runtime::new().unwrap();

    c.bench_function("echo tokio", |b| b.to_async(&rt).iter(|| tokio_client()));
}

async fn setup_reactors_server(reactor: reactors::io::IoReactor) -> anyhow::Result<()> {
    use futures::TryStreamExt;
    use futures::{AsyncReadExt, AsyncWriteExt};
    use reactors::io::socket::tcp::TcpAcceptor;

    let mut acceptor = TcpAcceptor::new(reactor, "127.0.0.1:1813".parse()?, None)?;

    while let Some((conn, remote)) = acceptor.try_next().await? {
        log::debug!("accept {}", remote);
        let mut reader = conn.to_read_stream(None);
        let mut writer = conn.to_write_stream(None);

        let mut buff = [0u8; 11];

        reader.read_exact(&mut buff).await?;

        assert_eq!(&buff, b"hello world");

        writer.write_all(&buff).await?;
    }

    Ok(())
}

async fn reactor_client(reactor: reactors::io::IoReactor) -> anyhow::Result<()> {
    use futures::{AsyncReadExt, AsyncWriteExt};
    use reactors::io::socket::tcp::TcpStream;

    let conn = TcpStream::connect(reactor, "127.0.0.1:1813".parse()?, None, None).await?;

    let mut reader = conn.to_read_stream(None);
    let mut writer = conn.to_write_stream(None);

    let mut buff = [0u8; 11];

    writer.write_all(&b"hello world"[..]).await?;

    reader.read_exact(&mut buff).await?;

    assert_eq!(&buff, b"hello world");

    Ok(())
}

fn bench_reactors(c: &mut Criterion) {
    // pretty_env_logger::init();
    use reactors::io::IoReactor;

    let reactor = IoReactor::default();

    let mut server_background_reactor = reactor.clone();

    spawn(move || loop {
        server_background_reactor
            .poll_once(Duration::from_millis(600))
            .unwrap();
    });

    let server_reactor = reactor.clone();

    spawn(move || {
        block_on(setup_reactors_server(server_reactor)).unwrap();
    });

    c.bench_function("echo reactors", |b| {
        b.to_async(FuturesExecutor)
            .iter(|| reactor_client(reactor.clone()))
    });
}

criterion_group!(benches, bench_reactors, bench_tokio);
criterion_main!(benches);