Tasks, generators and processes

This chapter describes how a Mica program does several things at once: the tasks it schedules and the tree they form, the concurrent block that owns them, where the work runs, on the program’s carriers or in worker processes, the generators and the streams their values travel through, what happens when a task or a stream fails, how a block cancels its tasks and gives them a deadline, and what a block does when a worker process ends before its work. Each construct names its rule in the grammar, says what it means, and quotes what the compiler answers when a source breaks the rule. The failure channel a task and a generator fail on is the one every routine has, which failure channel teaches, and the learn pages on tasks, generators, channel streams and tasks on processes walk each construct through a longer program.

Every example is complete, compiled and run by the compiler’s own test suite on AMD64 and AArch64, and the output shown beneath it is the output those runs are held to. Each runs on one carrier, the default, so the order of its lines is the order the scheduler gives them there.

Tasks and the task tree

A task is a routine declared with the word task (task-heading): it takes parameters as a procedure does, may state fails and an error domain, and returns no value. A task does not run when it is named; it runs when a schedule statement (schedule-statement) creates it and starts it, and from then on it runs beside the routine that scheduled it.

A concurrent block (concurrent-block) is the scope a task lives in. Its statements run in order and may schedule tasks, and its end waits until every task scheduled in it has finished. A task cannot outlive the block that scheduled it, and a task may hold blocks of its own, so the tasks of a program form a tree whose shape is the shape of the blocks in its source.

A task’s arguments are copied into the task when it is scheduled. The task reads its own copies, so the owner may change the variables it passed as soon as the schedule statement is behind it, and a parameter whose value shares memory, a pointer among them, is refused. A task touches a variable outside its own frame, one of the program’s globals, only in a synchronized statement (synchronized-statement), which runs without a task switch, so an update such as synchronized total := total + n is one step that no other task interleaves.

A task yields its turn at a suspension point: a call of Yield, a wait on a stream, a wait for a block’s tasks. A schedule runs its task at once, on the carrier that scheduled it, until the task reaches its first suspension point or its end, and its owner goes on from there; so the task stacks standing at once grow with the depth of the task tree, not its breadth. On one carrier a task that yields joins the tail of the ready list, and the tasks take their turns in the order they yielded. The language promises the join at a block’s end, not an order among the block’s tasks.

{
    Kitchen - the task tree. A 'concurrent' block schedules three cooks, each a task that runs beside the block's
    owner, and the block's 'end' waits for every one of them. A task's arguments are copied when it is scheduled, so
    each cook serves the table the owner named at that moment, whatever the owner writes afterwards. The cooks add to
    one shared count, and a task touches the program's state only in a 'synchronized' statement, so each addition is
    one step no other task interleaves. A second block schedules a baker who fails with a code of the kitchen's error
    domain: the failure cancels the block's other task, whose deferred line still runs, and the block's 'on fail'
    section receives the code after the join.
}
program Kitchen;

imp
    WriteLn, Ord, Yield : std;

type
    KitchenError = error (OutOfFlour, OvenCold);

const
    FirstTable = 1;
    SecondTable = 2;
    ThirdTable = 3;
    LateTable = 9;
    WaiterTable = 4;
    BakerTable = 5;
    FirstPlates = 2;
    SecondPlates = 3;
    ThirdPlates = 1;
    StartLine = "cook starts table %lld";
    DoneLine = "cook serves table %lld, %lld plates";
    TotalLine = "plates served %lld";
    BakerLine = "baker at table %lld is out of flour";
    ServedLine = "waiter serves table %lld";
    WaiterLeftLine = "waiter leaves table %lld";
    FailedLine = "the block failed with code %lld";

var
    platesServed : int64;
    table : int64;
    failure : KitchenError;

{ A cook starts an order, lets the other cooks run, and then adds its plates to the shared count. }
task Cook(order : int64, plates : int64);
begin
    WriteLn(StartLine, order);
    Yield();
    synchronized platesServed := platesServed + plates;
    WriteLn(DoneLine, order, plates);
end;

{ A baker runs out of flour, which fails the task with a code of the kitchen's domain. }
task Baker(order : int64) fails KitchenError;
begin
    WriteLn(BakerLine, order);
    fail OutOfFlour;
end;

{ A waiter would serve after the others have run, and is cancelled before it does. }
task Waiter(order : int64);
begin
    defer WriteLn(WaiterLeftLine, order);
    Yield();
    WriteLn(ServedLine, order);
end;

begin
    platesServed := 0;

    table := FirstTable;
    concurrent
        schedule Cook(table, FirstPlates);
        table := SecondTable;
        schedule Cook(table, SecondPlates);
        table := ThirdTable;
        schedule Cook(table, ThirdPlates);
        table := LateTable;
    end;
    WriteLn(TotalLine, platesServed);

    concurrent
        schedule Waiter(WaiterTable);
        schedule Baker(BakerTable);
    on fail failure do
        WriteLn(FailedLine, Ord(failure))
    end;
end.

It prints:

cook starts table 1
cook starts table 2
cook starts table 3
cook serves table 1, 2 plates
cook serves table 2, 3 plates
cook serves table 3, 1 plates
plates served 6
baker at table 5 is out of flour
waiter leaves table 4
the block failed with code 0

Each cook serves the table the owner named when it was scheduled, though the owner had moved table on to 9 before the block’s end.

Failure in a block

A task that states fails ends with a failure when it executes fail, and the block that scheduled it receives that failure. The first failure in a block wins: it cancels the block’s other tasks, and after the join the block’s section for that failure runs. A failure a cancelled task would report afterwards is a consequence of the cancellation and is dropped, so exactly one section of a block runs. The sections stand before the block’s end:

Every domain the block’s tasks fail is consumed: by its own section, by on fail continue, or by the enclosing routine when that routine itself fails the domain. No two sections name one domain, and no section names a domain no task of the block fails. A section whose statement is a call ends that statement with a semicolon, because a clause behind a call binds the call.

In the example above, the baker fails with OutOfFlour while the waiter is waiting for its next turn: the failure cancels the waiter, whose deferred line runs, and the section receives the code, 0, the ordinal of OutOfFlour.

Where the work runs

A program runs its tasks on carriers, the threads of the runtime. A program built with --tasking single, the default, runs on one carrier, and its tasks take turns at their suspension points; one built with --tasking multicore (the compiler’s flags) runs its tasks on a pool of carriers at once, sized by the carriers runtime flag (runtime flags). The source is the same on both: what changes is the order in which the lines of several tasks interleave, never the results the language promises, and carriers explains which order a program may rely on.

schedule ... on process (process-placement) runs a task in a worker process of its own. The worker receives a copy of the task’s arguments and writes to the program’s own standard output, and its work may not reach state outside its own frame. A worker can end before its task does, so a block that places a task on a process states in an on lost section what the program does then. The number of workers that run at once is the program’s process count, --processes or the processes runtime flag; a task placed beyond it waits until a worker ends.

Work placed only on processes shares no memory with the program, so it makes none of the program’s variables shared. In a program whose every task and producer runs on a process, the program’s statements touch its variables without synchronized and may take their address. One placement on a carrier anywhere in the source, in a routine that is never called as well, keeps the rules of a task on a carrier for the whole program, and so does a task that no statement schedules.

Generators and streams

A generator is a routine declared with the word generator (generator-heading) whose result is a stream (stream-type), stream of int64, and which may state fails and an error domain behind it. Its body hands each value to the stream’s reader with emit (emit-statement) and waits for the next pull, and its locals keep their values from one emit to the next. Its stream ends when its body ends. A for ... in loop over a stream (for-statement) takes one value per turn until the stream ends.

A stream delivers each value as a copy its reader owns, so an element type that shares memory, a pointer or a stream, is refused. A stream comes in two forms, and the statement that binds it decides which:

A generator that takes a stream as a parameter is a stage: it reads one stream and emits another, so Doubled(Countdown(3)) is a chain of two stages, and no value is computed that the last reader does not pull. The capacity behind buffer is an ordinal value of one or more, which the runtime rounds up to the ring’s slot count, and a ring bound without buffer takes the default capacity.

A ring may be filled by several producers: schedule Arrivals(200, 3) into platforms joins a second task to a ring a schedule already bound (schedule-statement), and the ring ends when every producer has ended. A ring may also be read by several readers: scheduled tasks that take it as a parameter, the owner’s own loop, select arms, and stages scheduled with schedule ... buffer over it, so that a := schedule Fwd(s) buffer 4 and b := schedule Fwd(s) buffer 4 split the values of s between them. Each value of a ring reaches exactly one of its readers. A pull chain keeps its one reader, and to share an upstream it is opened as a ring with schedule ... buffer.

A select statement (select-statement) waits on several rings at once. Each arm (select-arm) names a variable and a ring, on v in express do ..., and the first ring that holds a value gives it to its arm, whose statement runs; an on timeout arm runs instead when no ring speaks within the milliseconds it names. A select watches only rings, since a pull chain holds no value before its reader pulls it.

{
    Station - generators and streams. A generator's body runs only as far as its consumer pulls, and its locals keep
    their values from one 'emit' to the next. A generator that takes a stream is a stage, so 'Doubled(Countdown(3))'
    is a pull chain of two stages read by one loop. 'schedule ... buffer' runs a generator as a task that fills a ring
    of the given capacity, 'schedule ... into' joins a second producer to that ring, and the owner's loop takes every
    value of both, in an order the scheduler chooses, so the program prints their count and their sum. A 'select'
    waits on rings and takes the first value one of them holds, or runs its timeout arm when none speaks in time. A
    generator that fails ends its stream with the failure, which the reading loop's 'on fail' clause receives once the
    values before it are read.
}
program Station;

imp
    WriteLn, Ord, Yield : std;

type
    SensorError = error (Saturated);

const
    CountdownStart = 3;
    Factor = 2;
    MorningFirst = 100;
    EveningFirst = 200;
    TrainsEach = 3;
    RingRoom = 4;
    ExpressValue = 7;
    OneSlot = 1;
    PatienceMilliseconds = 100;
    ReadingCount = 2;
    PulledLine = "pulled %lld";
    ArrivalsLine = "arrivals %lld, sum %lld";
    ExpressLine = "express %lld";
    SilentLine = "silent %lld";
    QuietLine = "quiet";
    ReadingLine = "reading %lld";
    SaturatedLine = "the sensor failed with code %lld";

var
    n, arrivals, sum, v : int64;
    platforms, express, silent : stream of int64;
    failure : SensorError;

{ A countdown emits its start and every number below it down to one, keeping its place in a local between pulls. }
generator Countdown(start : int64) : stream of int64;
var
    remaining : int64;
begin
    remaining := start;
    while remaining > 0 do
    begin
        emit remaining;
        remaining := remaining - 1;
    end;
end;

{ A stage doubles every value it pulls from its source. }
generator Doubled(source : stream of int64) : stream of int64;
var
    item : int64;
begin
    for item in source do
        emit item * Factor;
end;

{ A platform's arrivals are a number of trains counted up from a first one. }
generator Arrivals(first : int64, count : int64) : stream of int64;
var
    k : int64;
begin
    for k := 0 to count - 1 do
        emit first + k;
end;

{ A silent line yields its turn and never emits. }
generator Silent(spins : int64) : stream of int64;
begin
    while spins > 0 do
        Yield();
    emit spins;
end;

{ A sensor gives its readings and then saturates, which fails its stream. }
generator Readings(count : int64) : stream of int64 fails SensorError;
var
    k : int64;
begin
    for k := 1 to count do
        emit k;
    fail Saturated;
end;

begin
    for n in Doubled(Countdown(CountdownStart)) do
        WriteLn(PulledLine, n);

    platforms := schedule Arrivals(MorningFirst, TrainsEach) buffer RingRoom;
    schedule Arrivals(EveningFirst, TrainsEach) into platforms;
    arrivals := 0;
    sum := 0;
    for n in platforms do
    begin
        arrivals := arrivals + 1;
        sum := sum + n;
    end;
    WriteLn(ArrivalsLine, arrivals, sum);

    express := schedule Arrivals(ExpressValue, OneSlot) buffer OneSlot;
    silent := schedule Silent(OneSlot) buffer OneSlot;
    select
        on v in express do WriteLn(ExpressLine, v);
        on timeout PatienceMilliseconds do WriteLn(QuietLine);
    end;
    select
        on v in silent do WriteLn(SilentLine, v);
        on timeout PatienceMilliseconds do WriteLn(QuietLine);
    end;

    for n in Readings(ReadingCount) do
    begin
        WriteLn(ReadingLine, n);
    end
    on fail failure do
        WriteLn(SaturatedLine, Ord(failure));
end.

It prints:

pulled 6
pulled 4
pulled 2
arrivals 6, sum 906
express 7
quiet
reading 1
reading 2
the sensor failed with code 0

A worker’s stream in a select

A stream that a producer on a process feeds can be an arm of a select, beside rings and a timeout. Its worker can be lost before the stream ends, so the select answers that with a lost arm that names the stream:

select
    on reading in counted do total := total + reading;
    on lost counted do WriteLn("the counter's worker was lost");
    on item in local do total := total + item;
    on timeout 1000 do WriteLn("quiet");
end;

The lost arm runs once for each loss, after every value the worker sent before it was lost. The stream’s arm then counts as ended, so a select inside a loop goes on waiting on the arms that are left. An arm needs its lost arm wherever a worker’s stream can reach the select, by the rule lost workers states for loops, and a failure of the stream reaches the select’s on fail form as the failure of a ring does.

Failure across a stream

A generator that states fails ends its stream with a failure when it executes fail. Its deferred statements run as it ends, and the reading loop receives the failure once it has read the values emitted before it. The loop carries the clause behind its body (statement-consumption), with the choices a call statement has: on fail e do handles the code, on fail leave passes the failure on as the reader’s own, and on fail continue discards it. A ring carries the same failure: a stream variable bound to a failing generator is declared stream of int64 fails SensorError, and the failure crosses the ring as the stream’s end. A loop whose body is a single call writes its body in begin ... end, so the clause binds the loop and not the call. The last lines of the example above are a failing stream read to its end.

A task on a process that reads a failing stream ends its loop with the failure in the same way, and on fail leave there hands the failure to the block’s section as the task’s own.

Cancellation and deadlines

A block cancels the tasks still running in it when it is left early, by a leave among its statements, or when one of its tasks fails. Cancellation is cooperative: a cancelled task is delivered its cancellation at its next suspension point, or at the back edge of a loop it is running, and it takes its leave there exactly as a leave of its own would. Its deferred statements run, and the statements behind that point do not.

A block is given a deadline by the routine ScopeDeadline of the standard library, called among the block’s statements with a budget in milliseconds. When the budget runs out, the block’s tasks are cancelled as above and the block ends; after the block, ScopeTimedOut answers true in the routine that owns it. A block whose tasks finish in time disarms its deadline, and ScopeTimedOut answers false. A deadline fires no sooner than its budget, and however many blocks have one armed, the program holds a single timer for all of them and for its sleeping tasks.

{
    Search - cancellation and deadlines. A 'leave' out of a 'concurrent' block cancels the tasks still running in it:
    each is delivered its cancellation at its next suspension point and takes its leave there, so the cleanup it
    deferred runs and the lines behind that point never print. 'ScopeDeadline' gives the block it stands in a budget in
    milliseconds: when the budget runs out, the block's tasks are cancelled the same way, a loop that never suspends
    included, and 'ScopeTimedOut' answers true in the routine that owns the block. A block whose work ends in time
    disarms its deadline, and 'ScopeTimedOut' answers false.
}
program Search;

imp
    WriteLn, Yield, ScopeDeadline, ScopeTimedOut : std;

const
    FirstProbe = 1;
    SecondProbe = 2;
    TightBudget = 25;
    GenerousBudget = 10000;
    EndlessSteps = 100000000000;
    FewSteps = 1000;
    StartLine = "probe %lld starts";
    CleanupLine = "probe %lld cleans up";
    ResumedLine = "probe %lld resumes";
    AfterBlockLine = "the block ran to its end";
    TimedOutLine = "timed out";
    CompletedLine = "completed";
    StepsLine = "spin took %lld steps";

{ The search schedules two probes and leaves the block at once when it is told to stop early. }
procedure Probe(stopEarly : bool);

    { A probe starts, yields its turn, and is cancelled at that point when the block is left. }
    task Seek(tag : int64);
    begin
        defer WriteLn(CleanupLine, tag);
        WriteLn(StartLine, tag);
        Yield();
        WriteLn(ResumedLine, tag);
    end;

begin
    concurrent
        schedule Seek(FirstProbe);
        schedule Seek(SecondProbe);
        leave when stopEarly;
    end;
    WriteLn(AfterBlockLine);
end;

{ The watch gives a spinning task a budget and says afterwards whether the budget ran out. }
procedure Watch(budget : int64, steps : int64);

    { A spin counts in a loop that never suspends, so only the deadline's check at the loop's back edge stops it. }
    task Spin(limit : int64);
    var
        i : int64;
    begin
        i := 0;
        while i < limit do
            i := i + 1;
        WriteLn(StepsLine, i);
    end;

begin
    defer if ScopeTimedOut() then WriteLn(TimedOutLine) else WriteLn(CompletedLine);

    concurrent
        ScopeDeadline(budget);
        schedule Spin(steps);
    end;
end;

begin
    Probe(True);
    Watch(TightBudget, EndlessSteps);
    Watch(GenerousBudget, FewSteps);
end.

It prints:

probe 1 starts
probe 2 starts
probe 1 cleans up
probe 2 cleans up
timed out
spin took 1000 steps
completed

The first spin counts in a loop that never suspends, toward a bound far out of reach of its budget, and its deadline stops it at the loop’s back edge, so its line never prints.

Lost workers

A worker process ends before its task when the task traps or when the process is killed. The block’s on lost section (lost-section) then runs after the join:

A task on a process that reads a stream of the program, a reader, asks for one value at a time, and a worker holds at most the one it asked for last. When a reader is lost, only its own share is lost: the block’s other tasks go on, the values the stream still holds go to the readers still running, and when every reader is lost the values nobody took stay in the stream, where the program reads them. A held value is never sent again. A lost task that reads no stream cancels the block as a failure does.

The name in on lost held do is a variable declared in the routine that holds the block, whose type is the element type of every stream the block’s readers read. A held value that holds text or dynamic arrays reaches the section whole. A loop that can meet a stream fed by a producer on a process carries an on lost section without a name, since a lost producer holds no value. The section’s statement may be any statement, a loop among them.

A stream variable holds the stream its latest bind opened, so the compiler judges each loop by the binds of its variable that can reach it, on every road to it: through either branch of an if and each arm of a case, past a leave, and around a loop that binds the variable again at its end. A bind ... on process or a join into ... on process on any of those roads makes the loop carry the section; a loop that can meet only streams of this program carries none, and a section there is refused, because it can never run:

readings := schedule Sample(sensor, 100) buffer 8;
for reading in readings do        { a ring this program fills: no section }
    total := total + reading;

readings := schedule Sample(sensor, 100) on process;
for reading in readings do        { a stream a worker feeds }
    total := total + reading
on lost do
    WriteLn("the sensor's worker was lost");

A select is judged the same way for its lost arms. Where something besides the routine’s own statements can bind the variable, a library’s unit variable that any routine of the library binds, or a variable a nested routine, task or generator reaches, the compiler keeps one judgment for every loop and select over it: one placement of its producer on a process anywhere makes every one of them answer a lost worker.

{
    Workshop - tasks on processes and lost workers. 'schedule ... on process' runs a task in a worker process of its
    own, which receives a copy of the task's arguments and writes to the program's standard output. A worker can end
    before its task does, so a block that places a task on a process states what happens then in its 'on lost' section,
    which runs after the join: here the second worker divides by zero, its trap ends that worker alone, and the program
    goes on. A task on a process may read a stream of the program, one value at a time. The fragile reader kills its
    own worker while it holds the third part, so 'on lost held do' receives exactly that part, and the parts nobody took
    stay in the stream, where the program reads them after the block.
}
program Workshop;

imp
    WriteLn : std;
    Kill, GetPid, SIGKILL : posix;

const
    Side = 7;
    Dividend = 10;
    Divisor = 0;
    PartCount = 10;
    RingRoom = 4;
    FatalPart = 3;
    NoStatus = -1;
    AreaLine = "area %lld";
    QuotientLine = "quotient %lld";
    SquareLostLine = "the squaring worker was lost";
    DivideLostLine = "the dividing worker was lost";
    HeldLine = "the lost worker held part %lld";
    RestLine = "the program read %lld parts, sum %lld";

var
    parts : stream of int64;
    held, part, count, sum : int64;

{ A square prints the area of a square with the given side. }
task Square(side : int64);
begin
    WriteLn(AreaLine, side * side);
end;

{ A divide prints a quotient, and traps when the divisor is zero. }
task Divide(dividend : int64, divisor : int64);
begin
    WriteLn(QuotientLine, dividend / divisor);
end;

{ The parts are numbered from one up to their count. }
generator Parts(total : int64) : stream of int64;
var
    k : int64;
begin
    for k := 1 to total do
        emit k;
end;

{ A fragile reader takes parts until the fatal one, and kills its own worker while it holds that part. }
task Fragile(source : stream of int64);
var
    v : int64;
    status : int32;
begin
    for v in source do
        if v = FatalPart then
            status := Kill(GetPid(), SIGKILL) on fail use NoStatus;
end;

begin
    concurrent
        schedule Square(Side) on process;
    on lost do
        WriteLn(SquareLostLine)
    end;

    concurrent
        schedule Divide(Dividend, Divisor) on process;
    on lost do
        WriteLn(DivideLostLine)
    end;

    parts := schedule Parts(PartCount) buffer RingRoom;
    concurrent
        schedule Fragile(parts) on process;
    on lost held do
        WriteLn(HeldLine, held)
    end;

    count := 0;
    sum := 0;
    for part in parts do
    begin
        count := count + 1;
        sum := sum + part;
    end;
    WriteLn(RestLine, count, sum);
end.

It prints:

area 49
the dividing worker was lost
the lost worker held part 3
the program read 7 parts, sum 49

The squaring worker’s line reaches the program’s output, and its section does not run. The dividing worker’s trap report goes to standard error and ends that worker alone. The reader is lost holding part 3, and the program reads the seven parts behind it.

What the checked build traps

A task runs under the same guards as any routine: the checked build, --optimize debug,checked or O2,checked, plants the arithmetic and conversion guards chapter 5 lists in the body of every task and generator, and a division by zero traps on every tier. Where a trap stops the program depends on where the work runs:

The traps list every trap, and reading a trap report explains the report’s lines.

What the compiler refuses

Where the source breaks a rule of this chapter, the compiler refuses the program, and the refusal begins with these words:

Each refusal names the task, the stream or the variable it judges, and most go on to name the remedy, as 5337 and 5384 do. The learn pages, from tasks on, show more of them in the programs they belong to.