Channels

What you’ll build

Two small pipelines. In the first, a thread produces lines of text and the main thread prints each as it arrives, until the producer says it’s done. In the second, ten tasks each send a result to one Channel, and the main thread adds them up. Last, you’ll try a Channel without waiting on it.

Sending and receiving

Channel<T>(capacity) makes a Channel that holds up to capacity values, 1 or more, of a Sendable T. Send puts a value in, waiting while the Channel is full, and Receive takes the oldest out, waiting while it’s empty. A Channel is Sendable itself, so a Const or a parameter hands it to another thread.

Import std.core

Function Produce(lines: Channel<String>)
    For i = 1 To 5
        lines.Send("line {i}")
    End For
    lines.Close()
End Function

Const lines = Channel<String>(2)
Local producer = Thread.Start(Function() => Produce(lines))
For line In lines
    Print("got {line}")
End For
producer.Join()
Print(lines.IsClosed())

got line 1
got line 2
got line 3
got line 4
got line 5
True

The Channel holds two lines at most, so the producer waits whenever the main thread falls behind. Close() says nothing more will come. For line In lines receives until the Channel is closed and empty, then the loop ends. Values come out in the order they went in.

Receive() gives a T?, which is Null once the Channel is closed and empty. Send on a closed Channel panics, since nobody would receive the value.

Many senders

Any number of threads can send to one Channel. Here ten tasks each send one square, in whatever order they finish, and the main thread receives ten values.

Import std.core

Function Square(n: Int, results: Channel<Int>)
    results.Send(n * n)
End Function

Const results = Channel<Int>(10)
For n = 1 To 10
    Spawn(Function() => Square(n, results))
End For

Local total = 0
For i = 1 To 10
    total += results.Receive() ?? 0
End For
Print("sum of squares: {total}")

sum of squares: 385

The squares arrive in any order, but their sum is the same every run.

Without waiting

TrySend and TryReceive never wait. TrySend gives False when the Channel is full or closed, and TryReceive() gives Null when there’s nothing to take. TryReceive(ByRef value) gives whether it took one, and sets the variable when it did.

Import std.core

Const box = Channel<Int>(1)
Print(box.TryReceive() ?? -1)
Print(box.TrySend(7), box.TrySend(8))

Local got = 0
Print(box.TryReceive(ByRef got), got)

-1
True False
True 7

The Channel holds one value, so the second TrySend finds it full. A loop that checks several Channels in turn uses these, with IsClosed() to know when one won’t get more.

A task on the pool that waits on a Channel keeps its worker thread while it waits, and the worker runs nothing else meanwhile, so a task that mostly waits on Channels belongs on a Thread of its own.

What you learned

• Channel<T>(capacity) passes Sendable values between threads, in order.
• Send waits while it’s full and Receive while it’s empty.
• Close() ends it, and For x In channel receives until it’s closed and empty.
• TrySend and TryReceive check a Channel without waiting.