Parallelism

Multiple processes: world

The built-in world module runs one program as several cooperating processes (ranks), on one machine or many, in the style of MPI. Every world call is collective: all ranks make it, in the same order, and the compiler checks that they do, so a Brine program can’t deadlock waiting for a message. Every program is also a valid one-rank program, so the same code runs with or without --ranks.

A first program

fn main():
    let rank = world.rank()
    let size = world.size()

    # each rank sums its own slice of the numbers
    var local = 0
    for i in 0..1_000_000:
        if i % size == rank:
            local += i
    let total = world.sum(local)

    # every rank contributes one value; everyone gets the whole list
    let all = world.gather_all(rank * rank)

    if rank == 0:
        print(f"{size} ranks, total = {total}, squares = {all}")

    # pass a message around the ring: each rank sends to rank + 1
    var token = "token"
    for step in 0..size:
        token = world.shift(token + f"->{rank}", 1)
    if rank == 0:
        print(f"after one lap: {token}")

    # all-to-all: element j of the array goes to rank j
    let outbox = [[rank, j] for j in 0..size]
    let inbox = world.exchange(outbox)
    if rank == size - 1:
        print(f"rank {rank} received {inbox}")

brine run world.br --ranks 4

4 ranks, total = 499999500000, squares = [0, 1, 4, 9]
after one lap: token->0->1->2->3
rank 3 received [[0, 3], [1, 3], [2, 3], [3, 3]]

Without --ranks, the same program runs as one rank:

1 ranks, total = 499999500000, squares = [0]
after one lap: token->0
rank 0 received [[0, 0]]

The API

Every call except rank() and size() communicates, and every rank must make it. T is a messageable type.

Call	Returns	What it does	Same on every rank?
`world.rank()`	`int`	this process’s rank, `0 .. size() - 1`	no
`world.size()`	`int`	the number of ranks (`1` without `--ranks`)	yes
`world.barrier()`	nothing	waits until every rank has reached it
`world.broadcast(root: int, v: T)`	`T`	every rank gets rank `root`’s `v`	yes
`world.sum(v)`	the type of `v`	the sum of every rank’s `v` (`int` or `float`)	yes
`world.min(v)`	the type of `v`	the smallest `v` (`int` or `float`)	yes
`world.max(v)`	the type of `v`	the largest `v` (`int` or `float`)	yes
`world.gather_all(v: T)`	`[T]`	every rank’s `v`, in rank order	yes
`world.shift(v: T, by: int)`	`T`	sends `v` to rank `rank + by` and returns the value from rank `rank - by`, wrapping around	no
`world.exchange(xs: [T])`	`[T]`	all-to-all: `xs[j]` goes to rank `j`; element `j` of the result came from rank `j`	no

world.sum, min and max combine on rank 0 in rank order, then send the result to everyone, so they are deterministic for a given number of ranks. Integer world.sum stops the program on overflow.
broadcast’s root and shift’s distance must be the same on every rank (E0511).
exchange needs exactly one element per rank. Any other length panics: world.exchange needs one element per rank: got 3 elements, but the world size is 2.
There is no send or recv (E0512). Point-to-point messages are written with shift and exchange; see deadlock freedom.

Every call at once, on three ranks:

fn main():
    let rank = world.rank()
    let size = world.size()
    world.barrier()
    let root = world.broadcast(0, f"hello from rank {rank}")
    let lo = world.min(float(rank) * 1.5)
    let hi = world.max(rank * 10)
    let all = world.gather_all([rank, rank * rank])
    let flags = world.gather_all(rank % 2 == 0)
    let left = world.shift(rank, 1)
    let right = world.shift(rank, -1)
    let inbox = world.exchange([f"{rank}->{j}" for j in 0..size])
    print(f"rank {rank}: root={root} lo={lo} hi={hi} left={left} right={right} inbox={inbox}")
    if rank == 0:
        print(all, flags)

rank 2: root=hello from rank 0 lo=0.0 hi=20 left=1 right=0 inbox=["0->2", "1->2", "2->2"]
rank 1: root=hello from rank 0 lo=0.0 hi=20 left=0 right=2 inbox=["0->1", "1->1", "2->1"]
rank 0: root=hello from rank 0 lo=0.0 hi=20 left=2 right=1 inbox=["0->0", "1->0", "2->0"]
[[0, 0], [1, 1], [2, 4]] [true, false, true]

Each rank’s lines are in order, but lines from different ranks come out in whatever order the ranks get there (rank 2 printed first in this run). Print results from one rank, usually 0, when the order matters.

Messageable types

Values sent between ranks can be int, float, bool, str, and arrays of those, nested to any depth ([int], [[float]], [str], …). world.sum, min and max take int or float only. Anything else is E0319:

struct P:
    x: int

fn main():
    let p = world.broadcast(0, P(x=1))
    print(p)

error[E0319]: values of type `P` can't be sent between processes yet
 --> point.br:5:32
  |
5 |     let p = world.broadcast(0, P(x=1))
  |                                ^^^^^^ not messageable
  |
  = note: messageable types: int, float, bool, str, and arrays of those (e.g. [int], [[float]], [str])
  = run `brine explain E0319` for more about this error
error: could not compile due to 1 error

Send a struct as its fields (or an array of them), and a map as its keys and values.

Running with –ranks

brine run world.br --ranks 4               # 4 processes on this machine
brine run world.br --ranks 4 --threads 8   # each with 8 threads for par

brine run --ranks N builds the program once, reserves N free local ports, and starts N copies with the environment variables set. Then:

all ranks print to the same terminal;
only rank 0 reads standard input (the others get an empty one);
brine run waits for all of them and exits with the lowest rank’s nonzero exit code (or 0);
if a rank fails, the others are stopped half a second later.

Without --ranks (or with --ranks 1), world.size() is 1, world.rank() is 0, and every call returns at once: sum(v) is v, gather_all(v) is [v], shift(v, by) is v.

When a rank fails

A rank that panics or exits early never sends its next message. The ranks waiting for it stop with an error instead of hanging:

fn main():
    let rank = world.rank()
    let weights = [1.5]
    let mine = weights[rank]
    let total = world.sum(mine)
    print(f"rank {rank}: total = {total}")

panic (rank 1): index 1 is out of bounds for an array of length 1
  --> crash.br:4:16

panic (rank 0): rank 1 exited before sending the expected message
  --> crash.br:5:17

The exit code is 101, as for any panic. Panic messages name the rank.

Combining world with par

world splits the work between processes; par splits each process’s share between its cores (and GPU). A common shape: each rank takes a block, reduces it with par, and world.sum adds up the ranks.

fn main():
    let rank = world.rank()
    let size = world.size()
    let n = 100_000_000
    # each rank takes a contiguous block and sums it on all of its cores
    let lo = n / size * rank
    let hi = n if rank == size - 1 else n / size * (rank + 1)
    let local = par sum(1.0 / (float(i) + 1.0) for i in lo..hi)
    let total = world.sum(local)
    if rank == 0:
        print(f"{size} ranks: sum of 1/i for i in 1..={n} = {total}")

1 ranks: sum of 1/i for i in 1..=100000000 = 18.9978964138539
2 ranks: sum of 1/i for i in 1..=100000000 = 18.9978964138539
4 ranks: sum of 1/i for i in 1..=100000000 = 18.997896413853898

Each line is the same on every run and any thread count. The last digit changed at 4 ranks because the blocks changed: each rank’s par sum has its own tree, and world.sum adds the ranks’ results left to right. A float total is reproducible for a given number of ranks, not across different numbers of ranks. If you need the same bits for any rank count, have each rank compute whole, fixed-size blocks and combine them in a fixed order (for example world.gather_all of the block sums, then a sum on every rank).

Communication can’t happen inside a par body (E0503): call world before or after the construct. On one machine, give each rank its share of the cores with --threads, or the ranks’ thread pools compete for them.

Multiple machines

Each rank is an ordinary process configured by three environment variables. To run across machines, start the built program on each machine with:

Variable	Meaning
`BRINE_RANK`	this process’s rank, `0 .. size - 1`
`BRINE_WORLD_SIZE`	the number of ranks
`BRINE_PEERS`	`host:port` for every rank, comma-separated, in rank order

Every rank gets the same BRINE_PEERS; entry r is where rank r listens. For two machines:

brine build world.br --release -o world
# on 10.0.0.1:
BRINE_RANK=0 BRINE_WORLD_SIZE=2 BRINE_PEERS=10.0.0.1:7101,10.0.0.2:7101 ./world
# on 10.0.0.2:
BRINE_RANK=1 BRINE_WORLD_SIZE=2 BRINE_PEERS=10.0.0.1:7101,10.0.0.2:7101 ./world

The same works on one machine with 127.0.0.1 and different ports, which is what --ranks does. Run that way with two ranks, the program above printed rank 0’s lines in one terminal and rank 1’s in the other:

2 ranks, total = 499999500000, squares = [0, 1]
after one lap: token->0->1

rank 1 received [[0, 1], [1, 1]]

Without all three variables (or with BRINE_WORLD_SIZE=1), a program runs as a single rank. Ranks wait up to 30 seconds for the lower-numbered ranks to start listening, so the machines don’t have to start at the same moment. A brine launch --hosts command that does this for you is on the roadmap.

Transport

Full-mesh TCP. Every pair of ranks has one connection: each rank connects to the ranks below it and accepts the ranks above it.
Sends never block. Each rank runs a background receiver thread per peer that queues incoming messages, so a send completes even if the receiver is busy.
FIFO. Messages between a pair of ranks arrive in the order they were sent.
Loopback when local. When every peer is 127.x.x.x or localhost, ranks listen on the loopback interface only: nothing outside the machine can reach them, and firewalls don’t ask. Otherwise they listen on all interfaces.
No encryption or authentication. Messages are plain TCP. Run multi-machine jobs on a network you trust.

Deadlock freedom: the compile-time check, and what it allows
Parallelism overview: par inside each rank
Environment variables