Threads & Atomics

// This example demonstrates parallel computation using threads and atomic operations in Zig.
// It calculates the sum of even numbers in an array by distributing work across multiple threads.
const std = @import("std");

// Arguments passed to each worker thread for parallel processing
const WorkerArgs = struct {
    slice: []const u64,                  // The subset of numbers this worker should process
    sum: *std.atomic.Value(u64),         // Shared atomic counter for thread-safe accumulation
};

// Worker function that accumulates even numbers from its assigned slice
// Each thread runs this function independently on its own data partition
fn accumulate(args: WorkerArgs) void {
    // Use a local variable to minimize atomic operations (performance optimization)
    var local_total: u64 = 0;
    for (args.slice) |value| {
        if (value % 2 == 0) {
            local_total += value;
        }
    }

    // Atomically add the local result to the shared sum using sequentially consistent ordering
    // This ensures all threads see a consistent view of the shared state
    _ = args.sum.fetchAdd(local_total, .seq_cst);
}

pub fn main() !void {
    // Set up memory allocator with automatic leak detection
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const allocator = gpa.allocator();

    // Allocate array of 64 numbers for demonstration
    var numbers = try allocator.alloc(u64, 64);
    defer allocator.free(numbers);

    // Initialize array with values following the pattern: index * 7 + 3
    for (numbers, 0..) |*slot, index| {
        slot.* = @as(u64, @intCast(index * 7 + 3));
    }

    // Initialize shared atomic counter that all threads will safely update
    var shared_sum = std.atomic.Value(u64).init(0);

    // Determine optimal number of worker threads based on available CPU cores
    const cpu_count = std.Thread.getCpuCount() catch 1;
    const desired = if (cpu_count == 0) 1 else cpu_count;
    // Don't create more threads than we have numbers to process
    const worker_limit = @min(numbers.len, desired);

    // Allocate thread handles for parallel workers
    var threads = try allocator.alloc(std.Thread, worker_limit);
    defer allocator.free(threads);

    // Calculate chunk size, rounding up to ensure all elements are covered
    const chunk = (numbers.len + worker_limit - 1) / worker_limit;

    // Spawn worker threads, distributing the array into roughly equal chunks
    var start: usize = 0;
    var spawned: usize = 0;
    while (start < numbers.len and spawned < worker_limit) : (spawned += 1) {
        const remaining = numbers.len - start;
        // Give the last thread all remaining elements to handle uneven divisions
        const take = if (worker_limit - spawned == 1) remaining else @min(chunk, remaining);
        const end = start + take;

        // Spawn thread with its assigned slice and shared accumulator
        threads[spawned] = try std.Thread.spawn(.{}, accumulate, .{WorkerArgs{
            .slice = numbers[start..end],
            .sum = &shared_sum,
        }});

        start = end;
    }

    // Track how many threads were actually spawned (may be less than worker_limit)
    const used_threads = spawned;

    // Wait for all worker threads to complete their work
    for (threads[0..used_threads]) |thread| {
        thread.join();
    }

    // Read the final accumulated result from the atomic shared sum
    const even_sum = shared_sum.load(.seq_cst);

    // Perform sequential calculation to verify correctness of parallel computation
    var sequential: u64 = 0;
    for (numbers) |value| {
        if (value % 2 == 0) {
            sequential += value;
        }
    }

    // Set up buffered stdout writer for efficient output
    var stdout_buffer: [256]u8 = undefined;
    var stdout_state = std.fs.File.stdout().writer(&stdout_buffer);
    const out = &stdout_state.interface;

    // Display results: thread count and both parallel and sequential sums
    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();
}

// This example demonstrates thread-safe one-time initialization using atomic operations.
// Multiple threads attempt to initialize a shared resource, but only one succeeds in
// performing the expensive initialization exactly once.

const std = @import("std");

// Represents the initialization state using atomic operations
const State = enum(u8) { idle, busy, ready };

// Global state tracking the initialization lifecycle
var once_state: State = .idle;
// The shared configuration value that will be initialized once
var config_value: i32 = 0;
// Counter to verify that initialization only happens once
var init_calls: u32 = 0;

// Simulates an expensive initialization operation that should only run once.
// Uses atomic operations to safely increment the call counter and set the config value.
fn expensiveInit() void {
    // Simulate expensive work with a sleep
    std.Thread.sleep(2 * std.time.ns_per_ms);
    // Atomically increment the initialization call counter
    _ = @atomicRmw(u32, &init_calls, .Add, 1, .seq_cst);
    // Atomically store the initialized value with release semantics
    @atomicStore(i32, &config_value, 9157, .release);
}

// Ensures expensiveInit() is called exactly once across multiple threads.
// Uses a state machine with compare-and-swap to coordinate thread access.
fn callOnce() void {
    while (true) {
        // Check the current state with acquire semantics to see initialization results
        switch (@atomicLoad(State, &once_state, .acquire)) {
            // Initialization complete, return immediately
            .ready => return,
            // Another thread is initializing, yield and retry
            .busy => {
                std.Thread.yield() catch {};
                continue;
            },
            // Not yet initialized, attempt to claim initialization responsibility
            .idle => {
                // Try to atomically transition from idle to busy
                // If successful (returns null), this thread wins and will initialize
                // If it fails (returns the actual value), another thread won, so retry
                if (@cmpxchgStrong(State, &once_state, .idle, .busy, .acq_rel, .acquire)) |_| {
                    continue;
                }
                // This thread successfully claimed the initialization
                break;
            },
        }
    }

    // Perform the one-time initialization
    expensiveInit();
    // Mark initialization as complete with release semantics
    @atomicStore(State, &once_state, .ready, .release);
}

// Arguments passed to each worker thread
const WorkerArgs = struct {
    results: []i32,
    index: usize,
};

// Worker thread function that calls the once-initialization and reads the result.
fn worker(args: WorkerArgs) void {
    // Ensure initialization happens (blocks until complete if another thread is initializing)
    callOnce();
    // Read the initialized value with acquire semantics
    const value = @atomicLoad(i32, &config_value, .acquire);
    // Store the observed value in the thread's result slot
    args.results[args.index] = value;
}

pub fn main() !void {
    // Reset global state for demonstration
    once_state = .idle;
    config_value = 0;
    init_calls = 0;

    // Set up memory allocation
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const allocator = gpa.allocator();

    const worker_count: usize = 4;

    // Allocate array to collect results from each thread
    const results = try allocator.alloc(i32, worker_count);
    defer allocator.free(results);
    // Initialize all result slots to -1 to detect if any thread fails
    for (results) |*slot| slot.* = -1;

    // Allocate array to hold thread handles
    const threads = try allocator.alloc(std.Thread, worker_count);
    defer allocator.free(threads);

    // Spawn all worker threads
    for (threads, 0..) |*thread, index| {
        thread.* = try std.Thread.spawn(.{}, worker, .{WorkerArgs{
            .results = results,
            .index = index,
        }});
    }

    // Wait for all threads to complete
    for (threads) |thread| {
        thread.join();
    }

    // Read final values after all threads complete
    const final_value = @atomicLoad(i32, &config_value, .acquire);
    const called = @atomicLoad(u32, &init_calls, .seq_cst);

    // Set up buffered output
    var stdout_buffer: [256]u8 = undefined;
    var stdout_state = std.fs.File.stdout().writer(&stdout_buffer);
    const out = &stdout_state.interface;

    // Print the value observed by each thread (should all be 9157)
    for (results, 0..) |value, index| {
        try out.print("thread {d} observed {d}\n", .{ index, value });
    }
    // Verify initialization was called exactly once
    try out.print("init calls: {d}\n", .{called});
    // Display the final configuration value
    try out.print("config value: {d}\n", .{final_value});
    try out.flush();
}

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

// Enum representing the possible states of task execution
// Uses explicit u8 backing to ensure consistent size across platforms
const TaskState = enum(u8) { idle, threaded_done, inline_done };

// Global atomic state tracking whether task ran inline or in a separate thread
// Atomics ensure thread-safe access even though single-threaded builds won't spawn threads
var task_state = std.atomic.Value(TaskState).init(.idle);

// Simulates a task that runs in a separate thread
// Includes a small delay to demonstrate asynchronous execution
fn threadedTask() void {
    std.Thread.sleep(1 * std.time.ns_per_ms);
    // Release ordering ensures all prior writes are visible to threads that acquire this value
    task_state.store(.threaded_done, .release);
}

// Simulates a task that runs inline in the main thread
// Used as fallback when threading is disabled at compile time
fn inlineTask() void {
    // Release ordering maintains consistency with the threaded path
    task_state.store(.inline_done, .release);
}

pub fn main() !void {
    // Set up buffered stdout writer for efficient output
    var stdout_buffer: [256]u8 = undefined;
    var stdout_state = std.fs.File.stdout().writer(&stdout_buffer);
    const out = &stdout_state.interface;

    // Reset state to idle with sequential consistency
    // seq_cst provides strongest ordering guarantees for initialization
    task_state.store(.idle, .seq_cst);

    // Check compile-time flag to determine execution strategy
    // builtin.single_threaded is true when compiled with -fsingle-threaded
    if (builtin.single_threaded) {
        try out.print("single-threaded build; running task inline\n", .{});
        // Execute task directly without spawning a thread
        inlineTask();
    } else {
        try out.print("multi-threaded build; spawning worker\n", .{});
        // Spawn separate thread to execute task concurrently
        var worker = try std.Thread.spawn(.{}, threadedTask, .{});
        // Block until worker thread completes
        worker.join();
    }

    // Acquire ordering ensures we observe all writes made before the release store
    const final_state = task_state.load(.acquire);
    
    // Convert enum state to human-readable string for output
    const label = switch (final_state) {
        .idle => "idle",
        .threaded_done => "threaded_done",
        .inline_done => "inline_done",
    };

    // Display final execution state and flush buffer to ensure output is visible
    try out.print("task state: {s}\n", .{label});
    try out.flush();
}