Tasks

A task is a function that runs alongside others. Each task has its own small stack, and the tasks on a thread take turns: one runs until it waits (for a timer, for another task, or by yielding), and then another runs. A runtime has a few such threads, its workers, so tasks also run in parallel. This is the model of Go’s goroutines and of libuv or tokio event loops, without async/await: a function that waits looks like any other function, and any function can wait, however deep in the call stack. A socket call that has to wait parks only its task, so one worker can serve thousands of connections.

WASI cannot switch stacks, so on WebAssembly (--target wasm32-wasip1) starting tasks (async.run, async.run_with) panics (see WebAssembly).

Tasks live in the async module:

import async;

main(): int {
    var log = Array<String>();
    async.run(() => {
        # A task starts when the running task next waits.
        var worker = async.Task<int>.spawn_local(() => {
            log.append("worker starts");
            async.yield_now();
            log.append("worker ends");
            return 42;
        });
        log.append("first task");
        var answer = worker.join().value();   # waits; the worker runs
        assert(answer == 42, "join returns the body's value");
        log.append("joined");
    });
    assert(log[0] == "first task" && log[1] == "worker starts", "tasks take turns");
    assert(log[2] == "worker ends" && log[3] == "joined", "join waited");
    return 0;
}

• async.run(body) runs body as the first task and returns once it has returned and every task it started has finished. run_with(options, body) does the same with settings; RuntimeOptions().stack_size(bytes) sets how much stack each task reserves (256 KiB by default).
• Task<T>.spawn_local(body) starts a task on the running task’s worker and returns its Task<T>. The new task runs the next time the running task waits. T is the body’s result type; use Task<void> for a body that returns nothing.
• task.join() waits for the task and returns a Result<T, TaskError>: the body’s value, or an error if the task was cancelled.
• async.yield_now() lets every other ready task run first.
• task.is_finished() says whether the body has returned.

A task from spawn_local stays on the thread of the task that started it, so it may use any value that task uses, as the closures above share log. Nothing needs to be thread-safe.

Several workers

run turns the calling thread into the first worker and starts one more thread per logical CPU after it (RuntimeOptions().workers(count) sets how many workers in all). async.spawn<T>(body) starts a task that whichever worker is free first runs, in parallel with the others:

import async;

sum(from: long, to: long): long {
    var total: long = 0;
    var i = from;
    while (i < to) {
        total = total + i;
        i = i + 1;
    }
    return total;
}

main(): int {
    async.run_with(async.RuntimeOptions().workers(4), () => {
        # Two halves of the work, on two workers at the same time.
        var low = async.spawn<long>(() => { return sum(0, 500000); });
        var high = async.spawn<long>(() => { return sum(500000, 1000000); });
        var total = low.join().value() + high.join().value();
        assert(total == 499999500000, "both halves were added");
    });
    return 0;
}

Because such a task may run on another thread, its body must be a shared function: it may capture only thread-safe values, as a thread’s body. Its result crosses threads too, so spawn constrains it T: Shared: a thread-safe type such as a number, a [Shared] struct, or a shared function type. async.spawn<String> is a compile error at String. join, cancel and is_finished work as for spawn_local, from the task that started it.

A task, once started, stays on its worker until it finishes, so the tasks it starts with spawn_local share values with it safely. An idle worker takes the next task that no worker has started yet. A task that computes without waiting still holds up the other tasks on its own worker (not the other workers), so a long computation should call yield_now() now and then.

async.spawn_on_each_worker(body) starts one task per worker, each on its own worker, and returns them in worker order. It is how a server runs an accept loop on every worker (see Sockets).

Blocking calls

Some work would hold a worker up: a call that blocks in the operating system (reading a file, a C library that waits) or a long computation. async.run_blocking<T>(body) runs body on a separate thread and parks the calling task until it returns, so the worker keeps running its other tasks meanwhile:

import async;

const ticks = async.Atomic<int>(0);

main(): int {
    async.run_with(async.RuntimeOptions().workers(1), () => {
        # Keeps running on the one worker while the blocking call waits.
        var ticker = async.Task<void>.spawn_local(() => {
            while (ticks.load() < 3) {
                async.sleep(5);
                ticks.fetch_add(1);
            }
        });
        var answer = async.run_blocking<int>(() => {
            async.Thread.sleep(50);   # blocks a pool thread, not the worker
            return 42;
        });
        assert(answer == 42, "the blocking call's result");
        assert(ticks.load() > 0, "the worker ran the ticker meanwhile");
        ticker.join();
    });
    return 0;
}

The runtime keeps a pool of threads for these calls. It starts one when a call finds none idle, up to RuntimeOptions().blocking_threads(count) (64 by default); past that, calls wait for a free one. As for spawn, body must be a shared function and T must be thread-safe. body always runs to the end: cancelling the task, or a timeout around the call, takes effect once the call has returned. Outside a task, run_blocking just calls body.

Waiting

async.sleep puts a task to sleep and runs the others meanwhile, so many tasks can wait at the same time on one thread. It takes an os.Duration (see The operating system), or a number of milliseconds; a duration that is not a whole number of milliseconds rounds up:

import async;
import os;

main(): int {
    var order = Array<int>();
    async.run(() => {
        var tasks = Array<async.Task<void>>();
        var i = 0;
        while (i < 3) {
            var n = i;
            # The later a task starts, the shorter it sleeps.
            tasks.append(async.Task<void>.spawn_local(() => {
                async.sleep(os.Duration.from_milliseconds(((3 - n) * 20) as ulong));
                order.append(n);
            }));
            i = i + 1;
        }
        for (var task in tasks) {
            task.join();
        }
    });
    # All three slept at the same time, on one thread.
    assert(order[0] == 2 && order[1] == 1 && order[2] == 0, "shortest sleep first");
    return 0;
}

Outside a task, sleep blocks the thread, and yield_now does nothing.

Tasks are never interrupted. A task that computes for a long time without waiting keeps every other task waiting, so a long loop should call yield_now() now and then.

Thread.sleep in a task is async.sleep.

Mutexes, channels and conditions

The synchronization types work in tasks too. When a task has to wait for a Mutex, for a value from a Channel or for room in a bounded one, in Condition.wait, for an RwLock, or in Semaphore.acquire, Barrier.wait or Latch.wait, it parks and its worker runs other tasks, just as it does for a socket. The task that holds the lock or will send the value can be on the same worker, on another worker, or a plain thread:

import async;

struct Log {
    var lines: Array<String>;
    construct() {
        this.lines = Array<String>();
    }
}

const log = async.Mutex<Log>(Log());
const jobs = async.Channel<int>(2);

main(): int {
    # One worker: every task below shares one thread.
    async.run_with(async.RuntimeOptions().workers(1), () => {
        # The consumer parks in the channel until a value arrives.
        var consumer = async.Task<int>.spawn_local(() => {
            var total = 0;
            for (var job in jobs) {
                total = total + job;
            }
            return total;
        });
        # A task may hold the lock while it waits; the other task parks
        # until it is free.
        var writer = async.Task<void>.spawn_local(() => {
            with (var l = log.lock()) {
                l.lines.append("first");
                async.sleep(10);
                l.lines.append("second");
            }
        });
        async.yield_now();
        with (var l = log.lock()) {
            l.lines.append("third");
        }
        writer.join();

        # The producer parks while the bounded channel is full.
        var i = 1;
        while (i <= 5) {
            jobs.send(i);
            i = i + 1;
        }
        jobs.close();
        assert(consumer.join().value() == 15, "every job arrived");
    });
    with (var l = log.lock()) {
        assert(l.lines[2] == "third", "the lock waited for the writer");
    }
    return 0;
}

• Each task counts as its own holder of a Mutex, so two tasks on one worker exclude each other, and a task may hold a lock across its own waits (a sleep, a socket call, another lock).
• A cancel, or a timeout around the call, ends a channel wait: send returns false and receive returns an error (and a for loop over the channel ends). A task waiting for a Mutex still gets the lock, and Condition.wait still waits for a notification, since the caller’s loop would only wait again; wait_for returns when its own time runs out.
• A Select parks its task on every channel at once, and a cancel or timeout ends its wait with an error. Condition.wait_while parks like wait.
• Outside a task they block the thread, as before.

Task-local values

A TaskLocal<T> is a variable with a separate value in every task, as a ThreadLocal<T> has one per thread. Tasks that take turns on one worker share its thread, so only a TaskLocal can tell them apart:

import async;

# The id of the request the running task is handling.
const request_id = async.TaskLocal<long>();

# Deep in the call chain, without the id passed down as a parameter.
current_request(): long {
    return request_id.get();
}

main(): int {
    # One worker: the three tasks take turns on one thread, so a
    # ThreadLocal could not tell them apart.
    async.run_with(
        async.RuntimeOptions().workers(1),
        () => {
            var tasks = Array<async.Task<bool>>();
            for (var id in [1, 2, 3]) {
                tasks.append(
                    async.Task<bool>.spawn_local(
                        () => {
                            request_id.set(id);
                            async.yield_now();  # the other tasks set their ids meanwhile
                            return current_request() == id;
                        }
                    )
                );
            }
            for (var task in tasks) {
                assert(task.join().value(), "each task sees its own id");
            }
            # This task never set one.
            assert(!request_id.is_set(), "a task starts with no value");
        }
    );
    return 0;
}

• set(value) stores the calling task’s value and get() returns it (the default of T, zero or null, until the task sets one). is_set() and clear() work as on a ThreadLocal.
• A task starts with no value, even one started by a task that has one, and its values are released when it finishes.
• Code outside any task (before async.run, or on a plain Thread) has one value per thread, like a ThreadLocal.
• For a T that is not thread-safe, such as a String, reach the value with with (var value = local.enter()) { ... }, as for a ThreadLocal; get() is then a compile error.

Sockets

A socket call in a task that would have to wait (for a connection, for data, or for room to send) parks the task and lets the others run until the socket is ready. The socket API is the same one a plain program uses; only the waiting differs:

import async;
import io;

# Echoes everything a client sends until it closes the connection.
serve(stream: io.TcpStream) {
    var buffer = Array<u8>();
    while (true) {
        var r = stream.read(buffer, 4096);   # parks this task until data arrives
        if (r.is_err() || r.value() == 0) {
            return;
        }
        stream.write_all(buffer);
        buffer.clear();
    }
}

main(): int {
    async.run(() => {
        var listener = io.TcpListener.bind(io.SocketAddress.parse("127.0.0.1:0").value()).value();
        var address = listener.local_address().value();
        var clients = 10;

        # The server: one task per connection, all on this thread.
        async.Task<void>.spawn_local(() => {
            while (true) {
                var accepted = listener.accept();
                if (accepted.is_err()) {
                    return;   # cancelled when the first task returns
                }
                var (stream, peer) = accepted.value();
                async.Task<void>.spawn_local(() => { serve(stream); });
            }
        });

        # The clients, as tasks on the same thread.
        var replies = Array<async.Task<ulong>>();
        var i = 0;
        while (i < clients) {
            var k = i;
            replies.append(async.Task<ulong>.spawn_local(() => {
                var stream = io.TcpStream.connect(address).value();
                var message = $"hello from {k}";
                stream.write_all(message);
                var buffer = Array<u8>();
                stream.read_exact(buffer, message.size);
                return buffer.count;
            }));
            i = i + 1;
        }
        for (var reply in replies) {
            assert(reply.join().value() > 0, "the server echoed the message");
        }
    });
    return 0;
}

Outside a task, the same calls block the thread as before. A socket belongs to the worker whose task made it: tasks started with spawn_local can share it, and each spawned task makes its own.

Waiting on several sockets

To wait on whichever of several sockets is ready first, give each socket a task that reads it and sends what it read to a channel, and wait on the channels with a Select. Closing a channel when its socket closes lets the select stop waiting on it. Tasks are cheap, so this costs little, and the reading code stays the ordinary blocking-style loop:

import async;
import io;

# What the select returns: bytes that arrived on one of the connections.
enum Event {
    Data(from: int, count: ulong),
}

# Reads `stream` until the peer closes it, sending the size of each read to
# `into`, then closes `into` so the select stops waiting on it.
forward(stream: io.TcpStream, into: async.Channel<ulong>) {
    var buffer = Array<u8>();
    while (true) {
        var r = stream.read(buffer, 4096);   # parks this task, not the others
        if (r.is_err() || r.value() == 0) {
            into.close();
            return;
        }
        into.send(r.value());
        buffer.clear();
    }
}

main(): int {
    async.run(() => {
        var listener = io.TcpListener.bind(io.SocketAddress.parse("127.0.0.1:0").value()).value();
        var address = listener.local_address().value();
        var first = io.TcpStream.connect(address).value();
        var (from_first, first_peer) = listener.accept().value();
        var second = io.TcpStream.connect(address).value();
        var (from_second, second_peer) = listener.accept().value();

        # One task per socket turns what it reads into channel messages.
        var first_bytes = async.Channel<ulong>();
        var second_bytes = async.Channel<ulong>();
        async.Task<void>.spawn_local(() => { forward(from_first, first_bytes); });
        async.Task<void>.spawn_local(() => { forward(from_second, second_bytes); });

        first.write_all("hello");
        first.shutdown_write();
        second.write_all("hi there");
        second.shutdown_write();

        # The select then waits on both sockets at once, through their channels.
        var select = async.Select<Event>();
        select.receive(first_bytes, (count: ulong): Event => Event.Data(0, count));
        select.receive(second_bytes, (count: ulong): Event => Event.Data(1, count));
        ulong first_total = 0;
        ulong second_total = 0;
        var running = true;
        while (running) {
            match (select.wait()) {
                Ok(event) => {
                    match (event) {
                        Data(from, count) => {
                            if (from == 0) {
                                first_total = first_total + count;
                            } else {
                                second_total = second_total + count;
                            }
                        }
                    }
                }
                Err(error) => {
                    running = false;   # both connections closed
                }
            }
        }
        assert(first_total == 5 && second_total == 8, "every byte arrived once");
    });
    return 0;
}

Files

File.read, write and write_all in a task park the task until the system call is done, and the worker runs other tasks meanwhile. With the IoUring backend the kernel does the read or write itself; with the others a thread from run_blocking’s pool makes the blocking call. Opening, seeking, metadata and the other File calls still block the thread, as they are quick on local disks. A cancel or a timeout does not cut a read or write short: it takes effect once the call returns.

The array a read fills or a write sends from is in use until the call returns, so growing it from another task meanwhile panics (“Array: the buffer is in use by an I/O operation”). A String passed to write_all is copied first.

Accepting on every worker

An io.SharedTcpListener is a listening socket that any thread can accept on. Give each worker an accept loop with spawn_on_each_worker; each connection goes to one of the waiting workers, and the stream it returns belongs to that worker:

import async;

import io;

const served = async.Atomic<int>(0);

main(): int {
    async.run_with(async.RuntimeOptions().workers(4), () => {
        var listener = io.SharedTcpListener.bind(io.SocketAddress.parse("127.0.0.1:0").value()).value();
        var port = listener.local_address().value().port();

        # One accept loop per worker; each connection goes to one of them.
        async.spawn_on_each_worker(() => {
            while (true) {
                var accepted = listener.accept();
                if (accepted.is_err()) {
                    return;   # cancelled when the first task returns
                }
                var (stream, peer) = accepted.value();
                served.fetch_add(1);
                async.Task<void>.spawn_local(() => {
                    stream.write_all("hello").value();
                });
            }
        });

        var i = 0;
        while (i < 8) {
            var client = io.TcpStream.connect_host("127.0.0.1", port).value();
            var reply = Array<u8>();
            while (client.read(reply, 64).value() > 0) {
            }
            assert(reply.count == 5, "greeted");
            i = i + 1;
        }
    });
    assert(served.load() == 8, "every connection was accepted once");
    return 0;
}

The listener has no close(), since other workers may be waiting on it. It closes when the last reference goes: here, once run has cancelled the loops.

Moving a stream to another worker

stream.into_transfer() moves a connection out of a TcpStream into an io.SocketTransfer, a thread-safe token that a shared function can capture or a Channel can carry, and io.TcpStream.adopt(transfer) takes it out again on the receiving worker. The old stream is closed afterwards, only the first adopt gets the connection, and a token no one adopts closes the connection when it goes:

import async;
import io;

main(): int {
    async.run_with(async.RuntimeOptions().workers(2), () => {
        var listener = io.TcpListener.bind(io.SocketAddress.parse("127.0.0.1:0").value()).value();
        var port = listener.local_address().value().port();
        var client = async.Task<ulong>.spawn_local(() => {
            var stream = io.TcpStream.connect_host("127.0.0.1", port).value();
            var reply = Array<u8>();
            while (stream.read(reply, 64).value() > 0) {
            }
            return reply.count;
        });

        var (stream, peer) = listener.accept().value();
        # A stream is not thread-safe; its transfer token is.
        var transfer = stream.into_transfer();
        var handler = async.spawn<void>(() => {
            var mine = io.TcpStream.adopt(transfer).value();
            mine.write_all("from any worker").value();
        });
        handler.join();
        assert(client.join().value() == 15, "the moved stream answered");
    });
    return 0;
}

When no task is ready to run, a worker blocks until a socket is ready, the next timer is due, or another worker wakes it; while tasks are running, it also checks the sockets every few dozen tasks, so a busy task cannot starve the network.

Backends

How the runtime waits for sockets is its I/O backend, an async.Backend:

Backend.Automatic, the default, picks epoll, kqueue or IOCP, whichever the system has. RuntimeOptions().backend(kind) asks for one (and run_with panics if the system does not have it), and the PEPPER_IO_BACKEND environment variable (epoll, io_uring, kqueue, iocp or poll) overrides the automatic choice without changing the program, for testing and benchmarks. async.current_backend() says which one is running.

io_uring pays off as the amount of multiplexed I/O grows: many operations in flight, submitted and completed in batches through the ring rather than one system call each. With the IoUring backend, a stream socket’s read and write in a task go to the kernel as a receive or send request, which it runs once the socket is ready; the worker submits every request queued since its last look, and collects every completion, in one system call. In our echo benchmark that makes about an eighth of epoll’s system calls, runs as fast as epoll from 64 to 1,024 connections (half again as fast as io_uring used only for readiness), and stays about 10% behind at 4,096.

A cancel or a timeout asks the kernel to stop a request and waits for it to finish, since the kernel is writing into or reading from the array until then. A receive that got data before the cancel took effect returns that data instead of failing, so no bytes are lost. As with files, growing the array from another task meanwhile panics. Accepting, connecting, UDP sockets, and a second task reading (or writing) a socket that one already reads (or writes) still wait for readiness.

Timeouts

async.timeout(milliseconds) limits how long the waits inside a with block may take:

import async;
import io;

main(): int {
    async.run(() => {
        var listener = io.TcpListener.bind(io.SocketAddress.parse("127.0.0.1:0").value()).value();

        # Nobody connects, so the accept gives up after 50 milliseconds.
        with (var deadline = async.timeout(50)) {
            var accepted = listener.accept();
            assert(accepted.is_err() && accepted.error().is_timed_out(), "accept timed out");
            assert(deadline.is_expired(), "the limit has passed");
        }

        # A join gives up too; the task itself carries on until cancelled.
        var slow = async.Task<int>.spawn_local(() => {
            async.sleep(1000000);
            return 42;
        });
        with (async.timeout(20)) {
            var result = slow.join();
            assert(result.error().is_timed_out(), "join timed out");
        }
        slow.cancel();
    });
    return 0;
}

Once the limit passes, a socket call that has to wait fails with an IoError for which is_timed_out() holds, join fails with a TaskError for which is_timed_out() holds (the task it waited for carries on), and sleep returns early. Code that does not wait is never interrupted. The limit ends with the block. Timeouts nest, and an inner one cannot extend an outer one. deadline.is_expired() says whether the limit has passed. timeout panics outside a task.

Cancelling

task.cancel() asks a task to stop. Nothing stops it by force: from then on async.is_cancelled() is true inside it, sleep returns at once, and join on it fails with a TaskError whose is_cancelled() is true. A task waiting in join returns the same error when it is cancelled itself, and a socket call waiting in a cancelled task fails with an IoError whose is_cancelled() is true. When the first task of a run returns, every task still running is cancelled, on every worker, and run waits for each to finish:

import async;

main(): int {
    var polls = Array<int>();
    async.run(() => {
        var poller = async.Task<void>.spawn_local(() => {
            while (!async.is_cancelled()) {
                polls.append(1);
                async.sleep(5);
            }
        });
        async.sleep(30);
        poller.cancel();
        var result = poller.join();
        assert(result.is_err() && result.error().is_cancelled(), "a cancelled task's join fails");

        # A task still running when the first task returns is cancelled too,
        # and run waits for it to finish.
        async.Task<void>.spawn_local(() => {
            async.sleep(1000000);   # returns at once when cancelled
        });
    });
    assert(polls.count > 0, "the poller ran until cancelled");
    return 0;
}

If every task is waiting for another task and none can continue, run panics instead of waiting forever.

How it works

Each task runs on a fiber: a stack of its own and a saved set of registers. Switching tasks saves the running task’s registers on its stack and restores the other’s, a few instructions of assembly on Linux and macOS (the __switch_context intrinsic) and the Fibers API on Windows. A task’s stack is reserved address space with a guard page below it; only the pages the task touches use memory, and a task that overflows its stack crashes the program as a thread would. Stacks are reused for later tasks.

Each worker has its own ready queue, timers and backend. Workers talk through a queue of messages each (a spawned task finished, was cancelled, or was woken by a Mutex, Condition or Channel) and a doorbell that ends a worker’s wait in its backend from another thread: an eventfd with epoll and io_uring, a user event with kqueue, a posted completion with IOCP, and a pipe (on Windows, a loopback UDP socket that a second socket sends to) with poll(). run_blocking’s pool threads are plain threads that report back through the same messages.

Current status

• File calls other than read, write and write_all (opening, seeking, sync, directories) block the thread in a task.
• On Windows, a ThreadLocal has a separate value in each task (it is built on fiber local storage); elsewhere the tasks of a thread share it.

The reference page documents every function.