runite 0.1.0

An event-loop-per-thread async runtime built on io_uring (Linux), kqueue (macOS), and IOCP (Windows)
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

//! Linux fd readiness backend.

use std::io;
use std::os::fd::RawFd;

use crate::op::completion::completion_for_current_thread;
use crate::platform::current::runtime::with_current_driver;
use crate::platform::linux::uring::{IORING_OP_POLL_ADD, IoUringCqe};

/// Waits until `fd` becomes readable or reports an error/hangup condition.
pub async fn wait_readable(fd: RawFd) -> io::Result<()> {
    submit_poll(fd, libc::POLLIN | libc::POLLERR | libc::POLLHUP).await
}

/// Waits until `fd` becomes writable or reports an error/hangup condition.
pub async fn wait_writable(fd: RawFd) -> io::Result<()> {
    submit_poll(fd, libc::POLLOUT | libc::POLLERR | libc::POLLHUP).await
}

async fn submit_poll(fd: RawFd, mask: i16) -> io::Result<()> {
    let (future, handle) = completion_for_current_thread::<io::Result<()>>();
    let callback_handle = handle.clone();
    let token = with_current_driver(|driver| {
        driver.submit_operation(
            move |sqe| {
                sqe.opcode = IORING_OP_POLL_ADD;
                sqe.fd = fd;
                sqe.len = 0;
                sqe.op_flags = mask as u32;
            },
            move |cqe| {
                callback_handle.complete(cqe_to_result(cqe));
            },
        )
    })?;

    handle.set_cancel(move || {
        let _ = with_current_driver(|driver| driver.cancel_operation(token));
    });

    future.await
}

fn cqe_to_result(cqe: IoUringCqe) -> io::Result<()> {
    if cqe.res < 0 {
        return Err(io::Error::from_raw_os_error(-cqe.res));
    }
    Ok(())
}