IO utilities: Learn Nuclei

Reads and writes as futures

Streams implement the futures-io traits (AsyncRead, AsyncWrite, AsyncBufRead, AsyncSeek), so code written against them works unchanged. nuclei::io adds the calls you await, in four extension traits:

Trait	Calls
`AsyncReadExt`	`read`, `read_exact`, `read_to_end`, `read_to_string`, `read_buf`, `read_u32` and every number (big-endian, or `_le`), `take`, `chain`
`AsyncWriteExt`	`write`, `write_all`, `write_vectored`, `write_u32` and every number, `flush`, `shutdown`
`AsyncBufReadExt`	`read_line`, `read_until`, `lines`, `split`, `fill_buf`, `consume`
`AsyncSeekExt`	`seek`, `rewind`, `stream_position`

src/main.rs

use nuclei::io::{self, AsyncReadExt, AsyncWriteExt};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // Two connected in-memory streams, 64 bytes of buffer each way.
    let (mut client, mut server) = io::duplex(64);

    // A frame: a length, big-endian, then that many bytes.
    let body = b"framed";
    client.write_u32(body.len() as u32).await?;
    client.write_all(body).await?;

    let len = server.read_u32().await?;
    let mut frame = vec![0u8; len as usize];
    server.read_exact(&mut frame).await?;
    assert_eq!(frame, body);

    // Ending one side ends the other's reads.
    drop(client);
    let mut rest = Vec::new();
    assert_eq!(server.read_to_end(&mut rest).await?, 0);
    Ok(())
}

Lines and buffers

BufReader puts a buffer in front of a reader, so small reads and line reads take from memory; BufWriter gathers small writes into one; BufStream is both, for a stream that reads and writes.

src/main.rs

use nuclei::io::{self, AsyncBufReadExt, AsyncWriteExt, BufReader};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    let (rx, mut tx) = io::simplex(1024);
    tx.write_all(b"GET /\nHost: nuclei.rs\n\nbody").await?;
    drop(tx);

    // A buffer in front of the reader, then a line at a time.
    let mut lines = BufReader::new(rx).lines();
    let mut head = Vec::new();
    while let Some(line) = lines.next_line().await? {
        if line.is_empty() {
            break;
        }
        head.push(line);
    }
    assert_eq!(head, ["GET /", "Host: nuclei.rs"]);
    Ok(())
}

src/main.rs

use nuclei::io::{AsyncReadExt, AsyncWriteExt, BufWriter};
use nuclei::net::{TcpListener, TcpStream};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let stream = TcpStream::connect(listener.local_addr()?).await?;
    let (mut server, _) = listener.accept().await?;

    // Many small writes gather in the buffer and go out as one.
    let mut out = BufWriter::new(stream);
    for word in ["one ", "two ", "three"] {
        out.write_all(word.as_bytes()).await?;
    }
    out.flush().await?;
    out.shutdown().await?;

    let mut text = String::new();
    server.read_to_string(&mut text).await?;
    assert_eq!(text, "one two three");
    Ok(())
}

Copies

io::copy moves everything a reader gives into a writer, then flushes; copy_buf does it from a buffered reader's own buffer. copy_bidirectional runs both directions between two streams at once, the heart of a proxy, and answers the bytes moved each way.

src/main.rs

use nuclei::io::{self, AsyncReadExt, AsyncWriteExt};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // copy: everything a reader gives, into a writer, then a flush.
    let mut source = io::repeat(b'x').take(1 << 20);
    let copied = io::copy(&mut source, &mut io::sink()).await?;
    assert_eq!(copied, 1 << 20);

    // copy_bidirectional: a proxy between two streams, both ways at once.
    let (mut client, mut front) = io::duplex(1024);
    let (mut back, mut backend) = io::duplex(1024);
    let proxy = nuclei::spawn(async move { io::copy_bidirectional(&mut front, &mut back).await });
    client.write_all(b"request").await?;
    let mut buf = [0u8; 7];
    backend.read_exact(&mut buf).await?;
    backend.write_all(b"reply").await?;
    let mut reply = [0u8; 5];
    client.read_exact(&mut reply).await?;
    assert_eq!(&reply, b"reply");

    // Each side that ends ends its direction; the proxy answers both counts.
    drop(client);
    drop(backend);
    assert_eq!(proxy.await?, (7, 5));
    Ok(())
}

Buffers you lend

The owned calls (read_owned, write_owned, write_zero_copy, and the datagram and file forms) lend the driver your buffer and give it back, so the kernel reads into and writes from your memory and no byte is copied (How it works). A write takes any IoBuf and a read any IoBufMut, and each gives back the same type: a Vec<u8> in, a Vec<u8> out.

src/main.rs

use std::sync::Arc;

use nuclei::io::{AlignedBuf, IoBuf};
use nuclei::net::{TcpListener, TcpStream};

// A message in a shared allocation, sent to many peers without a copy.
#[derive(Default)]
struct Shared(Arc<[u8]>);

// SAFETY: the bytes live in the Arc's allocation, which stays in place when
// the value moves, and every one of them is initialized.
unsafe impl IoBuf for Shared {
    fn stable_ptr(&self) -> *const u8 {
        self.0.as_ptr()
    }

    fn bytes_init(&self) -> usize {
        self.0.len()
    }

    fn bytes_total(&self) -> usize {
        self.0.len()
    }
}

#[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?;

    // A buffer of your own type goes in, and the same value comes back.
    let news: Arc<[u8]> = Arc::from(&b"shared, not copied"[..]);
    let (sent, back) = server.write_owned(Shared(Arc::clone(&news))).await;
    assert_eq!(sent?, news.len());
    assert!(Arc::ptr_eq(&back.0, &news));

    // An AlignedBuf fills like a vector: its room, up to its capacity.
    let room = AlignedBuf::new(64, 64)?;
    let (n, got) = client.read_owned(room).await;
    assert_eq!(&got[..n?], &news[..]);
    assert_eq!(got.as_ptr() as usize % 64, 0);
    Ok(())
}

AlignedBuf is a buffer of a fixed capacity at an address of the alignment you choose, made by a safe constructor (new or zeroed): the buffer for a file opened for direct IO. Its memory is yours, not charged to the runtime's io_buffer_cap.
A type of your own (a shared byte buffer, memory of a registered region) implements the two traits. They are unsafe: the memory must stay in place while the runtime holds the value, even when the value moves.
The runtime keeps a Vec<u8> or an AlignedBuf as it is while a call has it, and boxes a buffer of any other type once.
Where the compiler cannot see the buffer's type, write it: Vec::<u8>::new(), not Vec::new().into() or Default::default(), and send_to_owned::<A, _> where you name the address type.

Several buffers in one call

write_vectored_owned writes every buffer of an IoVecBuf in order as one write: no other write of the handle comes between two of them. read_vectored_owned fills the room of each buffer of an IoVecBufMut before the next one's. Both take a Vec or an array of any owned buffer type and give the buffers back together. TCP and Unix streams, pipes and files have both.

src/main.rs

use nuclei::fs::File;
use nuclei::io::AsyncSeekExt;

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    let mut file = File::create_new("record.bin").await?;

    // A header and a body from two buffers, as one write: nothing comes between.
    let parts = vec![b"LEN=4;".to_vec(), b"body".to_vec()];
    let (written, parts) = file.write_vectored_owned(parts).await;
    assert_eq!(written?, 10);
    // Both buffers come back, for the next record.
    assert_eq!(parts.len(), 2);

    // Read it back into two rooms: the first fills before the second.
    file.rewind().await?;
    let rooms = [Vec::<u8>::with_capacity(6), Vec::with_capacity(64)];
    let (n, [header, body]) = file.read_vectored_owned(rooms).await;
    assert_eq!(n?, 10);
    assert_eq!((&header[..], &body[..]), (&b"LEN=4;"[..], &b"body"[..]));
    nuclei::fs::remove_file("record.bin").await
}

Each call of the kernel takes every buffer at once: writev and readv, or a sendmsg and recvmsg of several buffers, on epoll and kqueue; their ring operations on io_uring; a WSASend or WSARecv of several buffers on I/O completion ports; pwritev and preadv for files. A call the kernel takes part of goes on from the byte it stopped at. Where the system has no vectored call of the kind, the buffers go one call each, in order: a buffered file on Windows does so, and a Windows handle that is not a socket writes that way and reads into its first buffer only.

In memory

Streams with no socket behind them, for tests and for joining parts of a program:

duplex(n): two connected streams, n bytes of buffer each way. simplex(n): one pipe, a reader and a writer.
io::split(stream): a read half and a write half of any stream, put back with unsplit.
empty(), repeat(byte), sink(): a reader at its end, a reader without end, a writer that keeps nothing.

The standard streams

io::stdin(), io::stdout() and io::stderr() are the process's own, as async streams.

src/main.rs

use nuclei::io::{self, AsyncWriteExt};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    let mut out = io::stdout();
    out.write_all(b"written to stdout\n").await?;
    out.flush().await?;
    Ok(())
}

output

written to stdout

Any std object

The socket and file types are Handles over std's own objects, and Handle::new takes over any other one: on Unix anything with a file descriptor, on Windows sockets, files and a child's pipes. get_ref reaches the object; into_inner gives it back once the bytes written behind went out.

src/main.rs

use nuclei::Handle;

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // Any std socket or file, taken over: Handle::new makes its IO async.
    let std_listener = std::net::TcpListener::bind("127.0.0.1:0")?;
    let listener = Handle::new(std_listener)?;
    let addr = listener.get_ref().local_addr()?;

    let client = Handle::new(std::net::TcpStream::connect(addr)?)?;
    let (server, _) = listener.accept().await?;
    client.send(b"hi").await?;
    let mut buf = [0u8; 2];
    server.recv(&mut buf).await?;

    // into_inner gives std's object back once the bytes written behind went out.
    let std_stream: std::net::TcpStream = client.into_inner();
    assert_eq!(std_stream.peer_addr()?, addr);
    Ok(())
}