Pipeline execution order

NOTE This is only theory right now. Full implementation will be later.

Sync

1-frame

Simplest example

"hello world" | stdout

All scalars are implicitly wrapped into 1-length array, so previous code is equivalent to

["hello world"] | stdout

This code will have coffee-script equivalent

for v in ["hello world"]
  console.log v

and can be optimized to

console.log "hello world"

because we detected that frame count == 1

Alias

You can use identifier for sources, sinks and intermediate connections. There are equivalent pipelines:

src = "hello world"
src | stdout

Same

"hello world" | dst
dst | stdout

Split

You can make some calculation once and reuse it with alias

gramm | (t)->t/1000 | kilogramm
gramm | (t)->t/100000 | ton

You can code less with joined pipes. Here is split example

gramm | (t)->t/1000 | kilogramm
      | (t)->t/100000 | ton

This can make your code 1 useless variable less.

Join

You can use 2 result from different alias

1 | a
2 | b
a+b | c

You can use join syntax

1 |
2 |
  | $1+$2 | c

This combos with split syntax

src | make_op1 |
    | make_op2 |
               | $1+$2 | dst

So you can make more visual-friendy diagramms. This code will have coffee-script equivalent

dst = []
for v in src
  tmp1 = make_op1 v
  tmp2 = make_op2 v
  dst.push tmp1+tmp2
  reduce_accumulator1 += v

BEWARE | symbol must be at same position. Tabs are not allowed. In strict mode | allowed only at %10==0 columns for forcing better formating.

Clocking

Clocking is important for understanding how pipeline will be unrolled into sequentian code. Let's try multiframe example

[1,2,3] | stdout

This code will have coffee-script equivalent

for v in [1,2,3]
  console.log v

We can add multiple functions. All functions by default accept 1 frame and produce 1 frame, so count frame in == count frame out.

[1,2,3] | (t)->t+1 | stdout

This code will have coffee-script equivalent

for v in [1,2,3]
  res2 = v+1
  console.log res2

Nothing special we just applied (t)->t+1 to all elements in array (in pipeline terms to all frames in stream). Let's meet reduce

[1,2,3] | reduce0 0, (a,b)->a+b | stdout
[1,2,3] | reduce0 0, (+) | stdout # same with syntax sugar

This code will have coffee-script equivalent

reduce_accumulator1 = 0
for v in [1,2,3]
  reduce_accumulator1 += v
console.log reduce_accumulator1

Note that some code generated before main loop and some after. That's because initial pipeline has different frame count on different stages.

[1,2,3] | reduce0 0, (a,b)->a+b | stdout
# 3     | 3                   1 | 1

Things became more complex with split/join.

[1,2,3] | reduce0 0, (+) | avg | 
        |                |     | (t)->t-avg | dst

This code will have coffee-script equivalent

reduce_accumulator1 = 0
src = [1,2,3]
for v in src
  reduce_accumulator1 += v
avg = reduce_accumulator1
dst = []
for v in src
  dst.push v-avg

Note that (t)->t-avg has exactly same frame count as [1,2,3], but other loop created. Join will check that all branches have same frame count. And only at this case it produce 1 loop. All other cases there would be multiple loops

Async

Any pipeline block produces a process that can be assigned to a variable. You can operate process with API.

proc = a | b | c
proc.start()
proc.pause()
proc.stop()

If pipeline is not assigned to any variable, it starts implicitly.\ Async makes this more interesting, but have pretty same syntax with a lot different code generated.

each 1h | "hello world" | stdout

Here we used new things. each is async function that use setInterval and emit frame each 1 hour. "hello world" const replaces current frame with const. CoffeeScript equivalent

setTimeout ()->
  console.log "hello world"
, 60*60*1000

using sink alias in non-pipeline code will force barrier creation. So you will wait until sink would be resolved

[1,2,3] | rate 1rps | dst
# some code
console.log dst

in CoffeeScript

src = [1,2,3]
anonymous_process = (on_end)->
  res = []
  for v in src
    dst.push v
    await setTimeout defer(), 1000
  on_end null res
# some code
await anonymous_process defer(err, dst); throw err if err
console.log dst

You can make lazy calculations with pipelines. Alias with no use will be cutted by compiler. Alias that are not 100% be accessed will call function-wrapper that have all needed logic only if that calculation is really needed.