Fibers, channels, and select

// spawn takes a function CALL and runs it on a new fiber.
// Fibers are lightweight green threads scheduled by the VM.
fn greet(name: string) {
    println("hello from ${name}")
}

fn main() {
    spawn greet("fiber A")
    spawn greet("fiber B")

    // The main fiber yields so the spawned fibers get a turn to run.
    fiber_yield()
    fiber_yield()
    println("main is done")
}

// Channels carry typed values between fibers. A function that receives a
// channel declares its element type with Channel<T>.
fn produce(out: Channel<int>) {
    send(out, 1)
    send(out, 2)
    send(out, 3)
}

fn main() {
    // chan(n) is buffered (holds up to n values); chan() is unbuffered.
    let ch = chan(3)
    spawn produce(ch)

    // Bind each received value with a select arm (see below for why).
    var seen = 0
    while seen < 3 {
        select {
            n = <-ch => {
                println("recv ${n}")
                seen = seen + 1
            }
        }
    }
}

// recv(ch) BLOCKS until a value is available -- but as an EXPRESSION it
// evaluates to an ok-bool (true while the channel is open), not the value.
// So you cannot bind the received value this way:
fn main() {
    let ch = chan(1)
    send(ch, 99)

    let ok = recv(ch)          // ok is `true`, NOT 99
    println("recv returned: ${ok}")

    // To BIND the value, use a select arm -- that is the idiom.
    send(ch, 99)
    select {
        v = <-ch => println("v is ${v}")
    }
}

// select waits on several channel operations at once and runs the first
// arm that is ready. It supports three kinds of arm:
//   v = <-ch  =>   receive arm  (binds the received value to v)
//   x -> ch   =>   send arm     (sends the in-scope value x into ch)
//   _         =>   default arm  (runs immediately if nothing else is ready)
fn main() {
    let ch = chan(1)

    // Nothing has been sent yet, so the default arm fires: select is
    // non-blocking when a default arm is present.
    select {
        v = <-ch => println("got ${v}")
        _ => println("nothing ready yet")
    }

    let payload = 7
    select {
        payload -> ch => println("sent ${payload}")
        _ => println("send would block")
    }

    // Now the value is buffered, so the receive arm wins.
    select {
        v = <-ch => println("got ${v}")
        _ => println("nothing ready yet")
    }
}

// A worker pool: one producer fills a jobs channel, three workers pull jobs,
// square them, and fan results back in. A WaitGroup tells main when every
// worker has drained the queue.
import std.sync

fn worker(jobs: Channel<int>, results: Channel<int>, wg: WaitGroup) {
    var running = true
    while running {
        select {
            j = <-jobs => send(results, j * j)
            _ => running = false   // no jobs left: this worker retires
        }
    }
    wg.done()
}

fn main() {
    let jobs = chan(8)
    let results = chan(8)
    let wg = new_wait_group()

    var i = 1
    while i <= 8 {
        send(jobs, i)
        i = i + 1
    }

    spawn worker(jobs, results, wg)
    spawn worker(jobs, results, wg)
    spawn worker(jobs, results, wg)

    wg.wait(3)   // block until all three workers signal done

    var total = 0
    var got = 0
    while got < 8 {
        select {
            r = <-results => {
                total = total + r
                got = got + 1
            }
        }
    }
    println("sum of squares 1..8 = ${total}")
}

// @expect-error
// The scheduler detects deadlock: if every fiber is blocked with no way to
// make progress, the program halts with a located error instead of hanging.
fn main() {
    let ch = chan()      // unbuffered: a send needs a matching receiver

    // No other fiber is running, so this recv can never be satisfied.
    let ok = recv(ch)
    println("unreachable: ${ok}")
}

// std.sync gives you shared-state primitives. Here two fibers each increment
// a shared IntMutex 1000 times; mutual exclusion guarantees no lost updates,
// so the total is exactly 2000.
import std.sync

fn worker(m: IntMutex, wg: WaitGroup, iters: int) {
    var i = 0
    while i < iters {
        let v = m.lock()       // lock returns the current value
        m.unlock(v + 1)        // unlock writes the new value back
        i = i + 1
    }
    wg.done()
}

fn main() {
    let m = new_int_mutex(0)
    let wg = new_wait_group()
    let iters = 1000

    spawn worker(m, wg, iters)
    spawn worker(m, wg, iters)

    wg.wait(2)

    let total = m.lock()
    m.unlock(total)
    println("total: ${total}")
    println("expected: ${2 * iters}")
}

// A Semaphore bounds how many fibers run a section at once. Eight workers
// share a Semaphore(2): each one bumps an "active" counter on entry and
// records the running maximum, so we can prove concurrency never exceeds 2.
import std.sync

fn worker(sem: Semaphore, active: IntMutex, maxseen: IntMutex, wg: WaitGroup) {
    sem.acquire()                 // blocks if 2 permits are already taken
    let now = active.lock() + 1
    active.unlock(now)
    let mx = maxseen.lock()
    if now > mx { maxseen.unlock(now) } else { maxseen.unlock(mx) }

    fiber_yield()                 // give peers a chance to overlap
    fiber_yield()

    let a = active.lock()
    active.unlock(a - 1)
    sem.release()
    wg.done()
}

fn main() {
    let limit = 2
    let sem = new_semaphore(limit)
    let active = new_int_mutex(0)
    let maxseen = new_int_mutex(0)
    let wg = new_wait_group()

    var i = 0
    while i < 8 {
        spawn worker(sem, active, maxseen, wg)
        i = i + 1
    }
    wg.wait(8)

    let mx = maxseen.lock()
    maxseen.unlock(mx)
    println("limit: ${limit}")
    println("max concurrent: ${mx}")
}