Readiness on a proactor
The driver completes IO for the task (How it works), so most code never asks whether a socket is ready: it awaits recv or read. Readiness is for code that wants to try first and wait only when it must, and for calls the driver does not make. On a Nuclei socket:
- A read is ready when the handle holds bytes the driver read ahead, or the stream ended.
- A write is ready when the stream's write-behind has room.
Wait, then try
readable(), writable() and ready(interest) wait until a read or a write would end at once; the try_ calls make it without waiting, or answer WouldBlock. They answer from the handle's own read-ahead and write-behind, which the driver keeps filling and draining: a try_read that finds nothing leaves the driver reading, so a later one finds what came.
use std::io::ErrorKind;
use nuclei::io::{Interest, Ready};
use nuclei::net::{TcpListener, TcpStream};
#[nuclei::main]
async fn main() -> std::io::Result<()> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let client = TcpStream::connect(listener.local_addr()?).await?;
let (server, _) = listener.accept().await?;
// try_write hands bytes to the write-behind, or says WouldBlock when it is full.
client.writable().await?;
assert_eq!(client.try_write(b"ready")?, 5);
// try_read answers from what the driver read ahead, never waiting.
let mut buf = [0u8; 16];
let n = loop {
server.readable().await?;
match server.try_read(&mut buf) {
Ok(n) => break n,
// Nothing read ahead yet: wait for readiness again.
Err(e) if e.kind() == ErrorKind::WouldBlock => continue,
Err(e) => return Err(e),
}
};
assert_eq!(&buf[..n], b"ready");
// ready waits for either, and says which it found.
let found: Ready = server
.ready(Interest::READABLE | Interest::WRITABLE)
.await?;
assert!(found.is_writable());
Ok(())
}Streams have try_read, try_read_vectored, try_write and try_write_vectored. A try_ call after a wait can still answer WouldBlock, when another task got there first: wait again, as the loop does.
Datagrams
Datagram sockets have try_send, try_send_to, try_recv and try_recv_from the same way, and UDP sockets try_peek and try_peek_from too.
use std::io::ErrorKind;
use nuclei::net::UdpSocket;
#[nuclei::main]
async fn main() -> std::io::Result<()> {
let a = UdpSocket::bind("127.0.0.1:0").await?;
let b = UdpSocket::bind("127.0.0.1:0").await?;
// Nothing came yet: the try_ call answers at once instead of waiting.
let mut buf = [0u8; 64];
let empty = b.try_recv_from(&mut buf).unwrap_err();
assert_eq!(empty.kind(), ErrorKind::WouldBlock);
a.try_send_to(b"datagram", b.local_addr()?)?;
// peek_sender names who sent the next datagram, which stays to be received.
assert_eq!(b.peek_sender().await?, a.local_addr()?);
b.readable().await?;
let (n, from) = b.try_recv_from(&mut buf)?;
assert_eq!((&buf[..n], from), (&b"datagram"[..], a.local_addr()?));
Ok(())
}For futures of your own
A future or stream written by hand polls instead: poll_accept, poll_peek, poll_read_ready, poll_write_ready, and for datagrams poll_recv_ready, poll_send_ready and poll_peek_sender. Each keeps the waker of its Context until the answer is ready.
use std::future::poll_fn;
use nuclei::io::ReadBuf;
use nuclei::net::{TcpListener, TcpStream};
#[nuclei::main]
async fn main() -> std::io::Result<()> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let client = nuclei::spawn(TcpStream::connect(listener.local_addr()?));
// poll_accept: the accept as a poll, for a future or a stream of your own.
let (server, _) = poll_fn(|cx| listener.poll_accept(cx)).await?;
let client = client.await?;
client.send(b"peek").await?;
// poll_peek: the bytes the next receive gives, left where they are.
let mut storage = [0u8; 4];
let mut buf = ReadBuf::new(&mut storage);
let n = poll_fn(|cx| server.poll_peek(cx, &mut buf)).await?;
assert_eq!(n, 4);
assert_eq!(buf.filled(), b"peek");
Ok(())
}A UDP socket also polls its datagrams themselves: poll_recv_from receives one into a ReadBuf and answers its sender, cutting what does not fit, and poll_send_to sends one to an address.
use std::future::poll_fn;
use nuclei::io::ReadBuf;
use nuclei::net::UdpSocket;
#[nuclei::main]
async fn main() -> std::io::Result<()> {
let server = UdpSocket::bind("127.0.0.1:0").await?;
let client = UdpSocket::bind("127.0.0.1:0").await?;
let to = server.local_addr()?;
// poll_send_to: a datagram's send as a poll.
let sent = poll_fn(|cx| client.poll_send_to(cx, b"ping", to)).await?;
assert_eq!(sent, 4);
// poll_recv_from: the datagram fills the ReadBuf, and the answer is its sender.
let mut storage = [0u8; 16];
let mut buf = ReadBuf::new(&mut storage);
let from = poll_fn(|cx| server.poll_recv_from(cx, &mut buf)).await?;
assert_eq!((buf.filled(), from), (&b"ping"[..], client.local_addr()?));
Ok(())
}Calls of your own on a socket
For a socket option or a system call Nuclei does not make, async_io(interest, f) runs f, your own call on the socket, each time it may be ready, until it answers other than WouldBlock; try_io runs it once, now (Unix).
use std::io::ErrorKind;
use nuclei::io::Interest;
use nuclei::net::UdpSocket;
#[nuclei::main]
async fn main() -> std::io::Result<()> {
let a = UdpSocket::bind("127.0.0.1:0").await?;
let b = UdpSocket::bind("127.0.0.1:0").await?;
let to = b.local_addr()?;
// A call of your own on the socket, made when it may be ready. The socket
// is non-blocking, so a call that cannot go on answers WouldBlock and the
// wait goes on.
let sent = a
.async_io(Interest::WRITABLE, || a.get_ref().send_to(b"own call", to))
.await?;
assert_eq!(sent, 8);
let mut buf = [0u8; 64];
let n = b
.async_io(Interest::READABLE, || b.get_ref().recv(&mut buf))
.await?;
assert_eq!(&buf[..n], b"own call");
// try_io makes the call once, now, or answers WouldBlock.
let again = b.try_io(Interest::READABLE, || b.get_ref().recv(&mut buf));
assert_eq!(again.unwrap_err().kind(), ErrorKind::WouldBlock);
Ok(())
}Nothing is reordered around your call. Bytes the handle wrote behind go out before it, and a read of your own while the handle holds bytes it read ahead answers ResourceBusy: read those first.
A descriptor of your own
nuclei::io::unix::AsyncFd drives any descriptor you own by readiness: an eventfd, a device, a socket of a kind Nuclei does not wrap. Wait with readable or writable, then make your call through the guard's try_io; a call that answers WouldBlock clears the readiness, and the next wait lasts until the driver says so again. The descriptor must be in non-blocking mode.
use std::io::{Read, Write};
use std::os::unix::net::UnixStream;
use nuclei::io::unix::AsyncFd;
#[nuclei::main]
async fn main() -> std::io::Result<()> {
// A descriptor of your own (here a std socket), in non-blocking mode.
let (mine, mut theirs) = UnixStream::pair()?;
mine.set_nonblocking(true)?;
let fd = AsyncFd::new(mine)?;
theirs.write_all(b"edge")?;
let mut buf = [0u8; 4];
let n = loop {
// Wait until the driver says it may be readable...
let mut guard = fd.readable().await?;
// ...then make the call; WouldBlock clears the readiness and waits again.
match guard.try_io(|inner| inner.get_ref().read(&mut buf)) {
Ok(result) => break result?,
Err(_would_block) => continue,
}
};
assert_eq!(&buf[..n], b"edge");
Ok(())
}