Async, Await and Concurrency in Zig with std.Io

What is std.Io for?

#

std.Io is the interface for any access to the outside world (files, networking, processes, time, etc). It also holds the tools you’ll need for concurrency (async/await, concellation, queues, mutex, etc).

The way to think about this is anything that can block goes through an io value, which gives you two things:

• A function’s signature will tell you what it does, if it takes an io it can wait on something.
• You (the caller) get to decide how the waiting happens. A function that takes an io can run on a thread pool, on a single thread, an event loop, etc, all without changing the function.

Io is a value you pass (like an allocator)

#

When working with allocators we know that passing an allocator to a function means that function has the ability to allocate. Likewise, a function that takes an io can block, touch files, read the clock, spawn workers, etc. Just like the caller decides the allocation strategy by picking an allocator prior to passing to the function, the caller also picks the implementation before passing io. The function written that uses the io runs unchanged on any implementation the caller chooses. It’s magical.

Getting an Io

#

Approach 1: short method with main

#

Since zig 0.16.0 and juicy main merge we can give main a std.process.Init parameter and the startup code builds everything for you (see the link for a deeper dive).

pub fn main(init: std.process.Init) !void {
  const io = init.io;
  const gpa = init.gpa;
  // also: init.arena, init.environ_map, init.minimal.args
}

Approach 2: the manual way

#

var threaded: std.Io.Threaded = .init(std.heap.page_allocator, .{});
defer threaded.deinit();
const io = threaded.io();

Take the second approach (the manual way) when you need to choose settings. Be aware that the allocator we pass is only used as the async/concurrent ledger and must be thread-safe.

Clocks, timestamps, and durations (oh my)

#

Time always comes from the operating system (the outside world) so it goes through io, and you need to specify which clock.

• .awake never goes backwards, use it to measure things like how long it took and to sleep.
• .real is used to tell the time, it’s essentially an IRL wall clock in unix time.
• .boot, .cpu_process, .cpu_thread exists for rarer cases we can get into in a different post.

const start = Io.Timestamp.now(io, .awake);
// do some work
std.debug.print("{f}\n", .{start.untilNow(io, .awake)});

Asynchronicity

#

var pot = io.async(cook, .{ io, step });
defer _ = pot.cancel(io) catch {};
// do other things
try pot.await(io);

• io.async(function, args_tuple) starts the call and returns an Io.Future(T), where T is the function’s return type, error union included.
• future.await(io) blocks until the task is done and returns what the direct call would’ve returned, so you try it the same way. Futures must be var type because awaiting changes the value.

Every future must be awaited or canceled, otherwise the resources leak.

async is a hint and concurrent is a guarantee

#

We’ll write some code later in this post to better explain why but here’s what to know:

• io.async promises only that the result is ready when you await. The io is allowed to run the function at that point in time, before async even returns.
• io.concurrent promises the function gets its own unit of concurrency and truly runs alongside the caller.

In the code we write later we’ll write a producer that fills a one-slot queue, and a consumer that drains it. If the consumer were started with async and run inline, it would wait forever for items the producer never gets to send. With concurrent, it fails cleanly at the start instead.

The rule of thumb here is use async when tasks can overlap, use concurrent when they must overlap. Producer/consumer pairs, servers accepting connections while handling other connections, timeouts, etc all need concurrent. If your code is still correct running inline then use async but if your code would be wrong or deadlock inline, then use concurrent.

Cancellation

#

Calling future.cancel(io) is essentially “await, but first ask the task to stop”. It never abandons a running task, instead it waits until the task actually returns, then gives you its results.

The request to cancel is delivered at cancellation points which is any io call whose error set includes error.Canceled.

Let’s write some code

#

The io value

#

const std = @import("std");
const Io = std.Io;
const print = std.debug.print;

// takes an io so it's allowed to wait.
fn boil(io: Io, minutes: u16) Io.Cancelable!void {
    // a minute is 10ms around these parts.
    try io.sleep(.fromMilliseconds(minutes * 10), .awake);
}

// no io, so it can't block, can't touch the clock, just potato math.
fn servings(potatoes: u32) u32 {
    return potatoes / 2;
}

pub fn main(init: std.process.Init) !void {
    const io = init.io;

    const clocked_in = Io.Timestamp.now(io, .real);
    const start = Io.Timestamp.now(io, .awake);

    try boil(io, 20);
    const plates = servings(9);

    print("boiled for 20 minutes in {f}\n", .{start.untilNow(io, .awake)});
    print("that's enough for {d} plates\n", .{plates});
    print("clocked in at unix {d}\n", .{clocked_in.toSeconds()});
}

The async await

#

const std = @import("std");
const Io = std.Io;
const print = std.debug.print;

const Step = struct { what: []const u8, minutes: u16 };

const dinner = [_]Step{
    .{ .what = "boil the potatoes", .minutes = 30 },
    .{ .what = "fry the onions", .minutes = 10 },
    .{ .what = "set the table", .minutes = 5 },
};

fn cook(io: Io, step: Step) Io.Cancelable!void {
    try io.sleep(.fromMilliseconds(step.minutes * 10), .awake);
    print("  done: {s}\n", .{step.what});
}

pub fn main(init: std.process.Init) !void {
    const io = init.io;

    // one cook, one thing at a time.
    var start = Io.Timestamp.now(io, .awake);
    for (dinner) |step| try cook(io, step);
    print("by myself: {f}\n", .{start.untilNow(io, .awake)});

    // same recipe, every step gets a future.
    start = Io.Timestamp.now(io, .awake);
    var futures: [dinner.len]Io.Future(Io.Cancelable!void) = undefined;
    for (&futures, dinner) |*f, step| f.* = io.async(cook, .{ io, step });
    // if an await fails early the rest still get cleaned up.
    defer for (&futures) |*f| {
        _ = f.cancel(io) catch {};
    };
    for (&futures) |*f| try f.await(io);
    print("with help: {f}\n", .{start.untilNow(io, .awake)});
}

Async vs concurrent

#

const std = @import("std");
const Io = std.Io;
const print = std.debug.print;

// weight of each potato in the sack, in grams.
const sack = [_]u32{ 180, 210, 150, 200, 170 };

// peels one potato at a time into a bowl that only fits one.
fn peel(io: Io, bowl: *Io.Queue(u32)) Io.Cancelable!void {
    defer bowl.close(io);
    for (sack) |grams| {
        bowl.putOne(io, grams) catch |err| switch (err) {
            error.Closed => return,
            error.Canceled => |e| return e,
        };
    }
}

// takes potatoes out of the bowl until the peeler closes it.
fn boil(io: Io, bowl: *Io.Queue(u32)) Io.Cancelable!u32 {
    var grams: u32 = 0;
    while (bowl.getOne(io)) |potato| {
        grams += potato;
    } else |err| switch (err) {
        error.Closed => return grams,
        error.Canceled => |e| return e,
    }
}

fn makeDinner(io: Io, kitchen: []const u8) !void {
    print("{s}\n", .{kitchen});

    var slot: [1]u32 = undefined;
    var bowl: Io.Queue(u32) = .init(&slot);

    // the pot has to be going while we peel, so we go concurrent here, not async.
    var pot = io.concurrent(boil, .{ io, &bowl }) catch |err| {
        print("  no second burner: error.{t}\n", .{err});
        return;
    };
    defer _ = pot.cancel(io) catch {};

    // we peel right here on the calling task.
    try peel(io, &bowl);
    print("  {d}g of potatoes went in the pot\n", .{try pot.await(io)});
}

pub fn main() !void {
    var threaded: Io.Threaded = .init(std.heap.page_allocator, .{});
    defer threaded.deinit();
    try makeDinner(threaded.io(), "kitchen with helpers");

    var single: Io.Threaded = .init_single_threaded;
    try makeDinner(single.io(), "kitchen with just me");
}

Later

#

In a future post I plan to dive into racing with Select, grouping threads with Group and shared memory with mutex and atomic.