Channels

● Advanced5 minBefore this: TasksSpec 9.4

A channel passes values from one task to another, in order. put adds a value, take takes the next, and close says no more will come.

const (send, receive) = Channel<T>(capacity: n)

fn main() {
    const (send, receive) = Channel<Int>(capacity: 4)
    const _producer = spawn {
        send.put(1)
        send.put(2)
        send.close()
    }
    for n in receive {
        print(n)
    }
}

kavach run pipe.kav

put waits when the channel already holds capacity values: a fast producer can't run ahead and fill memory.
for n in receive takes values until the channel is closed and empty.
Each value is copied into the channel, so the two tasks never share it.

Channels are how tasks talk without sharing anything: one task produces, another consumes, and neither waits for a lock.

Putting without waiting

send.try_put(v) adds v if the channel has room and gives true. When it is full or closed, it gives false at once, and v is let go:

fn main() {
    const (send, receive) = Channel<Int>(capacity: 2)
    for n in 1..=3 {
        print("try_put({n}): {send.try_put(n)}")
    }
    send.close()
    print("try_put(4) after close: {send.try_put(4)}")
    for n in receive {
        print("took {n}")
    }
}

kavach run try-put.kav

try_put(1): truetry_put(2): truetry_put(3): falsetry_put(4) after close: falsetook 1took 2

It never waits, so one slow taker can't hold up a task that sends to many, such as a server sending news to every listener.

tested

When the receivers are gone

A receiver is gone once nothing holds it: every copy of the Receiver was let go, as when the task that held it ended. Then try_put gives false, so a task that sends to many, such as a chat room, sees at once that a member left:

/// Joins the room, takes `wants` messages, then leaves.
fn member(joins: Sender<Sender<String>>, wants: Int) -> [String] {
    const (send, receive) = Channel<String>(capacity: 4)
    joins.put(send)
    let got: [String] = []
    for _ in 1..=wants {
        match receive.take() {
            Taken.Item(m) => got.push(m),
            Taken.Closed => return got,
        }
    }
    got
}

fn main() ! Error {
    const (joins, joined) = Channel<Sender<String>>(capacity: 2)
    const ann = spawn member(joins, 2)
    const bob = spawn member(joins, 1)
    let members: [Sender<String>] = []
    for _ in 1..=2 {
        match joined.take() {
            Taken.Item(s) => members.push(s),
            Taken.Closed => panic("nobody joined"),
        }
    }
    members = members.filter(m => m.try_put("news 1"))
    print("bob got {try bob.wait()}")
    members = members.filter(m => m.try_put("news 2"))   // bob left: his channel takes nothing
    print("{members.len()} member left")
    print("ann got {try ann.wait()}")
}

kavach run receivers-gone.kav

bob got ["news 1"]1 member leftann got ["news 1", "news 2"]

members.filter(m => m.try_put(…)) keeps the members that took the message: it drops those who left, and those too slow to keep up. A put nobody could ever take panics, and so does a put waiting for room when the last receiver goes:

fn main() {
    const send = {
        const (send, _receive) = Channel<Int>(capacity: 4)
        send
    }
    print(send.try_put(1))
    send.put(2)
    print("not reached")
}

kavach run put-nobody-takes.kav

falsepanic: `put` into a channel whose receivers are all gone

A channel's ends live to the end of their block, so the receiver above is gone once its block ends.

tested a waiting put, tested

Limits and sizes

What	Limit or size	Source
`Channel capacity`	1 or more; put waits while the channel is full	09-concurrency.md
`A task's stack`	512 MB of address space each, used only as calls need it	lib.rs

Every limit in Kavach, on one page: Limits and Sizes.

One page for each form

put and take →Many senders →Deadlocks →