Zig Thread Pool Deep

A thread pool is a fixed set of worker threads that run tasks handed to it. It reuses its workers instead of starting a new thread for each task.

Run tasks in a pool

Eight tasks each add 10 to a shared total. The main thread waits with waitAndWork, then prints the total.

const std = @import("std");
const Context = struct {
    mutex: std.Thread.Mutex = .{},
    total: i64 = 0,
    fn add(ctx: *@This(), n: i64) void {
        ctx.mutex.lock();
        defer ctx.mutex.unlock();
        ctx.total += n;
    }
};
pub fn main() !void {
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const alloc = gpa.allocator();
    var ctx = Context{};
    var pool: std.Thread.Pool = undefined;
    try pool.init(.{ .allocator = alloc, .n_jobs = 4 });
    defer pool.deinit();
    var wg: std.Thread.WaitGroup = .{};
    var i: usize = 0;
    while (i < 8) : (i += 1) {
        pool.spawnWg(&wg, Context.add, .{ &ctx, 10 });
    }
    pool.waitAndWork(&wg);
    const out = std.io.getStdOut().writer();
    try out.print("{d}\n", .{ctx.total});
}

Tasks write separate results

Six tasks each square one number into its own slot. No mutex is needed here, because no two tasks touch the same slot.

const std = @import("std");
fn square(slot: *i64, n: i64) void {
    slot.* = n * n;
}
pub fn main() !void {
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const alloc = gpa.allocator();
    var results: [6]i64 = .{ 0, 0, 0, 0, 0, 0 };
    var pool: std.Thread.Pool = undefined;
    try pool.init(.{ .allocator = alloc, .n_jobs = 2 });
    defer pool.deinit();
    var wg: std.Thread.WaitGroup = .{};
    var i: usize = 0;
    while (i < 6) : (i += 1) {
        pool.spawnWg(&wg, square, .{ &results[i], @as(i64, @intCast(i + 1)) });
    }
    pool.waitAndWork(&wg);
    var sum: i64 = 0;
    for (results) |v| {
        sum += v;
    }
    const out = std.io.getStdOut().writer();
    try out.print("{d}\n", .{sum});
}

One pool, two batches

The same pool runs a second batch after the first finishes. Each batch gets its own wait group.

const std = @import("std");
const Context = struct {
    mutex: std.Thread.Mutex = .{},
    total: i64 = 0,
    fn add(ctx: *@This(), n: i64) void {
        ctx.mutex.lock();
        defer ctx.mutex.unlock();
        ctx.total += n;
    }
};
pub fn main() !void {
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const alloc = gpa.allocator();
    var ctx = Context{};
    var pool: std.Thread.Pool = undefined;
    try pool.init(.{ .allocator = alloc, .n_jobs = 4 });
    defer pool.deinit();
    var wg1: std.Thread.WaitGroup = .{};
    var i: usize = 0;
    while (i < 4) : (i += 1) {
        pool.spawnWg(&wg1, Context.add, .{ &ctx, 1 });
    }
    pool.waitAndWork(&wg1);
    var wg2: std.Thread.WaitGroup = .{};
    i = 0;
    while (i < 4) : (i += 1) {
        pool.spawnWg(&wg2, Context.add, .{ &ctx, 10 });
    }
    pool.waitAndWork(&wg2);
    const out = std.io.getStdOut().writer();
    try out.print("{d}\n", .{ctx.total});
}

Example

const std = @import("std");
const Context = struct {
    mutex: std.Thread.Mutex = .{},
    total: i64 = 0,
    fn addSquare(ctx: *@This(), n: i64) void {
        ctx.mutex.lock();
        defer ctx.mutex.unlock();
        ctx.total += n * n;
    }
};
pub fn main() !void {
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const alloc = gpa.allocator();
    var ctx = Context{};
    var pool: std.Thread.Pool = undefined;
    try pool.init(.{ .allocator = alloc, .n_jobs = 4 });
    defer pool.deinit();
    var wg: std.Thread.WaitGroup = .{};
    var i: i64 = 1;
    while (i <= 10) : (i += 1) {
        pool.spawnWg(&wg, Context.addSquare, .{ &ctx, i });
    }
    pool.waitAndWork(&wg);
    const out = std.io.getStdOut().writer();
    try out.print("{d}\n", .{ctx.total});
}

Key points

Thread pool: a thread pool is a fixed group of worker threads that runs tasks from a shared list.
Worker: a worker is one thread in the pool that picks up tasks and runs them.
The pool is created with an allocator and a worker count, and it is shut down with deinit.
WaitGroup: a wait group is a counter that tracks tasks still running.
spawnWg hands a task to the pool and counts it in the wait group.
waitAndWork makes the calling thread help run tasks until the wait group is done.
Shared state touched by pool tasks still needs a mutex, because tasks run at the same time.

Ten tasks each add one square to a shared total under a mutex. The pool runs them on four workers. waitAndWork waits until every task is done, so the printed total is the full sum of squares from 1 to 10.

Try it Yourself

Edit the code, then press Run.

Running is simulated

const std = @import("std");
const Context = struct {
    mutex: std.Thread.Mutex = .{},
    total: i64 = 0,
    fn addSquare(ctx: *@This(), n: i64) void {
        ctx.mutex.lock();
        defer ctx.mutex.unlock();
        ctx.total += n * n;
    }
};
pub fn main() !void {
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const alloc = gpa.allocator();
    var ctx = Context{};
    var pool: std.Thread.Pool = undefined;
    try pool.init(.{ .allocator = alloc, .n_jobs = 4 });
    defer pool.deinit();
    var wg: std.Thread.WaitGroup = .{};
    var i: i64 = 1;
    while (i <= 10) : (i += 1) {
        pool.spawnWg(&wg, Context.addSquare, .{ &ctx, i });
    }
    pool.waitAndWork(&wg);
    const out = std.io.getStdOut().writer();
    try out.print("{d}\n", .{ctx.total});
}

Exercise

What does waitAndWork do?

It stops the pool forever. It makes the calling thread help run tasks until the wait group is done. It starts new worker threads. It frees the pool allocator.