Some basic questions about Channel, join and Condition

[voroskoi]

Hi,

Please note, that I am not a programmer, so these issues may not be problem for a real programmer, but I wanted to share my struggles, so we can enhance the documentation where it is necessary.

My starting point was the producer-consumer example, and after some try and error I came up with this little program. I have commented my issues with ASK when something happened which I did not expect.

The goal here was to start everything with an initial item in the channel and the workers may or may not put in additional items, which spawns even more workers (i can keep track of those in an atomic variable, not implemented here) and when there are no more running tasks we can close the channel and finish up everything.

const std = @import("std");
const zio = @import("zio");

const log = std.log.scoped(.producer_consumer);

fn producer(
    rt: *zio.Runtime,
    channel: *zio.Channel(i32),
    id: u32,
) zio.Cancelable!void {
    channel.send(rt, @intCast(id)) catch |err| switch (err) {
        error.ChannelClosed => {
            log.info("Producer {}: channel closed, exiting", .{id});
            return;
        },
        error.Canceled => {
            log.info("Producer {}: canceled, exiting", .{id});
            return;
        },
    };
    log.info("Produced: {}", .{id});
}

fn consumer(
    rt: *zio.Runtime,
    channel: *zio.Channel(i32),
    mutex: *zio.Mutex,
    cond: *zio.Condition,
    running_tasks: *std.atomic.Value(bool),
) zio.Cancelable!void {
    while (true) {
        const item = channel.receive(rt) catch |err| switch (err) {
            error.ChannelClosed => {
                log.info("Consumer: channel closed, exiting", .{});
                return;
            },
            error.Canceled => {
                log.info("Consumer: canceled, exiting", .{});
                return;
            },
        };
        log.info("Consumed: {}", .{item});

        // Condition signal
        {
            try mutex.lock(rt);
            running_tasks.store(false, .monotonic);
            defer mutex.unlock(rt);
            cond.signal(rt);
        }
    }
    log.info("Consumer finished", .{});
}

fn channelCloser(
    rt: *zio.Runtime,
    channel: *zio.Channel(i32),
    mutex: *zio.Mutex,
    condition: *zio.Condition,
    running_tasks: *const std.atomic.Value(bool),
) zio.Cancelable!void {
    {
        try mutex.lock(rt);
        defer mutex.unlock(rt);
        while (running_tasks.load(.monotonic) == true) {
            try condition.wait(rt, mutex);
        }
    }
    channel.close(false);
}

pub fn main() !void {
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();

    var runtime = try zio.Runtime.init(gpa.allocator(), .{});
    defer runtime.deinit();

    var buffer: [8]i32 = undefined;
    var channel = zio.Channel(i32).init(&buffer);

    // ASK1: You can not just tuck in a value in the channel. Why?
    // try channel.send(runtime, 42);

    // Prepare for 2 consumers
    const CONSUMERS = 2;
    var consumers: [CONSUMERS]zio.JoinHandle(zio.Cancelable!void) = undefined;
    var consumer_count: usize = 0;
    defer {
        for (consumers[0..consumer_count]) |*task| task.cancel(runtime);
    }

    var task = try runtime.spawn(producer, .{ runtime, &channel, 42 }, .{});
    defer task.cancel(runtime);

    // ASK2: this makes the runtime just wait: The consumer has not even started
    // yet, and the producer also did not put 42 in the channel. So we are
    // deadlocked.
    // try task.join(runtime);

    // ASK3: This line let's things go on without waiting for the result.
    // Using `_` for return value does the same.
    task.detach(runtime);

    var mutex: zio.Mutex = .init;
    defer mutex.unlock(runtime);
    var condition: zio.Condition = .init;
    var running_tasks: std.atomic.Value(bool) = .init(true);

    // Start 2 consumers.
    for (0..CONSUMERS) |i| {
        consumers[i] = try runtime.spawn(consumer, .{
            runtime,
            &channel,
            &mutex,
            &condition,
            &running_tasks,
        }, .{});
        consumer_count += 1;
    }

    var handler = try runtime.spawn(channelCloser, .{
        runtime,
        &channel,
        &mutex,
        &condition,
        &running_tasks,
    }, .{});
    handler.detach(runtime);

    try runtime.run();

    log.info("All tasks completed.", .{});
}

• ask1: It was not clear to me that zio.Channel is connected with the runtime, so I have tried to do this. This results in an unreachable call, I think it would be nice to replace that with panic("You can not use this outside a zio.Runtime") or something like that.
• ask2: It probably just me, but for me the difference between join and detach was not obvious. This is well written in the Tasks docs now.
• ask3: I am just curious here: What is the difference between using _ and detach for detaching? Maybe the docs could clarify which is the preferred one to use.
• ask4: When using a Condition we need an addition variable (ready in the example). Why we need that instead of modifying the Condition directly?

Thanks,

[lalinsky]

This is a fairly low-level library, so I would't recommend trying to use it unless you know why do you need all the bits.

However:

1. Yes, all functions that receive rt should be called from within a coroutine. I have plans to change this, but that's how it is for now.

2. This also kind of depends on the knowledge of threads, where the detach/join terminology is from, but I'm glad you found the docs useful. I'll keep improving them.

3. You should never just assign a task to _, you are leaking memory that way.

4. This is a complex topic, the Condition and a Mutex are tools to synchronize over the other variable, ready. See this: https://en.wikipedia.org/wiki/Monitor_(synchronization)#Condition_variables