006 rings vs channels

// Go Baseline: Producer/Consumer with Buffered Channels
//
// Tests idiomatic Go concurrency:
// - Buffered channel (1024 capacity, like MPMC ring)
// - Goroutines (2 threads: producer + consumer)
// - Channel send/receive (10M messages)
// - Synchronization (WaitGroup)
// - Data integrity (checksum validation)
//
// This is how you'd actually write concurrent Go code.

package main

import (
	"fmt"
	"sync"
)

const MESSAGES = 10_000_000
const BUFFER_SIZE = 1024

func main() {
	// Buffered channel - like MPMC ring with 1024 capacity
	messages := make(chan uint64, BUFFER_SIZE)

	var wg sync.WaitGroup
	wg.Add(1)

	// Producer goroutine - send 10M messages
	go func() {
		defer wg.Done()
		for i := uint64(0); i < MESSAGES; i++ {
			messages <- i
		}
		close(messages)
	}()

	// Consumer runs on MAIN THREAD (same as Zig, Rust, and Koru!)
	var sum uint64
	for msg := range messages {
		sum += msg
	}

	// Wait for producer to complete
	wg.Wait()

	// Validate checksum (sum of 0 to N-1 = N*(N-1)/2)
	expected := uint64(MESSAGES * (MESSAGES - 1) / 2)
	if sum == expected {
		fmt.Printf("✓ Go: Validated %d messages (checksum: %d)\n", MESSAGES, sum)
	} else {
		fmt.Printf("✗ Go: CHECKSUM MISMATCH! got %d, expected %d\n", sum, expected)
	}
}

// Rust Baseline: Producer/Consumer with Crossbeam Channels
//
// Tests idiomatic Rust concurrency:
// - Bounded channel (1024 capacity, like MPMC ring and Go channels)
// - Threads (2 threads: producer + consumer)
// - Channel send/receive (10M messages)
// - Zero-cost abstractions (no runtime overhead)
// - Data integrity (checksum validation)
//
// Uses crossbeam for fair comparison:
// - Bounded channels (matching Go's buffered channels)
// - Lock-free implementation (matching Zig's MPMC ring)
// - No async runtime overhead (matching Zig's approach)
//
// This is how you'd actually write concurrent Rust code with channels.

use crossbeam::channel::bounded;
use std::thread;

const MESSAGES: u64 = 10_000_000;
const BUFFER_SIZE: usize = 1024;

fn main() {
    // Bounded channel - like Go's buffered channel and Zig's MPMC ring
    let (tx, rx) = bounded(BUFFER_SIZE);

    // Producer thread - send 10M messages
    let producer = thread::spawn(move || {
        for i in 0..MESSAGES {
            tx.send(i).unwrap();
        }
        // Channel automatically closed when tx is dropped
    });

    // Consumer runs on MAIN THREAD (same as Zig and Koru!)
    let mut sum = 0u64;
    for msg in rx {
        sum += msg;
    }

    // Wait for producer to finish
    producer.join().unwrap();

    // Validate checksum (sum of 0 to N-1 = N*(N-1)/2)
    let expected = MESSAGES * (MESSAGES - 1) / 2;
    if sum == expected {
        println!(
            "✓ Rust: Validated {} messages (checksum: {})",
            MESSAGES, sum
        );
    } else {
        println!(
            "✗ Rust: CHECKSUM MISMATCH! got {}, expected {}",
            sum, expected
        );
    }
}

// Zig Baseline: Producer/Consumer with MPMC Ring
//
// Tests Vyukov's lock-free MPMC ring buffer:
// - MPMC ring (1024 capacity, lock-free atomics)
// - Threads (2 threads: producer + consumer)
// - Enqueue/dequeue (10M messages)
// - Synchronization (thread join)
// - Data integrity (checksum validation)
//
// MPMC ring vendored from beist-rings (https://github.com/...)
// Algorithm: Dmitry Vyukov's bounded MPMC queue

const std = @import("std");
const atomic = std.atomic;

const MESSAGES = 10_000_000;
const BUFFER_SIZE = 1024;

/// Vyukov's bounded MPMC ring buffer
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 {};
                }
            }
        }
    };
}

pub fn main() !void {
    var ring = MpmcRing(u64, BUFFER_SIZE).init();

    var sum: u64 = 0;

    // Producer thread
    const producer = try std.Thread.spawn(.{}, struct {
        fn run(r: *MpmcRing(u64, BUFFER_SIZE)) void {
            var i: u64 = 0;
            while (i < MESSAGES) : (i += 1) {
                while (!r.tryEnqueue(i)) {
                    std.Thread.yield() catch {};
                }
            }
        }
    }.run, .{&ring});

    // Consumer runs on MAIN THREAD (same as Koru!)
    var received: u64 = 0;
    while (received < MESSAGES) {
        if (ring.tryDequeue()) |value| {
            sum +%= value;
            received += 1;
        } else {
            std.Thread.yield() catch {};
        }
    }

    producer.join();

    // Validate checksum
    const expected: u64 = MESSAGES * (MESSAGES - 1) / 2;
    if (sum == expected) {
        std.debug.print("✓ Zig: Validated {} messages (checksum: {})\n", .{ MESSAGES, sum });
    } else {
        std.debug.print("✗ Zig: CHECKSUM MISMATCH! got {}, expected {}\n", .{ sum, expected });
    }
}

#!/bin/bash
# Benchmark: Concurrent Message Passing
# Compare Go channels vs Zig MPMC rings vs Rust channels vs Koru events vs Koru taps
#
# Tests:
# - Go: Buffered channels (idiomatic Go)
# - Zig: MPMC ring (Vyukov's lock-free algorithm)
# - Rust: Crossbeam bounded channels (lock-free)
# - Koru: Events/flows wrapping MPMC ring
# - Koru Taps: Pure event-based producer/consumer (no ring!)
#
# All send/receive 10M messages between producer/consumer threads
# Success criteria: Koru should match Zig (zero-cost abstraction!)

set -e

echo "============================================"
echo "  CONCURRENT MESSAGE PASSING BENCHMARK"
echo "  Go vs Zig vs Rust vs Koru vs Koru Taps"
echo "============================================"
echo ""

# Clean up previous builds
rm -f go_baseline zig_baseline rust_baseline bchan_baseline koru_output koru_taps_output backend backend.zig output_emitted.zig results.json
rm -rf zig-out .zig-cache target compile_backend.err backend.err Cargo.lock

echo "Building Go baseline (channels)..."
go build -o go_baseline baseline.go

echo "Building Zig baseline (MPMC ring)..."
zig build-exe baseline.zig -O ReleaseFast -femit-bin=zig_baseline

echo "Building Rust baseline (crossbeam channels)..."
cargo build --release --quiet
cp target/release/rust_baseline ./rust_baseline

echo "Building bchan baseline (MPSC)..."
zig build-exe -O ReleaseFast --dep bchan -Mroot=baseline_bchan.zig -Mbchan=vendor_bchan/src/lib.zig -femit-bin=bchan_baseline

echo "Building Koru version (events + MPMC)..."
# Two-pass compilation (see run_regression.sh for details)

# Pass 1: Frontend - Parse .kz -> backend.zig
koruc input.kz -o backend.zig

# Pass 2: Compile and run backend to generate final code
# koruc already generated build_backend.zig with all required modules
# Fix the REL_TO_ROOT path to point to the repo root
REL_TO_ROOT=$(realpath --relative-to="$(pwd)" /Users/larsde/src/koru)
sed -i.bak "s|const REL_TO_ROOT = \".\";|const REL_TO_ROOT = \"$REL_TO_ROOT\";|g" build_backend.zig
rm build_backend.zig.bak

# Compile backend using the generated build file
zig build --build-file build_backend.zig 2>compile_backend.err
if [ $? -ne 0 ]; then
    echo "ERROR: Failed to compile backend"
    cat compile_backend.err
    exit 1
fi

# Move backend to current directory
mv zig-out/bin/backend ./backend

# Run backend to generate and compile final executable
./backend koru_output 2>backend.err
if [ $? -ne 0 ]; then
    echo "ERROR: Backend execution failed"
    cat backend.err
    exit 1
fi

# Clean up build artifacts
rm -rf zig-out .zig-cache backend compile_backend.err backend.err

echo "Building Koru Taps version (pure events, no ring)..."
# Two-pass compilation for taps variant

# Pass 1: Frontend - Parse .kz -> backend.zig (same name, build_backend.zig references it)
koruc input_taps.kz -o backend.zig

# Pass 2: Compile and run backend to generate final code
# koruc already generated build_backend.zig with all required modules
# Fix the REL_TO_ROOT path to point to the repo root
REL_TO_ROOT=$(realpath --relative-to="$(pwd)" /Users/larsde/src/koru)
sed -i.bak "s|const REL_TO_ROOT = \".\";|const REL_TO_ROOT = \"$REL_TO_ROOT\";|g" build_backend.zig
rm build_backend.zig.bak

# Compile backend using the generated build file
zig build --build-file build_backend.zig 2>compile_backend.err
if [ $? -ne 0 ]; then
    echo "ERROR: Failed to compile taps backend"
    cat compile_backend.err
    exit 1
fi

# Move backend to current directory
mv zig-out/bin/backend ./backend

# Run backend to generate and compile final executable
./backend koru_taps_output 2>backend.err
if [ $? -ne 0 ]; then
    echo "ERROR: Taps backend execution failed"
    cat backend.err
    exit 1
fi

# Clean up build artifacts
rm -rf zig-out .zig-cache backend compile_backend.err backend.err

echo ""
echo "Running benchmarks with hyperfine..."
echo ""

# Check if hyperfine is installed
if ! command -v hyperfine &> /dev/null; then
    echo "ERROR: hyperfine not installed"
    echo "Install with: brew install hyperfine (macOS) or cargo install hyperfine"
    exit 1
fi

# Run benchmark
# - warmup: 3 runs to stabilize (message passing can vary)
# - runs: 10 (fewer than simple loop since this takes longer)
# - shell=none: avoid shell overhead
hyperfine --warmup 3 --runs 10 --shell=none \
    --export-json results.json \
    --command-name "Go (channels)" './go_baseline' \
    --command-name "Zig (MPMC)" './zig_baseline' \
    --command-name "bchan (MPSC)" './bchan_baseline' \
    --command-name "Rust (crossbeam)" './rust_baseline' \
    --command-name "Koru (events)" './koru_output' \
    --command-name "Koru (taps)" './koru_taps_output'

echo ""
echo "============================================"
echo "Benchmark complete! Results saved to results.json"
echo "============================================"

// ============================================================================
// Koru Taps Implementation: Pure Event-Based Producer/Consumer
// ============================================================================
// Tests the RAW POWER of event taps:
// - Simple counting loop (the "producer")
// - Tap observes each count and accumulates (the "consumer")
// - Tap observes completion to validate
//
// NO RING. NO CHANNEL. NO SHARED MEMORY STRUCTURE.
// Just events and taps proving zero-cost observation.
// ============================================================================

const std = @import("std");

const MESSAGES: u64 = 10_000_000;

// Global accumulator
var sum: u64 = 0;

// ============================================================================
// The counting loop - this is the "producer"
// ============================================================================

~tor count { i: u64 }
| next u64

~proc count|zig {
    if (i >= MESSAGES) {
    }
    return .{ .next = i };
}

// ============================================================================
// Void events for tap actions
// ============================================================================

~tor accumulate { value: u64 }

~proc accumulate|zig {
    sum += value;
}

~tor validate {}

~proc validate|zig {
    const expected: u64 = MESSAGES * (MESSAGES - 1) / 2;
    if (sum == expected) {
        std.debug.print("✓ Taps: Validated {} messages (checksum: {})\n", .{ MESSAGES, sum });
    } else {
        std.debug.print("✗ Taps: CHECKSUM MISMATCH! got {}, expected {}\n", .{ sum, expected });
    }
}

// Starting event to kick off the loop
~tor start {}

~proc start|zig {
}

// ============================================================================
// EVENT TAPS - The "consumer" pattern via observation
// ============================================================================

// TAP: Observe count, accumulate on "next"
~count -> * | next v |> accumulate(v.value)

// TAP: Observe count completion, validate
~count -> * | done |> validate()

// ============================================================================
// Main Flow - Just a simple loop
// ============================================================================

~start() |> #loop count(i: 0)
    | next n |> @loop(i: n.value + 1)