zig-concurrency

Zig Concurrent Programming

Concurrency primitives based on std.Thread and std.atomic (Zig 0.16.0).

Capability Boundaries

✅ Strengths

1. Thread creation and management (spawn, join, detach)
2. Synchronization primitives (Mutex, RwLock, Condition, Semaphore, ResetEvent, WaitGroup)
3. Thread pool (Thread.Pool) for parallel tasks
4. Atomic operations (std.atomic.Value)
5. Common concurrency patterns (producer-consumer, work queue)

⚠️ Prerequisites

1. Confirm the Zig version (zig version）
2. Understand basic concurrency concepts (race conditions, deadlocks)

❌ Out of Scope

1. Asynchronous I/O → use zig-http or zig-0.16 's std.Io
2. Non-concurrent scenarios → use the zig-0.16 skill

When to Use

• "Multithreaded data processing"
• "Execute multiple tasks in parallel"
• "Thread-safe shared state"

Data Privacy

This skill does not collect, store, or transmit any user data.

Workflow

Step 1. Identify the concurrency requirement — CPU-intensive / I/O waiting / parallel computation? Step 2. Choose the primitive — threads / thread pool / atomic operations? Step 3. Implement synchronization — Mutex / RwLock / WaitGroup Step 4. Test for correctness — race conditions, deadlock checks

Module Overview

std.Thread                  // 线程创建与管理
std.Thread.Mutex            // 互斥锁
std.Thread.Mutex.Recursive  // 可重入互斥锁
std.Thread.RwLock           // 读写锁
std.Thread.Condition        // 条件变量
std.Thread.Semaphore        // 计数信号量
std.Thread.ResetEvent       // 布尔事件标志（阻塞式）
std.Thread.WaitGroup        // 等待多任务完成
std.Thread.Pool             // 线程池
std.Thread.Futex            // 底层 futex（高级）
std.atomic.Value(T)         // 原子类型包装

Thread Creation

Basic Thread

fn worker(id: usize) void {
    std.debug.print("Worker {d}\n", .{id});
}

const thread = try std.Thread.spawn(.{}, worker, .{42});
thread.join();  // 等待完成

With Allocator Configuration

const thread = try std.Thread.spawn(.{ .allocator = allocator }, worker, .{42});
defer thread.join();

Detached Thread (no join)

const thread = try std.Thread.spawn(.{}, worker, .{1});
thread.detach();  // 线程自动清理

Synchronization Primitives

Mutex (mutual exclusion lock)

var mutex: std.Thread.Mutex = .{};
var shared: i32 = 0;

fn increment() void {
    mutex.lock();
    defer mutex.unlock();
    shared += 1;
}

RwLock (read-write lock)

var rwlock: std.Thread.RwLock = .{};
var data: i32 = 0;

fn reader() void {
    rwlock.lockShared();
    defer rwlock.unlockShared();
    _ = data;  // 可并发读
}

fn writer() void {
    rwlock.lock();
    defer rwlock.unlock();
    data += 1;  // 独占写
}

Condition (condition variable)

var mutex: std.Thread.Mutex = .{};
var cond: std.Thread.Condition = .{};
var ready: bool = false;

fn waiter() void {
    mutex.lock();
    defer mutex.unlock();
    while (!ready) {
        cond.wait(&mutex);  // 等待通知
    }
}

fn notifier() void {
    mutex.lock();
    defer mutex.unlock();
    ready = true;
    cond.signal();  // 或 cond.broadcast() 通知所有等待者
}

Semaphore

var sem = std.Thread.Semaphore{ .permits = 3 }; // 最多 3 个并发

fn worker() void {
    sem.wait();   // 获取许可
    defer sem.post();  // 释放许可
    // 执行工作...
}

WaitGroup

var wg: std.Thread.WaitGroup = .{};
wg.reset();  // 初始化计数器为 0

fn worker(wg: *std.Thread.WaitGroup) void {
    defer wg.finish();
    // 执行工作
}

// 启动 5 个任务
wg.start();  // +1
const t1 = try std.Thread.spawn(.{}, worker, .{&wg});
wg.start();  // +1
const t2 = try std.Thread.spawn(.{}, worker, .{&wg});
wg.wait();   // 等待所有 finish()

ResetEvent

var event: std.Thread.ResetEvent = .{};

fn waiter() void {
    event.wait();  // 阻塞直到被设置
}

fn setter() void {
    // ... 准备工作
    event.set();  // 通知等待者
}

Thread Pool

var pool: std.Thread.Pool = .{};
try pool.init(.{ .max_threads = 4 });
defer pool.deinit();

// 提交任务
for (0..100) |i| {
    try pool.spawn(worker, .{i});
}
// pool.deinit() 等待所有任务完成

Atomic Operations

var counter: std.atomic.Value(u64) = .{ .raw = 0 };

// 原子递增
_ = counter.fetchAdd(1, .acq_rel);

// 原子加载/存储
const val = counter.load(.acquire);
counter.store(42, .release);

// CAS（比较并交换）
const prev = counter.cmpxchgWeak(42, 100, .acq_rel, .acquire);

Atomic Ordering

// 从弱到强
.monotonic    // 仅保证原子性，无同步语义
.acquire      // 读取后所有后续操作可见
.release      // 写入前所有操作已完成
.acq_rel      // acquire + release
.seq_cst      // 全局顺序一致（默认）

Common Patterns

Producer-Consumer

const Queue = struct {
    items: std.ArrayList(i32),
    mutex: std.Thread.Mutex = .{},
    cond: std.Thread.Condition = .{},
    done: bool = false,
};

fn producer(q: *Queue) void {
    for (0..10) |i| {
        q.mutex.lock();
        q.items.append(@intCast(i));
        q.cond.signal();
        q.mutex.unlock();
    }
    q.mutex.lock();
    q.done = true;
    q.cond.broadcast();
    q.mutex.unlock();
}

fn consumer(q: *Queue) void {
    q.mutex.lock();
    while (!q.done or q.items.items.len > 0) {
        while (q.items.items.len == 0 and !q.done)
            q.cond.wait(&q.mutex);
        if (q.items.popOrNull()) |item| { /* 处理 */ }
    }
    q.mutex.unlock();
}

Gotchas

1. Zig threads do not return values — the thread function is fn(Args) void, and results are passed through shared memory
2. Mutex is not copyable — pass it using a pointer *std.Thread.Mutex
3. Spurious wakeups of condition variables — cond.wait() After returning, the condition must be rechecked (use while instead of if）
4. pool.spawn may block — when the thread pool is full, spawn will wait for an idle thread
5. std.Thread.spawn 's configuration parameters — the first parameter is .{}(default configuration), and you can set stack_size , etc.
6. defer is the best practice for unlocking — mutex.lock() immediately after defer mutex.unlock() to prevent omission

FAQ

Q: When should I use Mutex vs RwLock? A: For many reads and few writes, use RwLock (readers can run concurrently); when the read-write ratio is close, use Mutex (simpler implementation, no lock contention overhead).

Q: How should the thread pool size be set? A: For CPU-intensive workloads, use std.Thread.getCpuCount() or manually set it to the number of cores; for I/O-intensive workloads, it can be set larger (e.g., 4× the number of cores).

Q：std.Thread.spawn and pool.spawn difference? A: The former creates independent threads, which is suitable for a small number of long-term tasks; the latter delegates to a thread pool to reuse threads, which is suitable for a large number of short tasks.