async await demo

example.zig

const std = @import("std");
const assert = std.debug.assert;
const Allocator = std.mem.Allocator;
const Io = std.Io;

pub fn main() !void {
    var debug_allocator: std.heap.DebugAllocator(.{}) = .init;
    const gpa = debug_allocator.allocator();
    //const gpa = std.heap.smp_allocator;

    var thread_pool: std.Thread.Pool = undefined;
    try thread_pool.init(.{
        .allocator = gpa,
        .n_jobs = std.Thread.getCpuCount() catch 1,
    });
    defer thread_pool.deinit();

    const io = thread_pool.io();

    //var event_loop: std.Io.EventLoop = undefined;
    //try event_loop.init(gpa);
    //defer event_loop.deinit();

    //const io = event_loop.io();

    var first_half = io.@"async"(calcSum, .{ 0, 100 });
    var second_half = io.@"async"(calcSum, .{ 100, 200 });

    var rot13 = io.@"async"(processTextFile, .{io});

    var select_example = io.@"async"(selectExample, .{ io, Io.Dir.cwd() });
    defer select_example.cancel(io) catch {};

    var queue: Io.Queue(i32) = .init(&.{});
    var producer = io.@"async"(producerRun, .{ io, &queue });
    defer producer.cancel(io) catch {};
    var consumer = io.@"async"(consumerRun, .{ io, &queue });
    defer _ = consumer.cancel(io);

    const total = first_half.@"await"(io) + second_half.@"await"(io);
    std.log.info("total: {d}", .{total});

    try rot13.@"await"(io);

    const consumer_sum = consumer.@"await"(io);
    std.log.info("consumer sum = {d}", .{consumer_sum});

    try select_example.@"await"(io);
}

fn producerRun(io: Io, queue: *Io.Queue(i32)) !void {
    for (0..100) |i| {
        try queue.putOne(io, @intCast(i * 2 + 1));
    }
    try queue.putOne(io, 999);
}

fn consumerRun(io: Io, queue: *Io.Queue(i32)) i32 {
    var sum: i32 = 0;
    while (true) {
        const n = queue.getOne(io) catch return undefined;
        if (n == 999) return sum;
        sum += n;
    }
}

fn calcSum(start: usize, end: usize) usize {
    var sum: usize = 0;
    for (start..end) |i| {
        sum += i;
    }
    std.log.debug("calcSum returning {d}", .{sum});
    return sum;
}

fn processTextFile(io: Io) !void {
    var future_in_file = io.@"async"(Io.Dir.openFile, .{ .cwd(), io, "example.txt", .{} });
    defer if (future_in_file.cancel(io)) |f| f.close(io) else |_| {};

    var future_out_file = io.@"async"(Io.Dir.createFile, .{ .cwd(), io, "output.txt", .{} });
    defer if (future_out_file.cancel(io)) |f| f.close(io) else |_| {};

    const in_file = try future_in_file.@"await"(io);
    var buffer: [5000]u8 = undefined;
    const n = try in_file.readAll(io, &buffer);

    const contents = buffer[0..n];
    for (contents) |*elem| elem.* = elem.* +% 1;

    const out_file = try future_out_file.@"await"(io);
    try out_file.writeAll(io, contents);
}

fn selectExample(io: Io, dir: Io.Dir) !void {
    var a_future = io.@"async"(Io.Dir.writeFile, .{ dir, io, .{ .sub_path = "a.txt", .data = "a contents" } });
    defer _ = a_future.cancel(io) catch {};
    var b_future = io.@"async"(Io.Dir.writeFile, .{ dir, io, .{ .sub_path = "b.txt", .data = "b contents" } });
    defer _ = b_future.cancel(io) catch {};
    var timeout = io.@"async"(Io.sleepDuration, .{ io, .ms(100) });
    defer timeout.cancel(io) catch {};

    switch (io.select(.{
        .a_future = &a_future,
        .b_future = &b_future,
        .timeout = &timeout,
    })) {
        .a_future => |res| {
            std.log.info("a future won the race", .{});
            try res;
        },
        .b_future => |res| {
            std.log.info("b future won the race", .{});
            try res;
        },
        .timeout => {
            return error.Timeout;
        },
    }
}