Worker pool

A worker pool shares a stream of jobs between a fixed number of goroutines: each job goes to whichever worker is free. Spreading the work out is fan-out; collecting the results back into one place is fan-in.

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

'' Each worker takes jobs until the jobs channel is closed and empty.
sub Worker( byval jobs as tasks.RecvChan( of int32 ), _
            byval results as tasks.SendChan( of int32 ), _
            byval wg as tasks.WaitGroup ptr )
    dim as int32 n
    do while jobs.Receive( n )
        results.Send( n * n )
    loop
    wg->Done( )                       '' this worker has finished
end sub

'' Closes results once every worker has called Done( ).
sub CloseWhenDone( byval wg as tasks.WaitGroup ptr, _
                   byval results as tasks.SendChan( of int32 ) )
    wg->Wait( )
    results.Close( )
end sub

'' Sends the jobs 1 to upTo, then closes the channel.
sub Feed( byval upTo as int32, byval jobs as tasks.SendChan( of int32 ) )
    for n as int32 = 1 to upTo
        jobs.Send( n )
    next
    jobs.Close( )
end sub

var jobs = tasks.Channel( of int32 )( 4 )
var results = tasks.Channel( of int32 )( 4 )
dim wg as tasks.WaitGroup

'' Fan out: three workers share one jobs channel.
for w as int32 = 1 to 3
    wg.Add( )                         '' count it before it starts
    spawn Worker( jobs, results, @wg )
next
spawn CloseWhenDone( @wg, results )

'' Feed the jobs from another goroutine, so the main program is free
'' to collect results while they are still being sent.
spawn Feed( 10, jobs )

'' Fan in: one loop collects from all three.
dim squares as containers.Array( of int32 )
for each sq in results
    squares.Push( sq )
next
containers.Sort( squares )            '' arrival order varies
print squares.Count( ); " results:";
for each sq in squares
    print sq;
next
print

Output

 10 results: 1 4 9 16 25 36 49 64 81 100

Three Workers receive from the one jobs channel and send to the one results channel. Which worker squares which number, and the order the squares arrive in, change from run to run; the sorted list does not.

The main program's for each over results ends when results is closed, and it must be closed exactly once, after the last worker has finished. No worker knows that it is the last, so a tasks.WaitGroup counts them:

CloseWhenDone waits on the group and then closes results. Two rules keep it correct. Call Add before spawn, never inside the goroutine: otherwise Wait can run before the goroutine has counted itself and return early. And pass the WaitGroup by pointer: it cannot be copied, because a copy would be a second counter, so passing one byval is refused with TKO0746: the operator let of 'waitgroup' is private or protected and cannot be reached from here.

Feed runs as a goroutine too, so the main program can collect results while jobs are still going out. Had the main program sent every job itself before collecting anything, it would work only while the jobs fit in the channels' buffers and the workers' hands. With 20 jobs instead of 10 it stops for good: both buffers are full, every worker waits to send a result, and the main program waits to send a job.

Activities

1. Run the pool with one worker, then with ten. The output is the same; time the two with a job that sleeps for 10 ms.
2. Make each result say which worker produced it, and count the jobs each one took.