Nuclei

Chapter 9 of 17

Files

Files and directories, as async calls that never hold a shard.

Whole files

nuclei::fs mirrors std::fs, with each call a future: read, read_to_string, write, copy, rename, remove_file, metadata, try_exists and the rest take a path and do the whole job.

src/main.rs
use nuclei::fs;

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // Whole files in one call each.
    fs::write("notes.txt", "first line\n").await?;
    let text = fs::read_to_string("notes.txt").await?;
    assert_eq!(text, "first line\n");

    fs::copy("notes.txt", "copy.txt").await?;
    fs::rename("copy.txt", "moved.txt").await?;
    assert!(fs::try_exists("moved.txt").await?);
    assert_eq!(fs::read("moved.txt").await?, b"first line\n");

    let meta = fs::metadata("moved.txt").await?;
    assert!(meta.is_file());
    assert_eq!(meta.len(), 11);

    fs::remove_file("moved.txt").await?;
    fs::remove_file("notes.txt").await
}

Opening a file

A File is std's file in a Handle: File::open reads, File::create writes (creating or cutting the file), File::create_new creates a file to read and write and fails if it exists, and OpenOptions sets the rest (append, truncate, the mode and the open flags on Unix). A file reads and writes through the IO helpers and has a cursor that seek moves.

src/main.rs
use std::io::SeekFrom;

use nuclei::fs::{File, OpenOptions};
use nuclei::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    let mut file = File::create("log.txt").await?;
    file.write_all(b"one\n").await?;
    // flush hands the bytes to the system; sync_all puts them on the device.
    file.flush().await?;
    file.sync_all().await?;
    drop(file);

    // Open for appending: every write lands at the end.
    let mut log = OpenOptions::new().append(true).open("log.txt").await?;
    log.write_all(b"two\n").await?;
    log.flush().await?;
    drop(log);

    let mut file = File::open("log.txt").await?;
    let mut text = String::new();
    file.read_to_string(&mut text).await?;
    assert_eq!(text, "one\ntwo\n");

    // A file has a cursor: seek moves it.
    file.seek(SeekFrom::Start(4)).await?;
    let mut second = String::new();
    file.read_to_string(&mut second).await?;
    assert_eq!(second, "two\n");
    nuclei::fs::remove_file("log.txt").await
}
  • flush waits until the system has the bytes written behind; sync_all and sync_data put them on the device (Surviving a crash).
  • metadata, set_len, set_permissions, set_times and set_modified work on the open file; try_clone gives a second handle with a cursor of its own.
  • A read at the cursor sees the changes made beside it: bytes it read ahead before a positional write, a cut by set_len, an allocate or a copy into the file are never handed out after the change.

Lending buffers

As on streams (How it works), read_owned and write_owned take your buffer and give it back, so no byte is copied. The buffer is a Vec<u8>, an AlignedBuf, or a type of your own (IO utilities).

src/main.rs
use nuclei::fs::File;

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // Lend a vector to the write: the bytes are written from it, no copy.
    let file = File::create("lent.bin").await?;
    let (written, _buf) = file.write_owned(vec![7u8; 4096]).await;
    assert_eq!(written?, 4096);
    drop(file);

    // And to the read: they land in its spare capacity.
    let file = File::open("lent.bin").await?;
    let (n, buf) = file.read_owned(Vec::with_capacity(8192)).await;
    assert_eq!(n?, 4096);
    assert!(buf.iter().all(|&b| b == 7));
    nuclei::fs::remove_file("lent.bin").await
}

Reads and writes at an offset

read_at and write_at read and write at an offset you give, on a shared reference to the handle, so any number of tasks may use one file at once. They never read or move the cursor. read_exact_at and write_all_at do the whole length, and read_at_owned and write_at_owned lend your buffer.

src/main.rs
use std::sync::Arc;

use nuclei::fs::File;
use nuclei::io::AsyncReadExt;

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    // One handle shared by four tasks, each writing its record at its offset.
    let file = Arc::new(File::create_new("records.bin").await?);
    let writers: Vec<_> = (0..4u8)
        .map(|i| {
            let file = Arc::clone(&file);
            nuclei::spawn(async move { file.write_all_at(&[i; 16], u64::from(i) * 16).await })
        })
        .collect();
    for writer in writers {
        writer.await?;
    }

    let mut record = [0u8; 16];
    file.read_exact_at(&mut record, 32).await?;
    assert_eq!(record, [2; 16]);

    // Many reads in one call, each into a buffer of its own, out together.
    let reads = (0..4u64)
        .map(|i| (Vec::<u8>::with_capacity(16), i * 16))
        .collect();
    for (i, (n, buf)) in (0u8..).zip(file.read_many_at(reads).await) {
        assert_eq!(n?, 16);
        assert!(buf.iter().all(|&b| b == i));
    }

    // Positional calls never move the cursor: a read at it starts at 0.
    let mut all = Vec::new();
    (&*file).read_to_end(&mut all).await?;
    assert_eq!(all.len(), 64);
    nuclei::fs::remove_file("records.bin").await
}
  • A positional call lands after every write accepted at the cursor before it, and a read sees every positional write that returned before it. Two positional writes running at once land in either order, as two threads' pwrite calls do.
  • A positional write whose future is dropped still lands, and every later call of the handle waits for it.
  • read_at and write_at go through a buffer of the runtime's pool, at most 64 KiB a call; the owned forms use your buffer itself.
  • read_many_at makes many reads in one call and gives each buffer back in order with its result. On io_uring they go to the kernel together, at most 64 out at once.
  • These are the async calls, and they shadow std's FileExt::read_at and write_at, which the handle reaches through Deref and which block the worker. std's stay reachable as (*file).read_at(..).

Surviving a crash

sync_all writes the file's data and metadata through to the device, and sync_data its data, once every write accepted before the call landed. A file's sync does not cover its name in its directory; fs::sync_dir does. To replace a file so a crash leaves the old contents or the new and never a mix, write a temporary file in the same directory, sync it, rename it over the target, then sync the directory.

src/main.rs
use nuclei::fs::{self, File};
use nuclei::io::AsyncWriteExt;

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    fs::write("config.toml", "version = 1\n").await?;

    // Replace it so a crash leaves the old contents or the new, never a mix.
    let mut tmp = File::create("config.toml.tmp").await?;
    tmp.write_all(b"version = 2\n").await?;
    tmp.sync_all().await?; // the bytes, on the device
    drop(tmp);
    fs::rename("config.toml.tmp", "config.toml").await?;
    fs::sync_dir(".").await?; // the new name, on the device

    assert_eq!(fs::read_to_string("config.toml").await?, "version = 2\n");
    fs::remove_file("config.toml").await
}

Once a sync of a handle failed, every later sync of it fails with the same error, since the system reports a lost write-back only once. Open the file again to start over.

Space, write-back and copies

allocate changes the space a range of the file uses, as its Allocate mode says: Extend reserves it and grows the file, KeepSize reserves it and keeps the length, PunchHole frees it, and ZeroRange makes it read as zeros. A system that cannot do a mode answers Unsupported (macOS cannot zero a range in place), and a file system may refuse one with its own error.

sync_range writes a range back and waits for it; it writes no metadata, so it is not a durability guarantee. copy_range_to copies a range into another file, in the kernel where the system can, in chunks of at most 16 MiB, and moves neither cursor.

src/main.rs
use nuclei::fs::{Allocate, File};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    let log = File::create_new("log.bin").await?;
    // Reserve 1 MiB up front, so writes into it cannot run out of space.
    log.allocate(0, 1 << 20, Allocate::KeepSize).await?;
    assert_eq!(log.metadata().await?.len(), 0);

    log.write_all_at(b"entry", 0).await?;
    // Start writing the range back now; sync_data is what makes it durable.
    log.sync_range(0, 5).await?;

    // Copy the range into another file, in the kernel where it can.
    let copy = File::create_new("copy.bin").await?;
    assert_eq!(log.copy_range_to(&copy, 0, 0, 5).await?, 5);
    let mut back = [0u8; 5];
    copy.read_exact_at(&mut back, 0).await?;
    assert_eq!(&back, b"entry");

    nuclei::fs::remove_file("log.bin").await?;
    nuclei::fs::remove_file("copy.bin").await
}

Locks

lock_file takes an advisory lock on the whole file, shared or exclusive, and waits for it on the blocking pool so the worker runs on. A future dropped while it waits never leaves the file locked with nobody to release it. try_lock_file answers at once, with WouldBlock when another holder has the lock, and unlock_file releases it once the writes before it landed.

src/main.rs
use nuclei::fs::{File, TryLockError};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    let holder = File::create("app.lock").await?;
    let other = File::open("app.lock").await?;

    // Wait for the lock on the blocking pool; the worker runs on meanwhile.
    holder.lock_file(false).await?;
    // While it is held, another handle cannot have it.
    let refused = other.try_lock_file(false).await;
    assert!(matches!(refused, Err(TryLockError::WouldBlock)));

    // Released once the writes before it landed; then free for the other.
    holder.unlock_file().await?;
    other.try_lock_file(true).await?;
    other.unlock_file().await?;
    nuclei::fs::remove_file("app.lock").await
}

std's own lock, try_lock and unlock stay reachable through Deref; std's lock blocks the worker while it waits.

Direct IO

A file opened for direct IO (O_DIRECT on Linux, FILE_FLAG_NO_BUFFERING on Windows) skips the system's page cache. Nuclei reads and writes it with exactly your lengths and offsets, with no read-ahead and no write-behind, so the kernel's alignment rules apply to your calls as they would to your own system calls: the memory's address, the offset and the length are multiples of the device's block size. AlignedBuf gives memory at the alignment you ask for.

src/main.rs
use nuclei::fs::OpenOptions;
use nuclei::io::AlignedBuf;

// Direct IO asks the memory, the offset and the length to be multiples of
// the device's block size; 4096 meets that on the common ones.
const BLOCK: usize = 4096;

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    let file = OpenOptions::new()
        .read(true)
        .write(true)
        .create(true)
        .custom_flags(libc::O_DIRECT)
        .open("direct.bin")
        .await?;

    // Two blocks in memory aligned to a block, handed to the kernel as they are.
    let mut blocks = AlignedBuf::new(2 * BLOCK, BLOCK)?;
    blocks.extend_from_slice(&[7u8; 2 * BLOCK]);
    let (written, _blocks) = file.write_at_owned(blocks, 0).await;
    assert_eq!(written?, 2 * BLOCK);

    // The second block, read straight into an aligned buffer of its own.
    let room = AlignedBuf::new(BLOCK, BLOCK)?;
    let (n, block) = file.read_at_owned(room, BLOCK as u64).await;
    assert_eq!(n?, BLOCK);
    assert!(block.iter().all(|&b| b == 7));
    nuclei::fs::remove_file("direct.bin").await
}

Directories

create_dir, create_dir_all and DirBuilder (recursive, with a mode on Unix) make them; remove_dir and remove_dir_all take them away. read_dir streams the entries a page at a time, so a directory of any size is read without holding it all.

src/main.rs
use nuclei::fs::{self, DirBuilder};

#[nuclei::main]
async fn main() -> std::io::Result<()> {
    fs::create_dir_all("site/assets/img").await?;
    DirBuilder::new()
        .recursive(true)
        .create("site/pages")
        .await?;
    fs::write("site/index.html", "<h1>hi</h1>").await?;

    // read_dir is a stream of entries, read a page at a time.
    let mut names = Vec::new();
    let mut entries = fs::read_dir("site").await?;
    while let Some(entry) = entries.next_entry().await? {
        let mut name = entry.file_name().to_string_lossy().into_owned();
        if entry.file_type().await?.is_dir() {
            name.push('/');
        }
        names.push(name);
    }
    names.sort();
    assert_eq!(names, ["assets/", "index.html", "pages/"]);

    fs::remove_dir_all("site").await
}

Links: hard_link, read_link and canonicalize, symlink on Unix, symlink_file and symlink_dir on Windows.

Where file IO runs

A regular file is always ready to the kernel, so no poller can say when its IO would wait. On io_uring a file's reads and writes, at the cursor and at an offset, its syncs, sync_range and allocate are operations of the shard's ring; on the other backends each is a job of the blocking pool. Calls by path, lock_file, copy_range_to and the file's other calls run on the blocking pool; try_lock_file runs on the calling thread, since it never waits. On Linux, File::open opens a regular file on a local file system on the calling thread when the kernel's caches resolve its path at once (openat2 with RESOLVE_CACHED, Linux 5.12), with no job; any other open, a FIFO, a device or a file on a network file system among them, goes to the blocking pool. A File that is not a regular file (a FIFO, a terminal, a pipe) is a byte stream the driver serves like a socket.