Zig Barrier Patterns
A thread barrier is a meeting point where every thread waits until all threads have arrived. It splits work into phases that no thread starts early.
Meet, then double
Three threads store 1, 2, and 3, then meet at the barrier. Only after all three arrive does each thread double its slot.
const std = @import("std");
const Barrier = struct {
mutex: std.Thread.Mutex = .{},
cond: std.Thread.Condition = .{},
count: usize = 0,
total: usize,
generation: usize = 0,
fn wait(self: *Barrier) void {
self.mutex.lock();
defer self.mutex.unlock();
const gen = self.generation;
self.count += 1;
if (self.count == self.total) {
self.count = 0;
self.generation += 1;
self.cond.broadcast();
} else {
while (gen == self.generation) {
self.cond.wait(&self.mutex);
}
}
}
};
const Context = struct {
barrier: *Barrier,
slots: [3]i64 = .{ 0, 0, 0 },
mutex: std.Thread.Mutex = .{},
total: i64 = 0,
fn work(ctx: *@This(), i: usize) void {
ctx.slots[i] = @as(i64, @intCast(i + 1));
ctx.barrier.wait();
ctx.mutex.lock();
defer ctx.mutex.unlock();
ctx.total += ctx.slots[i] * 2;
}
};
pub fn main() !void {
var barrier = Barrier{ .total = 3 };
var ctx = Context{ .barrier = &barrier };
var threads: [3]std.Thread = undefined;
var i: usize = 0;
while (i < 3) : (i += 1) {
threads[i] = try std.Thread.spawn(.{}, Context.work, .{ &ctx, i });
}
for (threads) |t| {
t.join();
}
const out = std.io.getStdOut().writer();
try out.print("{d}\n", .{ctx.total});
}
Reuse across two rounds
The same barrier is crossed twice, once per round. The generation count keeps the two rounds apart, and every slot ends at 2.
const std = @import("std");
const Barrier = struct {
mutex: std.Thread.Mutex = .{},
cond: std.Thread.Condition = .{},
count: usize = 0,
total: usize,
generation: usize = 0,
fn wait(self: *Barrier) void {
self.mutex.lock();
defer self.mutex.unlock();
const gen = self.generation;
self.count += 1;
if (self.count == self.total) {
self.count = 0;
self.generation += 1;
self.cond.broadcast();
} else {
while (gen == self.generation) {
self.cond.wait(&self.mutex);
}
}
}
};
const Context = struct {
barrier: *Barrier,
slots: [3]i64 = .{ 0, 0, 0 },
fn work(ctx: *@This(), i: usize) void {
var round: usize = 0;
while (round < 2) : (round += 1) {
ctx.slots[i] += 1;
ctx.barrier.wait();
}
}
};
pub fn main() !void {
var barrier = Barrier{ .total = 3 };
var ctx = Context{ .barrier = &barrier };
var threads: [3]std.Thread = undefined;
var i: usize = 0;
while (i < 3) : (i += 1) {
threads[i] = try std.Thread.spawn(.{}, Context.work, .{ &ctx, i });
}
for (threads) |t| {
t.join();
}
var sum: i64 = 0;
for (ctx.slots) |s| {
sum += s;
}
const out = std.io.getStdOut().writer();
try out.print("{d}\n", .{sum});
}
Partials meet, then combine
Three threads sum blocks of ten numbers each. The barrier holds them until all three blocks are done, then each adds its partial to the total.
const std = @import("std");
const Barrier = struct {
mutex: std.Thread.Mutex = .{},
cond: std.Thread.Condition = .{},
count: usize = 0,
total: usize,
generation: usize = 0,
fn wait(self: *Barrier) void {
self.mutex.lock();
defer self.mutex.unlock();
const gen = self.generation;
self.count += 1;
if (self.count == self.total) {
self.count = 0;
self.generation += 1;
self.cond.broadcast();
} else {
while (gen == self.generation) {
self.cond.wait(&self.mutex);
}
}
}
};
const Context = struct {
barrier: *Barrier,
mutex: std.Thread.Mutex = .{},
total: i64 = 0,
fn work(ctx: *@This(), k: i64) void {
var sum: i64 = 0;
var n: i64 = k * 10 + 1;
while (n <= k * 10 + 10) : (n += 1) {
sum += n;
}
ctx.barrier.wait();
ctx.mutex.lock();
defer ctx.mutex.unlock();
ctx.total += sum;
}
};
pub fn main() !void {
var barrier = Barrier{ .total = 3 };
var ctx = Context{ .barrier = &barrier };
var threads: [3]std.Thread = undefined;
var i: i64 = 0;
while (i < 3) : (i += 1) {
threads[@intCast(i)] = try std.Thread.spawn(.{}, Context.work, .{ &ctx, i });
}
for (threads) |t| {
t.join();
}
const out = std.io.getStdOut().writer();
try out.print("{d}\n", .{ctx.total});
}
Example
const std = @import("std");
const Barrier = struct {
mutex: std.Thread.Mutex = .{},
cond: std.Thread.Condition = .{},
count: usize = 0,
total: usize,
generation: usize = 0,
fn wait(self: *Barrier) void {
self.mutex.lock();
defer self.mutex.unlock();
const gen = self.generation;
self.count += 1;
if (self.count == self.total) {
self.count = 0;
self.generation += 1;
self.cond.broadcast();
} else {
while (gen == self.generation) {
self.cond.wait(&self.mutex);
}
}
}
};
const Context = struct {
barrier: *Barrier,
slots: [4]i64 = .{ 0, 0, 0, 0 },
fn work(ctx: *@This(), i: usize) void {
const n: i64 = @intCast(i + 1);
ctx.slots[i] = n * n;
ctx.barrier.wait();
ctx.slots[i] *= 2;
}
};
pub fn main() !void {
var barrier = Barrier{ .total = 4 };
var ctx = Context{ .barrier = &barrier };
var threads: [4]std.Thread = undefined;
var i: usize = 0;
while (i < 4) : (i += 1) {
threads[i] = try std.Thread.spawn(.{}, Context.work, .{ &ctx, i });
}
for (threads) |t| {
t.join();
}
var sum: i64 = 0;
for (ctx.slots) |s| {
sum += s;
}
const out = std.io.getStdOut().writer();
try out.print("{d}\n", .{sum});
}
Key points
- Thread barrier: a thread barrier is a point where each thread waits until every thread in the group has arrived.
- Arrival: an arrival is one thread reaching the barrier and counting itself in.
- Release: a release is the moment the last arrival wakes every waiting thread.
- Generation: a generation is one round of arrivals at a reusable barrier, counted so a late thread cannot join the wrong round.
- A barrier needs a mutex and a condition variable, because arrivals and releases share a count.
- Every thread in the group must reach the barrier, or the rest never get released.
- Phase: a phase is one stretch of work between two barriers.
Four threads square 1, 2, 3, and 4 into separate slots, then meet at the barrier. The second phase doubles every slot, and no thread doubles early. The main thread joins all four threads, then sums the slots to 60.
Try it Yourself
Edit the code, then press Run.
Running is simulated
Exercise
What happens at a barrier when the last thread arrives?