How to run an async task as detached

Help
1

I will preface this by saying that i have no experience with async in any language. I mostly understand how its supposed to be implemented and I do have some experience with callback async in c++ with asio.

I’m playing around with the new Io in the master branch and trying to build a simple echo server. What I’m stuck on is how I’m supposed to register multiple in-flight tasks from the accept loop. I mean in here:

while (true) {
    const stream = io.async(net.Server.accept, .{ server, io });
    defer if (stream.cancel(io)) |s| s.close(io);

    handleConnection(io, try stream.await()); // How??
}

With callback it’s easy, just register a callback in the loop and handle everything in the callback. But in the code above, it would need to await anything that might be inside handleConnection before starting to accept the next connection. Also I might want to do something more complex inside it, like echoing each line, not just the first. For this I would either need a loop or to re-register it.

The real issue that I’m running into is that io.async() returns a future that I need to take care off, and that I need to await in order to have it execute its task. Either I still await it inside the accept loop or keep an array of all connection futures and only await at the end (this is essentially a memory leak). I’ve also looked into std.Io.Group but that also seems to be a await-at-the-end kind of deal.

Is Zig missing some sort of yield or am I just missing an obvious solution?

2

You can’t detach futures, you must handle them eventually, but it doesn’t have to be right away either.

Your more immediate problem is, you should async handleConnection instead of accept.
Well, you can do both, but the code you provided doesn’t take advantage of that, so it’s just adding unnecessary overhead to a call.

std.Io.Group is exactly what you want to await the futures later, think of it as an arena for futures.
wait or cancel the group outside the loop, basically before your sever exits. you can do so with a defer.

1 Like
3

I assumed that std.Io.Group would eventually fill up with futures after a lot of connection, but thinking more about it, it’d make more sense for it to start executing them as soon as possible. Even so, it doesn’t seem to integrate with normal error handling. If this is the idiomatic way, it seems very cumbersome. What if I want to bubble an error up to the group’s scope? Would I need to use a std.Io.Queue(MyError!void)?

For anyone trying to achieve a similar thing, I did manage to scrape something together:

const std = @import("std");
const log = std.log;
const net = std.Io.net;
const Dir = std.Io.Dir;

pub fn main() !void {
    var thread_pool: std.Io.Threaded = .init(std.heap.smp_allocator);
    defer thread_pool.deinit();
    const io = thread_pool.io();

    const port = 6969;
    const address = net.IpAddress{ .ip4 = .unspecified(port) };
    log.info("Listening on port {}", .{port});

    var server = try address.listen(io, .{});
    defer server.deinit(io);

    var group: std.Io.Group = .init;
    defer group.cancel(io);

    while (true) {
        var accept = io.async(net.Server.accept, .{ &server, io });
        defer _ = accept.cancel(io) catch {};
        const stream = try accept.await(io);

        log.info("Accepted connection from {f}", .{stream.socket.address});
        group.async(io, handleConnection, .{ io, stream });
    }
}

fn handleConnection(io: std.Io, stream: net.Stream) void {
    defer {
        log.info("Closing connection {f}", .{stream.socket.address});
        stream.close(io);
    }
    var read_buffer: [512]u8 = undefined;
    var write_buffer: [512]u8 = undefined;
    var reader = stream.reader(io, &read_buffer);
    var writer = stream.writer(io, &write_buffer);

    while (true) {
        const size = reader.interface.streamDelimiter(&writer.interface, '\n') catch |err| {
            log.err("Stream error: {t}", .{err});
            break;
        };
        _ = reader.interface.discardDelimiterInclusive('\n') catch |err| {
            log.err("Discard delimiter error: {t}", .{err});
            break;
        };
        writer.interface.writeByte('\n') catch |err| {
            log.err("Write newline error: {t}", .{err});
            break;
        };
        writer.interface.flush() catch |err| {
            log.err("Flush error: {t}", .{err});
            break;
        };
        log.info("Recieved {} bytes from {f}", .{ size, stream.socket.address });
    }
}

It works fine, but I don’t love all the catch |err|s. Also I’m not sure that it’s impossible to leak a stream from defer _ = accept.cancel(io) catch {}; or that operations on the reader and writer integrate implicitly into the event loop, though I would assume so.

1 Like
4

Those details are up to the Io implementation, Threaded uses a linked list with heap allocated nodes.

You can propagate the error through an out parameter, but it is just better practice to deal with them in handleConnection.

it is, cancel does return the result of the task if it finished before it could be cancelled, you have to deal with that defer if(accept.cance(io)) |stream| stream.close() else |_| {};
I’m surprised that discarding hides the type mismatch of the success case being a stream and the error case being void (your catch {}). Though I can see how It’d make sense to do that, I dont think it should as it hides resource managment errors like this.

the interfaces have no knowledge of Io, and they shouldn’t, otherwise they wouldn’t be arbitrary streams of data.
The implementation however should go through Io, which you can see as you have to give it io when you create them.

1 Like