Channel streams: the push half

Generators told half the stream story: the consumer’s loop pulls, the body advances one emit per pull, and everything runs on one thread of control. This tutorial is the other half — the push — and its whole surface is one change at the bind site. The body does not change. The loop does not change. What changes is who drives.

Elsewhere this is assembled by hand from separate primitives — channels, lightweight tasks, a select. Mica derives it: a stream is already a typed sequence with an honest end, a task is already a child of its block, so a task-fed stream is one word connecting two things the language has — schedule, applied to a generator, with a ring buffer in between.

The example is ChannelStreams; it assumes tasks and generators.

The bind decides: pull or push

generator Numbers(limit : int64) : stream of int64;
var
    n : int64;
begin
    n := 1;
    while n <= limit do
    begin
        emit n;
        n := n + 1;
    end;
end;

s := Numbers(5);                    { pull: the loop drives the body      }
s := schedule Numbers(5) buffer 8;  { push: the body runs as a task,      }
                                    { its emits land in a ring of eight   }

One body, two binds, zero changes to either loop. Under schedule … buffer, the generator becomes a producing task of the enclosing block — the same task rules as always: child of the block, joined at its end, cancellation delivered at suspension points. Its emits push into a bounded ring; the consumer’s for pops, parking when the ring is empty and resuming as values arrive:

The bind: a producer task feeds a bounded ring
  drained 1..5, sum 15

The ring being bounded is a design sentence, not a tuning knob: a producer that outruns its consumer parks at emit instead of growing a queue, so backpressure is built in and memory stays fixed — the same discipline the fixed arena applies to allocation, applied to buffering. (The buffer clause is optional; without it the bind opens a default-sized ring. A capacity is one or more, and the runtime rounds it up to a power of two, so buffer 1 holds two values and buffer 5 eight. A capacity below one, or a ring larger than memory can hold, is refused: by the compiler when the capacity is a constant, and as an allocation failure when the stream starts when the program computes it.)

The stream still ends honestly: exactly when the ring is empty and no producer remains. No sentinel values, no closed-flag protocol to hand-roll.

Fan-in: a second attach, not a second mechanism

s := schedule Ones(4) buffer 16;
schedule Twos(4) into s;

into attaches another producer to a ring that is already open. Both push; one loop pops; the end still waits for the last producer standing. What the scheduler now owns is the arrival order — so the example reports what is actually promised, an order-free total:

Fan-in: two producers, one stream
  popped 8 values, sum 12

Every pushed value is popped exactly once — four ones and four twos, whatever the interleaving. When a tutorial prints an order in a section like this, distrust it; the count and the sum are the honest oracle.

A producer in another process

A producer is placed on a process the way a task is, and the stream is the stream it always was — the bind, the capacity, the pull, the close and the failure word do not change:

readings := schedule Sample(sensor, 10000) on process;

for reading in readings do
begin
    total := total + reading;
end
on lost do
    WriteLn("the sensor's worker was lost");

The producer runs in a worker process and its elements cross to this program’s ring as frames. The worker fills a frame with what it pulled rather than sending one element at a time, so a for loop is not one round trip per element; a loop that consumes faster than the worker produces still gets each frame as it arrives. into ... on process joins a second worker to the same ring, exactly as into joins a second producer here. A stream’s worker takes no place among the worker processes that run tasks placed on processes, because the loop paces it, so the loop may place tasks on processes for what it reads.

Two things follow, and the compiler holds you to both:

• A worker can end before the stream does — a trap, a kill, the machine running out of memory — and that is no failure of the producer’s own domain. So every loop over such a stream carries an on lost section, and a section on a stream no worker feeds is refused as dead. The elements a lost worker had pulled but not yet sent are lost with it.
• A select arm has no place to answer a lost worker, so a stream a worker feeds is consumed by a loop. A select over rings of this process is unchanged.

The producer itself answers the rule a task on a process answers: it reaches no state outside its own frame, because a worker shares none with the program. What it is handed and what it emits may hold text and arrays like any other stream’s: the request carries an argument’s text with it, and every frame carries the text of the elements inside it. A connection, a listener or any other value holding a descriptor that it is handed arrives as a descriptor of the worker’s own, naming the same socket, pipe or file, and the program may close its copy the moment the placement returns. What it emits holds no descriptor: its elements come back as bytes, so the compiler refuses such an element, naming where in it the descriptor stands.

Pipelines: stages that overlap in time

A stage is a generator whose parameter is itself a stream — it pulls its source, emits its answers. Generators composed these sequentially. Schedule them, and each stage is its own task pulling the stage before it across its own ring:

plain   := Numbers(4);
doubled := schedule Double(plain) buffer 4;
shifted := schedule AddOne(doubled) buffer 4;

A pipeline: each stage its own task
  4 values through two stages, sum 24

Same composition, now concurrent, by changing only the binds — and every ring bounded, so the whole pipeline flows at the pace of its slowest stage instead of ballooning between two fast ones.

One rule keeps this safe, and the compiler states it better than most documentation would. Hand a stream variable to a scheduled stage and that hand-off must be the variable’s only consuming appearance — a second pull, hand-off, or bind of the same variable is refused:

analyzer error 5411: the stream variable 's' feeds the scheduled producer
'Half', which pulls it from another task: that hand-off must be the
variable's only consuming appearance in this subprogram, so another pull,
hand-off, or bind of 's' would race the running producer over one frozen
activation — open the upstream directly inside the schedule's argument, or
give this consumer its own upstream

Every stage has exactly one puller; a chain is a line, never a diamond. That is the reason the pipeline needs no locks — there is nothing shared to lock.

select: whoever speaks first, and the timeout that rules silence

fast := schedule Quick(7) buffer 4;
slow := schedule Silent(7) buffer 4;
got := 0;
pump : while got < 2 do
begin
    select
        on v in fast do got := got + 1;
        on w in slow do got := got + 100;
        on timeout 2000 do leave pump;
    end;
    WriteLn("  heard %lld", v);
end;

select parks on several streams at once and runs the arm whose stream has a value. The bindings are declared variables, like every binding in the language; the timeout arm counts milliseconds against the monotonic clock and fires only if nothing arrived. The quick producer speaks twice; the silent one only ever yields; and a live stream that will never speak is exactly what the second select demonstrates:

Select: the ready arm wins, the timeout rules the silence
  heard 7
  heard 14
  quiet after 100 ms
  ready arms served 2 values

The quiet line is deterministic in the strongest sense: nothing can arrive, so no scheduling order can change when the timeout fires.

And the silent producer? It is still parked at a yield when the program body ends — so the binding scope’s end cancels it there, the same delivery-at-suspension-points rule every task lives under, and the program exits cleanly. Producer cleanup is structural, not a close() you must remember.

What the compiler (and runtime) refused

What this does not do

• No unbounded channels. Every ring is bounded; a producer parks rather than queue without limit. If you want an unbounded queue, that is a data structure you write and own, not a default you fall into.
• Fan-out is the ring’s, never the chain’s. One stream into N consumers — yes, when the variable is a task-fed ring: a stream of parameter stands on a task, and the schedule’s argument must be a ring bind (s := schedule G(...) buffer 8), because a ring’s pop hands each value to exactly one puller, however many tasks share it. A pull chain still cannot fan out — one frozen activation cannot serve two pulling tasks, and 5448 refuses the hand-off with the ring bind spelled out as the fix. (The exclusivity rule 5411 is a different wall, and a permanent one: it stops two consumers racing over one pull chain’s frozen activation, which no ring is.)
• Arrival order across producers is the scheduler’s. Per producer, order is program order; across fan-in producers, only completeness is promised.
• select is for streams. It parks on task-fed streams and a timeout — it is not a general event system, and there is no arm for “a task finished” (the block’s end already owns that story).

Try it

make -C mica-container/examples/ChannelStreams run

Add a third heard by making Quick emit three values — the pump loop’s condition is the only other line that knows the count. Then hand plain to a second scheduled stage and meet error 5411 in its full, explanatory form.

Next

With this pair the concurrency group tells both halves of the stream story: pull in generators, push here, over tasks that join structurally and carriers that decide the thread count at build time. The series continues with strings — the two-field value, the builder, and the encoding as a build-time choice.