Eve Concurrency

Keywords

Exclusive and concurrent aspects

An aspect is independent (see Aspects): it shares nothing with the other aspects and talks to the primary aspect only through its parameters. That makes it the unit of concurrency: two aspects that share nothing can run at the same time. But an aspect also imports modules, and a module is shared by the whole program. So an aspect says how it may run, with a word in front of aspect:

• exclusive aspect name is: runs alone, with apply. It may call any function and use any module, thread safe or not. Starting it is a compile error: the aspect 'worker' can not be started: worker is exclusive, start needs a concurrent aspect. A script that says only aspect is exclusive: exclusive is the default;
• concurrent aspect name is: may run beside other tasks, started in a parallel group, and may still be applied. The compiler checks the aspect at its declaration and reports every call that is not thread safe, with the chain of calls that leads to it (see Thread safety).

concurrent aspect worker is
  from "lib" use (plain);
  process main(n: Integer) is
    plain.hit();     ** error: plain.hit() changes a plain module variable,
                     ** it is not thread safe
  return;
end worker;

The default is the safe one: an aspect becomes concurrent only when its author writes the word, and then the compiler holds it to that promise. The mistake is found where the aspect is written, not later, at the place where somebody starts it. If the aspect really needs such a function, it is honest: write exclusive aspect, or make the module thread safe with an atomic variable.

Parallel aspects

Eve does concurrent work with aspects. An aspect is an encapsulated machine: it has one process, main, and nothing public. apply name(args); runs it in serial mode and the caller waits. start name(args);, written in a parallel group, runs it as a task, beside the others. Methods and functions always run in serial mode, on the core of their caller: they can't be started. These are the concepts:

• Driver: the master. Its process main opens the parallel groups, one after the other;
• Parallel group: a block parallel … do … done that starts aspects and waits for all of them;
• Task: one aspect started with start. It has its own state, created at start and dropped when its main returns, and its own copy of the input arguments;
• Barrier: the closing done. The primary aspect waits there until every task of the group has returned; then the outputs (@) are ready;
• Channel: a queue that connects tasks running at the same time. A task sends values into it, another task receives them.

With these elements Eve supports three models of concurrent work: task groups (different aspects side by side), Bulk Synchronous Parallel (the same aspect on slices of the data, in supersteps) and pipelines (stages connected by channels). An aspect can't apply another aspect, so the work of a task is done by its own methods and functions, and by the modules it imports.

Parallel groups

A parallel group starts aspects that run at the same time. The primary aspect executes the groups one by one: a group begins after the previous group has finished. Each group can have an optional label.

Syntax:

[label:] parallel [on error cancel] [within time]
  ** declarations, local to the group
do
  start aspect_name(arguments);
  ...
done [label];

parallel … do … start … done

Example: task group

The work of one worker is an aspect, in its own file task.eve:

# work done by one worker
concurrent aspect task is
  process main(n: Integer) is
    print ("task ", n, " is running");
  return;
end task;

The primary aspect starts it four times, in two groups:

# parallel groups demo
primary aspect parallel_demo is
  process main is
    ** the primary aspect is a series of sequential groups
    group1: parallel
      ** local variables for group1
    do
      start task(1);
      start task(2);
    done group1;

    ** group1 is now complete, group2 starts
    group2: parallel
      ** local variables for group2
    do
      start task(3);
      start task(4);
    done group2;

    ** Both groups are now complete
  return;
end parallel_demo;

Notes:

• Each group executes sequentially. Within a group, the started aspects run at the same time, on several cores;
• start is allowed only in the do region of a group, at any depth of if, for, while and match, but not inside a function called from there. It starts only concurrent aspects: start method() is an error;
• The declarations between parallel and do belong to the group and are dropped at done. A value needed after done is declared before the group;
• The closing done waits for all the aspects started in the group, and it is always for a single group;
• The label of a group is optional. A group without a label starts with parallel and ends with a naked done;; a labeled group ends with done label;.

Where a group is allowed

Each violation is a compile error that says what is allowed: parallel is allowed only in an aspect of the first two levels. The limit of three levels and two starts keeps the tree of tasks small enough to read: the aspect that is run, its workers, and at most two helpers for each worker. A call of a private process of the same aspect is not a level; only the call of an aspect is.

A primary aspect is serial: its processes are applied, never started, and its parallel groups start concurrent aspects. Nothing applies or starts a primary aspect; it is run by a command.

The primary aspect can start the same aspect several times with different arguments, to compute different parts of the data. Every started aspect writes its own element of the array s: two started aspects never receive the same output variable.

# compute one result
concurrent aspect square is
  process main(n: Integer, @result: Integer) is
    let result := n * n;
  return;
end square;

## create parallel work
primary aspect parallel_loop is
  process main is
    new s: [4]Integer; ** one result for each worker

    ** start the same aspect 4 times
    demo: parallel
    do
      for i in (1..4) do
        start square(i, @s[i]); ** runs in parallel
      done;
    done demo;
    ** all started aspects have finished, s is ready
    print s; ** [1,4,9,16]
  return;
end parallel_loop;

Data rules

Tasks run at the same instant, so two tasks must never write the same data. Eve guarantees this with a few rules; it needs no locks and no shared mutable objects, except channels.

• Inputs by value: the arguments are evaluated at start. An argument without "@" is passed by value, also a collection: the task works on its own copy, and the primary aspect can go on changing its variables. The virtual machine shares a large collection and copies it only when one side writes into it (copy on write), so passing by value is cheap;
• One owner for each output: an output argument (@) is passed by reference and belongs to one task. Two tasks of the same group can't receive the same variable, the same element or two overlapping slices as output. The compiler rejects the same variable, or the same element with a constant index, given to two starts. The virtual machine checks an element with a computed index, @s[i], at start. Both report Output s is given to two tasks of the group ($err_output 43);
• Channels are the exception: a channel is the only object that several tasks of a group can receive with "@". It is built for that: its operations are safe when tasks use it at the same time;
• Encapsulated: an aspect sees no globals of the aspect that calls it and has no public members but main; its state belongs to one task. It calls its methods by name, and they run in the same task. Its private processes work on the same state: a private process that is started in a parallel group runs as its own task, so it may use only the thread-safe names of the aspect (atomic variables, channels, constants and thread-safe functions); any other name is a compile error that names it;
• Three levels: the aspect that is run opens parallel groups. An aspect it starts may open its own parallel group and start at most 2 leaf aspects; a leaf aspect starts none. An aspect can't call itself, and no chain of calls comes back to an aspect already in it, so recursion between aspects is impossible.
• Modules are shared: all the tasks use the same copy of a module. Its exported constants, classes and plain functions are safe to share, and an exported channel of a module connects tasks without being passed as a parameter. The compiler decides which functions of a module are thread safe; an aspect that calls one that is not, directly or through other calls, can't be started. Module state that tasks share is atomic, guarded by a mutex or sent through a channel (see Thread safety).

Output of a group

Tasks run at the same time, so their print statements would mix in any order. Eve does not let that happen. Each task writes print and write into its own buffer, and at done the buffers are written to the standard output in the order of start, task after task. A task that ends first does not print first. Only log_err and log_wrn write at once, to the error output, with the name of the aspect in front, because a log line is needed when it happens.

** slow sleeps 50 ms and prints "slow"; fast prints "fast" at once
parallel
do
  start slow();
  start fast();
done;

slow
fast

This rule came out of the implementation. Without it, the tests of the virtual machine could not compare their output: the same program printed its lines in another order at every run. With it, the output of a group is the same on one core or on sixteen, and a test can check it line by line.

The same result at every run

A concurrent program is hard to test when its result depends on which task is faster. Eve fixes everything it can:

• the output of a group is in the order of start;
• the failed tasks in $error.errors are in the order of start;
• every output @ has one owner, so its value does not depend on the timing;
• the values of one sender keep their order in a channel.

What is not fixed: the order in which values from several senders arrive in one channel, the moment a timer ends, and ch.count(). A program that combines the values of several senders with an operation that does not depend on the order (a sum, a count, a maximum), or sends an index with each value, gives the same result at every run. The tests of level 4 rely on nothing else, and each one must pass 100 runs in a row.

Errors and time-out

A failing task never blocks its siblings: the other tasks go on, and done waits for all of them. Then the group raises one error in the primary aspect, ParallelError. No error is lost, and the order does not depend on the scheduling:

  process main is
    parallel
    do
      for i in (1..4) do
        start check(i);           ** check fails for 2 and 4
      done;
    done;
  recover
    print $error.message;         ** 2 of 4 tasks failed
    for e in $error.errors do
      print "{e.index} {e.aspect}: {e.message}";
    done;
  return;

In recover, retry runs the whole group again and resume goes on after its done. Without recover the next groups do not begin: the process ends with exit code 4, and the error output lists the failed tasks, one per line:

error 45: 2 of 4 tasks failed
task 2 check: error 4: bad 2
task 4 check: error 4: bad 4

• on error cancel changes the policy: after the first failure, the tasks not yet begun are not begun, and the running ones are stopped at their next wait or statement. A cancelled task is not an error and is not listed; $error.cancelled counts them. Without it, the group waits for all (the default);
• within time sets a deadline: parallel within 30s do … done; stops every task still running after 30 seconds, and each one appears in $error.errors with a TimeoutError (code 41), Time-out after 30s;
• A task that waits on a channel longer than $timeout (60 seconds by default, set in the configuration file of the primary aspect) gets a TimeoutError. An aspect may set its own value, set $timeout = 5s;, which shadows the primary aspect value inside that aspect only;
• When every unfinished task waits on a channel and no timer is pending, nothing can move any more: a deadlock. The virtual machine does not wait for the time-out. It stops the waiting tasks at once, and each gets a DeadlockError (code 42) that names it and its channel: Deadlock: consume waits to receive from jobs.

Workers

The number of started aspects is not limited by the number of cores. The virtual machine keeps a pool of worker threads, one for each core, and queues the extra tasks: when no core is free, a start waits until one is. Two system variables, set in the configuration file of the primary aspect, bound the work: $cores, the size of the pool (by default the number of hardware cores, at most 16), and $max_parallel, the number of aspects one parallel group of the primary aspect may start (8 by default). The compiler detects, when it can, a group that starts too many aspects and reports a warning.

A waiting aspect gives its core away. A started aspect that waits, for a file, the network, a database, a wait 10ms; or a channel, releases its worker to another task and continues when the wait is over. The virtual machine does this; the code of the aspect stays plain sequential code. So a group can start more aspects than there are cores and overlap their waits: calling 50 services is one group of 50 started aspects on a few cores, and a pipeline with more stages than cores still works.

How version 0.3 runs a group. The first machine takes the simplest road that keeps every rule of this page (stage A). Each task has its own thread, but only one task runs at a time: it runs until it waits, on a channel, on the done of its own group, or at its end, and then the next ready task runs, in the order of start. A group uses one core, but the results, the output and the errors are exactly those of the machine with many cores, because none of the rules above depends on the timing. Stage B spreads the tasks on $cores cores; stage C gives each task its own region of memory, freed when it ends.

The number of tasks can be dynamic. You can read it from the configuration file, compute it from the size of the data or from $cores, or hard code it as a constant inside the primary aspect. This gives you flexibility to tune your driver.

Modules shared by tasks

When several aspects run in a parallel group, they all use the same copy of a module (see Modules are singletons). Constants, classes and plain functions are safe to share. A module may also export a channel, declared with set: the reference to the channel is a constant, the values that flow through it change, and the aspects exchange data without passing the channel as a parameter.

# a shared log for parallel workers
module journal is
  export (lines);
  set lines := Channel(:String)(capacity: 1000);
end journal;

# a worker that writes into the shared log
concurrent aspect worker is
  from "lib" use (journal);

  process main(n: Integer) is
    journal.lines.send("worker {n} started");
    ...
  return;
end worker;

The private variables of a module are shared too. The compiler decides which functions and procedures are thread safe, on the call graph: one that reads or changes a plain module variable is not, and a concurrent aspect that calls it, directly or through other calls, is a compile error. An exclusive aspect, which is the default kind, can call anything. In the example counter, counter.tick() and counter.count!() become safe when total is declared atomic: new total = 0 :Atomic(:Integer); (or with a class derived from it, class SageInteger <: Atomic(:Integer);). A channel or an atomic variable is how concurrent aspects share state: see Thread safety. The functions of a module that are not thread safe stay available to the primary aspect and to aspects run with apply.

Managed and direct modules

The header of a module can promise that all of it is thread safe. A module is direct by default; managed is written in front of module:

# a managed module: no plain variable, no unsafe call
managed module pure is
  export (FACTOR, twice);

  set FACTOR = 2 :Integer;

  function twice(x: Integer) => (@result: Integer) is
    let result := x * FACTOR;
  return;
end pure;

The compiler checks a managed module as a whole and rejects it if a variable is not atomic, or if it uses an unsafe function or procedure, also from another module. A direct module can hold plain variables; its members are checked one by one, so a concurrent aspect may still call the members that the compiler proves safe. The words managed and direct are keywords only in front of module.

Thread safety

Two tasks that write the same variable at the same instant can lose an update: both read 5, both write 6, and one hit is gone. A function is thread safe when many tasks can call it at the same time and the result is always right. Thread safety is a different question from the one ! answers. ! marks a stochastic function, one whose result can differ for the same arguments, like roll!(). A function that is not stochastic is deterministic. A stochastic function can be thread safe, and a deterministic one may not be: what decides is the data it touches, not whether its result repeats.

The compiler decides

You do not mark a function or a procedure as thread safe. The compiler reads the call graph and decides for every subprogram, in every module:

• A subprogram that uses only its arguments and its own local variables is thread safe;
• A subprogram that reads or changes a plain variable of a module is not;
• A subprogram that uses a module variable only through an atomic variable, a mutex or a channel is thread safe;
• A subprogram that calls another is thread safe only if every subprogram it calls is.

The check is made on the aspect, where it is written (see Marking an aspect). For a concurrent aspect the compiler looks at everything the aspect can call, directly or through other calls. If one function in that set is not thread safe, the program does not compile, and the message names the function and the variable. An exclusive aspect has no such rule, because it is never started.

** a plain module variable: not thread safe
module plain is
  export (hit);
  new total = 0 :Integer;
  procedure hit() is
    let total += 1;              ** read, add, write: two tasks can lose an update
  return;
end plain;

The whole example is the Eve macro unsafe_counter, one file: the shared procedure hit() changes a plain shared variable, so the aspect worker that calls it compiles as an exclusive aspect and is applied; declared concurrent aspect, it does not compile.

Generators and threads

• A generator declared in a primary aspect or in an aspect is private to it. It runs on the core of its task and has one owner, so it can use any variable of the aspect. It needs to be neither thread safe nor deterministic (local_generator);
• A generator exported by a module is called by many tasks, so the compiler requires it to be thread safe, with the same rules as a function. The generator object itself still belongs to one task. A generator that hands out numbers from a shared counter uses an atomic variable. The same rule holds for a generator shared by the aspects of an Eve macro (shared_generator).

Atomic variables

An atomic variable holds one value, and each operation on it is indivisible: the operating system and the processor guarantee that no other thread sees it half done. It needs no lock, so it can never be part of a deadlock. It is declared with the class Atomic(:T) of the standard library, where T is a Logic, a number, a reference or an ordinal. The class is a wrapper of the atomic of the machine. A variable is initialized with a value, and a type is declared by derivation:

new total = 0 :Atomic(:Integer);    ** a variable
class SageInteger <: Atomic(:Integer);   ** a type, a class derived from Atomic(:Integer)

total.add(1);                      ** no result: add, nothing is lost
let n := total.get!();             ** read: the value changes, so the name ends with !
total.set(0);                      ** replace the value
let old := total.swap!(7);         ** replace the value and return the previous one
let done := total.compare_swap!(7, 8); ** set 8 only if the value is 7: True when it was set
let first := total.fetch_add!(1);  ** add and return the previous value

Operations with a result read a value that other tasks change, so they are stochastic and end with !. The complete example is the Eve macro atomic_counter: four workers count 4000 hits with a shared atomic variable, all in one file.

Advantage: the fastest tool, no waiting, no deadlock. Limit: one variable at a time. Two atomic variables change one after the other, so another task can see one changed and the other not.

Mutex

A mutex (mutual exclusion) lets one task at a time into a piece of code. A task calls acquire() and the others wait until it calls release(). Use it when several variables must change together:

new lock = Mutex();

procedure transfer(amount: Integer) is
  lock.acquire();                  ** waits while another task holds the mutex
  defer lock.release();            ** runs on every path, also after an error
  let alice -= amount;
  let bob   += amount;             ** nobody sees alice changed and bob not yet
return;

The complete example is the Eve macro mutex_account: the two accounts, the mutex, the tellers and the primary aspect in one file. A pipeline of channels is in channel_pipeline.

Rules that prevent deadlocks:

• Write defer lock.release(); right after acquire(): the mutex is never kept by mistake;
• A task holds one mutex at a time. The compiler rejects a function that acquires a second mutex while it holds the first, directly or through a call. Without a circle of tasks waiting for each other, a deadlock can't happen;
• A wait for a mutex ends after $timeout (see Errors and time-out) with a time-out error, so a mistake becomes an error and not a frozen program;
• Keep the guarded code short: no input, output, network or other wait while the mutex is held.

Other measures

• Do not share. Arguments passed by value are copies, and every task works on its own data. This costs a copy only when a task writes (copy on write) and never needs a lock;
• Channels pass values from one task to another (see Channels). Their operations are safe by design;
• One owner for each output (@): two tasks never write the same variable (see Data rules);
• Constants and exported classes are read only, so any number of tasks can read them.

Choosing, with the cost

The costs are typical orders of magnitude on a desktop processor, not promises: measure your own program before you decide.

As a rule, take the first row that fits: do not share; else an atomic variable; else a channel; else a mutex. A mutex is the most general and the most expensive to get right.

Streams and batches

Stream(:T) is a library trait that extends Iterable(:T): for x in s do reads a stream value by value and waits when the next value is not ready. In version 0.3, Channel(:T) is the only stream; generators, the lines of a file and the rows of a query become streams later.

A group splits its data by hand, with two methods of lists. The size is always written at the call, so the reader sees how the work is cut:

new parts := rows.split(4);
new sums: [4]Integer;
parallel
do
  for i in (1..4) do
    start sum_part(parts[i], @sums[i]);
  done;
done;
new total := 0;
for s in sums do let total += s; done;

Bulk Synchronous Parallel

The recommended pattern for parallel work is Bulk Synchronous Parallel (BSP). The computation is a sequence of supersteps. A superstep has three phases:

1. Compute: every worker computes on its own part of the data, without talking to the others;
2. Communicate: every worker writes its results into its own output;
3. Synchronize: a barrier waits for all the workers. After the barrier, all the results are visible.

In Eve, one parallel block is one superstep, and its done is the barrier. Between two blocks the primary aspect runs alone: it combines the partial results or prepares the data for the next superstep. An iterative algorithm puts the parallel block in a loop.

Example: parallel sum

Next program computes the sum of 32,000 random numbers with 32 workers. The data is split into 32 slices, one for each worker, and every worker writes one element of the array partials. The worker is the aspect partial_sum:

# local computation: one slice in, one partial sum out
concurrent aspect partial_sum is
  process main(data: []Integer, @partial: Integer) is
    let partial := 0;
    for x in data do
      let partial += x;
    done;
  return;
end partial_sum;

# parallel sum with one BSP superstep
primary aspect parallel_sum is
  set workers = 32 :Integer;
  set size = 1000 :Integer; ** elements for one worker

  process main is
    new data: [workers * size]Integer;
    new partials: [workers]Integer;

    ** produce the data
    for i in (1..workers * size) do
      let data[i] := (0..99).random();
    done;

    ** superstep: 32 workers, one slice each
    sums: parallel
      new first: Integer;
    do
      for w in (1..workers) do
        let first := (w - 1) * size + 1;
        start partial_sum(data[first..first + size - 1], @partials[w]);
      done;
    done sums; ** barrier: all the partial sums are ready

    ** combine on the main core
    new total := 0;
    for p in partials do
      let total += p;
    done;
    print total;
  return;
end parallel_sum;

The same job can be written as a pipeline, with a producer, 32 consumers and channels between them (see Channels). For data that is already in memory, the BSP version is simpler and safer:

• no shared object: every worker reads its own slice and writes its own element;
• no blocking: a worker never waits for another worker, only the primary aspect waits, at done. There is no buffer that can fill up, so there is no deadlock;
• deterministic: the partial results are combined in index order, so the result is the same on every run, also for floating point numbers, where the order of a sum changes the rounding;
• one error point: errors arrive at the barrier, where the recover region of the primary aspect can handle them;
• predictable cost: a superstep lasts as long as its slowest worker, plus the barrier.

Example: iterative supersteps

Next program simulates the heat that flows along a bar. The bar is an array of 1002 cells. The first cell is hot and the last cell is cold; they don't change. At every step, a cell takes the average of itself and its two neighbors. The 1000 inner cells are split into 10 slices. A worker, the aspect smooth, receives the current state cur by value and writes its slice of the next state nxt by reference. After the barrier the primary aspect clones nxt into cur with ::; an assignment := would share the array instead of copying it, and the next superstep would read and write the same cells. This copy is the communication between supersteps: in the next superstep every worker sees the new values of its neighbor slices.

# compute one slice of the next state
concurrent aspect smooth is
  process main(cur: []Real, first: Integer, @part: []Real) is
    for k in (1..part.count()) do
      new i := first + k - 1;
      let part[k] := (cur[i - 1] + cur[i] + cur[i + 1]) / 3;
    done;
  return;
end smooth;

# heat diffusion with iterative BSP supersteps
primary aspect heat_bar is
  set workers = 10 :Integer;
  set width = 100 :Integer; ** cells for one worker

  process main is
    new cur: [workers * width + 2]Real;
    new nxt: [workers * width + 2]Real;
    let cur[1] := 100.0; ** hot end, the cold end stays 0
    let nxt[1] := 100.0;

    for n in (1..500) do
      ** compute in parallel
      superstep: parallel
        new first: Integer;
      do
        for w in (1..workers) do
          let first := (w - 1) * width + 2;
          start smooth(cur, first, @nxt[first..first + width - 1]);
        done;
      done superstep; ** barrier

      ** communicate: the next state becomes the current state
      let cur :: nxt; ** clone: cur is a new copy of nxt
    done;
    print cur[2..11]; ** temperature near the hot end
  return;
end heat_bar;

Channels

A channel is a queue with a fixed capacity that connects tasks running in the same group. A producer task sends values into the channel; a consumer task receives them in the same order. When the channel is full, the sender waits; when it is empty, the receiver waits. In this way a fast stage can't run away from a slow one, and data flows through the tasks while it is produced. This is the right tool for streams: data that arrives over time or does not fit in memory.

Create a channel

Channel(:T) is a generic class of the core library. The type parameter is the type of the values; the constructor takes the capacity and, optionally, the number of senders.

  new jobs := Channel(:Integer)(capacity: 100);
  new results := Channel(:Real)(capacity: 100, senders: 4);

Operations

Rules:

• A channel is created by the primary aspect, before the group or in the declarations of the group, and it is passed to the tasks with "@": start produce(@jobs);. The parameter of main is declared @jobs: Channel(:Integer);
• A sender closes the channel when it has finished, ch.close();. With several senders, create the channel with senders: n, so the receivers stop after the last one. When a task ends without closing, normally or with an error, the machine closes for it every channel it has sent into: a failing producer never leaves its consumers waiting. A second close() by the same task does nothing;
• send into a closed channel, receive from a closed and empty channel, and a close beyond senders raise a ChannelError (code 46), Channel jobs is closed. The name is the name of the variable that created the channel;
• Values sent into a channel are copied, like arguments passed by value: a receiver never shares data with the sender;
• The order of the values from one sender is kept. With several senders the values are interleaved in any order, so a result that depends on the order is not deterministic;
• After done all the tasks have finished. The primary aspect can still read what is left in a closed channel with for x in ch do, but it is better to collect the results inside the group, as below.

Example: pipeline

Next program has three stages, each one an aspect. One producer sends 10,000 random numbers into jobs. Four workers receive from jobs, square every number and send it into results. One collector receives from results and adds the values into total. The three stages run at the same time, and the barrier done flow waits for all six tasks.

produce → jobs → square ×4 → results → collect

# stage 1: produce the numbers
concurrent aspect produce is
  process main(count: Integer, @jobs: Channel(:Integer)) is
    for i in (1..count) do
      jobs.send((0..99).random()); ** waits while jobs is full
    done;
    jobs.close(); ** no more numbers
  return;
end produce;

# stage 2: square every number
concurrent aspect square is
  process main(@jobs, @results: Channel(:Integer)) is
    for x in jobs do ** ends when jobs is closed and empty
      results.send(x * x);
    done;
    results.close(); ** one of the 4 senders has finished
  return;
end square;

# stage 3: collect the results
concurrent aspect collect is
  process main(@results: Channel(:Integer), @total: Integer) is
    let total := 0;
    for y in results do ** ends after the 4th close
      let total += y;
    done;
  return;
end collect;

# pipeline: producer, 4 workers and a collector
primary aspect pipeline is
  set workers = 4 :Integer;

  process main is
    new total: Integer;
    new jobs := Channel(:Integer)(capacity: 100);
    new results := Channel(:Integer)(capacity: 100, senders: workers);

    flow: parallel
    do
      start produce(10000, @jobs);
      for w in (1..workers) do
        start square(@jobs, @results);
      done;
      start collect(@results, @total);
    done flow; ** barrier: all the stages have finished

    print total;
  return;
end pipeline;

Notes:

• The capacity limits the memory: at most 100 numbers wait in each channel, whatever the size of the stream;
• The collector runs inside the group. If the primary aspect tried to read results after done, the workers would wait on a full channel and the group would never end: the virtual machine would report a deadlock;
• Only the collector writes total, so the output rule (one owner) holds;
• A sum does not depend on the order of the values. For results that do, send an index with each value, or use BSP.

Choosing a model

Inside one process, use a generator for a sequence and a job with tasks for many waits (see Multitasking). For work that needs several cores, or long computations side by side, start with a task group or BSP: a worker never waits for another worker, so they can't deadlock and they give the same result on every run. Use channels when the data is a stream. A primary aspect can combine the models: for example a pipeline that reads a large file in one group, then BSP supersteps on the data it has collected.

Read next: Templates