30. pargrep
The four projects before this one are tools you could have written in 1978. This one is not, and the reason is not that wolf is newer. It is that the sequential version of this program is forty lines shorter, works perfectly, and is the version you should ship for the input in this chapter — so a parallel grep has to earn its extra forty lines out loud, in front of a reader holding the shorter one.
Part 3 taught tasks, channels, select, and freeze. This chapter spends
all four on one program, and then counts what they cost.
30.1 Sharding the input
pargrep searches a set of files for a pattern and prints every matching
line with its file name and line number, in the order the lines appear in
the input. It is grep, narrowed: no regular expressions, a |-separated
list of literal needles instead, and a match is a line containing any one
of them.
Here is the whole job, done sequentially:
struct Table { lines: List[str], names: List[str], nums: List[int] }
fn main() -> !int {
let args = env_args()
var pattern = "wolf"
var names = List[str]()
if args.len > 0 { pattern = args[0] }
var k = 1
while k < args.len {
(mut names).push(args[k])
k += 1
}
if names.is_empty() {
fs_write_text("a.log", "06:12 the wolf runs\n06:31 the moon watches\n07:02 the wolf howls\n07:40 the pack waits\n")?
fs_write_text("b.log", "08:15 alone\n08:52 the wolf sleeps\n09:03 moonrise\n09:44 the wolf wakes\n")?
(mut names).push("a.log")
(mut names).push("b.log")
}
let needles = pattern.split("|")
var lines = List[str]()
var files = List[str]()
var nums = List[int]()
var f = 0
while f < names.len {
let name = names[f]
var no = 1
for line in fs_read_text(name)?.lines() {
(mut lines).push(line)
(mut files).push(name)
(mut nums).push(no)
no += 1
}
f += 1
}
let t = Table { lines: lines, names: files, nums: nums }
var hits = 0
var i = 0
while i < t.lines.len {
if matches(t.lines[i], needles) {
print("{t.names[i]}:{t.nums[i]}: {t.lines[i]}")
hits += 1
}
i += 1
}
if hits == 0 { 1 } else { 0 }
}
fn matches(line: str, needles: List[str]) -> bool {
var k = 0
while k < needles.len {
if line.contains(needles[k]) { return true }
k += 1
}
false
}
$ wolf build seqgrep.lu && ./seqgrep
a.log:1: 06:12 the wolf runs
a.log:3: 07:02 the wolf howls
b.log:2: 08:52 the wolf sleeps
b.log:4: 09:44 the wolf wakes
$ ./seqgrep 'moon|pack' a.log b.log
a.log:2: 06:31 the moon watches
a.log:4: 07:40 the pack waits
b.log:3: 09:03 moonrise
Two things in that program are new to this part and both are visible in
the transcript. env_args() hands back the command line as a List[str],
so this is the first project in the book that is a tool rather than a
demonstration: it takes a pattern and file names the way every other
program on your machine does. And when it is given nothing, it writes two
small logs and searches those, so the binary you built a moment ago has
something to do before you have written a file for it.
Now make it parallel. The first decision is not “how do I spawn a task” — it is what does a task get, and that decision is arithmetic, made before anything spawns:
fn main() -> !int {
let n = 9
let b1 = n / 4
let b2 = (n * 2) / 4
let b3 = (n * 3) / 4
print("0..{b1} {b1}..{b2} {b2}..{b3} {b3}..{n}")
0
}
$ lupin shards.lu
0..2 2..4 4..6 6..9
Nine lines, four shards, and the fourth one takes the remainder. Integer division does the whole job: three boundaries, four half-open ranges, every index in exactly one of them. There is no shared cursor, no work queue, and nothing for two tasks to disagree about, because the disagreement was settled by three lines of arithmetic while the program was still single-threaded.
That is the shape worth taking away from this section, and it is worth more than the syntax that follows it. Sharding is deciding the division of labour before you have anyone to give it to. A parallel program that shards first has one concurrency problem — collecting results. A parallel program that hands out work as it goes has two, and the second one is the hard one.
Four is written into the program as four spawns, not computed from the input. The shard count is a constant here, and §30.4 charges us for it.
30.2 A task per shard
The parallel version starts exactly where the sequential one did — the command line, the two demo logs, and the input read into a table:
struct Table { lines: List[str], names: List[str], nums: List[int] }
fn main() -> !int {
let args = env_args()
var pattern = "wolf"
var names = List[str]()
if args.len > 0 { pattern = args[0] }
var k = 1
while k < args.len {
(mut names).push(args[k])
k += 1
}
if names.is_empty() {
fs_write_text("a.log", "06:12 the wolf runs\n06:31 the moon watches\n07:02 the wolf howls\n07:40 the pack waits\n")?
fs_write_text("b.log", "08:15 alone\n08:52 the wolf sleeps\n09:03 moonrise\n09:44 the wolf wakes\n")?
(mut names).push("a.log")
(mut names).push("b.log")
}
let needles = freeze region { pattern.split("|") }
let t = freeze region {
var lines = List[str]()
var files = List[str]()
var nums = List[int]()
var f = 0
while f < names.len {
let name = names[f]
var no = 1
for line in fs_read_text(name)?.lines() {
(mut lines).push(line)
(mut files).push(name)
(mut nums).push(no)
no += 1
}
f += 1
}
Table { lines: lines, names: files, nums: nums }
}
The table is one value with three parallel lists in it: every line of every file, the file each line came from, and its line number. Flattening the files into one table before sharding is what makes the shards even. Shard by file and four tasks split two files badly; shard by line and the split is as good as integer division gets.
Both freeze region { … } blocks are §30.4’s subject. Read them here as
“build this value, then make it permanent.”
The rest of main is the concurrency, and it is short:
let n = t.lines.len
let b1 = n / 4
let b2 = (n * 2) / 4
let b3 = (n * 3) / 4
let hits = channel[int](0)
let done = channel[int](0)
var found = List[int]()
scope s {
s.spawn(fn() {
for i in 0..b1 { if matches(t.lines[i], needles) { hits.send(i) } }
done.send(1)
})
s.spawn(fn() {
for i in b1..b2 { if matches(t.lines[i], needles) { hits.send(i) } }
done.send(1)
})
s.spawn(fn() {
for i in b2..b3 { if matches(t.lines[i], needles) { hits.send(i) } }
done.send(1)
})
s.spawn(fn() {
for i in b3..n { if matches(t.lines[i], needles) { hits.send(i) } }
done.send(1)
})
scope s { … } is chapter 10’s structured scope: the four tasks are
spawned into it, and the closing brace does not run until all four have
finished. There is no join to remember and no handle to keep, which is the
whole point of a scope — you cannot leak a task out of one, because the
brace that ends the block is the join.
Each task walks its own half-open range of the table and sends the index
of every line it matches. Four ranges, four tasks, no overlap: t.lines[i]
is read by exactly one task for each i, so the four bodies are reading
one table and never the same element twice.
Notice what the four bodies do not have. No lock, no atomic counter, no
mutable shared accumulator. Each task’s only route out is hits.send(i),
and the accumulating happens somewhere else.
30.3 Results through a channel
Somewhere else is here:
var live = 4
while live > 0 {
select {
i from hits => { (mut found).push(i) },
_ from done => { live -= 1 },
}
}
}
var i = 0
while i < n {
if member(found, i) { print("{t.names[i]}:{t.nums[i]}: {t.lines[i]}") }
i += 1
}
if found.is_empty() { 1 } else { 0 }
}
fn matches(line: str, needles: List[str]) -> bool {
var k = 0
while k < needles.len {
if line.contains(needles[k]) { return true }
k += 1
}
false
}
fn member(xs: List[int], v: int) -> bool {
var k = 0
while k < xs.len {
if xs[k] == v { return true }
k += 1
}
false
}
Two channels, both channel[int](0), and the zero is doing real work. A
zero-capacity channel is a rendezvous: a send blocks until a receiver
takes the value, so a task’s done.send(1) cannot overtake its own hits.
By the time the collector sees a task’s completion it has already received
everything that task found. live therefore counts down to zero exactly
once, and the while loop ends knowing the work is finished rather than
guessing.
The collector is one select over two channels, and it is the only code
in the program that touches found. This is what “collect matches without
a lock” means in practice: the shared mutable state is not shared. One
task owns it, three lines own the accumulation, and the other four tasks
communicate by sending.
A task reports the index of a matching line rather than the line itself,
and that choice pays for itself twice. It keeps the message small, and —
much more importantly — it makes the report independent of the order the
messages arrived in. The final loop walks the table from 0 to n and
prints the indices that are in found, so the output is in input order
whatever the tasks did. §30.5 is that sentence, measured.
The exit status follows the C convention every grep on your machine follows: zero when something matched, one when nothing did.
$ wolf build pargrep.lu && ./pargrep
a.log:1: 06:12 the wolf runs
a.log:3: 07:02 the wolf howls
b.log:2: 08:52 the wolf sleeps
b.log:4: 09:44 the wolf wakes
$ ./pargrep 'moon|pack' a.log b.log
a.log:2: 06:31 the moon watches
a.log:4: 07:40 the pack waits
b.log:3: 09:03 moonrise
Byte for byte what the sequential version printed in §30.1, which is the only acceptable result. A parallel tool whose output you have to squint at is a slower tool with extra steps.
30.4 The frozen pattern table
Four tasks read t and needles. Neither is copied, and nothing in the
program says “copy”. What says it is freeze.
freeze region { … } evaluates the block, then promotes everything the
block produced to imm — chapter 8’s third mode: immutable, shareable
from anywhere, permanently. An imm value is the one thing a task may
read from its enclosing function without a copy and without a lock,
because there is nothing to synchronize. Nobody can write it, including
the code that built it:
fn main() -> !int {
let needles = freeze region { "wolf|moon".split("|") }
(mut needles).push("pack")
print("{needles.len}")
0
}
error[E1012]: `needles` is frozen, so it cannot be passed as `mut`
--> ./s3.lu:6:5
|
5 | let needles = freeze region { "wolf|moon".split("|") }
| ---------------------------------------- the freeze happens here — the promotion to `imm` is deep and permanent
6 | (mut needles).push("pack")
| ^^^^^^^^^^^^^ this needs the data to be mutable
|
= note: `freeze` promotes the whole graph to `imm`: shareable from anywhere, forever, and never
writable again. Build the value completely before freezing it, or keep a mutable copy
(`copy`) alongside the frozen one.
Read the underline on line 2. The freeze is not a property of the
needles binding; it is a property of the data, and the data remembers
where it happened. That is what makes the sharing safe to do implicitly:
four tasks reading one imm graph cannot be racing, because there is no
second kind of access for them to race with.
The table is the same argument at a larger size. t holds every line of
every input file, and it crosses into four tasks as a shared read. Had it
been ordinary mutable data, each task would need its own copy of the whole
input, which is the version of this program that gets slower as you add
tasks.
What four tasks cost
The two programs, measured:
$ wc -l samples/projects/seqgrep/seqgrep.lu samples/projects/pargrep/pargrep.lu
57 samples/projects/seqgrep/seqgrep.lu
97 samples/projects/pargrep/pargrep.lu
Forty lines, seventy percent, to search eight lines of log four times faster in a way that is not measurable on eight lines of log. This is the part’s opening rule collecting on the chapter that most wanted to avoid it: on this input the sequential program is better, and it is better by every measure a reader can apply.
Three of the forty lines are the honest cost of concurrency — two channel
declarations and a scope. The other thirty-seven are the shard count
being a constant. Four spawns are written out because four is written out,
and each is four lines that differ from its neighbour in two identifiers.
A shard count taken from the input would collapse those sixteen lines to
about five and would make the program worth running on a directory instead
of on a demonstration.
So the honest summary is narrower than a slogan, and it was never going to be anything else. What wolf does here is remove the two failure modes that make people frightened of writing this program at all: there is no lock to forget and no shared accumulator to corrupt, and the reason is not discipline but that the compiler declines the shapes those bugs are made of. What it does not do is write the fan-out for you.
30.5 Testing with a seed
Run pargrep twice and you get the same four lines twice:
$ wolf build pargrep.lu && ./pargrep
a.log:1: 06:12 the wolf runs
a.log:3: 07:02 the wolf howls
b.log:2: 08:52 the wolf sleeps
b.log:4: 09:44 the wolf wakes
$ ./pargrep
a.log:1: 06:12 the wolf runs
a.log:3: 07:02 the wolf howls
b.log:2: 08:52 the wolf sleeps
b.log:4: 09:44 the wolf wakes
Two runs prove nothing about a concurrent program, so here are two hundred of them, hashed:
$ for i in $(seq 1 200); do ./pargrep | sha256sum; done | sort -u | wc -l
1
One hash. Two hundred runs of a four-task program produced one output, and the interleaving was different nearly every time. Both halves of that sentence are checkable, and the second one is the interesting half.
Ask the compiler to run the program under a fixed seed and it reports what it saw:
$ wolf conform-run ./pargrep.lu --seed=7 --native
{… "seeded":true, "verdict":"exit(0)",
"stdout_sha256":"70b450e487d98eb503b9f6214098fb5cdf7bfc2fb06bf6ecb1c40b0012b42317" …}
Every seed reports that same stdout_sha256. Not “usually” and not “for
the seeds we tried” — the hash does not depend on the seed, because the
output does not depend on the schedule, because the collector stores
indices and the report walks the input in order. The property was designed
in during §30.3 and the seed is how you check that the design held.
Now the honest half, and it is the part of this section worth reading
twice. A seed does not reproduce a parallel run. Instrument the
collector to print each index as it arrives, fix the seed, and run the
same binary twenty times: three different arrival orders come out. The
seed pins the decisions the runtime makes with its own random number
generator — chiefly which of two simultaneously ready select arms wins —
and it does not pin the operating system’s scheduler, the order four
worker tasks reach a rendezvous, or how long a read takes.
What the seed does reproduce, it reproduces exactly. Two select arms
that are both ready are a genuine coin flip, and it is the runtime’s coin:
fn main() -> !int {
let a = channel[int](1)
let b = channel[int](1)
a.send(1)
b.send(2)
var got = 0
select {
v from a => { got = v },
v from b => { got = v },
}
print("{got}")
0
}
Under --seed=1 that program prints 2 eight times out of eight; under
--seed=3 it prints 1 eight times out of eight. The arm is the seed’s
to choose and the choice is stable, which is what makes a select
tie-break reproducible in a bug report.
Put the two facts together and you have the working rule for testing concurrent programs in wolf, which is the same rule as testing them anywhere and is easier to obey here than in most places:
Do not try to reproduce the run. Make the answer not depend on it.
Every design decision in pargrep that looked fussy in §30.3 — indices
instead of lines, rendezvous instead of buffering, a report that walks the
input instead of the results — exists to move one more thing out of the
schedule’s reach. What is left over is one coin flip, and the seed pins
that.
A test that runs pargrep and asserts four exact lines is therefore not a
flaky test. That is not luck, and it is not a claim about the scheduler.
It is a claim about the program, and it is the claim the two hundred runs
above check.
Exercise 30-1 (fingers · wolf) — Build both programs and run them
against the same two files with three patterns of your own. Then run
wc -l on both and write down, in one sentence, what you would tell a
colleague who proposed the parallel one for a log directory of four files.
Exercise 30-2 (comprehension · wolf) — hits and done are both
channel[int](0). Give hits a buffer — channel[int](64) — and predict
what happens before you run it. Then run the binary twenty times and count
the report lines each time. Two questions: what can done.send(1) do now
that it could not do before, and which line of the collector is the one
that loses the hits?
Exercise 30-3 (comprehension · wolf) — Delete freeze from the
needles binding, leaving let needles = pattern.split("|"). Predict
whether the program still compiles before you try it. Then explain, in two
sentences, what the four tasks are allowed to do with needles in each
version.
Exercise 30-4 (extension · wolf) — Instrument the collector: print each index as it arrives, before pushing it. Run the binary twenty times and count the distinct arrival orders you see; then confirm that the four report lines never move. Which of the two outputs would you put in a test?
Exercise 30-5 (extension · wolf) — Make the report order-dependent on purpose: delete both channels and the collector, and have each task print its own matches directly. Run the binary twenty times and hash the output. You will get more damage than you predicted — say what the extra damage is, and then say what you have broken in terms of §30.5’s rule rather than in terms of tasks.
Exercise 30-6 (spelunking · wolf) — Read the E1012 note in §30.4 in
full. It offers two ways out — build the value completely before freezing,
or keep a mutable copy alongside. Say which one pargrep uses and what
the other one would cost in a four-task program.
Exercise 30-7 (design) — The shard count is a constant. Sketch the
version that takes it from the input: what the ranges become, what the
collector’s live counter becomes, and what pargrep would need from the
language to spell the fan-out in one loop instead of four spawns. Then say
whether four shards on two files was ever the right number.