zruntime 0.1.0

A simple single-threaded Rust async runtime
Documentation

A simple, single-threaded async runtime for Rust.

A Runtime runs async tasks, timers and I/O on a single thread. Runtime::block_on runs a future to completion on the calling thread. While it waits for the future, the same thread also runs the runtime's tasks, timers and I/O. A runtime comes in two flavours:

  • LocalRuntime is the default. It stays on the thread that created it, and runs any 'static future, Send or not. Prefer it unless the runtime, or a future built on it, has to cross threads or be polled by another executor.
  • SharedRuntime can be used from, and run on, any thread. It runs Send futures only.

The flavour is a type parameter: LocalRuntime is Runtime<Local>, and SharedRuntime is Runtime<Shared>. A runtime is cheap to clone, and all its clones refer to the same runtime.

Overview

The runtime, and what runs on it:

What needs no runtime, and works under any executor:

The runtime and Event are on by default. The rest is behind cargo features.

Examples

A local runtime, used by the thread that created it:

use std::{cell::RefCell, rc::Rc, time::Duration};

use zruntime::LocalRuntime;

let runtime = LocalRuntime::new().expect("a runtime for this thread");
let doubled = runtime.block_on(async {
    let count = Rc::new(RefCell::new(0));
    let sleeper = runtime.clone();
    let task = runtime.spawn("double", {
        let count = count.clone();
        async move {
            sleeper.sleep(Duration::from_millis(1)).await;
            *count.borrow_mut() += 21;
            *count.borrow() * 2
        }
    });

    task.await.expect("the task did not panic")
});

assert_eq!(doubled, 42);

A shared runtime, spawned on from another thread and run by the main thread:

use std::{sync::mpsc, thread};

use zruntime::SharedRuntime;

let runtime = SharedRuntime::new().expect("a runtime for this thread");
let (sender, receiver) = mpsc::channel();
let spawner = thread::spawn({
    let runtime = runtime.clone();
    move || {
        let task = runtime.spawn("double", async { 21 * 2 });
        sender.send(task).expect("the main thread is still waiting for it");
    }
});

let task = receiver.recv().expect("the other thread sent the task");
let doubled = runtime.block_on(task).expect("the task did not panic");
spawner.join().expect("the other thread did not panic");

assert_eq!(doubled, 42);

Create a runtime with LocalRuntime::new() or SharedRuntime::new(). The compiler cannot infer the flavour of a plain Runtime::new().

Tasks

Runtime::spawn starts a task on the runtime. It takes a name, which the log message uses if the task panics.

Code that has no runtime handle can use the free spawn_local and spawn functions instead. They spawn on the runtime that is running the calling code, and name the task after the place it was spawned from.

use zruntime::{LocalRuntime, Task};

// Has no runtime handle: it spawns on the runtime that runs its caller.
fn double(number: u32) -> Task<u32> {
    zruntime::spawn_local(async move { number * 2 })
}

let runtime = LocalRuntime::new().expect("a runtime for this thread");
let doubled = runtime.block_on(async { double(21).await.expect("the task did not panic") });

assert_eq!(doubled, 42);
  • spawn_local takes any 'static future. Call it from code running on a LocalRuntime: from the future passed to its block_on, or from one of its tasks.
  • spawn takes a Send future, and spawns it on the SharedRuntime that runs the calling code. If there is none, it panics, unless the helper feature is on (see Per-thread runtimes).

Both return a Task, a handle to the task:

  • Awaiting it returns the task's output, or an error if the task panicked or its runtime was dropped.
  • Dropping it cancels the task.
  • Task::cancel cancels the task, and returns a future that completes once it has stopped: with the output, if the task had already finished.
  • Task::detach lets the task run on without a handle.
  • Task::is_finished tells whether the task has ended, without taking its output.

Timers

use std::{future, time::Duration};

use zruntime::LocalRuntime;

let runtime = LocalRuntime::new().expect("a runtime for this thread");
runtime.block_on(async {
    let mut interval = runtime.interval(Duration::from_millis(1));
    for _ in 0..3 {
        interval.tick().await;
    }

    let never = runtime.timeout(Duration::from_millis(2), future::pending::<()>());
    assert!(never.await.is_err());
});

Sockets

The net module has TCP and UDP sockets: TcpListener, TcpStream and UdpSocket. On unix, its unix module has UnixListener, UnixStream and UnixDatagram. They are like the sockets of smol::net, and work on both flavours of runtime. Streams implement AsyncRead and AsyncWrite from futures-io. Connecting never blocks the thread.

The TCP and UDP sockets take socket addresses, not host names. To look up a name without blocking the thread, run the lookup with unblock().

Other I/O sources

AsyncIo makes any source the runtime can watch async, like smol::Async. On unix, that is anything with a file descriptor: a pipe, a terminal, an eventfd, an inotify instance, or a socket type that net does not cover. On Windows, it can only be a socket. AsyncIo::new takes the runtime and the source, and switches the source to non-blocking mode.

  • AsyncIo::read_with and AsyncIo::write_with run an operation on the source until it no longer fails with WouldBlock.
  • AsyncIo::readable and AsyncIo::writable only wait for readiness. They are useful when another library does the I/O on the descriptor.
  • AsyncIo<T> implements AsyncRead from futures-io if &T implements Read, and AsyncWrite if &T implements Write.

Any number of tasks can wait through the four methods at once, but only one task per direction through the AsyncRead and AsyncWrite impls.

AsyncIo is not for regular files: reading or writing one can block the thread, whatever its readiness says. Use Unblock or the fs module for files.

For code that does its own polling, Runtime::register watches a source and returns a Registration to poll it through.

Blocking work and files

unblock() runs blocking work on a pool of threads, and returns a future of its result. Unblock wraps a blocking I/O handle, such as a file or the standard input. It implements the async I/O traits by running each operation on that pool.

The fs module is async filesystem access, in the shape of std::fs and like smol::fs. It runs each operation on the same pool.

None of these needs a runtime: they work under any executor.

Child processes

The process module runs child processes, like smol::process. Its Command works like std::process::Command, but spawns the child on a runtime that you pass to it. The Child it returns has async status and output methods. The pipes to the child, ChildStdin, ChildStdout and ChildStderr, implement AsyncWrite and AsyncRead.

Reading or writing a pipe, and waiting for the child to exit, never block the thread. Dropping a Child leaves the process running, unless kill_on_drop(true) was set on its Command. The process module documentation has the details.

Events, locks and channels

These work under any executor, not only zruntime's. Tasks on different runtimes and threads can share them.

  • Event is a notification that tasks can wait for. A task gets an EventListener from the event and awaits it. Code that changes what the task waits for, for example by releasing a lock, notifies the event. That wakes as many listeners as it asks for.
  • The lock module has an async Mutex, RwLock and Semaphore, whose guards can be held across an .await. It also has a Barrier, and a OnceCell whose initialiser can be async.
  • mpmc is a multi-producer, multi-consumer channel, bounded or unbounded. Each message goes to one receiver.
  • broadcast is a multi-producer, multi-consumer broadcast channel. Each message goes to every receiver.

Running on several threads

One thread at a time runs a runtime, so all its tasks share one CPU core. A task that keeps the CPU busy holds up every other task on the same runtime. Two threads cannot run one runtime together: a second block_on on a runtime from SharedRuntime::new panics, and one on a runtime from SharedRuntime::current waits for its turn.

To use more cores, use several runtimes, one per thread, each run by its own block_on. With the helper feature, the free block_on already works like this: each thread that calls it runs its own runtime.

To spread work over the threads, give each thread a clone of the same mpmc receiver. Each piece of work goes to whichever thread receives it first, as the module's example shows. To run a task on a specific thread, spawn it on that thread's runtime. A SharedRuntime can be spawned on from any thread, as the second example above shows.

Per-thread runtimes

The examples above run their runtime with one block_on call for the whole program. A library such as zbus works differently: it may call block_on once per operation, or have its futures polled by another executor, and its tasks and timers must keep running in between.

The helper feature supports this:

  • The free block_on runs a future on the calling thread's own runtime.
  • SharedRuntime::current returns the runtime for the calling code. From the future passed to the free block_on, or from code running on a runtime that current returned, it is that runtime. From anywhere else, code on a runtime from SharedRuntime::new included, it is one runtime that the whole process shares.
  • If such a runtime has work and no thread is running block_on on it, a helper thread runs it in the meantime. A block_on call takes the runtime back for as long as it lasts.
  • spawn works from any thread. If no shared runtime runs the calling code, the task goes on the runtime that SharedRuntime::current returns.

Combinators

zruntime has no combinators of its own. Use those of the futures crate instead. They work with zruntime's tasks, timers, sockets and pipes, which implement the Future, Stream, AsyncRead and AsyncWrite traits.

  • futures::future joins and races futures.
  • futures::stream adapts streams, such as an interval or the connections of a listener.
  • futures::io adds methods such as read_to_end and write_all, and a BufReader to read lines with. The Unblock documentation shows it reading the standard input line by line.

The default features of futures bring an executor, which you do not need with zruntime. They also bring the join! and select! macros, which build a proc-macro crate. To leave both out:

[dependencies]
futures = { version = "0.3", default-features = false, features = ["std"] }

std brings the I/O extension traits. Add async-await to get the macros back.

use std::time::Duration;

use futures::future;
use zruntime::LocalRuntime;

let runtime = LocalRuntime::new().expect("a runtime for this thread");
runtime.block_on(async {
    // Two tasks after the same answer, one of them much faster than the other.
    let soon = runtime.clone();
    let near = runtime.spawn("near", async move {
        soon.sleep(Duration::from_millis(1)).await;
        "near"
    });
    let late = runtime.clone();
    let far = runtime.spawn("far", async move {
        late.sleep(Duration::from_secs(60)).await;
        "far"
    });

    // The first to answer wins. The other one is returned with the answer, and dropping it
    // cancels it.
    let (answer, _) = future::select(near, far).await.factor_first();
    assert_eq!(answer.expect("the task did not panic"), "near");

    // Two futures awaited together, keeping the outputs of both.
    let (a, b) = future::join(async { 21 }, async { 2 }).await;
    assert_eq!(a * b, 42);
});

To race a future against a timer, use Runtime::timeout.

Cargo features

runtime and event do not depend on each other. A crate that only needs Event, or the locks and channels built on it, can turn off the default features. It then enables only what it uses, such as features = ["lock"], and builds none of the runtime.

Why?

The project grew out of the need for a single-threaded runtime in zbus that it would use by default. It was split into a separate project so non-zbus users can use it too.

License

MIT

Sponsors