线程与原子

// 此示例演示在Zig中使用线程和原子操作进行并行计算。
// 它通过在多个线程之间分配工作来计算数组中偶数的和。
const std = @import("std");

// 传递给每个工作线程用于并行处理的参数
const WorkerArgs = struct {
    slice: []const u64,              // 该工作线程应处理的数字子集
    sum: *std.atomic.Value(u64),     // 用于线程安全累加的共享原子计数器
};

// 工作函数，从其分配的切片中累加偶数
// 每个线程在其自己的数据分区上独立运行此函数
fn accumulate(args: WorkerArgs) void {
    // 使用局部变量以最小化原子操作（性能优化）
    var local_total: u64 = 0;
    for (args.slice) |value| {
        if (value % 2 == 0) {
            local_total += value;
        }
    }

    // 使用顺序一致性排序原子性地将局部结果添加到共享总和
    // 这确保所有线程都能看到共享状态的一致视图
    _ = args.sum.fetchAdd(local_total, .seq_cst);
}

pub fn main() !void {
    // 设置带自动泄漏检测的内存分配器
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const allocator = gpa.allocator();

    // 分配64个数字的数组用于演示
    var numbers = try allocator.alloc(u64, 64);
    defer allocator.free(numbers);

    // 用模式：index * 7 + 3 的值初始化数组
    for (numbers, 0..) |*slot, index| {
        slot.* = @as(u64, @intCast(index * 7 + 3));
    }

    // 初始化所有线程将安全更新的共享原子计数器
    var shared_sum = std.atomic.Value(u64).init(0);

    // 根据可用CPU核心数确定最优工作线程数
    const cpu_count = std.Thread.getCpuCount() catch 1;
    const desired = if (cpu_count == 0) 1 else cpu_count;
    // 不要创建比要处理的数字更多的线程
    const worker_limit = @min(numbers.len, desired);

    // 分配并行工作线程的线程句柄
    var threads = try allocator.alloc(std.Thread, worker_limit);
    defer allocator.free(threads);

    // 计算块大小，向上舍入以确保覆盖所有元素
    const chunk = (numbers.len + worker_limit - 1) / worker_limit;

    // 生成工作线程，将数组分配为大致相等的块
    var start: usize = 0;
    var spawned: usize = 0;
    while (start < numbers.len and spawned < worker_limit) : (spawned += 1) {
        const remaining = numbers.len - start;
        // 让最后一个线程处理所有剩余元素以处理不均匀的除法
        const take = if (worker_limit - spawned == 1) remaining else @min(chunk, remaining);
        const end = start + take;

        // 使用其分配的切片和共享累加器生成线程
        threads[spawned] = try std.Thread.spawn(.{}, accumulate, .{WorkerArgs{
            .slice = numbers[start..end],
            .sum = &shared_sum,
        }});

        start = end;
    }

    // 跟踪实际生成了多少线程（可能少于worker_limit）
    const used_threads = spawned;

    // 等待所有工作线程完成其工作
    for (threads[0..used_threads]) |thread| {
        thread.join();
    }

    // 从原子共享总和读取最终累加结果
    const even_sum = shared_sum.load(.seq_cst);

    // 执行顺序计算以验证并行计算的正确性
    var sequential: u64 = 0;
    for (numbers) |value| {
        if (value % 2 == 0) {
            sequential += value;
        }
    }

    // 设置缓冲stdout写入器以实现高效输出
    var stdout_buffer: [256]u8 = undefined;
    var stdout_state = std.fs.File.stdout().writer(&stdout_buffer);
    const out = &stdout_state.interface;

    // 显示结果：线程数和并行与顺序总和
    try out.print("spawned {d} worker(s)\n", .{used_threads});
    try out.print("even sum (threads): {d}\n", .{even_sum});
    try out.print("even sum (sequential check): {d}\n", .{sequential});
    try out.flush();
}

// 此示例演示使用原子操作的线程安全一次性初始化。
// 多个线程尝试初始化共享资源，但只有一个成功
// 精确执行一次昂贵的初始化。

const std = @import("std");

// 使用原子操作表示初始化状态
const State = enum(u8) { idle, busy, ready };

// 跟踪初始化生命周期的全局状态
var once_state: State = .idle;
// 将被一次性初始化的共享配置值
var config_value: i32 = 0;
// 验证初始化只发生一次的计数器
var init_calls: u32 = 0;

// 模拟只应运行一次的昂贵初始化操作。
// 使用原子操作安全地增加调用计数器并设置配置值。
fn expensiveInit() void {
    // 通过睡眠模拟昂贵的工作
    std.Thread.sleep(2 * std.time.ns_per_ms);
    // 原子性地增加初始化调用计数器
    _ = @atomicRmw(u32, &init_calls, .Add, 1, .seq_cst);
    // 使用发布语义原子性地存储初始化值
    @atomicStore(i32, &config_value, 9157, .release);
}

// 确保expensiveInit()在多个线程中只被调用一次。
// 使用带有比较和交换的状态机来协调线程访问。
fn callOnce() void {
    while (true) {
        // 使用获取语义检查当前状态以查看初始化结果
        switch (@atomicLoad(State, &once_state, .acquire)) {
            // 初始化完成，立即返回
            .ready => return,
            // 另一个线程正在初始化，让出并重试
            .busy => {
                std.Thread.yield() catch {};
                continue;
            },
            // 尚未初始化，尝试声明初始化责任
            .idle => {
                // 尝试从空闲原子性地转换到忙碌
                // 如果成功（返回null），此线程获胜并将初始化
                // 如果失败（返回实际值），另一个线程获胜，因此重试
                if (@cmpxchgStrong(State, &once_state, .idle, .busy, .acq_rel, .acquire)) |_| {
                    continue;
                }
                // 此线程成功声明了初始化
                break;
            },
        }
    }

    // 执行一次性初始化
    expensiveInit();
    // 使用发布语义将初始化标记为完成
    @atomicStore(State, &once_state, .ready, .release);
}

// 传递给每个工作线程的参数
const WorkerArgs = struct {
    results: []i32,
    index: usize,
};

// 工作线程函数，调用一次性初始化并读取结果。
fn worker(args: WorkerArgs) void {
    // 确保初始化发生（如果另一个线程正在初始化则阻塞直到完成）
    callOnce();
    // 使用获取语义读取初始化值
    const value = @atomicLoad(i32, &config_value, .acquire);
    // 将观察到的值存储在线程的结果槽中
    args.results[args.index] = value;
}

pub fn main() !void {
    // 重置全局状态用于演示
    once_state = .idle;
    config_value = 0;
    init_calls = 0;

    // 设置内存分配
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const allocator = gpa.allocator();

    const worker_count: usize = 4;

    // 分配数组以收集每个线程的结果
    const results = try allocator.alloc(i32, worker_count);
    defer allocator.free(results);
    // 初始化所有结果槽为-1以检测是否有任何线程失败
    for (results) |*slot| slot.* = -1;

    // 分配数组以保存线程句柄
    const threads = try allocator.alloc(std.Thread, worker_count);
    defer allocator.free(threads);

    // 生成所有工作线程
    for (threads, 0..) |*thread, index| {
        thread.* = try std.Thread.spawn(.{}, worker, .{WorkerArgs{
            .results = results,
            .index = index,
        }});
    }

    // 等待所有线程完成
    for (threads) |thread| {
        thread.join();
    }

    // 所有线程完成后读取最终值
    const final_value = @atomicLoad(i32, &config_value, .acquire);
    const called = @atomicLoad(u32, &init_calls, .seq_cst);

    // 设置缓冲输出
    var stdout_buffer: [256]u8 = undefined;
    var stdout_state = std.fs.File.stdout().writer(&stdout_buffer);
    const out = &stdout_state.interface;

    // 打印每个线程观察到的值（应该都是9157）
    for (results, 0..) |value, index| {
        try out.print("thread {d} observed {d}\n", .{ index, value });
    }
    // 验证初始化只被调用一次
    try out.print("init calls: {d}\n", .{called});
    // 显示最终配置值
    try out.print("config value: {d}\n", .{final_value});
    try out.flush();
}

const std = @import("std");
const builtin = @import("builtin");

// 表示任务执行可能状态的枚举
// 使用显式 u8 后备以确保跨平台一致的大小
const TaskState = enum(u8) { idle, threaded_done, inline_done };

// 全局原子状态跟踪任务是内联运行还是在单独线程中运行
// 原子操作确保线程安全访问，即使单线程构建不会生成线程
var task_state = std.atomic.Value(TaskState).init(.idle);

// 模拟在单独线程中运行的任务
// 包括一个小延迟来演示异步执行
fn threadedTask() void {
    std.Thread.sleep(1 * std.time.ns_per_ms);
    // 发布排序确保获取此值的所有线程都能看到之前的写入
    task_state.store(.threaded_done, .release);
}

// 模拟在主线程内联运行的任务
// 用作编译时禁用线程时的后备方案
fn inlineTask() void {
    // 发布排序保持与线程路径的一致性
    task_state.store(.inline_done, .release);
}

pub fn main() !void {
    // 为高效输出设置缓冲标准输出写入器
    var stdout_buffer: [256]u8 = undefined;
    var stdout_state = std.fs.File.stdout().writer(&stdout_buffer);
    const out = &stdout_state.interface;

    // 使用顺序一致性重置状态为空闲
    // seq_cst 为初始化提供最强的排序保证
    task_state.store(.idle, .seq_cst);

    // 检查编译时标志以确定执行策略
    // 当使用 -fsingle-threaded 编译时，builtin.single_threaded 为 true
    if (builtin.single_threaded) {
        try out.print("single-threaded build; running task inline\n", .{});
        // 直接执行任务而不生成线程
        inlineTask();
    } else {
        try out.print("multi-threaded build; spawning worker\n", .{});
        // 生成单独线程以并发执行任务
        var worker = try std.Thread.spawn(.{}, threadedTask, .{});
        // 阻塞直到工作线程完成
        worker.join();
    }

    // 获取排序确保我们观察到发布存储之前的所有写入
    const final_state = task_state.load(.acquire);
    
    // 将枚举状态转换为人类可读的字符串用于输出
    const label = switch (final_state) {
        .idle => "idle",
        .threaded_done => "threaded_done",
        .inline_done => "inline_done",
    };

    // 显示最终执行状态并刷新缓冲区以确保输出可见
    try out.print("task state: {s}\n", .{label});
    try out.flush();
}