Channels & cancellation

Send values between coroutines and threads, and stop work early.

On this page

valk.sync has channels to pass values between coroutines and threads, and cancel tokens to stop waiting work early.

Send and receive#

Valk

use valk.sync
use valk.thread

fn main() {
    let jobs = sync.Channel[uint].new(8) ! panic("channel")
    let results = sync.Channel[uint].new() ! panic("channel")

    let worker = thread.start(fn() {
        each jobs as job {
            results.send(job * 2) ! break
        }
    }) ! panic("thread")

    jobs.send(1) ! panic("closed")
    jobs.send(2) ! panic("closed")
    jobs.close()

    let total = (results.recv(1000) !? 0) + (results.recv(1000) !? 0)
    worker.await()
    println(total) // 6
}

new(8) holds at most 8 values, so send waits while it is full. new() has no limit.
new(0) makes a channel where send waits until a receiver takes the value.
each channel as value receives until the channel is closed and empty.
A channel works the same between coroutines on one thread.

Channel methods#

Method	Does
`send(value, timeout_ms, cancel)`	queues a value, waiting while full
`recv(timeout_ms, cancel)`	returns the next value, waiting for one
`try_send(value)` / `try_recv()`	never wait; throw `full` or `empty`
`close()`	receivers get what is queued, then `closed`
`is_closed()`, `length`	state of the channel

timeout_ms of 0 waits forever. Failures are sync.SyncError codes: closed, timeout, cancelled, empty, full.

Send objects#

Objects travel as atomic T. send makes a read-only copy, so later changes to the original are not sent:

Valk

use valk.sync

class Order {
    id: uint
    item: String
}

fn main() {
    let orders = sync.Channel[atomic Order].new() ! panic("channel")
    let order = Order { id: 1, item: "book" }
    orders.send(order) ! panic("closed")
    order.item = "changed"

    let received = orders.recv() ! panic("empty")
    println(received.item) // book
}

Wait on several channels#

sync.select returns the index of a channel that has a value or was closed. Take the value with try_recv. sync.after(ms) gives a channel that receives once, which makes a timeout:

Valk

use valk.sync

fn main() {
    let messages = sync.Channel[String].new() ! panic("channel")
    let deadline = sync.after(50) ! panic("timer")
    messages.send("hello") ! panic("closed")

    while true {
        let ready = sync.select(.{ messages, deadline }) ! break
        if ready == 0 : println(messages.try_recv() ! continue) // hello
        if ready == 1 {
            println("time is up")
            break
        }
    }
}

sync.ticker(ms) returns a Ticker whose channel receives every interval until stop().

Cancel work#

A CancelToken tells waiting work to stop. A send or recv given the token ends with cancelled:

Valk

use valk.sync

fn main() {
    let queue = sync.Channel[uint].new() ! panic("channel")
    let stop = sync.CancelToken.new() ! panic("token")
    stop.cancel_after(20)

    let value = queue.recv(0, stop) ! {
        if error_is(E.code, cancelled) : println("gave up") // gave up
        return
    }
    println(value)
}

Method	Does
`cancel()`	cancels the token and its children
`cancel_after(ms)`	cancels after a delay
`is_cancelled()`	whether it was cancelled
`wait(timeout_ms)`	waits until cancelled
`child()`	a token that is cancelled together with this one
`on_cancel(fn)`	runs a callback when cancelled

Sockets accept a token too, with connection.set_cancel(token); see TCP & UDP.

Wait for a group of tasks#

A WaitGroup counts running tasks: add() before starting one, done() when it finishes, wait() until all are done:

Valk

use valk.sync
use valk.time

fn main() {
    let group = sync.WaitGroup.new() ! panic("group")
    each 0..3 as i {
        group.add()
        co fn() {
            time.sleep_ms(5)
            group.done()
        }()
    }
    group.wait() ! panic("wait")
    println(group.count) // 0
}

To limit how many tasks run at once, use a Semaphore.

View page source ↗