Zig Condition Variables

A condition variable is a tool that lets a thread sleep until another thread says something changed. It always works together with a mutex.

Wait for one signal

The waiter sleeps until the main thread sets the value and signals. The waiter then copies the value into its own slot.

const std = @import("std");
const Shared = struct {
    mutex: std.Thread.Mutex = .{},
    cond: std.Thread.Condition = .{},
    ready: bool = false,
    value: i32 = 0,
    taken: i32 = 0,
    fn waitFor(sh: *@This()) void {
        sh.mutex.lock();
        defer sh.mutex.unlock();
        while (!sh.ready) {
            sh.cond.wait(&sh.mutex);
        }
        sh.taken = sh.value;
    }
};
pub fn main() !void {
    var sh = Shared{};
    const t = try std.Thread.spawn(.{}, Shared.waitFor, .{&sh});
    sh.mutex.lock();
    sh.value = 7;
    sh.ready = true;
    sh.cond.signal();
    sh.mutex.unlock();
    t.join();
    const out = std.io.getStdOut().writer();
    try out.print("{d}\n", .{sh.taken});
}

Wake every waiter

Two waiters sleep on the same condition. One broadcast wakes both, so the count of woken threads is 2.

const std = @import("std");
const Shared = struct {
    mutex: std.Thread.Mutex = .{},
    cond: std.Thread.Condition = .{},
    ready: bool = false,
    slots: [2]i32 = .{ 0, 0 },
    fn waitFor(sh: *@This(), i: usize) void {
        sh.mutex.lock();
        defer sh.mutex.unlock();
        while (!sh.ready) {
            sh.cond.wait(&sh.mutex);
        }
        sh.slots[i] = 1;
    }
};
pub fn main() !void {
    var sh = Shared{};
    var threads: [2]std.Thread = undefined;
    var i: usize = 0;
    while (i < 2) : (i += 1) {
        threads[i] = try std.Thread.spawn(.{}, Shared.waitFor, .{ &sh, i });
    }
    sh.mutex.lock();
    sh.ready = true;
    sh.cond.broadcast();
    sh.mutex.unlock();
    for (threads) |t| {
        t.join();
    }
    var woke: i32 = 0;
    for (sh.slots) |s| {
        woke += s;
    }
    const out = std.io.getStdOut().writer();
    try out.print("{d}\n", .{woke});
}

Wait for a threshold

The waiter loops until the count reaches 2. The first signal wakes it, but the loop sends it back to sleep.

const std = @import("std");
const Shared = struct {
    mutex: std.Thread.Mutex = .{},
    cond: std.Thread.Condition = .{},
    count: i32 = 0,
    fn waitForTwo(sh: *@This()) void {
        sh.mutex.lock();
        defer sh.mutex.unlock();
        while (sh.count < 2) {
            sh.cond.wait(&sh.mutex);
        }
    }
};
pub fn main() !void {
    var sh = Shared{};
    const t = try std.Thread.spawn(.{}, Shared.waitForTwo, .{&sh});
    sh.mutex.lock();
    sh.count = 1;
    sh.cond.signal();
    sh.mutex.unlock();
    sh.mutex.lock();
    sh.count = 2;
    sh.cond.signal();
    sh.mutex.unlock();
    t.join();
    const out = std.io.getStdOut().writer();
    try out.print("{d}\n", .{sh.count});
}

Example

const std = @import("std");
const Slot = struct {
    mutex: std.Thread.Mutex = .{},
    cond: std.Thread.Condition = .{},
    ready: bool = false,
    value: i32 = 0,
    taken: i32 = 0,
    fn produce(s: *@This()) void {
        s.mutex.lock();
        defer s.mutex.unlock();
        s.value = 99;
        s.ready = true;
        s.cond.signal();
    }
    fn consume(s: *@This()) void {
        s.mutex.lock();
        defer s.mutex.unlock();
        while (!s.ready) {
            s.cond.wait(&s.mutex);
        }
        s.taken = s.value;
        s.ready = false;
    }
};
pub fn main() !void {
    var s = Slot{};
    const p = try std.Thread.spawn(.{}, Slot.produce, .{&s});
    const c = try std.Thread.spawn(.{}, Slot.consume, .{&s});
    p.join();
    c.join();
    const out = std.io.getStdOut().writer();
    try out.print("{d}\n", .{s.taken});
}

Key points

Condition variable: a condition variable is a wait-and-signal tool that releases its mutex while a thread waits.
Wait: a wait is a call that releases the mutex, sleeps, and locks the mutex again when it wakes.
Signal: a signal is a call that wakes one waiting thread.
Broadcast: a broadcast is a call that wakes every waiting thread.
A waiter must recheck its condition in a loop, because a wakeup does not prove the condition holds.
The shared condition itself is always read and changed while holding the mutex.
Signal after changing the condition, while still holding the mutex, so no wakeup is lost.

The producer fills the single slot, marks it ready, and signals. The consumer waits in a loop until the slot is ready, then takes the value. Both threads are joined before the main thread prints, so the result is always 99.

Try it Yourself

Edit the code, then press Run.

Running is simulated

const std = @import("std");
const Slot = struct {
    mutex: std.Thread.Mutex = .{},
    cond: std.Thread.Condition = .{},
    ready: bool = false,
    value: i32 = 0,
    taken: i32 = 0,
    fn produce(s: *@This()) void {
        s.mutex.lock();
        defer s.mutex.unlock();
        s.value = 99;
        s.ready = true;
        s.cond.signal();
    }
    fn consume(s: *@This()) void {
        s.mutex.lock();
        defer s.mutex.unlock();
        while (!s.ready) {
            s.cond.wait(&s.mutex);
        }
        s.taken = s.value;
        s.ready = false;
    }
};
pub fn main() !void {
    var s = Slot{};
    const p = try std.Thread.spawn(.{}, Slot.produce, .{&s});
    const c = try std.Thread.spawn(.{}, Slot.consume, .{&s});
    p.join();
    c.join();
    const out = std.io.getStdOut().writer();
    try out.print("{d}\n", .{s.taken});
}

Exercise

Why must a waiter recheck its condition in a loop?

Because wait is slow. Because a wakeup does not prove the condition holds. Because the mutex unlocks itself. Because signal needs a loop.