Communication methods between concurrent tasks, modelling recovery / error handling

[kj4tmp]

Good evening my esteemed colleagues.

I am implementing a communication library which requires concurrency to work correctly.

The structure of my application could be represented as the following diagram:

setup -> taskAcyclic: configuration state machine
            -> taskCyclic

• setup allocates the memory / resources / arranges various const structures
• after setup we begin taskAcyclic
• taskAcyclic begets taskCyclic.
• The lifetime of taskAcyclic exceeds taskCyclic. (structured concurrency)

taskCyclic is rather simple:

fn taskCyclic(
    io: std.Io,
    md: *gcat.MainDevice,
    pdi_write_mutex: *std.Io.Mutex,
    max_recv_timeouts: u32,
    maybe_zh: ?*ZenohHandler,
    config: Config,
    stop_event: std.Io.Event,
) error{NonRecoverable}!void {
    var recv_timeouts: u32 = 0;
    while (!stop_event.isSet()) {
        // must protect the process data
        // from zenoh during the recv()
        {
            pdi_write_mutex.lockUncancelable(io);
            defer pdi_write_mutex.unlock(io);

            if (md.recv(io)) |cyclic_result| {
                md.check(cyclic_result) catch |err| switch (err) {
                    error.NotAllSubdevicesInOP,
                    error.TopologyChanged,
                    error.Wkc,
                    => |err2| {
                        std.log.err("failure out of .op: {s}", .{@errorName(err2)});
                        // TODO: emit notification?
                    },
                };
                recv_timeouts = 0;
            } else |err| switch (err) {
                error.RecvTimeout => {
                    std.log.warn("recv timeout!", .{});
                    recv_timeouts +|= 1;
                    if (recv_timeouts >= max_recv_timeouts) {
                        std.log.err("recv timeouts exhausted, rescanning bus", .{});
                        // TODO: emit notification?
                    }
                },
                error.LinkError,
                => return error.NonRecoverable,
            }
            // to reduce jitter, the lock is maintained though send
            md.send(io) catch |err| switch (err) {
                error.LinkError, error.Unexpected => return error.NonRecoverable,
            };
        }

        if (maybe_zh) |*zh| {
            zh.publishInputsOutputs(&md, config.value) catch |err| {
                std.log.err("failed to publish inputs / outputs on zenoh: {s}", .{@errorName(err)});
                return error.NonRecoverable; // TODO: correct action here?
            };
        }

        md.sleep(io) catch |err| switch (err) {
            error.Unexpected, error.UnsupportedClock, error.Canceled => |err2| {
                std.log.err("sleep failure: {}", .{err2});
                return error.NonRecoverable;
            },
        };
    }
}

The key feature of taskCyclic is that it has a fixed cycle time.
On an ideal computer:

• md.recv(io) consumes approx 100 us
• md.send(io) consumes approx 100 us
• other application code in the loop consumes approx 100 us
• md.sleep(io) sleeps for the remainder of the time in the cycle
• repeat

The problems with taskCyclic:

• there exists a certain set of “non-recoverable” errors that can only be recovered by unwinding up to taskAcyclic for reinitialization / retry logic.
• there exists a certain set of “recoverable” errors that can be recovered via prompt action by taskAcyclic, but nevertheless taskCyclic must remain operating at fixed cycle time.

Now for taskAcyclic, I am unsure of what the shape of this task looks like:

perhaps some pseudocode:

fn taskAcyclc(io: std.Io, md: *MainDevice) !void {

    spawnCyclicTask()

    stay_online: while (true) {
        md.maintainOperation(io)
        waitForBadThingInTaskCyclic(io) catch |err| switch (err) {
            error.NonRecoverable => {
                stop_event.set();
                cyclic_task.join()
                break;
            },
            error.Recoverable => {
                doRecoveryAction();
                continue :stay_online;
            },
        }
    }
    return error.NonRecoverable;

}

Questions:

1. In taskCyclic: how should I be modelling “non-recoverable” versus “recoverable” errors? Proposal: “Non-recoverable” errors are a return error.NonRecoverable while I should somehow provide some notification to taskAcyclic for the “recoverable” errors?
2. Should I structure the code differently between these two tasks?
3. How should I model notification from taskAcyclic to taskCyclic?

[naxlavun]

I think you’re overthinking it. Just handle the errors that are possible to handle (i.e. recoverable errors) and propagate the rest? What am I missing?

[kj4tmp]

its still pretty ugly and likely has another architectural iteration, but this works:

fn taskCyclic(io: std.Io, md: *gcat.MainDevice, pdi_write_mutex: *std.Io.Mutex, maybe_zh: *?ZenohHandler, config: Config, notification_queue: *std.Io.Queue(Notification)) error{ NonRecoverable, Canceled }!void {
    var recv_timeouts: u32 = 0;
    while (true) {
        // must protect the process data
        // from zenoh during the recv()
        {
            pdi_write_mutex.lockUncancelable(io);
            defer pdi_write_mutex.unlock(io);

            if (md.recv(io)) |cyclic_result| {
                md.check(cyclic_result) catch |err| switch (err) {
                    error.NotAllSubdevicesInOP,
                    error.TopologyChanged,
                    error.Wkc,
                    => |err2| {
                        // std.log.err("failure out of .op: {s}", .{@errorName(err2)});
                        _ = notification_queue.put(io, &.{Notification{ .@"error" = err2 }}, 0) catch |err3| switch (err3) {
                            error.Canceled => |e| return e,
                            error.Closed => return error.NonRecoverable,
                        };
                    },
                };
                recv_timeouts = 0;
            } else |err| switch (err) {
                error.RecvTimeout => {
                    std.log.warn("recv timeout!", .{});
                    recv_timeouts +|= 1;
                    _ = notification_queue.put(io, &.{Notification{ .recv_timeout = recv_timeouts }}, 0) catch |err3| switch (err3) {
                        error.Canceled => |e| return e,
                        error.Closed => return error.NonRecoverable,
                    };
                },
                error.LinkError => {
                    _ = notification_queue.put(io, &.{Notification{ .@"error" = error.NonRecoverable }}, 0) catch |err3| switch (err3) {
                        error.Canceled => |e| return e,
                        error.Closed => return error.NonRecoverable,
                    };
                    return error.NonRecoverable;
                },
            }
            // to reduce jitter, the lock is maintained though send
            md.send(io) catch |err| switch (err) {
                error.LinkError, error.Unexpected => {
                    _ = notification_queue.put(io, &.{Notification{ .@"error" = error.NonRecoverable }}, 0) catch |err3| switch (err3) {
                        error.Canceled => |e| return e,
                        error.Closed => return error.NonRecoverable,
                    };
                    return error.NonRecoverable;
                },
            };
        }

        if (maybe_zh.*) |*zh| {
            zh.publishInputsOutputs(md, config) catch |err| {
                std.log.err("failed to publish inputs / outputs on zenoh: {s}", .{@errorName(err)});
                _ = notification_queue.put(io, &.{Notification{ .@"error" = error.NonRecoverable }}, 0) catch |err3| switch (err3) {
                    error.Canceled => |e| return e,
                    error.Closed => return error.NonRecoverable,
                };
                return error.NonRecoverable; // TODO: correct action here?
            };
        }

        md.sleep(io) catch |err| switch (err) {
            error.Unexpected, error.UnsupportedClock => |err2| {
                std.log.err("sleep failure: {}", .{err2});
                _ = notification_queue.put(io, &.{Notification{ .@"error" = error.NonRecoverable }}, 0) catch |err3| switch (err3) {
                    error.Canceled => |e| return e,
                    error.Closed => return error.NonRecoverable,
                };
                return error.NonRecoverable;
            },
            error.Canceled => |e| return e,
        };
    }
}

So Im returning errors for non-recoverable stuff and emitting notifications into a queue for recoverable stuff from the main thread:

pub fn run(parsed: cli.Parsed(command)) !void {
    const thread_safe_allocator: bool = true;
    var gpa = std.heap.DebugAllocator(.{ .safety = true, .thread_safe = thread_safe_allocator }){};
    defer _ = gpa.deinit();
    const allocator = gpa.allocator();

    comptime assert(thread_safe_allocator); // std.Io.Threaded requires a thread-safe allocator
    const concurrent_limit: usize = 1;
    var io_impl = std.Io.Threaded.init(gpa.allocator(), .{
        .concurrent_limit = .limited(concurrent_limit),
        .async_limit = .nothing,
    });
    const io = io_impl.io();


    bus_scan: while (true) {
        var md = try gcat.MainDevice.init(...);
        defer md.deinit(io, allocator);

        comptime assert(concurrent_limit > 0);
        var notification_buffer: [1]Notification = undefined;
        var notification_queue: std.Io.Queue(Notification) = .init(&notification_buffer);
        var future_cyclic = io.concurrent(taskCyclic, .{ io, &md, &pdi_write_mutex, &maybe_zh, config.value, &notification_queue }) catch |err| switch (err) {
            error.ConcurrencyUnavailable => return error.NonRecoverable,
        };
        defer future_cyclic.cancel(io) catch {};

        try md.transitionToState(io, .op);

        var erase: [1]Notification = undefined;
        _ = notification_queue.get(io, &erase, 0) catch |err| switch (err) {
            error.Canceled => return error.NonRecoverable,
            error.Closed => return error.NonRecoverable,
        };

        processNotifications(
            io,
            &notification_queue,
            parsed.kind.args.@"max-recv-timeouts-before-rescan",
        ) catch |err| switch (err) {
            error.Canceled, error.Closed, error.NonRecoverable => {
                std.log.err("Notification failure: {}", .{err});
                return error.NonRecoverable;
            },
            error.Recoverable => {
                std.log.err("Notification failure: {}", .{err});
                continue :bus_scan;
            },
        };
    }
}

fn processNotifications(io: std.Io, queue: *std.Io.Queue(Notification), max_recv_timeouts: u32) error{ Canceled, NonRecoverable, Recoverable, Closed }!void {
    while (true) {
        const notif = try queue.getOne(io);
        std.log.err("Notification: {any}", .{notif});
        switch (notif) {
            .recv_timeout => |count| {
                if (count > max_recv_timeouts) {
                    return error.Recoverable;
                }
            },
            .@"error" => |err| switch (err) {
                error.NonRecoverable => |e| return e,
                error.NotAllSubdevicesInOP => {}, // TODO
                error.TopologyChanged, error.Wkc => return error.Recoverable,
            },
        }
    }
}

Likely a few more passes of simplification still required.