monoio 0.2.4

A thread per core runtime based on iouring.
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

use std::{
    io::{Error, ErrorKind, Read, Result, Write},
    net, thread,
};

use monoio::{
    io::{AsyncReadRent, AsyncWriteRentExt, Splitable},
    net::{TcpListener, TcpStream},
    try_join,
};

#[monoio::test_all]
async fn split() -> Result<()> {
    const MSG: &[u8] = b"split";

    let listener = TcpListener::bind("127.0.0.1:0")?;
    let addr = listener.local_addr()?;

    let (stream1, (mut stream2, _)) = try_join! {
        TcpStream::connect(&addr),
        listener.accept(),
    }?;
    let (mut read_half, mut write_half) = stream1.into_split();

    let ((), (), ()) = try_join! {
        async {
            let len = stream2.write_all(MSG).await.0?;
            assert_eq!(len, MSG.len());

            let read_buf = vec![0u8; 32];
            let (read_res, read_buf) = stream2.read(read_buf).await;
            assert_eq!(read_res.unwrap(), MSG.len());
            assert_eq!(&read_buf[..MSG.len()], MSG);
            Result::Ok(())
        },
        async {
            let len = write_half.write_all(MSG).await.0?;
            assert_eq!(len, MSG.len());
            Ok(())
        },
        async {
            let read_buf = vec![0u8; 32];
            let (read_res, read_buf) = read_half.read(read_buf).await;
            assert_eq!(read_res.unwrap(), MSG.len());
            assert_eq!(&read_buf[..MSG.len()], MSG);
            Ok(())
        },
    }?;

    Ok(())
}

#[monoio::test_all(enable_timer = true)]
async fn reunite() -> Result<()> {
    let listener = net::TcpListener::bind("127.0.0.1:0")?;
    let addr = listener.local_addr()?;

    let handle = thread::spawn(move || {
        drop(listener.accept().unwrap());
        drop(listener.accept().unwrap());
    });

    let stream1 = TcpStream::connect(&addr).await?;
    let (read1, write1) = stream1.into_split();

    let stream2 = TcpStream::connect(&addr).await?;
    let (_, write2) = stream2.into_split();

    let read1 = match read1.reunite(write2) {
        Ok(_) => panic!("Reunite should not succeed"),
        Err(err) => err.0,
    };

    read1.reunite(write1).expect("Reunite should succeed");

    handle.join().unwrap();
    Ok(())
}

/// Test that dropping the write half actually closes the stream.
#[monoio::test_all(enable_timer = true, entries = 1024)]
async fn drop_write() -> Result<()> {
    const MSG: &[u8] = b"split";

    let listener = net::TcpListener::bind("127.0.0.1:0")?;
    let addr = listener.local_addr()?;

    let handle = thread::spawn(move || {
        let (mut stream, _) = listener.accept().unwrap();
        stream.write_all(MSG).unwrap();

        let mut read_buf = [0u8; 32];
        let res = match stream.read(&mut read_buf) {
            Ok(0) => Ok(()),
            Ok(len) => Err(Error::new(
                ErrorKind::Other,
                format!("Unexpected read: {len} bytes."),
            )),
            Err(err) => Err(err),
        };

        drop(stream);

        res
    });

    let stream = TcpStream::connect(&addr).await?;
    let (mut read_half, write_half) = stream.into_split();

    let read_buf = vec![0u8; 32];
    let (read_res, read_buf) = read_half.read(read_buf).await;
    assert_eq!(read_res.unwrap(), MSG.len());
    assert_eq!(&read_buf[..MSG.len()], MSG);
    // drop it while the read is in progress
    monoio::spawn(async move {
        monoio::time::sleep(std::time::Duration::from_millis(10)).await;
        drop(write_half);
    });
    match read_half.read(read_buf).await.0 {
        Ok(0) => {}
        Ok(len) => panic!("Unexpected read: {len} bytes."),
        Err(err) => panic!("Unexpected error: {err}."),
    }
    handle.join().unwrap().unwrap();
    Ok(())
}