io: add AsyncFd::try_io() and try_io_mut()

[de-vri-es]

This allows to provide APIs like try_recv() and try_send() in custom types built on top of AsyncFd.

Closes #4719.

Motivation

Custom types built on top of AsyncFd may want to provide functions like try_recv() and try_send(). Currently, I don't think this can be done cleanly, because the poll_[read|write]_ready() functions will register the current task to be woken. They use the reserved waker slot, so they would interfere with other users of poll_{ready,write}_ready(), such as AsyncRead and AsyncWrite implementations.

I suppose you currently could poll the future from {read,write}_ready() once with a fake waker. But that is quite a hacky workaround, and if the future is not ready, the fake waker would actually be registered only to be removed again directly after the polling.

Solution

Add AsyncFd::try_io() and AsyncFd::try_io_mut() to mirror the API from UnixDatagram other sockets. There is a mut and non-mut version to allow mut or const access to the wrapped resource as required, just like AsyncFd::async_io[_mut]().

Example

To illustrate why this is useful, consider the current implementation of a try_recv() function:

impl CanSocket {
    /// Receive a frame from the socket, without waiting for one to become available.
    pub fn try_recv(&self) -> std::io::Result<CanFrame> {
        use std::future::Future;

        let waker = unsafe { std::task::Waker::from_raw(nop_waker_new()) };
        let mut context = std::task::Context::from_waker(&waker);

        let work = self.io.async_io(tokio::io::Interest::READABLE, |socket| Ok(CanFrame {
            inner: socket.recv()?,
        }));
        let work = std::pin::pin!(work);
        match work.poll(&mut context) {
            std::task::Poll::Ready(result) => result,
            std::task::Poll::Pending => Err(std::io::ErrorKind::WouldBlock.into()),
        }
    }
}

const NOP_WAKER_VTABLE: std::task::RawWakerVTable = std::task::RawWakerVTable::new(nop_waker_clone, nop_waker_wake, nop_waker_wake_by_ref, nop_waker_drop);

fn nop_waker_new() -> std::task::RawWaker {
    std::task::RawWaker::new(std::ptr::null(), &NOP_WAKER_VTABLE)
}
fn nop_waker_clone(_waker: *const ()) -> std::task::RawWaker {
    nop_waker_new()
}
fn nop_waker_wake(_waker: *const ()) { }
fn nop_waker_wake_by_ref(_waker: *const ()) { }
fn nop_waker_drop(_waker: *const ()) { }

Or, if we have try_io:

impl CanSocket {
    /// Receive a frame from the socket, without waiting for one to become available.
    pub fn try_recv(&self) -> std::io::Result<CanFrame> {
        self.io.try_io(|io| Ok(CanFrame {
            inner: io.get_ref().recv()?,
        }));
    }
}

No unnecessary wakers being registered, no unsafe code to avoid Arcs in the waker.

[Darksonn]

Overall looks good to me. Are you able to fix the merge conflict?

[de-vri-es]

Rebased on current master, and switched the test to use yield_now() instead of a tiny sleep. Thanks for looking at the PR!

[Darksonn]

Thanks.