Io.async / io.concurrent lacks per-call context argument, making priority scheduling difficult to express with `std.Io`

Help
,
1

Recently I started dabbling with zig and in particular with the std.Io. Comming from the rust async world, where I use non mainstream async runtimes such as monoio or glommio (glommio will be subject of this topic in particular), I found the interface in it’s current shape to be biased towards certain implementations.

One of the inspirations for glommio was a C++ framework seastar, which uses scheduling groups. Each of the group has a configurable number of shares in order to proportionally divide CPU time across different type of work. Glommio api for spawning a task looks as follows (I’ve trimmed some extra stuff for simplicity):

// Create low priority task queue queue
let low_prio_tq = glommio::executor().create_task_queue(
    Shares::Static(100),
    Latency::NotImportant,
);

// Create high priority task queue queue
let high_prio_tq = glommio::executor().create_task_queue(
    Shares::Static(1000),
    Latency::NotImportant,
);

// Create the task (in zig) `io.async` / `io.concurrent` call on high priority task queue
let high_task = glommio::spawn_local_into(
    high_priority_work(),
    high_prio_tq,
)
.unwrap();

// Create the task (in zig) `io.async` / `io.concurrent` call on low priority task queue
let low_task = glommio::spawn_local_into(
    low_priority_work(),
    low_prio_tq,
)
.unwrap();

// Await those tasks (in zig) high_task.await(io); low_task.await(io);
high_task.await;
low_task.await;

As you can see from the perspective of consumer of that API, there is quite a lot of choice when it comes to giving the scheduler “hints”, each of those teach the scheduler how you’d like that particular class of work to behave.

I am still a noob when it comes to Zig, so my understanding is quite limited, but as far as I can grasp the current architecture of std.Io and having not found an example of a priority-aware std.Io implementation, I conjure in order to emulate such setup, one would need multiple std.Io instances, where each of them in the init function specify their priority class, but also have a way to communicate between each other and coordinate it’s own scheduling.

A potential solution to this problem would be to extend io.async and io.concurrent with an type erased context that one could pass to each of those calls, I am aware of this being very leaky and not elegant, but the entire problem space of desiging an unified Io interface is full of such tradeoffs (the reader/writer cancellation story).

Alternatively, the std.Io interface instead of relying on Io itself as the scheduling primitive, could require an Scheduler interface, which would be used for the async and concurrent calls, but then there is a question on how to design such interface so it satisfies all the possible implementations of an scheduler, I am intentionally leaving it vague, so somebody more experienced could interrogate that idea further and accept/reject it based on how feasible it is at all.

5 Likes
2

If there is some context argument, it needs to be interpreted by the Io implementation, e.g. struct { priority: enum { low, high }, … }. But if this is not part of the unified Io API, then you are already specifying your code for some Io implementations (the ones that understand this struct). But then you could also just not use the abstract Io API, but instead refer to the actual priority-aware Io implementation itself (hypothetically):

pub fn main() void {
  var priority_uring : Io.PriorityUring = .init(.{});
  const io = priority_uring.io();
  priority_uring.asyncPriority(.low, testFn, .{ io });
  priority_uring.asyncPriority(.high, testFn, .{ io });
}

You could also fallback to a general Io implementation, if e.g. PriorityUring is not supported on your platform with some compile time checking. Meanwhile other functions could still rely on the Io API:

pub fn testFn(io: Io) void {
  std.io.sleep(io, …);
}

Of course, expanding the Io API is also an option. I don’t have an opinion on this, never needed this distinction myself. (-:

4 Likes
3

After playing with that idea for a little bit longer I think I’ve arrived to somewhat provable solution, using multiple Io and an Runtime. Turns out that the old saying still rings true

Any problem in computer science can be solved with another level of indirection.

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

fn Runtime(comptime size: usize) type {
    return struct {
        queues: [size]TaskQueue = undefined,
        current_size: usize = 0,

        const TaskQueue = struct { shares: usize };
        pub const Shared = struct {
            rt: *Self,
            queue_id: usize,
        };
        const Self = @This();

        pub const Error = error{TaskQueueOverflow};

        pub fn createTaskQueue(self: *Self, s: *Shared, shares: usize) Error!Io {
            if (self.current_size >= size) return error.TaskQueueOverflow;

            const idx = self.current_size;
            self.queues[idx] = .{ .shares = shares };
            self.current_size += 1;
            s.* = .{ .rt = self, .queue_id = idx };

            return .{ .userdata = s, .vtable = &vtable };
        }

        const vtable: Io.VTable = blk: {
            var vt = Io.failing.vtable.*;
            vt.async = asyncImpl;
            break :blk vt;
        };

        fn asyncImpl(
            userdata: ?*anyopaque,
            result: []u8,
            result_alignment: std.mem.Alignment,
            context: []const u8,
            context_alignment: std.mem.Alignment,
            start: *const fn (context: *const anyopaque, result: *anyopaque) void,
        ) ?*Io.AnyFuture {
            _ = result_alignment;
            _ = context_alignment;
            const s: *Shared = @ptrCast(@alignCast(userdata));
            const queue_id = s.queue_id;
            const q = s.rt.queues[queue_id];
            _ = q;
            // Actual logic.

            start(context.ptr, result.ptr);
            return null;
        }
    };
}

pub fn main() !void {
    const Rt = Runtime(5);

    var rt: Rt = .{};
    var hi_shared: Rt.Shared = undefined;
    var high_io = try rt.createTaskQueue(&hi_shared, 1000);

    var lo_shared: Rt.Shared = undefined;
    var low_io = try rt.createTaskQueue(&lo_shared, 100);

    var high_task = high_io.async(high_prio_work, .{});
    var low_task = low_io.async(low_prio_work, .{});

    high_task.await(high_io);
    low_task.await(low_io);
}

fn high_prio_work() void {
    std.debug.print("high priority work ran\n", .{});
}

fn low_prio_work() void {
    std.debug.print("low priority work ran\n", .{});
}

There are a few things such as the leakage of Shared type, I could get rid of it with heap allocation. The API is not type safe (can pass wrong std.Io to awaits) which could be solved by another wrapper. Since that wrapper would accept std.Io, in cases where Io implementation with no priority aware scheduling is suplied the shares become noops. Overall this resolves my complaints.

I was thinking about std.Io in a wrong way, the interface does not shape the executor itself, just how the executor interacts with the operating system.

3 Likes