The concurrency pages before this one took the model apart: a block that joins every task it started, generators that suspend between values, streams that tasks feed, and tasks that run in worker processes. This page puts it back together. One program reads a text, transforms it on carriers, does the heavy part of every line in a worker process, and merges what comes back, and every hand-over between those parts is a stream of T.

Three lines go wrong on purpose: one worker is killed, one refuses its line, one never answers. Each is answered by a rule of the language, and each shows in the output.

The example is examples/OneModel in the tutorial repository, with the text it reads beside it in lines.txt.

Getting the file

make -C mica-container/examples/OneModel run
OneModel - one text through a generator, tasks on carriers and worker processes

What became of each line
  line 1: 35 characters, nonce 174378
  line 2: 32 characters, nonce 55531
  line 3: lost with its worker, and nothing else
  line 4: 28 characters, nonce 36201
  line 5: refused by its worker as 'Empty'
  line 6: 31 characters, nonce 12483
  line 7: no answer within 3000 ms, so its worker was cancelled
  line 8: 32 characters, nonce 5730
  line 9: 27 characters, nonce 10653

After its title, the program pauses for three seconds: that is line 7’s worker running out of time. Every line is accounted for, and the three that went wrong cost exactly themselves.

A generator reads the text

generator Lines(path : string) : stream of string;
var
    f       : File;
    buffer  : stringbuffer;
    got     : bool;
    problem : OsError;
begin
    f := Open(path, ModeRead) on fail problem do
    begin
        WriteLn(Unreadable, path);
        leave;
    end;

    defer Close(address f);

    repeat
        got := ReadLine(f, address buffer) on fail problem do leave;

        if got then
            emit ToString(buffer);

        Clear(address buffer);
    until not got;
end;

Lines is an ordinary generator: each emit hands one line to whoever pulls and suspends right there, its file still open, until the next pull. Nothing reads the file further ahead than its reader asks, and the defer closes it on whichever way the generator ends.

Stages on carriers share one ring

jobs := schedule Tidy(Lines(InputPath)) buffer RingSlots;
left := schedule Lane(jobs) buffer RingSlots;
right := schedule Lane(jobs) buffer RingSlots;

Tidy pulls Lines and emits each line numbered and measured, as a Job. Scheduled with buffer, the same kind of body becomes a task on a carrier that runs ahead of its readers by as many jobs as its ring holds: the push half of the same streams. Two lanes read that one ring, so each job lands in exactly one lane, whichever asks first, as the worker pool shares one ring among its workers.

The heavy part runs in a worker process

generator Crunch(job : Job) : stream of Answer fails Refusal;

For each job, a lane calls Place, which places Crunch on a worker process. Crunch searches for the smallest nonce whose hash with the line is a multiple of 65,536, a small proof of work that gives the same digits on every machine. It runs in a process of its own, so it reaches nothing of the program: the job arrives as a copy, its text included, and the answer it emits travels back as bytes, exactly as the tasks on processes page describes.

Each line gets a worker of its own. A producer placed on a process cannot pull a stream, because a stream names memory of the program, so the lanes read the ring on carriers and hand each worker its one line as an argument. That is also what makes a loss cheap: a worker holds one line, never more.

One select, four outcomes

concurrent
    results := schedule Crunch(job) on process;

    select
        on got in results do answer := got;
        on lost results do answer.fate := Lost;
        on timeout Budget do leave;
    end
    on fail reason do answer.fate := Refused;
end;

select waits on everything that can happen to the worker at once, and runs the arm of whichever happens first:

  • The answer arrives. The worker’s one Answer is bound to got.
  • The worker is lost. A trap, a kill or the machine running out of memory ends a worker without an answer, and the on lost arm runs. A stream a worker feeds is refused in a select without such an arm.
  • The worker fails. A failure of the stream’s own domain, here Empty, reaches the on fail form behind the end, the same form that catches it from a producer on a carrier.
  • Nothing happens in time. The timeout arm runs after Budget milliseconds of silence.

The program merges the two lanes with a second select, whose arms are the two lanes. Which lane answers first belongs to the scheduler, so the program files each answer under its line number and prints the table in line order, the one order it can promise.

Three lines that go wrong

Line 3: the worker is killed.

if job.number = KilledLine then
    status := Kill(GetPid(), SIGKILL) on fail use -1;

The worker sends itself SIGKILL, the signal kill -9 sends from a shell, which no process can catch: it ends at once, holding line 3. The on lost arm answers, and the loss costs that one line. The program, the other lane and every other worker go on, because the worker was a process of its own.

Line 5: the worker refuses. The text’s fifth line is the empty line between its two stanzas, and Crunch refuses it with fail Empty before any work starts. The failure crosses from the worker process to the select’s on fail form: one failure channel, whichever side of a process boundary the task ran on.

Line 7: the worker never answers.

if job.number = StuckLine then
    Sleep(60 * Second);

Its worker waits a minute, parked rather than spinning. After Budget milliseconds, the timeout arm runs leave, which ends Place’s block early, and leaving a concurrent block cancels everything still running in it. The cancellation arrives where the task inside waits for the worker, a suspension point, and the worker is ended on the way out. Place set its result to a cancelled line before the block began, so the early way out hands that back, and the program moves on without waiting out the minute.

The rules that held

  • The block joins every task, on every path. Place’s block ends either at its end or early through leave, and on both it waits for what it started. The program’s own block ends only when the reader, the stage, both lanes and every worker have ended, which is why the table is complete when it prints.
  • Cancellation arrives at suspension points. The stuck worker’s line is cancelled where its task waits, never between two instructions.
  • One failure channel. Empty travels from a worker process to on fail exactly as it would from a carrier.
  • A loss is named, and costs what the worker held. on lost answers the killed worker, and every other line is answered.

What this does not do

  • No shared memory. A worker sees a copy of its line and nothing else, so Crunch takes everything it needs as arguments.
  • No order the language does not promise. The lanes race, and the program prints in line order only because it collects the answers first.
  • No restart. A lost line stays lost. Placing it again on a fresh worker is one more call to Place, if a program wants that policy.

Try it

1. Kill another line’s worker. Set KilledLine = 1; and run. Line 1 is lost now, and line 3 has a nonce:

  line 1: lost with its worker, and nothing else
  line 2: 32 characters, nonce 55531
  line 3: 29 characters, nonce 113271

2. Let line 7 finish. Set StuckLine = 0;, so that no line waits, and run. The pause is gone, and line 7 has its answer:

  line 7: 41 characters, nonce 54415

3. Kill a worker yourself. Set Budget = 30000;, so that line 7’s worker waits half a minute, and run the program. While it waits, kill the worker from a second terminal. Every other worker has answered by then, so the newest process named OneModel is the one that holds line 7:

pkill -9 -n -x OneModel

The program answers your kill -9 exactly as it answered line 3’s:

  line 7: lost with its worker, and nothing else

4. Run the heavy part on a carrier. Replace on process in Place with buffer RingSlots. The compiler refuses the on lost arm, because no worker can be lost any more:

no producer of the stream 'results' runs on a process, so no worker can be lost: remove the arm 'on lost results'

Remove the arm too, and run: this time SIGKILL at line 3 ends the program itself, since the process that holds the line is now the program’s own.

Next

The tasks-generators-processes report states the whole model and its guarantees in one place. A program that starts other programs and watches them live, rather than workers of its own, is the Supervisor example in the tutorial repository.