Concurrency

Beans threads are OS threads. thread.spawn runs a closure on a new OS thread, so it is the tool for CPU-heavy work and for reaching another core. Closures plus std.thread do the whole job.

For work that mostly waits — thousands of connections, timers, channel handoffs — reach for a fiber instead: brew starts a green thread that parks rather than blocking. See Fibers and brew.

import std.thread

// spawn: run a closure on another thread
let t: Thread<int> = thread.spawn(fn() -> int {
    return heavy_work()
})
let n: int = t.join()               // wait, then take the value

// a mutex wraps the data itself, so you cannot touch it without the lock
let ledger: Mutex<Ledger> = new Mutex(new Ledger())
ledger.with_lock(fn(l: Ledger) {
    l.post(entry)                   // locked for exactly this block, auto-unlock
})

// channels move work between threads
let ch: Channel<string> = new Channel(64)      // buffered
ch.send("job")
let job: Option<string> = ch.receive()         // none when closed and empty

// atomics for plain counters
let hits: AtomicInt = new AtomicInt(0)
hits.add_and_get(1)

spawn and join

thread.spawn(fn() -> T) runs a closure on a new OS thread and returns a Thread<T>. join() waits for the thread to finish and returns its value; detach() lets it finish independently and discards the result. The returned value T must be Send, and so must everything the closure captures.

Send and Sync

A thread.spawn closure may capture only Send values and must return a Send value.

• Local by default: plain class references and plain fn closures.
• Conditional: List<T>, Box<T>, and Arena<T> are Send when T is. Map<K, V> and OrderedMap<K, V> need both types to be Send.
• Move-only owners: Bytes, File, MMap, TCP/UDP sockets, pollers, and HTTP/HTTP2/WebSocket handles are Send, but mutable owners are not Sync.
• Shared tools: scalars, immutable strings, atomics, a Mutex or Channel of Send values, and Shared<T>/Weak<T> where T is Send & Sync.

This makes a class a local reference by default. Capturing a non-Send value in a spawned closure is a compile error that names the value and its type:

import std.thread

class Local {}

fn main() {
    let item: Local = new Local()
    var xs: List<Local> = [item]
    let t: Thread<int> = thread.spawn(fn() -> int {
        return xs.len()            // error: List<Local> is not Send
    })
    t.join()
}

To share mutable data across threads, wrap it in a Mutex, which is Send. See Memory and ownership.

For one-owner handoff, move the value into the closure. A direct closure passed to spawn is inferred as send fn; a stored sendable closure can name that type:

let data: Bytes = Bytes.filled(4096, 0)
let work: send fn() -> int =
    fn() move(data) -> int { return data.len() }
let worker: Thread<int> = thread.spawn(move work)

Plain fn values stay local and cloneable. A send fn is move-only.

Mutex

Mutex<T> holds the protected value. with_lock acquires the lock, passes the value to your closure, and unlocks when the closure exits. The value is accessible only inside the closure.

Channels

Channel<T> moves values between threads. new Channel(capacity) makes a buffered channel; send(x) puts a value in, receive() returns some(v) or none when the channel is closed and empty, and close() closes it. Pairing defer ch.close() with the channel closes it on every exit path.

Atomics

For plain counters and flags, use atomics:

• AtomicInt is a simple sequentially-consistent integer: new AtomicInt(0), load(), store(v), add_and_get(v).
• Atomic<T> is a typed atomic over any integer or bool, with explicit memory orders and compare_exchange, fetch_add, wait/notify, and Atomic.fence. See Atomics and MemoryOrder.

A complete program

Spawn a worker, guard shared state with a Mutex, and hand a value over a Channel:

import std.io
import std.thread

class Ledger {
    total: int = 0
    fn add(n: int) { self.total += n }
}

fn main() {
    let t: Thread<int> = thread.spawn(fn() -> int {
        return 21 * 2
    })
    io.println("worker said {t.join()}")

    let ledger: Mutex<Ledger> = new Mutex(new Ledger())
    ledger.with_lock(fn(l: Ledger) {
        l.add(10)
    })

    let ch: Channel<int> = new Channel(4)
    ch.send(7)
    ch.close()
    match ch.receive() {
        some(v) => io.println("got {v}"),
        none    => io.println("empty"),
    }

    let hits: AtomicInt = new AtomicInt(0)
    hits.add_and_get(1)
    io.println("hits {hits.load()}")
}

Readiness waits

To wait on many sockets from one thread, register them with the std.poll poller and wait for readiness events.