Channels

A tasks.Channel( of T ) is a typed queue that goroutines use to hand values to one another. It is created with a capacity, the number of values it can hold before a sender has to wait.

#import "std/tasks"
#cmdline "-mt"

'' A channel made with a capacity of 3 buffers up to three values.
var box = tasks.Channel( of int32 )( 3 )
box.Send( 10 )
box.Send( 20 )
box.Send( 30 )
print "queued:"; box.Count( ); " of"; box.Capacity( )

'' TrySend never waits: with the buffer full it fails instead.
print "try to send 40: "; box.TrySend( 40 )

'' Values come out in the order they went in.
dim as int32 v
box.Receive( v )
print "first out:"; v

'' Close: no more sends, but what is queued can still be received.
box.Close( )
for each x in box                 '' drains, then ends
    print "drained:"; x
next

'' Receive on a closed, empty channel returns false at once.
v = -1
print "receive after close: "; box.Receive( v ); ", v is still " & v

'' ReceiveFor gives up after a number of milliseconds.
var quiet = tasks.Channel( of int32 )( 1 )
print "wait 50 ms for a value: "; quiet.ReceiveFor( v, 50 )

Output

queued: 3 of 3
try to send 40: false
first out: 10
drained: 20
drained: 30
receive after close: false, v is still -1
wait 50 ms for a value: false

Receive returning false is not an error: it is how a receiver learns that the sender has finished, and for each over a channel is built on it. The variable keeps its old value when that happens. Sending on a closed channel stops the program with runtime error 1: it means two parts of the program disagree about who sends.

A channel is a handle. Copying one, or passing it to a procedure, gives another handle to the same queue, which is how one channel is shared between goroutines.

Pipelines and channel directions

Channels join goroutines into a pipeline, each stage reading from one channel and writing to the next:

#import "std/tasks"
#cmdline "-mt"

'' Each stage says which end of a channel it holds.
sub Generate( byval upTo as int32, byval out_ as tasks.SendChan( of int32 ) )
    for i as int32 = 1 to upTo
        out_.Send( i )
    next
    out_.Close( )                 '' tells the next stage there is no more
end sub

sub KeepOdd( byval in_ as tasks.RecvChan( of int32 ), _
             byval out_ as tasks.SendChan( of int32 ) )
    dim as int32 v
    do while in_.Receive( v )     '' false once in_ is closed and empty
        if v mod 2 = 1 then out_.Send( v )
    loop
    out_.Close( )
end sub

'' Capacity 0: every send waits for a receiver, a handover.
var numbers = tasks.Channel( of int32 )( 0 )
var odds = tasks.Channel( of int32 )( 0 )

spawn Generate( 9, numbers )
spawn KeepOdd( numbers, odds )

for each v in odds                '' ends when KeepOdd closes odds
    print v;
next
print

Output

 1 3 5 7 9

A capacity of 0 makes every send wait until a receiver takes the value: a handover, with no buffer between the stages. Generate closes numbers when it has sent everything; KeepOdd sees Receive return false and closes odds in turn; and the main program's for each ends. The close travels down the pipeline, and no stage needs to know how many values there are.

The parameter types say which end each stage holds. A tasks.SendChan( of T ) can send and close; a tasks.RecvChan( of T ) can only receive. A Channel converts to either when it is passed. Closing is the sender's decision, so a receiving end that tries is refused:

#import "std/tasks"
#cmdline "-mt"

sub Consume( byval inbox as tasks.RecvChan( of int32 ) )
    inbox.Close( )            '' closing is the sender's decision
end sub

recvclose.tiko(5,11): error TKO0739: 'recvchan' has no member named 'close'

Activities

1. Add a third stage between KeepOdd and the main program that squares each value.
2. Give numbers a capacity of 4 instead of 0. The output does not change; what does?