Channel fails on the first send

[split ocean]

The channel works fine for the first send/receiver then it proceeds to do unwanted behavior

Channel:

pub const TaggedHead = extern struct {
    index: usize = 0,
    tag: usize = 0,
};

pub fn Channel(comptime T: type, comptime capacity: usize) type {
    return struct {
        const Self = @This();

        head: Atomic(*TaggedHead),
        tail: Atomic(usize),
        buffer: [capacity]T = undefined,

        pub fn init(self: *Self, empty_tagged_head: *TaggedHead) void {
            self.head = Atomic(*TaggedHead).init(empty_tagged_head);
            self.tail = Atomic(usize).init(0);
        }

        pub fn send(self: *Self, value: T) !void {
            var old_tail = self.tail.load(.monotonic);
            var new_tail: usize = undefined;
            while (true) {
                new_tail = (old_tail + 1) % capacity;
                if (new_tail == self.head.load(.monotonic).index) {
                    return error.ChannelFull;
                }
                old_tail = self.tail.cmpxchgWeak(old_tail, new_tail, .release, .monotonic) orelse break;
            }
            self.buffer[old_tail] = value;
            self.tail.store(new_tail, .release);
        }

        pub fn recv(self: *Self) ?T {
            var old_tagged_head = self.head.load(.acquire);
            while (true) {
                if (old_tagged_head.index == self.tail.load(.monotonic)) {
                    return null;
                }
                var new_head = TaggedHead{
                    .index = (old_tagged_head.index + 1) % capacity,
                    .tag = old_tagged_head.tag + 1,
                };
                old_tagged_head = self.head.cmpxchgWeak(old_tagged_head, &new_head, .acquire, .monotonic) orelse {
                    const value = self.buffer[old_tagged_head.index];
                    return value;
                };
            }
        }
    };
}

main:

const IntChannel = Channel(i32, 65);

pub fn sendValue(channel: *IntChannel, v: i32) !void {
    try channel.send(v);
    std.debug.print("value sent {d}\n", .{v});
}

pub fn work(channel: *IntChannel) !void {
    var count: usize = 0;
    while (true) {
        if (count == 5) break;
        const v = channel.recv() orelse {
            std.debug.print("empty channel sleeping...\n", .{});
            std.time.sleep(std.time.ns_per_ms * 100);
            continue;
        };
        count += 1;
        std.debug.print("value received {d}\n", .{v});
    }
}

// Philo chopsticks
pub fn main() !void {
    var empty_tagged_head = TaggedHead{};
    var int_channel: IntChannel = undefined;
    int_channel.init(&empty_tagged_head);

    const handle = try std.Thread.spawn(.{}, work, .{&int_channel});
    const arr_val: [5]i32 = .{ 1, 3, 4, 5, 9 };
    for (arr_val) |v| {
        // _ = try std.Thread.spawn(.{}, sendValue, .{ &int_channel, v });
        try sendValue(&int_channel, v);
    }
    handle.join();
}

after first iteration of work the value of self.head.index
lldb:

(unsigned long) $6 = 1
  Fix-it applied, fixed expression was:
    channel->head.raw->index

in the third iteration of work since it will sleep on the second iteration:

(lldb) print channel.head.raw.index
(unsigned long) $7 = 8589934594
  Fix-it applied, fixed expression was:
    channel->head.raw->index

[raw iris]

in recv(), you're doing var new_head = ...; self.head.cmpxchg(old, &new) which stores a pointer to a local variable that's invalidated on the next loop iteration and/or when the function returns

[split ocean]

so the only way to solve this is by either using an allocator, or pass a pointer to a tagged_head as a param to recv (to solve the lifetime issue) ?

[raw iris]

or redesign your queue to not store pointers; try packing index and tag into something that can fit in an atomic

[split ocean]

OHH wait i think i remember something from your zap implementation

[split ocean]

basically something like this?

const Sync = packed struct {
    /// Tracks the number of threads not searching for Tasks
    idle: u14 = 0,
    /// Tracks the number of threads spawned
    spawned: u14 = 0,
    /// What you see is what you get
    unused: bool = false,
    /// Used to not miss notifications while state = waking
    notified: bool = false,
    /// The current state of the thread pool
    state: enum(u2) {
        /// A notification can be issued to wake up a sleeping as the "waking thread".
        pending = 0,
        /// The state was notifiied with a signal. A thread is woken up.
        /// The first thread to transition to `waking` becomes the "waking thread".
        signaled,
        /// There is a "waking thread" among us.
        /// No other thread should be woken up until the waking thread transitions the state.
        waking,
        /// The thread pool was terminated. Start decremented `spawned` so that it can be joined.
        shutdown,
    } = .pending,
};

and bitcast it whenever i need to use the fields

[raw iris]

you can use a packed struct if you like or just decode the fields from the atomic-representation manually

[split ocean]

im not sure if i do know what the second approach is

[raw iris]

e.g. (intFromBool(notified) << 16) | (spawned << 14) | idle

[split ocean]

so the representation would be something like this ?
Bit positions: [17.......16] [15..........2] [1..0]
Values: [notified] [spawned] [idle]

or am i missing something

[split ocean]

btw this seem to error out in zig

src/main.zig:152:55: error: type 'u0' cannot represent integer value '16'
    const combined_value = (@intFromBool(notified) << 16) | (spawned << 14) | idle;

[raw iris]

its pseudo code

(but youd solve that by doing @as(u32, @intFromBool(notified))

[split ocean]

that did solve it but i think i am doing something wrong as i am not getting the value of spawn back right

    const notified = true;
    const spawned: u14 = 1123;
    const idle: u14 = 1123;

    const combined_value = (@as(u32, @intFromBool(notified)) << 16) | (@as(u32, spawned) << 14) | idle;
    std.debug.print("combined_value: {d}\n", .{combined_value});

    const notified_mask: u32 = 1 << 16;
    const new_notified: bool = (combined_value & notified_mask) != 0;

    const spawned_mask: u32 = (1 << 14) - 1;
    const new_spawned: u14 = @intCast((combined_value >> 14) & spawned_mask);

    const idle_mask: u32 = (1 << 14) - 1;
    const new_idle: u14 = @intCast(combined_value & idle_mask);

    std.debug.print("notified: {}\n", .{new_notified});
    std.debug.print("spawned: {d}\n", .{new_spawned});
    std.debug.print("idle: {d}\n", .{new_idle});

the values are:

notified: true
spawned: 1127
idle: 1123

notified and idle are correct

[split ocean]

    const combined_value = (@as(u32, @intFromBool(notified)) << 28) | (@as(u32, spawned) << 14) | idle;
    std.debug.print("combined_value: {d}\n", .{combined_value});

    const new_notified: bool = (combined_value >> 28) & 1 != 0;

    const spawned_mask: u32 = (1 << 14) - 1;
    const new_spawned: u14 = @intCast((combined_value >> 14) & spawned_mask);

    const idle_mask: u32 = (1 << 14) - 1;
    const new_idle: u14 = @intCast(combined_value & idle_mask);

    std.debug.print("notified: {}\n", .{new_notified});
    std.debug.print("spawned: {d}\n", .{new_spawned});
    std.debug.print("idle: {d}\n", .{new_idle});

this what worked for me

u1 (bool) + u14 + u14

[split ocean]

It s working now

pub const TaggedHead = packed struct {
    index: usize = 0,
    tag: usize = 0,
};

pub fn Channel(comptime T: type, comptime capacity: usize) type {
    return struct {
        const Self = @This();

        head: Atomic(u128),
        tail: Atomic(usize),
        buffer: [capacity]T = undefined,

        pub fn init(self: *Self) void {
            const head: u128 = @bitCast(TaggedHead{});
            self.head = Atomic(u128).init(head);
            self.tail = Atomic(usize).init(0);
        }

        pub fn send(self: *Self, value: T) !void {
            var old_tail = self.tail.load(.monotonic);
            var new_tail: usize = undefined;
            while (true) {
                new_tail = (old_tail + 1) % capacity;
                const head: TaggedHead = @bitCast(self.head.load(.monotonic));
                if (new_tail == head.index) {
                    return error.ChannelFull;
                }
                old_tail = self.tail.cmpxchgWeak(old_tail, new_tail, .release, .monotonic) orelse break;
            }
            self.buffer[old_tail] = value;
            self.tail.store(new_tail, .release);
        }

        pub fn recv(self: *Self) ?T {
            var old_tagged_head: TaggedHead = @bitCast(self.head.load(.acquire));
            while (true) {
                if (old_tagged_head.index == self.tail.load(.monotonic)) {
                    return null;
                }
                const new_head = TaggedHead{
                    .index = (old_tagged_head.index + 1) % capacity,
                    .tag = old_tagged_head.tag + 1,
                };
                if (self.head.cmpxchgWeak(@bitCast(old_tagged_head), @bitCast(new_head), .acquire, .monotonic)) |v| {
                    old_tagged_head = @bitCast(v);
                } else {
                    const value = self.buffer[old_tagged_head.index];
                    return value;
                }
            }
        }
    };
}

@raw iris thanks for the help man !

[long spade]

Was wondering about channels in Zig and wanted to ask about it in the discord but then I came across this thread.
I saw that in 0.11 there seems to be some api that would indicate the existence of a channel in std, but they are no longer in 0.13.
Does this mean we would have to write our channel?

[raw iris]

yes. It can just be a mutex + LinearFifo

[split ocean]

You can as well just check other people implementations using sourcegraph, it s neat and at the same time most of them are well documented example:
https://sourcegraph.com/search?q=context:global+Channel+lang:Zig+&patternType=keyword&sm=0

[long spade]

Wow thank you for that. I did not even know about source graph!