010 threading library

○ Planned This feature is planned but not yet implemented.

Feature: Threading library support

Code

// ============================================================================
// Threading Library Demo: Using std/threading
// ============================================================================
// Demonstrates using the generic threading library to solve the same
// multi-consumer problem as test 2005, but with REUSABLE primitives!
//
// This proves that progressive disclosure is not just a one-off pattern,
// but a GENERAL SOLUTION for expressing concurrency in Koru.
//
// Pattern:
// 1. Define work context (contains input AND output)
// 2. Define work function (takes context, writes result)
// 3. Use worker.spawn.async to spawn workers
// 4. Use worker.spawn.join to wait for all workers
// 5. Extract results from contexts
//
// All threading at SEMANTIC level (events from std/threading)!
// ============================================================================

const std = @import("std");
~import std/threading

// ============================================================================
// MPMC Ring Implementation (same as test 2005)
// ============================================================================

const atomic = std.atomic;
const BUFFER_SIZE = 1024;

fn MpmcRing(comptime T: type, comptime capacity: usize) type {
    if (capacity & (capacity - 1) != 0) {
        @compileError("Ring capacity must be power of 2");
    }

    const CacheLine = 64;

    const Slot = struct {
        seq: atomic.Value(usize),
        value: T,
    };

    return struct {
        const Self = @This();

        head: atomic.Value(usize) align(CacheLine),
        _pad1: [CacheLine - @sizeOf(atomic.Value(usize))]u8 = undefined,

        tail: atomic.Value(usize) align(CacheLine),
        _pad2: [CacheLine - @sizeOf(atomic.Value(usize))]u8 = undefined,

        slots: [capacity]Slot align(CacheLine),

        pub fn init() Self {
            var self = Self{
                .head = atomic.Value(usize).init(0),
                .tail = atomic.Value(usize).init(0),
                .slots = undefined,
            };

            for (&self.slots, 0..) |*slot, i| {
                slot.seq = atomic.Value(usize).init(i);
                slot.value = undefined;
            }

            return self;
        }

        pub fn tryEnqueue(self: *Self, value: T) bool {
            var pos = self.head.load(.monotonic);

            while (true) {
                const slot = &self.slots[pos & (capacity - 1)];
                const seq = slot.seq.load(.acquire);
                const dif = @as(isize, @intCast(seq)) -% @as(isize, @intCast(pos));

                if (dif == 0) {
                    if (self.head.cmpxchgWeak(
                        pos,
                        pos + 1,
                        .monotonic,
                        .monotonic,
                    ) == null) {
                        slot.value = value;
                        slot.seq.store(pos + 1, .release);
                        return true;
                    }
                    pos = self.head.load(.monotonic);
                } else if (dif < 0) {
                    return false;
                } else {
                    pos = self.head.load(.monotonic);
                    std.Thread.yield() catch {};
                }
            }
        }

        pub fn tryDequeue(self: *Self) ?T {
            var pos = self.tail.load(.monotonic);

            while (true) {
                const slot = &self.slots[pos & (capacity - 1)];
                const seq = slot.seq.load(.acquire);
                const dif = @as(isize, @intCast(seq)) -% @as(isize, @intCast(pos + 1));

                if (dif == 0) {
                    if (self.tail.cmpxchgWeak(
                        pos,
                        pos + 1,
                        .monotonic,
                        .monotonic,
                    ) == null) {
                        const value = slot.value;
                        slot.seq.store(pos + capacity, .release);
                        return value;
                    }
                    pos = self.tail.load(.monotonic);
                } else if (dif < 0) {
                    return null;
                } else {
                    pos = self.tail.load(.monotonic);
                    std.Thread.yield() catch {};
                }
            }
        }
    };
}

const Ring = MpmcRing(u64, BUFFER_SIZE);
const TOTAL_MESSAGES: u64 = 1_000_000;
const NUM_CONSUMERS: u64 = 4;
const MESSAGES_PER_CONSUMER: u64 = TOTAL_MESSAGES / NUM_CONSUMERS;

// ============================================================================
// Worker Context and Function
// ============================================================================
// This is the key difference from test 2005:
// We define a context that works with the generic threading library!

const ConsumerContext = struct {
    // Inputs (immutable, set before spawning)
    ring: *Ring,
    target: u64,

    // Outputs (mutable, written by worker)
    sum: u64,
    count: u64,
};

// Generic work function that matches WorkFn signature
fn consumerWorker(ctx: *anyopaque) *anyopaque {
    const consumer_ctx = @as(*ConsumerContext, @ptrCast(@alignCast(ctx)));

    var sum: u64 = 0;
    var received: u64 = 0;

    while (received < consumer_ctx.target) {
        if (consumer_ctx.ring.tryDequeue()) |value| {
            sum += value;
            received += 1;
        } else {
            std.Thread.yield() catch {};
        }
    }

    // Write results to context
    consumer_ctx.sum = sum;
    consumer_ctx.count = received;

    return ctx;
}

// ============================================================================
// Ring and Producer Setup (same as test 2005)
// ============================================================================

~tor createRing {} -> *Ring

~proc createRing|zig {
    const ring_storage = std.heap.page_allocator.create(Ring) catch unreachable;
    ring_storage.* = Ring.init();
    return ring_storage;
}

~tor spawnProducer { ring: *Ring }

~proc spawnProducer|zig {
    const producer = std.Thread.spawn(.{}, struct {
        fn run(r: *Ring) void {
            var i: u64 = 0;
            while (i < TOTAL_MESSAGES) : (i += 1) {
                while (!r.tryEnqueue(i)) {
                    std.Thread.yield() catch {};
                }
            }
        }
    }.run, .{ring}) catch unreachable;

    producer.detach();
}

// ============================================================================
// Consumer Spawning Using Threading Library
// ============================================================================

~tor spawnConsumer { ring: *Ring, target: u64 } -> *ConsumerContext

~proc spawnConsumer|zig {
    // Allocate context on heap (must outlive this function)
    const ctx = std.heap.page_allocator.create(ConsumerContext) catch unreachable;
    ctx.* = .{
        .ring = ring,
        .target = target,
        .sum = 0,
        .count = 0,
    };

    return ctx;
}

// ============================================================================
// Result Processing
// ============================================================================

~tor combineResults { contexts: []*ConsumerContext }
| combined { total_sum: u64, total_count: u64 }

~proc combineResults|zig {
    var total_sum: u64 = 0;
    var total_count: u64 = 0;

    for (contexts) |ctx| {
        total_sum += ctx.sum;
        total_count += ctx.count;
    }

    return .{ .combined = .{ .total_sum = total_sum, .total_count = total_count } };
}

~tor validate { sum: u64, count: u64 }

~proc validate|zig {
    const expected_sum: u64 = TOTAL_MESSAGES * (TOTAL_MESSAGES - 1) / 2;
    const expected_count: u64 = TOTAL_MESSAGES;

    if (sum == expected_sum and count == expected_count) {
        std.debug.print("✓ Threading Library: Validated {} messages across {} consumers (checksum: {})\n", .{ count, NUM_CONSUMERS, sum });
    } else {
        std.debug.print("✗ Threading Library: MISMATCH! got sum={}, count={}, expected sum={}, count={}\n", .{ sum, count, expected_sum, expected_count });
    }
}

// ============================================================================
// Helper Events: Wrap context casting in events
// ============================================================================

~tor castToOpaque { context: *ConsumerContext } -> *anyopaque

~proc castToOpaque|zig {
    return @as(*anyopaque, @ptrCast(context));
}

~tor extractContexts { results: []*anyopaque } -> []*ConsumerContext

~proc extractContexts|zig {
    const contexts = std.heap.page_allocator.alloc(*ConsumerContext, results.len) catch unreachable;

    for (results, 0..) |result, i| {
        contexts[i] = @as(*ConsumerContext, @ptrCast(@alignCast(result)));
    }

    return contexts;
}

// ============================================================================
// Main Flow: Using std/threading!
// ============================================================================
// This is the beautiful part - we use the LIBRARY, not inline implementations!
//
// Pattern:
// 1. Create 4 consumer contexts
// 2. Use threading:worker.spawn.async to spawn each with consumerWorker function
// 3. Use threading:worker.spawn.join to wait for all 4
// 4. Extract contexts and combine results
//
// All threading primitives come from the LIBRARY! 🎉

// Helper to collect handles into array
~tor collectHandles { h1: threading:WorkerHandle, h2: threading:WorkerHandle, h3: threading:WorkerHandle, h4: threading:WorkerHandle }
| collected { handles: []const threading:WorkerHandle }

~proc collectHandles|zig {
    const handles = std.heap.page_allocator.alloc(threading:WorkerHandle, 4) catch unreachable;
    handles[0] = h1;
    handles[1] = h2;
    handles[2] = h3;
    handles[3] = h4;
    return .{ .collected = .{ .handles = handles } };
}

~createRing(): r |> spawnProducer(r.ring) |> spawnConsumer(r.ring, target: MESSAGES_PER_CONSUMER): ctx1 |> castToOpaque(ctx1.context): c1 |> threading:worker.spawn.async(work_fn: consumerWorker, context: c1.ptr)
            | awaitable h1 |> spawnConsumer(r.ring, target: MESSAGES_PER_CONSUMER): ctx2 |> castToOpaque(ctx2.context): c2 |> threading:worker.spawn.async(work_fn: consumerWorker, context: c2.ptr)
                        | awaitable h2 |> spawnConsumer(r.ring, target: MESSAGES_PER_CONSUMER): ctx3 |> castToOpaque(ctx3.context): c3 |> threading:worker.spawn.async(work_fn: consumerWorker, context: c3.ptr)
                                    | awaitable h3 |> spawnConsumer(r.ring, target: MESSAGES_PER_CONSUMER): ctx4 |> castToOpaque(ctx4.context): c4 |> threading:worker.spawn.async(work_fn: consumerWorker, context: c4.ptr)
                                                | awaitable h4 |> collectHandles(h1: h1.handle, h2: h2.handle, h3: h3.handle, h4: h4.handle)
                                                    | collected handles |> threading:worker.spawn.join(handles.handles)
                                                        | all-completed results |> extractContexts(results.results): e |> combineResults(e.contexts)
                                                                | combined c |> validate(sum: c.total_sum, count: c.total_count)

input.kz

Supporting Files

# Threading Library Test - Blocked by Codegen Bugs

## Status: BLOCKED ⚠️

This test demonstrates using `std/threading` - a generic threading library based on the progressive disclosure pattern proven in test 2005.

**The library design is CORRECT and the pattern WORKS!**

However, we're hitting **codegen bugs** that prevent the test from compiling to valid Zig.

---

## What WORKS ✅

1. **Library exists**: `koru_std/threading.kz` with progressive disclosure:
   - `worker.spawn` (fire-and-forget)
   - `worker.spawn.async` (returns handle)
   - `worker.spawn.await` (wait for single)
   - `worker.spawn.join` (wait for multiple)

2. **Import works**: `~import std/threading` resolves correctly

3. **Type checking passes**: All event calls type-check correctly

4. **Frontend compilation succeeds**: Koru → AST → validation all pass

---

## What's BLOCKED ❌

### Bug 1: Slice Types from Imported Modules

**Problem**: When generating Zig code, slice types from imported modules produce invalid syntax.

**Example**:
```koru
// In koru_std/threading.kz
pub const WorkerHandle = struct { ... };

~event collectHandles { ... }
| collected { handles: []const threading:WorkerHandle }
```

**Generated Zig** (INVALID):
```zig
handles: koru_[]const threading.WorkerHandle,  // ❌ Should be: []const koru_threading.WorkerHandle
```

**Impact**: Cannot return/pass slices of imported types

**Isolated in**: test 990_slice_type_imported_module (to be created)

### Bug 2: Array Literal Types with Imported Modules

**Problem**: Array literals with types from imported modules use `:` syntax in generated Zig.

**Example**:
```koru
~worker.spawn.join(handles: &[_]threading:WorkerHandle{ h1, h2, h3, h4 })
```

**Generated Zig** (INVALID):
```zig
.handles = &[_]threading:WorkerHandle{ h1, h2, h3, h4 }  // ❌ Should be: koru_threading.WorkerHandle
```

**Impact**: Cannot create arrays of imported types in flows

**Isolated in**: test 991_array_literal_imported_type (to be created)

---

## Workarounds Attempted

1. **✅ Use wrapper events for casts** - Works! (`castToOpaque` event)
2. **❌ Use inline array literals** - Fails (Bug #2)
3. **❌ Use slice return types** - Fails (Bug #1)
4. **🔄 Use subflow plus helper proc events** - Not yet tried (suggested by user)

---

## Solution Path

### Short-term: Fix Codegen

File these as regression tests in the 990 range:
- `990_slice_type_imported_module` - Isolated bug reproduction
- `991_array_literal_imported_type` - Isolated bug reproduction

Fix the code generator to:
1. Emit `[]const koru_modulename.TypeName` for slice types
2. Emit `koru_modulename.TypeName` in array literals

### Alternative: Subflow Pattern

User suggestion: Define the main flow as a subflow and then call it from a
top-level flow.

**Example**:
```koru
~mainFlow = createRing()
| created r |> spawnProducer(ring: r.ring)
    | spawned |> spawnConsumer(ring: r.ring, target: MESSAGES_PER_CONSUMER)
        | spawned ctx1 |> threading:worker.spawn.async(
            work_fn: consumerWorker,
            context: @as(*anyopaque, @ptrCast(ctx1.context))  // ✅ Zig expr allowed!
        )
        // ... spawn 3 more
        | awaitable h4 |> threading:worker.spawn.join(
            handles: &[_]koru_threading.WorkerHandle{ h1.handle, h2.handle, h3.handle, h4.handle }  // ✅ Zig expr allowed!
        )
        // ... rest

~mainFlow()  // Top-level call
```

If this needs raw Zig expressions that are not legal in Koru flow arguments,
introduce helper events with `~proc` host-code implementations for those pieces.

---

## What This Proves

**Progressive disclosure for threading CAN be expressed as a reusable library!**

- ✅ Pattern is sound
- ✅ Library design is correct
- ✅ Type system handles it
- ✅ Import system works
- ⚠️ Codegen needs fixes OR helper proc events around host-only expressions

Test 2005 proves the PATTERN works (inline).
Test 2006 proves the LIBRARY works (modulo codegen).

Once codegen is fixed (or helper proc events isolate host-only expressions), **Koru will have a real, working, generic threading library!** 🚀

---

## References

- **Working pattern**: test 2005_multi_consumer_async
- **Library source**: koru_std/threading.kz
- **Related bugs**: tests 990, 991 (to be created)
- **Blog post**: "Async Without the Color" explains the theory

TODO.md

Expected output

✓ Threading Library: Validated 1000000 messages across 4 consumers (checksum: 499999500000)

Flows

flow ~createRing click a branch to expand · @labels scroll to their anchor

createRing

Test Configuration

MUST_RUN

← 009 multi consumer async 011 loop optimization basic →