Parallel each

Note: Not built yet: parallel each inside a generator; a source of your own that is a ParallelIterable; the parallel builder inside an actor's method; and a reduction over a number range.

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:

import actors;

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

The loop has finished when control reaches the statement after it. It 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, and the code that wrote the loop works on pieces of its own loop while it waits. A module that writes one imports actors.

parallel is a word only directly before each: anywhere else it is a name like any other (a variable, a function, a field).

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.

What it can walk#

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, and the runs chunks(n) and mutChunks(n) give, which it cuts between runs (below). 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#

The passes run at the same time, so the body keeps rules an ordinary each does not:

• 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, or 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 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 (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) that is not its own. 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 (Rust's Sync: a borrow of it may be held by several threads at once; Race condition safety), and like Sendable it is worked out from what a type holds, with nothing to write: a value is not Shareable when, through a field, a variant, an element, a borrow or what an owner points at, it holds an owner that may be copied of a region that does not declare ThreadSafe (an Rc, or a region of your own that counts without atomic operations), a reference into a traced region, or a reference that writes through a shared path (a &mut held in a field, an element, a type argument or a closure's captured variable: Cone's &mut is shared mutable, so what holds one writes through a borrow of it), or a reference to a trait or a callable that does not say what is behind it (&<Shape, &<fn(i32) i32, So[Shape]: Rust's &dyn Trait is not Sync either; write &<Shape + Shareable where the value qualifies). An instance of a generic type holds one when a type argument does (List[Rc[T]], Dict[K, Rc[V]], Option[Rc[T]]), whatever the collection keeps its items behind, as in Rust a Vec<T> is shareable only if T is. A raw pointer in a type that is not generic is trusted, and nothing behind it is followed; an Arc, an Atomic, a number and a region that declares ThreadSafe are free. A name that is itself a borrow (self, a &mut parameter) is read through as a &: what it reaches is judged, and the body's writes through it are refused above. 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 (imm n = self.n;). For an Rc itself, borrow the contents before the loop: imm mesh = &*shared;, then use mesh in the body.
• 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. 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 imm 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. Another pass is writing that item, so reading it would race, and flow analysis sees only a local, not a parameter, a reborrow or a struct's field. Copy what you need before the loop: imm 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.
• 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 (ErrorActorGcPiece).
• Results come back through something that is changed by its methods, an Atomic, or as the list the parallel builder builds.
• 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.
• 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).
• 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.

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). A buffer is filled by handing each pass the part it alone may write. 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 rules for the body are those above: 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 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 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 (as Rayon's zip is), and the 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 (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 zip whose sources are reached from two of them is refused too, when the loop changes what they lend. Zip buffers the function owns, or the fields of one value it is lent (p.first.mutChunks(n).zip(p.second.mutChunks(n))).

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
}

In an actor's method#

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) {
    imm 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
  }
}

• 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: 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 (imm 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.

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:

imm weights = new List[u64] <- parallel each x in xs yield weigh(*x);
imm keep = new List[u64] <- parallel each x in xs if x > 9 yield weigh(*x);
imm 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 above 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. 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 loop cannot hand a result back: a parallel each may not write a variable outside it, since which pass wrote last would vary from run to run. A result is a reduction, and is written as a method called on parallel(), a view of any source that can be cut into pieces:

import actors;

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

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, a plain struct: a bounds, a total; a List or String is a panic, and one is built in parallel by the builder). The closures give each item as the view does: 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: one that borrows a variable it changes, or holds an Rc, is refused.

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 is not the plain loop's last bits, and is identical from one run, and one machine, to the next.