Zig Work Queue
A work queue is a shared list of jobs that worker threads pull from until the list is closed. It turns many small jobs into steady work for a fixed set of threads.
Drain three jobs
Two workers pull jobs 10, 20, and 30 from the list and add them to a shared total. Closing wakes any worker still waiting.
const std = @import("std");
const Queue = struct {
mutex: std.Thread.Mutex = .{},
cond: std.Thread.Condition = .{},
jobs: []const i64,
next: usize = 0,
closed: bool = false,
total: std.atomic.Value(i64) = std.atomic.Value(i64).init(0),
fn worker(q: *@This()) void {
while (true) {
q.mutex.lock();
while (q.next >= q.jobs.len and !q.closed) {
q.cond.wait(&q.mutex);
}
if (q.next >= q.jobs.len and q.closed) {
q.mutex.unlock();
break;
}
const job = q.jobs[q.next];
q.next += 1;
q.mutex.unlock();
_ = q.total.fetchAdd(job, .acq_rel);
}
}
};
pub fn main() !void {
var q = Queue{ .jobs = &[_]i64{ 10, 20, 30 } };
var threads: [2]std.Thread = undefined;
for (&threads) |*t| {
t.* = try std.Thread.spawn(.{}, Queue.worker, .{&q});
}
q.mutex.lock();
q.closed = true;
q.cond.broadcast();
q.mutex.unlock();
for (threads) |t| {
t.join();
}
const out = std.io.getStdOut().writer();
try out.print("{d}\n", .{q.total.load(.seq_cst)});
}
Workers transform jobs
Three workers pull the numbers 1 to 6 and add each square to the total. The total is always 91, whatever order the workers run in.
const std = @import("std");
const Queue = struct {
mutex: std.Thread.Mutex = .{},
cond: std.Thread.Condition = .{},
jobs: []const i64,
next: usize = 0,
closed: bool = false,
total: std.atomic.Value(i64) = std.atomic.Value(i64).init(0),
fn worker(q: *@This()) void {
while (true) {
q.mutex.lock();
while (q.next >= q.jobs.len and !q.closed) {
q.cond.wait(&q.mutex);
}
if (q.next >= q.jobs.len and q.closed) {
q.mutex.unlock();
break;
}
const job = q.jobs[q.next];
q.next += 1;
q.mutex.unlock();
_ = q.total.fetchAdd(job * job, .acq_rel);
}
}
};
pub fn main() !void {
var q = Queue{ .jobs = &[_]i64{ 1, 2, 3, 4, 5, 6 } };
var threads: [3]std.Thread = undefined;
for (&threads) |*t| {
t.* = try std.Thread.spawn(.{}, Queue.worker, .{&q});
}
q.mutex.lock();
q.closed = true;
q.cond.broadcast();
q.mutex.unlock();
for (threads) |t| {
t.join();
}
const out = std.io.getStdOut().writer();
try out.print("{d}\n", .{q.total.load(.seq_cst)});
}
Closing first keeps pending jobs
The list is closed before any worker starts, yet all four jobs are still drained. Closing stops new jobs, it never discards pending ones.
const std = @import("std");
const Queue = struct {
mutex: std.Thread.Mutex = .{},
cond: std.Thread.Condition = .{},
jobs: []const i64,
next: usize = 0,
closed: bool = false,
total: std.atomic.Value(i64) = std.atomic.Value(i64).init(0),
fn worker(q: *@This()) void {
while (true) {
q.mutex.lock();
while (q.next >= q.jobs.len and !q.closed) {
q.cond.wait(&q.mutex);
}
if (q.next >= q.jobs.len and q.closed) {
q.mutex.unlock();
break;
}
const job = q.jobs[q.next];
q.next += 1;
q.mutex.unlock();
_ = q.total.fetchAdd(job, .acq_rel);
}
}
};
pub fn main() !void {
var q = Queue{ .jobs = &[_]i64{ 5, 5, 5, 5 }, .closed = true };
var threads: [3]std.Thread = undefined;
for (&threads) |*t| {
t.* = try std.Thread.spawn(.{}, Queue.worker, .{&q});
}
for (threads) |t| {
t.join();
}
const out = std.io.getStdOut().writer();
try out.print("{d}\n", .{q.total.load(.seq_cst)});
}
Example
const std = @import("std");
const Queue = struct {
mutex: std.Thread.Mutex = .{},
cond: std.Thread.Condition = .{},
jobs: []const i64,
next: usize = 0,
closed: bool = false,
total: std.atomic.Value(i64) = std.atomic.Value(i64).init(0),
fn worker(q: *@This()) void {
while (true) {
q.mutex.lock();
while (q.next >= q.jobs.len and !q.closed) {
q.cond.wait(&q.mutex);
}
if (q.next >= q.jobs.len and q.closed) {
q.mutex.unlock();
break;
}
const job = q.jobs[q.next];
q.next += 1;
q.mutex.unlock();
_ = q.total.fetchAdd(job, .acq_rel);
}
}
};
pub fn main() !void {
var q = Queue{ .jobs = &[_]i64{ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 } };
var threads: [4]std.Thread = undefined;
for (&threads) |*t| {
t.* = try std.Thread.spawn(.{}, Queue.worker, .{&q});
}
q.mutex.lock();
q.closed = true;
q.cond.broadcast();
q.mutex.unlock();
for (threads) |t| {
t.join();
}
const out = std.io.getStdOut().writer();
try out.print("{d}\n", .{q.total.load(.seq_cst)});
}
Key points
- Work queue: a work queue is a shared list of jobs plus the rules for adding jobs, taking jobs, and closing the list.
- A worker takes one job at a time while holding the mutex, then releases the mutex before running the job.
- An empty list makes a worker wait on the condition variable instead of polling.
- Closing: closing is the step where the producer marks the list done and wakes every worker.
- A worker that finds the list closed and empty leaves its loop and can be joined.
- Jobs must be independent, because workers run them in no fixed order.
- The main thread joins every worker before reading the final result.
Four workers drain the jobs 1 to 10 and add each one to a shared total. The producer closes the list and wakes every worker, so no worker waits forever. The main thread joins all four workers before printing the total.
Try it Yourself
Edit the code, then press Run.
Running is simulated
Exercise
What wakes a worker waiting on an empty list?