parallel each in depth

On this page

Writing a parallel each
What it can walk
What the body may do
The parallel builder
Reductions: parallel().sum()
Filling a buffer: indexed, chunks and mutChunks
Two buffers in one pass: zip
Where a loop that changes its items may lend them from
In an actor's behaviour

Note: Not built yet: parallel each inside a generator; a source of your own that can be split (a ParallelIterable you write); the parallel builder inside an actor's method; and a reduction over a number range. The actors package it runs on runs on Windows and Linux (x86-64), not yet on macOS.

The Guide's parallel each found primes on every core and added them up. Real work asks more of it: filling a picture or a buffer by position, writing two buffers in one pass, finding the nearest of many things, running a loop inside an actor. This page has the whole of it: what a loop can walk, the rules its body keeps and why, and every way a result comes back. Afterwards you can write a parallel loop for any job whose pieces stand on their own. Here a picture is filled a row to a pass:

mod mandel;

import stdio use *;
import collections use *;
import actors;

const width = 60usize;
const height = 24usize;

// How many steps before the point (x, y) escapes, up to 30
fn depth(x usize, y usize) i32 {
  let cr = f64.from(x) / f64.from(width) * 3.0d - 2.2d;
  let ci = f64.from(y) / f64.from(height) * 2.4d - 1.2d;
  var zr = 0d;
  var zi = 0d;
  var n = 0;
  while n < 30 and zr * zr + zi * zi < 4d {
    let t = zr * zr - zi * zi + cr;
    zi = 2d * zr * zi + ci;
    zr = t;
    n += 1;
  }
  n;
}

fn main() {
  // A picture of width x height cells, every one 0 to begin with
  var picture = new List[i32](width * height) <- fill 0;

  // One pass a row: each pass is lent a row of its own to fill
  parallel each y, row in picture.mutChunks(width).indexed() {
    each x, cell in row.mutItems().indexed() {
      *cell = depth(x, y);
    }
  }

  each y in 0usize ..< height {
    each x in 0usize ..< width {
      let d = picture[y * width + x];
      print <- if d >= 30 { "#"; } elif d > 6 { "+"; } elif d > 3 { "."; } else { " "; };
    }
    print <- "\n";
  }
}

It prints the Mandelbrot set:


                                  ......++.....
                               .........++++#.....
                            ...........++#+++.......
                          ...........+++####++.......
                       .........++++++++####++++...+...
                     ..........++##++##########++++++..
                  ............+++##################++...
             .....++++++++++++++###################+++...
        ..........++++++#++++++#####################++...
      ...........++++#######++#######################+....
     .........++++++################################+.....
     #############################################+++.....
     .........++++++################################+.....
      ...........++++#######++#######################+....
        ..........++++++#++++++#####################++...
             .....++++++++++++++###################+++...
                  ............+++##################++...
                     ..........++##++##########++++++..
                       .........++++++++####++++...+...
                          ...........+++####++.......
                            ...........++#+++.......
                               .........++++#.....
                                  ......++.....

Try this: make the picture 600 by 240, keep only the count of # cells, and time it with and without parallel.

Writing a parallel each#

Writing parallel in front of an each runs the loop's passes at the same time, on the actors package's workers, one for each CPU. A module that writes one imports actors. The loop has finished when control reaches the statement after it.

Nobody divides the work ahead of time. The loop starts as one task for the whole range of passes; whoever runs a task splits it in half while it is bigger than a small share of the range, queues one half and works the other, and an idle worker takes the oldest, largest half. So uneven passes spread themselves (the rows through the middle of the picture take far longer than those at the edges), and the code that wrote the loop works on pieces of its own loop while it waits.

The filter, parallel each x in xs if cond { … }, skips the items for which the condition is false, as it does for each. A parallel each takes no else and is a statement. parallel is a word only directly before each: anywhere else it is a name like any other (a variable, a function, a field).

What it can walk#

The passes can only be shared out if the source can be cut. So the source must be something that can be split into independent pieces and says how big it is: a ParallelIterable, met by its methods alone, with no declaration, as Iterator is (len, and split(at) giving the items before at and the items from it on). Today the loop walks:

a number range counting up by one (parallel each i in 0 ..< n, or .. to include n). A count down or a by step is refused, since which pass runs first is not defined in parallel;
an array, a slice and a list, which it cuts into ranges of positions itself;
the runs chunks(n) and mutChunks(n) give, which it cuts between runs (Filling a buffer).

A cursor, a generator, a file or a channel hands out its items one after another and cannot be split: collect it into a list first, and walk the list. A chain of adapters (xs.iter().map(f)) is a cursor too: walk the list it starts from, and put the adapters' work in the loop's own header, if and yield. As in the other loops, the variable is a new one on every pass, and is a borrow of the item for an array, a slice or a list.

What the body may do#

A plain loop's passes run one after another, so each sees what the last one did. A parallel loop's passes run at the same time, and if two of them wrote one variable, which wrote last would vary from run to run, and an update could be lost. So the body keeps rules an ordinary each does not, and the compiler checks every one. Each refusal says what to do instead:

var total = 0;
parallel each n in 0 ..< 1000 {
  total = total + n;                 // refused
}

Error 1382: A 'parallel each' body may not write total, which is declared outside the loop: the passes run at
the same time. Outside variables are read-only for the loop. Hand results back with the parallel builder, or
'mutItems()' on the items, or count with an Atomic.

The rules, each with its reason:

What is declared outside the loop is read-only in it, borrowed for as long as the loop runs. Writing it is refused: an assignment, an operator that assigns, a field, an element, a swap, a method that changes it, a borrow that may write, and a write through a &mut reference declared outside. A parameter of the enclosing function is declared outside, and so is whatever is reached from one: a &mut parameter handed on as it is, to a method that takes &mut self (l.mutItems(), l[0].n = …) or to a function whose parameter is a &mut (bump(l)), is lent for writing as much as &mut l[0] is, and is refused alike. Handed to a &, it is only read. What the pass declares itself is its own to change. A write through a raw pointer is not checked: the loop trusts its writer.
A var global is outside the loop too, and the body may not change it through a function either. A call of a function the compiler can read the body of is followed through the bodies it reaches, and is refused (Error 1382, naming the global and the function that writes it) if any of them assigns to or swaps a place of a var global, or lends one for writing, which a method that changes it does. A @threadlocal global is each thread's own, and a @workgroup one a workgroup's, so both are allowed; so are reads, and an Atomic in a let global. What the compiler cannot read is not followed, and is taken as writing nothing: a call through a virtual reference or a function value, an extern function, and a function of another package (its body is not in the compile that checks the loop). A global written there still races.
A copy of a value from outside is a write when it counts. Copying an Rc adds to its count, which passes running at the same time cannot do without atomic operations. So a body may not copy an Rc that is not its own, or a struct, tuple, array, enum or list element that holds one, or a traced Gc reference, whose roots are the collector's one chain. An Arc counts atomically and may be copied.
Nor may the body reach such a value at all. A function the body hands it to could copy what it was lent, and so could any code reached through a field, an element or a borrow. So a value declared outside the loop (or the pass's own item) whose type holds an Rc or a traced reference may not be named in the body: not copied, not read, not lent, not passed. This is the marker Shareable, which says a borrow of a value may be held by several threads at once, and the compiler works it out from what a type holds (Sendable and Shareable, the next page, has the full rule). Naming the value is enough, so a struct or an actor's self that holds an Rc cannot be read for a number beside it: copy what the loop needs into a local before it (let n = self.n;). For an Rc itself, borrow the contents before the loop, as the message says:
```
Error 1390: This 'parallel each' reaches 'shared', a Rc[imm, Table] declared outside the loop, which is or holds
a counted owner whose count is not atomic (an Rc, for example) or a traced reference (a Gc, for example): its
passes run at the same time, and a copy made anywhere they can reach it, in this body or in a function it hands
it to, would write a count or a root that every pass shares without an atomic operation. Reading through it,
lending it and passing it count too. Borrow what the body needs before the loop ('let mesh = &*shared;') and
use 'mesh' in the body, or hold it in an Arc.
```
A loop that writes through its source (mutChunks, mutItems, a zip of them, or a &mut slice) may not name, in its body, filter or yield, a place from outside that could be one of the items it writes: another pass is writing that item, so reading it would race. It is judged by the path the body reads: first where first is a reference to the item's type; a struct reached through a reference that is itself of the item's type; an index into a collection or array reached through a reference whose elements are of the item's type (hr.cells[0], s.items[0] after let s = &*self;). A scalar field beside the buffer (self.scale) reads none of them. A name used whole, or reached through a call, is judged by its whole type: a reference to the item's type, or a struct or collection holding one or elements of that type, reached through a reference. A raw pointer is trusted. The check goes by type because flow analysis sees only a local, not a parameter, a reborrow or a struct's field. Copy what you need before the loop: let f = *first;, then use f. A value already copied, a reference to another type, and every outside reference when nothing is written (chunks) are free. A closure handed to a function that runs such a loop is read by every pass: one that borrows a variable of the written type ((x, y) => scale * f32.from(x) into a list of f32) is refused, naming the variable; copy it into the closure with the state list, fn(x i32, y i32) [scale] f32 { scale * f32.from(x); }, where it is a value of its own.
A pass may change its own item: parallel each x in list.mutItems() (or an array's or a slice's, or a mutable slice as the source itself) gives x as a &mut of the pass's item, and the items are disjoint. Then the body may not name the list (or slice, or the struct holding it, or what is inside the place the items are lent from) as well: &mut is shared in Cone, so another pass may be changing the item it would read. Another field of the same struct is free. Where the items may be lent from is below.
The collector package's Gc may not be made or held in the loop: its heap and its one chain of roots are not safe on the workers the passes run on. An allocation (new Gc[...]), a call that returns one, or a literal holding one (a struct, tuple, array or enum holding a traced reference counts) is refused in the body, and in a parallel builder's source, if and yield, with a message saying to hold an ActorGc instead. A reference into an actor's own region (The actor's own garbage-collected region) is read in the passes of a loop in a behaviour if it is read-only, ActorGc[imm, T]: it counts as Shareable, since nothing collects or moves the object while the passes run. One that can write, ActorGc[mut, T], is not shared into them, like any &mut, and nor is the state of an actor that holds one; and nothing allocates in the actor's region from a pass, neither new self[T](...) nor a direct allocation.
continue skips the rest of a pass. A break or continue of a loop inside the body is the pass's own. A break and a return are refused (which items had run by then would vary from run to run), and so is an await (a pass runs to completion and has no mailbox):
```
Error 1381: 'break' is not allowed in a 'parallel each' body: which items had already run when it was reached
would vary from run to run. 'continue' skips the rest of an item and is allowed.
```
A message sent to an actor is allowed, and is not sent where it is written: the passes' messages are kept and sent after the loop, by the code that wrote it, in iteration order (the first pass's, then the second's), so an actor sees the same order every run. A loop inside a pass's body is gathered into that pass in the same way.

So a result comes back by one of four routes: a list the parallel builder builds, a reduction, the items or runs the passes change in place (Filling a buffer), or something changed only by its own methods, an Atomic:

fn total(xs &Array[u64]) u64 {
  let sum = Atomic[u64].of(0u64);
  parallel each x in xs {
    sum.add(weigh(*x));
  }
  sum.load();               // every pass has finished
}

The parallel builder#

A list is built by a parallel each as it is by an each inside <-: put parallel in front of the each in the list of entries, and give the value each pass appends with yield:

let weights = new List[u64] <- parallel each x in xs yield weigh(*x);
let keep = new List[u64] <- parallel each x in xs if x > 9 yield weigh(*x);
let squares = new List[i32] <- parallel each i in 0 ..< n yield i * i;   // n an i32
xs <- 0, parallel each x in other yield x + 1, 99;    // in the middle of a list

The list holds the values in the order of the passes, the order a plain each would give them, however the passes were spread over the workers; with a filter, the values of the passes that kept their items, in order. The loop has finished, and the list is whole, when control reaches the statement after it. The body's rules hold for the header, the filter and the yield: nothing outside the loop is written, an Rc from outside is not reached, and there is no break, return or await. The items of a collection come as borrows, as in a loop, so the yield writes its own *; a value that moves (a String made in the body) is handed to the list. With only a source, new List[T] <- parallel each src, every item is appended as a copy.

Only a List is built this way. A dictionary or any other collection is built from a list of pairs after the loop: a receiver that is not a List is refused, and so is yield k: v into a list. A builder inside an actor's method or a generator is refused, as a parallel each is there. The list being appended to may be the one the loop reads: the pieces are appended once the loop has read everything.

Underneath, each piece of the loop (a run of consecutive passes) appends to a list of its own, and when the loop has finished, the pieces' lists are joined onto the end of the list being built in the order of where each piece began. List has four methods the compiler calls for this, pieceBag, emptyPiece, depositPiece and joinPieces: they are not for a program to call.

Reductions: parallel().sum()#

A total, a largest value, a first match: each boils a whole source down to one value, which is a reduction. A loop cannot hand one back, since it may not write a variable outside it. A reduction is written as a method called on parallel(), a view of any source that can be cut into pieces:

fn stats(xs &List[f64]) {
  let total = xs.parallel().sum();                 // the total of the numbers (0 for none)
  let top = xs.parallel().fold(0.0d, (m, x) => if *x > m { *x; } else { m; },
                               (a, b) => if b > a { b; } else { a; });
  let hit = xs.parallel().findFirst(x => **x > 100.0d);   // an Option of the item
}

Given 3, 150, 7 and 220, total is 380, top is 220, and hit holds 150. Each piece of the source keeps a result of its own, and the results of neighbouring pieces are joined pairwise up the tree the loop's splitting makes, on the workers, so that only the final value reaches the code after the call. A module that calls one imports actors, as one that writes a parallel each does. The reduction is called directly on parallel(): a view held in a variable has no methods of its own, and a source that cannot be cut (a chain of adapters, a generator, a file or a channel) has no parallel(); collect it into a list first.

A reduction reads, so a view of mutItems() or mutChunks(n) has none. The views that have them are those of a list, an array or a slice, of the runs of chunks(n), of iter(), indexed() and a zip of those.

sum() is the total of a view of numbers (integers, f32, f64), or of borrows of them, as a list's items are.
fold(init, step, combine): step(acc, item) folds a run of items, front to back, into an accumulator that starts as a copy of init, and combine(a, b) joins the accumulators of two neighbouring runs, a the earlier. combine must be associative, and need not be commutative (it may join text or multiply matrices in order). init must be its identity, since each run begins from a copy of it, so it may not have a finalizer: a number, or a plain struct such as a bounds or a total. A List or String as init is a panic; build one in parallel with the builder. The closures are given each item as the view gives it: a borrow for a list's, a tuple for a zip's.
findFirst(pred) is the earliest item by position that pred (given a borrow of the item) is true of, as an Option, though a later piece may find a match first. The pieces after a match are not scanned, and a piece under way stops at its next look.

The closures run on several workers at once, so each must be Shareable, as a parallel each's body must be: one that borrows a variable it changes, or holds an Rc, is refused. In a method taking self &mut, a closure that names a field borrows self and is refused; borrow the field first (let bodies = &bodies;) and name that.

The nearest of many, with where it was found, is a fold. The accumulator is a plain struct (the distance, the position, a payload), the identity has a distance above every real one, and both step and combine keep the later candidate only if it is strictly nearer, so ties go to the lowest position whatever the pieces:

let best = hits.indexed().parallel().fold(new Nearest(1.0e30, 0usize),
  (acc, p) => if p.1.d < acc.d { new Nearest(p.1.d, p.0); } else { acc; },
  (a, b) => if b.d < a.d { b; } else { a; });

With distances 5, 2, 9 and 2, best is the 2 at position 1, never the one at 3.

The result is the same on every run, with any number of workers, a float sum included. The items are cut into runs by their count alone (at least 128 items to a run, and at most 256 runs; a view of 128 or fewer is one run, folded at once on the caller's thread), each run is folded front to back, and the runs are joined by halving. So a float sum may differ from the plain loop's in its last bits, and is identical from one run, and one machine, to the next.

Filling a buffer: indexed, chunks and mutChunks#

A counting range cannot write the elements of a buffer by position: out[i] = … writes something declared outside the loop, and the compiler does not try to prove the passes' writes disjoint. So a buffer is filled by handing each pass the part it alone may write, as the picture above was. Two variables from indexed() are the position and the item, and the position is global: the item's place in the whole source, whichever piece of the loop it falls in.

parallel each i, p in buf.mutItems().indexed() { *p = f(i); }     // one pass an element

parallel each y, row in pixels.mutChunks(size).indexed() {      // one pass a row
  each x, p in row.mutItems().indexed() { *p = shade(x, y); }
}

mutChunks(n) cuts only between runs, so each pass is given a &mut slice of its own run (the last may be shorter) that no other pass can reach; the position of mutChunks(n).indexed() is the run's number. chunks(n) is the same for reading. indexed() over a list, an array, a slice or these runs gives two variables; every other source gives one. The body's rules hold: a pass writes through its own run or item, naming the buffer the runs come from in the body is refused (it may be a run another pass is writing), and a run that chunks gives cannot be written.

Two buffers in one pass: zip#

A zip of cursors is walked by a parallel each when both its sources can be cut and walked by position (core's RandomAccess and ParallelIterable, met by the methods alone): the cursors that chunks(n), mutChunks(n) and mutItems() give, and those of arrays, slices and lists that zip(&a, &b) takes. The zip is as long as the shorter source, is cut at the same place in both, and pass k gets what each gives at k. Its indexed() adds the pass's number, global, so three variables:

parallel each y, ra, rb in a.mutChunks(n).zip(b.mutChunks(3usize * n)).indexed() {
  each x, p in ra.mutItems().indexed() { *p = f(y * n + x); }               // a's part of the pass
  each x, q in rb.mutItems().indexed() { *q = g(y * 3usize * n + x); }      // b's part, three times as long
}

Without indexed() a zip gives two variables. What each source lends, & or &mut, is what its own cursor says. A source that cannot be cut or walked by position (a Deque's cursor, a cursor of your own, a generator, a file or a channel) is refused with the reason: collect it into a list and zip the list. The body rules stand for both buffers: neither may be named in the body (the passes change their own parts of them at the same time; naming a buffer is refused even when its source only reads), and a zip of two cursors over the same buffer, where the loop changes what they lend, is refused, since two passes could be given the same part and write it at the same time. Two different &mut references, two parameters say, may be the one buffer, because the caller may pass it twice, so a source the loop changes may not be reached from one (next): take it as &uni. Zip buffers the function owns, or the fields of one value it is lent (p.first.mutChunks(n).zip(p.second.mutChunks(n)), with p &uni).

A zip takes a third source with its own zip, a flat triple, three variables (four with indexed(), the position first). It is as long as the shortest, cut at the same place in all three, and the rules above hold for all three buffers:

parallel each y, ca, cb, cc in a.mutChunks(n).zip(b.mutChunks(m)).zip(c.mutChunks(k)).indexed() {
  ...                          // a part of each of three buffers, one pass
}

Where a loop that changes its items may lend them from#

A loop that changes the items of its source (mutItems, mutArrays, mutChunks, a &mut slice, in the header or inside a zip) must lend them from something exclusive, a place that no other name reaches while the passes run:

a local of the function, or a field of one held by value (grid.cells.mutItems());
a &uni parameter, or a field of one held by value, taken fn fill(out &uni List[T]) and called with fill(&uni out) (a var local lends itself &uni);
the self of an actor's behaviour, which runs alone on its actor (its parameters cannot be borrows) and whose state the passes borrow for the loop.

A list reached through a &mut or & reference (a parameter, the &mut self of a synchronous method, a field that is one, a local that holds one), an Rc, or a global is refused, since &mut is one of several names for a list: a caller may pass the one list as two parameters, a zip may read it beside the writer, another thread may hold it, and a pass would then read or write an item another pass is writing. Take the list as &uni instead, which no other name can reach, so the call that passes the one list under two names is refused where it calls. A global is copied or moved into a local first. Only the sources the loop writes are judged: a source that only reads, beside the writer in a zip, may be anything, but it cannot be a second name for the written list.

fn lift(a &List[Item], b &uni List[Item]) {
  parallel each i, x, pair in a.iter().zip(b.mutArrays[2usize]().mutItems()).indexed() {
    pair[0usize].n = x.n + 1000;     // b is the one name for what the passes write
  }
}
lift(&l, &uni m);                  // l and m are different lists
lift(&l, &uni l);                  // refused

Error 1168: 'l' is borrowed (at 19:8) for a call or value still being made, which uses that borrow. It may not
be borrowed as 'uni' before then.

A list lends its items as a slice from a &uni path with viewUni, the way viewMut lends them from a &mut one: a &uni list parameter by l.viewUni(), a local or a field of a &uni struct by (&uni list).viewUni(). So a function taking the slice as &uni Array[T] takes a list's items or an array's alike. A String lends its bytes the same way, (&uni s).viewUni(). A viewMut slice handed to a function that takes &mut Array[T] is the &mut slice above, which is refused, and a list reached through a &mut or & reference lends no viewUni. The list is lent to the slice, so it may not be read or pushed onto until the slice's last use.

fn number(s &uni Array[Item]) {
  parallel each i, x in s.mutItems().indexed() { x.n = i64.from(i) + 1; }
}
number((&uni items).viewUni());        // items is a List[Item]

A cursor held in a variable is followed to the call that made it when the variable is let and has its value (let c = list.mutItems(); then parallel each x in c lends from list). A cursor the loop cannot trace, in a var variable that may be assigned, a parameter or a field, is refused: build a writable cursor in the loop's header.

In an actor's behaviour#

An actor's own data is often exactly what wants changing in parallel. But a behaviour that held its worker while the loop ran would stall the actor's thread for the whole loop. So a parallel each written directly in a behaviour (a method declared async do) cuts the behaviour where the loop ends, as an await does. The behaviour returns to the actor's dispatcher while the passes run on the workers, and the rest of it runs when the last pass has finished:

actor Scaler {
  vals List[u64];
  scale u64;

  pub async do run(self, add u64) {
    let k = add * 2u64;
    parallel each v in vals.mutItems() {
      *v = *v * scale + k;      // the actor's fields are read through self; k and add are carried
    }
    // here the loop has finished, and this is the same behaviour, the same call
    print <- "scaled: ", vals[0], " ", vals[1], " ", vals[2], "\n";
  }
}

Made with the values 1, 2 and 3 and a scale of 10, and sent run(5u64), it prints scaled: 20 30 40.

The actor is not run again until the loop has finished, the rest of the behaviour included. Its mailbox goes on accepting messages, and they are handled after the behaviour, in order. A message sent after the behaviour began is handled after all its parallel work.
The passes borrow the actor's fields directly, through self, for the loop: nothing is moved out or copied, and no update is lost. This is the one borrow a behaviour holds across the cut. Nothing else runs the actor, and a pass can neither await nor receive, so the loop always finishes.
What the behaviour holds across the cut is carried as it is across an await (What crosses an await): the parameters and locals it uses afterwards, and every droppable value, which dies where it would have died had the method not been cut. What the passes read of the behaviour's variables is copied as the behaviour returns, so a pass may not read a borrow held in a variable (let t = &total; read total through self instead), and the array or slice the loop walks may not lie in the behaviour's own frame (an array in a local, or made for the loop; walk a list, or an array in a field).
The messages the passes send are sent when the last pass has finished, before the rest of the behaviour runs, in iteration order.
A behaviour that returns a value answers it from after the loop. A loop in a loop of the behaviour, and several loops, are a cut each. A loop inside a pass's body runs inside that pass.
A synchronous fn of an actor cannot be cut. A loop there runs where it stands, the worker running passes meanwhile, so its actor is not run again until the method that called it has returned.
The parallel builder is not built inside an actor's method yet.

Underneath, a parallel each becomes a call of the runtime's own parallel loop, which How the runtime works describes, and which a program may also call directly.

You can now fill buffers, build lists and reduce on every core, and run a parallel loop inside an actor, and you know why each body rule is there. Several of those rules, and every behaviour's parameters, come down to one question: may this value cross to another thread, or be read from several at once? The next page answers it.