Milo

Appearance

std/sync ​

Channel, wait-group, once, and atomic primitives for coordinating green tasks and Promise.blocking workers.

milo
from "std/sync" import { Channel, WaitGroup, Once, AtomicI64, AtomicI32, AtomicU64, AtomicBool }

There is no Mutex or RwLock for you to hold. The locking still happens: Channel is a shared bounded queue over a pthread mutex and two condition variables, written once here and audited behind unsafe impl Send/Sync, so producers and consumers coordinate without a lock at your call sites. What crosses the boundary is ownership, since send moves a value in and recv moves it out. Channel<T> is Send only when T is.

When several cores transform one large buffer, divide its ownership with std/shard rather than queueing the elements. See Concurrency.

Every type here is a reference-counted handle. .clone() gives another task or worker its own owner, and the shared object frees itself when the last owner drops — there is no destroy.

Every atomic operation on every type below is sequentially consistent (seq_cst), including both the success and failure orderings of a cas. There is no ordering parameter and no acquire/release/relaxed form. add/sub wrap on overflow, unlike ordinary Milo arithmetic, which traps.

Example: Lazy static ​

A module-level global whose initializer runs code is already initialized lazily, on its first read and exactly once, so a lazy static needs no Once at all. Reach for Once when the one-time work is a side effect at a point you choose, or fills several globals together. The shape is a global plus a guard function, because a getter cannot hand back a &T:

milo
from "std/sync" import { Once }

var gTable: Vec<i64> = []
var gTableOnce: Once = Once.new()

pub fn ensureTable(): void {
    gTableOnce.run((): void => {
        gTable = buildTable()
    })
}

Callers do ensureTable() and then read gTable directly. There is no Lazy<T> or OnceCell<T>: with no way to return a reference, every get() would deep-copy the cached value.

Example: Producer-Consumer ​

The producer runs on a Promise.blocking worker so it makes progress while main consumes on the channel (a green producer would only run while the scheduler is driven):

milo
from "std/runtime" import { Promise }
from "std/sync" import { Channel }

fn main(): i32 {
    var ch = Channel<i64>.new(8)!

    let producer = Promise<i64>.blocking(move (): i64 => {
        ch.send(10)!
        ch.send(20)!
        ch.send(30)!
        ch.close()
        return 0
    })

    for val in ch {
        print(val)
    }

    producer.await()!
    print("done")
    return 0
}

API reference ​

AtomicBool ​

milo
pub struct AtomicBool

Lock-free atomic boolean. There is no AtomicPtr: a safe-looking one would be a lifetime claim nothing checks. Share a Vec index or an arena Handle instead.

AtomicBool.cas ​

milo
fn AtomicBool.cas(self: &AtomicBool, expected: bool, desired: bool): bool

Returns the value that was there — equal to expected exactly when the swap happened, which is how a caller claims a one-shot flag: f.cas(false, true) == false.

AtomicBool.clone ​

milo
fn AtomicBool.clone(self: &AtomicBool): AtomicBool

Share this atomic with another owner; freed when the last owner drops.

AtomicBool.load ​

milo
fn AtomicBool.load(self: &AtomicBool): bool

Atomic read.

AtomicBool.new ​

milo
fn AtomicBool.new(initial: bool): AtomicBool

An atomic holding initial.

AtomicBool.store ​

milo
fn AtomicBool.store(self: &AtomicBool, val: bool): void

Atomic write.

AtomicBool.swap ​

milo
fn AtomicBool.swap(self: &AtomicBool, val: bool): bool

Atomic swap. Returns the old value.

AtomicI32 ​

milo
pub struct AtomicI32

Lock-free signed 32-bit atomic integer.

AtomicI32.add ​

milo
fn AtomicI32.add(self: &AtomicI32, val: i32): i32

Returns the OLD value. Wraps on overflow.

AtomicI32.cas ​

milo
fn AtomicI32.cas(self: &AtomicI32, expected: i32, desired: i32): i32

Store desired only if the current value is expected. Returns the value that was there — equal to expected exactly when the swap happened.

AtomicI32.clone ​

milo
fn AtomicI32.clone(self: &AtomicI32): AtomicI32

Share this atomic with another owner; freed when the last owner drops.

AtomicI32.load ​

milo
fn AtomicI32.load(self: &AtomicI32): i32

Atomic read.

AtomicI32.new ​

milo
fn AtomicI32.new(initial: i32): AtomicI32

An atomic holding initial.

AtomicI32.store ​

milo
fn AtomicI32.store(self: &AtomicI32, val: i32): void

Atomic write.

AtomicI32.sub ​

milo
fn AtomicI32.sub(self: &AtomicI32, val: i32): i32

Returns the OLD value. Wraps on underflow.

AtomicI32.swap ​

milo
fn AtomicI32.swap(self: &AtomicI32, val: i32): i32

Atomic swap. Returns the old value.

AtomicI64 ​

milo
pub struct AtomicI64

Lock-free signed 64-bit atomic integer.

AtomicI64.add ​

milo
fn AtomicI64.add(self: &AtomicI64, val: i64): i64

Atomic add. Returns the OLD value. Wraps on overflow.

AtomicI64.cas ​

milo
fn AtomicI64.cas(self: &AtomicI64, expected: i64, desired: i64): i64

Store desired only if the current value is expected. Returns the value that was there — equal to expected exactly when the swap happened.

AtomicI64.clone ​

milo
fn AtomicI64.clone(self: &AtomicI64): AtomicI64

Share this atomic with another owner (e.g. a spawned task). Each clone must be dropped exactly once; the underlying storage is freed when the last owner drops.

AtomicI64.load ​

milo
fn AtomicI64.load(self: &AtomicI64): i64

Atomic read.

AtomicI64.new ​

milo
fn AtomicI64.new(initial: i64): AtomicI64

An atomic holding initial.

AtomicI64.store ​

milo
fn AtomicI64.store(self: &AtomicI64, val: i64): void

Atomic write.

AtomicI64.sub ​

milo
fn AtomicI64.sub(self: &AtomicI64, val: i64): i64

Atomic subtract. Returns the OLD value. Wraps on underflow.

AtomicI64.swap ​

milo
fn AtomicI64.swap(self: &AtomicI64, val: i64): i64

Atomic swap. Returns the old value.

AtomicU64 ​

milo
pub struct AtomicU64

Lock-free unsigned 64-bit atomic integer. Uses the same instructions as AtomicI64 (64-bit atomics have no notion of sign), so add/sub wrap through the full u64 range.

AtomicU64.add ​

milo
fn AtomicU64.add(self: &AtomicU64, val: u64): u64

Returns the OLD value. Wraps on overflow.

AtomicU64.cas ​

milo
fn AtomicU64.cas(self: &AtomicU64, expected: u64, desired: u64): u64

Store desired only if the current value is expected. Returns the value that was there — equal to expected exactly when the swap happened.

AtomicU64.clone ​

milo
fn AtomicU64.clone(self: &AtomicU64): AtomicU64

Share this atomic with another owner; freed when the last owner drops.

AtomicU64.load ​

milo
fn AtomicU64.load(self: &AtomicU64): u64

Atomic read.

AtomicU64.new ​

milo
fn AtomicU64.new(initial: u64): AtomicU64

An atomic holding initial.

AtomicU64.store ​

milo
fn AtomicU64.store(self: &AtomicU64, val: u64): void

Atomic write.

AtomicU64.sub ​

milo
fn AtomicU64.sub(self: &AtomicU64, val: u64): u64

Returns the OLD value. Wraps on underflow.

AtomicU64.swap ​

milo
fn AtomicU64.swap(self: &AtomicU64, val: u64): u64

Atomic swap. Returns the old value.

Channel ​

milo
pub struct Channel<T>

Bounded FIFO channel for streaming values between green tasks and Promise.blocking workers. send blocks when full, recv when empty; for v in ch drains it until close.

Channel.clone ​

milo
fn Channel.clone(self: &Channel): Channel<T>

Share this channel with another owner (a spawned producer/consumer). Each clone must be dropped exactly once; the queue is torn down when the last owner drops, and any payload still queued at that point is destroyed (its Drop runs once). Values already received are the receiver's and are not touched. send/recv take &Self, so a handle only needs cloning when moved into a task while the parent still uses it.

Channel.close ​

milo
fn Channel.close(self: &Channel): void

Signal no more values will be sent. Pending items are still delivered.

Channel.len ​

milo
fn Channel.len(self: &Channel): i64

Number of values currently queued.

Channel.new ​

milo
fn Channel.new(capacity: i64): Result<Channel<T>>

A channel holding at most capacity values.

Channel.next ​

milo
fn Channel.next(self: &mut Channel): Option<T>

Iterator protocol — enables for val in channel { ... } Uses match, not let-else: std must stay within the subset milo-self parses (src-milo has no let-else yet), or self-host can't compile std.

Channel.rawPtr ​

milo
fn Channel.rawPtr(self: &Channel): *u8

Raw ChannelInner pointer, for std/select arm hooks (channelArm*).

Channel.recv ​

milo
fn Channel.recv(self: &Channel): Result<T>

Receive a value, blocking while the channel is empty.

Channel.retainHandle ​

milo
fn Channel.retainHandle(self: &Channel): ChannelHandle

A T-erased owner of the queue, for std/select: a channel arm keeps one so the channel outlives the arm even when every Channel<T> is dropped before wait(). If this handle ends up the last owner, the queue is freed but payloads still in it are not destroyed (only Channel<T>.drop knows T): they leak, they never dangle.

Channel.send ​

milo
fn Channel.send(self: &Channel, val: T): Result<i32>

Send val, blocking while the channel is full.

Channel.tryRecv ​

milo
fn Channel.tryRecv(self: &Channel): Option<T>

Receive without blocking: None if the channel is empty.

Channel.trySend ​

milo
fn Channel.trySend(self: &Channel, val: T): bool

Send without blocking: false if the channel is full.

ChannelHandle ​

milo
pub struct ChannelHandle

Non-generic reference-counted owner of a ChannelInner. Channel<T> holds one of these; the handle's Drop is the fallback for a handle that was never wrapped, and the refcount/free logic lives here once, free of T.

ChannelHandle.retain ​

milo
fn ChannelHandle.retain(self: &ChannelHandle): ChannelHandle

ChannelInner ​

milo
pub struct ChannelInner

Bounded FIFO channel for safe message passing between threads. Channel is a reference-counted handle — .clone() to share it with another owner (a task/worker); the queue frees itself when the last owner drops. No manual destroy. Payloads still queued when the last owner drops are destroyed (their Drop runs, once each); payloads already received belong to the receiver and are untouched.

let ch = Channel<i64>.new(16)! let chW = ch.clone() let p = Promise<i64>.blocking(move (): i64 => { chW.send(42)! return 0 }) let val = ch.recv()! p.await()! // ch (and chW, dropped by the worker) free the queue automatically

Fields:

  • mtx: *u8
  • condNotEmpty: *u8
  • condNotFull: *u8
  • buf: *u8
  • capacity: i64
  • len: i64
  • head: i64
  • tail: i64
  • closed: i64
  • recvWaitHead: *u8: intrusive lists of parked green tasks; nodes live on the parked task's stack: [0..8) task ptr, [8..16) next ptr. pthread waiters use the conds.
  • sendWaitHead: *u8
  • refcount: i64: inlined Arc-style refcount. Channel<T>.drop takes the last-owner decrement itself (it must destroy queued payloads, which needs T) and then zeroes its ChannelHandle so the handle's own Drop below is a no-op.

Once ​

milo
pub struct Once

Run-exactly-once initialization guard, correct under green tasks and Promise.blocking threads alike: a green waiter parks, an OS-thread waiter blocks on a condition variable, and the main thread with a live scheduler drives it.

Once.clone ​

milo
fn Once.clone(self: &Once): Once

Share this Once with another owner (a worker task/thread); freed when the last owner drops. run takes &Self, so a module-level Once never needs a clone.

Once.isDone ​

milo
fn Once.isDone(self: &Once): bool

True once the initializer has completed. Still false while it is running, so this is a progress hint, never a substitute for run.

Once.new ​

milo
fn Once.new(): Once

A guard whose initializer has not run yet.

Once.run ​

milo
fn Once.run(self: &Once, f: () => void): void

Run f if nobody has yet, otherwise block until whoever did is finished. Returns only once the initializer has completed exactly once, process-wide. @synchronized: f is a critical section. The embedded mutex serializes it and the atomic state word publishes its writes, so globals mutated in here are not racing.

Re-entering run from inside its own initializer would wait for itself forever, so it aborts with that message instead of hanging.

WaitGroup ​

milo
pub struct WaitGroup

Counting barrier: add before spawning, done from each task, wait until the counter reaches zero. Works across green tasks and OS threads alike.

WaitGroup.add ​

milo
fn WaitGroup.add(self: &WaitGroup, n: i64): void

Add n to the counter. Call it before spawning the tasks it tracks.

WaitGroup.clone ​

milo
fn WaitGroup.clone(self: &WaitGroup): WaitGroup

Share this WaitGroup with another owner (e.g. a worker task); freed when the last owner drops. add/done/wait take &Self, so most uses need no clone.

WaitGroup.done ​

milo
fn WaitGroup.done(self: &WaitGroup): void

Decrement the counter by one. Call it from each task when it finishes.

WaitGroup.new ​

milo
fn WaitGroup.new(): WaitGroup

A wait group with a zero counter.

WaitGroup.wait ​

milo
fn WaitGroup.wait(self: &WaitGroup): void

Block until the counter reaches zero.