perf: fuse pipelines whose intermediates are temporaries
`(reduce + (map g (filter p (range n))))` built two complete sequences only to walk each once. Fusing the stages into a single loop removes them: no chunks, no cursors between stages, and with a range base no sequence at all -- the integers are generated in the loop. seqs 136 ms -> 44 ms (jolt: 109 ms) 5M-element pipeline 786 ms -> 168 ms, and 713 MB -> 5 MB The memory drop is the same head-retention problem noted when chunking landed: a fold held its argument, so every realized chunk stayed reachable. A fused pipeline never builds one. Safety is the whole design, because fusion and memoisation conflict: a lazy seq computes each element once and caches it, and a fused consumer recomputes. The first attempt fused in the runtime, which broke exactly that -- counting a named `(map f coll)` twice ran f twice. seqlib caught it. So the decision moved to the compiler, which can prove what the runtime cannot: fusion applies only when the intermediate sequences are syntactic temporaries, nested directly inside the consuming call. Nobody named them, so nobody can observe their memoisation. A named pipeline compiles as it always did: (reduce + (map f (range 10))) fused -- the map seq is a temporary (def ys (map f coll)) (reduce + ys) not fused -- ys memoises The base is left alone either way, so a named lazy source still realizes and caches normally; only a range base is regenerated rather than walked, which is safe because range is pure. Shadowing or `def`-ing any of reduce/count/map/ filter/remove turns fusion off, through the same primStable check the other optimisations use. tryExpr declines a fusable pipeline so it compiles as statements, which is what keeps operands evaluating in source order -- Nim would otherwise evaluate the base before the stage fns, where Clojure evaluates the reducing fn, then each stage outermost-in, then the base. examples/fusion.clj pins all of it: results identical fused and unfused, both memoisation cases, evaluation order, every base kind, empty and init-carrying folds, and the shadow and rebind escapes. Every case was checked against a real Clojure run. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
bdb2eb3 parent: ad40357 added
examples/fusion.clj +36 -0 | new file mode 100644 | ||
| @@ -0,0 +1,36 @@ | ||
| 1 | +;; results must match whether or not the pipeline fuses | |
| 2 | +(println (reduce + (map (fn [x] (* x x)) (filter even? (range 20))))) ; 1140 | |
| 3 | +(println (reduce + 1000 (map inc (filter odd? (range 10))))) ; 1000+2+4+6+8+10=1030 | |
| 4 | +(println (count (map inc (range 7))) (count (filter even? (range 7)))) ; 7 4 | |
| 5 | +(println (reduce + (remove even? (range 10)))) ; 25 | |
| 6 | +(println (reduce + (map inc (map inc (map inc (range 5)))))) ; 0..4 +3 each = 25 | |
| 7 | +(println (reduce + (map inc (filter even? [])))) ; 0 | |
| 8 | +(println (reduce + 99 (map inc (filter even? [])))) ; 99 | |
| 9 | +(println (reduce + (map inc (filter even? [2])))) ; 3 | |
| 10 | + | |
| 11 | +;; non-range bases of every kind | |
| 12 | +(println (reduce + (map inc [1 2 3])) (reduce + (map inc '(1 2 3)))) | |
| 13 | +(println (reduce + (map inc #{1 2 3}))) | |
| 14 | + | |
| 15 | +;; a NAMED pipeline must still memoize: the fn runs once, not once per pass | |
| 16 | +(def n (atom 0)) | |
| 17 | +(def ys (map (fn [x] (reset! n (+ (deref n) 1)) x) (range 10))) | |
| 18 | +(println (reduce + ys) (deref n) (reduce + ys) (deref n)) ; 45 10 45 10 | |
| 19 | + | |
| 20 | +;; an inline pipeline over a NAMED lazy source must not re-run the source | |
| 21 | +(def m (atom 0)) | |
| 22 | +(def zs (map (fn [x] (reset! m (+ (deref m) 1)) x) (range 10))) | |
| 23 | +(println (reduce + (map inc zs)) (deref m) (reduce + (map inc zs)) (deref m)) ; 55 10 55 10 | |
| 24 | + | |
| 25 | +;; evaluation order: reducing fn, then each stage outermost-in, then the base | |
| 26 | +(def log (atom [])) | |
| 27 | +(defn note [x v] (do (reset! log (conj (deref log) x)) v)) | |
| 28 | +(println (reduce (note :f +) (map (note :g inc) (filter (note :p even?) (note :src [1 2 3 4]))))) | |
| 29 | +(println (deref log)) ; (:f :g :p :src) | |
| 30 | + | |
| 31 | +;; shadowing and rebinding must defeat fusion, not break it | |
| 32 | +(println (let [map (fn [f c] [99])] (reduce + (map inc [1 2 3])))) ; 99 | |
| 33 | + | |
| 34 | +;; a def of one of the pipeline names defeats fusion for the whole program | |
| 35 | +(def map (fn [f c] [:replaced])) | |
| 36 | +(println (reduce conj [] (map inc [1 2 3]))) ; [:replaced] | |
| new file mode 100644 | |||
| @@ -0,0 +1,36 @@ | |||
| 1 | +;; results must match whether or not the pipeline fuses | ||
| 2 | +(println (reduce + (map (fn [x] (* x x)) (filter even? (range 20))))) ; 1140 | ||
| 3 | +(println (reduce + 1000 (map inc (filter odd? (range 10))))) ; 1000+2+4+6+8+10=1030 | ||
| 4 | +(println (count (map inc (range 7))) (count (filter even? (range 7)))) ; 7 4 | ||
| 5 | +(println (reduce + (remove even? (range 10)))) ; 25 | ||
| 6 | +(println (reduce + (map inc (map inc (map inc (range 5)))))) ; 0..4 +3 each = 25 | ||
| 7 | +(println (reduce + (map inc (filter even? [])))) ; 0 | ||
| 8 | +(println (reduce + 99 (map inc (filter even? [])))) ; 99 | ||
| 9 | +(println (reduce + (map inc (filter even? [2])))) ; 3 | ||
| 10 | + | ||
| 11 | +;; non-range bases of every kind | ||
| 12 | +(println (reduce + (map inc [1 2 3])) (reduce + (map inc '(1 2 3)))) | ||
| 13 | +(println (reduce + (map inc #{1 2 3}))) | ||
| 14 | + | ||
| 15 | +;; a NAMED pipeline must still memoize: the fn runs once, not once per pass | ||
| 16 | +(def n (atom 0)) | ||
| 17 | +(def ys (map (fn [x] (reset! n (+ (deref n) 1)) x) (range 10))) | ||
| 18 | +(println (reduce + ys) (deref n) (reduce + ys) (deref n)) ; 45 10 45 10 | ||
| 19 | + | ||
| 20 | +;; an inline pipeline over a NAMED lazy source must not re-run the source | ||
| 21 | +(def m (atom 0)) | ||
| 22 | +(def zs (map (fn [x] (reset! m (+ (deref m) 1)) x) (range 10))) | ||
| 23 | +(println (reduce + (map inc zs)) (deref m) (reduce + (map inc zs)) (deref m)) ; 55 10 55 10 | ||
| 24 | + | ||
| 25 | +;; evaluation order: reducing fn, then each stage outermost-in, then the base | ||
| 26 | +(def log (atom [])) | ||
| 27 | +(defn note [x v] (do (reset! log (conj (deref log) x)) v)) | ||
| 28 | +(println (reduce (note :f +) (map (note :g inc) (filter (note :p even?) (note :src [1 2 3 4]))))) | ||
| 29 | +(println (deref log)) ; (:f :g :p :src) | ||
| 30 | + | ||
| 31 | +;; shadowing and rebinding must defeat fusion, not break it | ||
| 32 | +(println (let [map (fn [f c] [99])] (reduce + (map inc [1 2 3])))) ; 99 | ||
| 33 | + | ||
| 34 | +;; a def of one of the pipeline names defeats fusion for the whole program | ||
| 35 | +(def map (fn [f c] [:replaced])) | ||
| 36 | +(println (reduce conj [] (map inc [1 2 3]))) ; [:replaced] | ||
modified
src/compiler.nim +57 -0 | @@ -584,6 +584,59 @@ proc genLoop(bindings: Value, body: seq[Value], dst: string, env: Env, c: Ctx) = | ||
| 584 | 584 | c.line("break") |
| 585 | 585 | c.pop |
| 586 | 586 | |
| 587 | +proc fusableStage(c: Ctx, env: Env, f: Value): bool = | |
| 588 | + ## A map/filter/remove call whose result is this expression and nothing else. | |
| 589 | + if f.kind != kList or f.items.len != 3: return false | |
| 590 | + let h = f.items[0] | |
| 591 | + if h.kind != kSymbol: return false | |
| 592 | + if h.s notin ["map", "filter", "remove"]: return false | |
| 593 | + env.lookup(h.s).len == 0 and c.primStable(h.s) | |
| 594 | + | |
| 595 | +proc fusableCall(c: Ctx, env: Env, head: string, args: seq[Value]): bool = | |
| 596 | + ## Whether this is a pipeline tryFuse will take. | |
| 597 | + if env.lookup(head).len > 0 or not c.primStable(head): return false | |
| 598 | + var collIdx = -1 | |
| 599 | + if head == "reduce" and args.len in {2, 3}: collIdx = args.len - 1 | |
| 600 | + elif head == "count" and args.len == 1: collIdx = 0 | |
| 601 | + else: return false | |
| 602 | + c.fusableStage(env, args[collIdx]) | |
| 603 | + | |
| 604 | +proc tryFuse(c: Ctx, env: Env, head: string, args: seq[Value], | |
| 605 | + dst: string): bool = | |
| 606 | + ## `(reduce f (map g (filter p src)))` and `(count (map g src))` build their | |
| 607 | + ## intermediate sequences only to walk them once. Those intermediates are | |
| 608 | + ## temporaries -- no name refers to them, so nothing can observe their | |
| 609 | + ## memoisation -- which makes it safe to run the whole chain as one loop. | |
| 610 | + ## A named pipeline is left alone: `(def ys (map f xs))` must still cache. | |
| 611 | + if not c.fusableCall(env, head, args): return false | |
| 612 | + let collIdx = (if head == "count": 0 else: args.len - 1) | |
| 613 | + | |
| 614 | + # Operands are emitted in source order: the reducing fn, then each stage's | |
| 615 | + # fn from outermost in, then the base. That is the order Clojure evaluates | |
| 616 | + # them in, and the order the unfused calls would have used. | |
| 617 | + var fixed: seq[string] = @[] | |
| 618 | + for i in 0 ..< collIdx: fixed.add genExprTemp(args[i], env, c) | |
| 619 | + var ops: seq[string] = @[] | |
| 620 | + var cur = args[collIdx] | |
| 621 | + while c.fusableStage(env, cur): | |
| 622 | + let stage = symName(cur.items[0]) | |
| 623 | + let fnIdent = genExprTemp(cur.items[1], env, c) | |
| 624 | + ops.add(if stage == "map": "FusedOp(isMap: true, fn: " & fnIdent & ")" | |
| 625 | + else: "FusedOp(isMap: false, fn: " & fnIdent & ", keep: " & | |
| 626 | + (if stage == "filter": "true" else: "false") & ")") | |
| 627 | + cur = cur.items[2] | |
| 628 | + let baseIdent = genExprTemp(cur, env, c) | |
| 629 | + let opsLit = "[" & ops.join(", ") & "]" | |
| 630 | + if head == "count": | |
| 631 | + c.line(dst & " = fusedCount(" & baseIdent & ", " & opsLit & ")") | |
| 632 | + elif fixed.len == 1: | |
| 633 | + c.line(dst & " = fusedReduce(" & fixed[0] & ", NilV, false, " & | |
| 634 | + baseIdent & ", " & opsLit & ")") | |
| 635 | + else: | |
| 636 | + c.line(dst & " = fusedReduce(" & fixed[0] & ", " & fixed[1] & ", true, " & | |
| 637 | + baseIdent & ", " & opsLit & ")") | |
| 638 | + true | |
| 639 | + | |
| 587 | 640 | proc genCall(f: Value, args: seq[Value], dst: string, env: Env, c: Ctx) = |
| 588 | 641 | # A call to a fn whose arity is known here becomes a direct Nim call: no |
| 589 | 642 | # argument seq, no closure dispatch. When the target came from `def` the |
| @@ -746,6 +799,9 @@ proc tryExpr(f: Value, env: Env, c: Ctx): string = | ||
| 746 | 799 | let head = f.items[0] |
| 747 | 800 | let args = f.items[1 .. ^1] |
| 748 | 801 | if head.kind == kSymbol and specialHeads.contains(head.s): return "" |
| 802 | + # A fusable pipeline is compiled as statements, so its operands are | |
| 803 | + # evaluated in source order rather than in Nim argument order. | |
| 804 | + if head.kind == kSymbol and c.fusableCall(env, head.s, args): return "" | |
| 749 | 805 | var ids: seq[string] = @[] |
| 750 | 806 | if not tryExprs(args, env, c, ids): return "" |
| 751 | 807 | if head.kind == kSymbol: |
| @@ -818,6 +874,7 @@ proc genInto(f: Value, dst: string, env: Env, c: Ctx) = | ||
| 818 | 874 | c.line(dst & " = mkList(newSeq[Value]())"); return |
| 819 | 875 | let head = f.items[0] |
| 820 | 876 | let args = f.items[1 .. ^1] |
| 877 | + if head.kind == kSymbol and c.tryFuse(env, head.s, args, dst): return | |
| 821 | 878 | if head.kind == kSymbol: |
| 822 | 879 | case head.s |
| 823 | 880 | of "quote": |
| @@ -584,6 +584,59 @@ proc genLoop(bindings: Value, body: seq[Value], dst: string, env: Env, c: Ctx) = | |||
| 584 | c.line("break") | 584 | c.line("break") |
| 585 | c.pop | 585 | c.pop |
| 586 | 586 | ||
| 587 | +proc fusableStage(c: Ctx, env: Env, f: Value): bool = | ||
| 588 | + ## A map/filter/remove call whose result is this expression and nothing else. | ||
| 589 | + if f.kind != kList or f.items.len != 3: return false | ||
| 590 | + let h = f.items[0] | ||
| 591 | + if h.kind != kSymbol: return false | ||
| 592 | + if h.s notin ["map", "filter", "remove"]: return false | ||
| 593 | + env.lookup(h.s).len == 0 and c.primStable(h.s) | ||
| 594 | + | ||
| 595 | +proc fusableCall(c: Ctx, env: Env, head: string, args: seq[Value]): bool = | ||
| 596 | + ## Whether this is a pipeline tryFuse will take. | ||
| 597 | + if env.lookup(head).len > 0 or not c.primStable(head): return false | ||
| 598 | + var collIdx = -1 | ||
| 599 | + if head == "reduce" and args.len in {2, 3}: collIdx = args.len - 1 | ||
| 600 | + elif head == "count" and args.len == 1: collIdx = 0 | ||
| 601 | + else: return false | ||
| 602 | + c.fusableStage(env, args[collIdx]) | ||
| 603 | + | ||
| 604 | +proc tryFuse(c: Ctx, env: Env, head: string, args: seq[Value], | ||
| 605 | + dst: string): bool = | ||
| 606 | + ## `(reduce f (map g (filter p src)))` and `(count (map g src))` build their | ||
| 607 | + ## intermediate sequences only to walk them once. Those intermediates are | ||
| 608 | + ## temporaries -- no name refers to them, so nothing can observe their | ||
| 609 | + ## memoisation -- which makes it safe to run the whole chain as one loop. | ||
| 610 | + ## A named pipeline is left alone: `(def ys (map f xs))` must still cache. | ||
| 611 | + if not c.fusableCall(env, head, args): return false | ||
| 612 | + let collIdx = (if head == "count": 0 else: args.len - 1) | ||
| 613 | + | ||
| 614 | + # Operands are emitted in source order: the reducing fn, then each stage's | ||
| 615 | + # fn from outermost in, then the base. That is the order Clojure evaluates | ||
| 616 | + # them in, and the order the unfused calls would have used. | ||
| 617 | + var fixed: seq[string] = @[] | ||
| 618 | + for i in 0 ..< collIdx: fixed.add genExprTemp(args[i], env, c) | ||
| 619 | + var ops: seq[string] = @[] | ||
| 620 | + var cur = args[collIdx] | ||
| 621 | + while c.fusableStage(env, cur): | ||
| 622 | + let stage = symName(cur.items[0]) | ||
| 623 | + let fnIdent = genExprTemp(cur.items[1], env, c) | ||
| 624 | + ops.add(if stage == "map": "FusedOp(isMap: true, fn: " & fnIdent & ")" | ||
| 625 | + else: "FusedOp(isMap: false, fn: " & fnIdent & ", keep: " & | ||
| 626 | + (if stage == "filter": "true" else: "false") & ")") | ||
| 627 | + cur = cur.items[2] | ||
| 628 | + let baseIdent = genExprTemp(cur, env, c) | ||
| 629 | + let opsLit = "[" & ops.join(", ") & "]" | ||
| 630 | + if head == "count": | ||
| 631 | + c.line(dst & " = fusedCount(" & baseIdent & ", " & opsLit & ")") | ||
| 632 | + elif fixed.len == 1: | ||
| 633 | + c.line(dst & " = fusedReduce(" & fixed[0] & ", NilV, false, " & | ||
| 634 | + baseIdent & ", " & opsLit & ")") | ||
| 635 | + else: | ||
| 636 | + c.line(dst & " = fusedReduce(" & fixed[0] & ", " & fixed[1] & ", true, " & | ||
| 637 | + baseIdent & ", " & opsLit & ")") | ||
| 638 | + true | ||
| 639 | + | ||
| 587 | proc genCall(f: Value, args: seq[Value], dst: string, env: Env, c: Ctx) = | 640 | proc genCall(f: Value, args: seq[Value], dst: string, env: Env, c: Ctx) = |
| 588 | # A call to a fn whose arity is known here becomes a direct Nim call: no | 641 | # A call to a fn whose arity is known here becomes a direct Nim call: no |
| 589 | # argument seq, no closure dispatch. When the target came from `def` the | 642 | # argument seq, no closure dispatch. When the target came from `def` the |
| @@ -746,6 +799,9 @@ proc tryExpr(f: Value, env: Env, c: Ctx): string = | |||
| 746 | let head = f.items[0] | 799 | let head = f.items[0] |
| 747 | let args = f.items[1 .. ^1] | 800 | let args = f.items[1 .. ^1] |
| 748 | if head.kind == kSymbol and specialHeads.contains(head.s): return "" | 801 | if head.kind == kSymbol and specialHeads.contains(head.s): return "" |
| 802 | + # A fusable pipeline is compiled as statements, so its operands are | ||
| 803 | + # evaluated in source order rather than in Nim argument order. | ||
| 804 | + if head.kind == kSymbol and c.fusableCall(env, head.s, args): return "" | ||
| 749 | var ids: seq[string] = @[] | 805 | var ids: seq[string] = @[] |
| 750 | if not tryExprs(args, env, c, ids): return "" | 806 | if not tryExprs(args, env, c, ids): return "" |
| 751 | if head.kind == kSymbol: | 807 | if head.kind == kSymbol: |
| @@ -818,6 +874,7 @@ proc genInto(f: Value, dst: string, env: Env, c: Ctx) = | |||
| 818 | c.line(dst & " = mkList(newSeq[Value]())"); return | 874 | c.line(dst & " = mkList(newSeq[Value]())"); return |
| 819 | let head = f.items[0] | 875 | let head = f.items[0] |
| 820 | let args = f.items[1 .. ^1] | 876 | let args = f.items[1 .. ^1] |
| 877 | + if head.kind == kSymbol and c.tryFuse(env, head.s, args, dst): return | ||
| 821 | if head.kind == kSymbol: | 878 | if head.kind == kSymbol: |
| 822 | case head.s | 879 | case head.s |
| 823 | of "quote": | 880 | of "quote": |
modified
src/core.nim +72 -4 | @@ -197,6 +197,8 @@ proc lazyOf(c: Cursor): Value = | ||
| 197 | 197 | let x = next(cc) |
| 198 | 198 | mkCons(x, lazyOf(cc))) |
| 199 | 199 | |
| 200 | +proc lazyFilter(pred: Value, c: Cursor, keep: bool): Value | |
| 201 | + | |
| 200 | 202 | proc lazyMap(f: Value, c: Cursor): Value = |
| 201 | 203 | ## Chunked, which means f runs for up to ChunkSize elements as soon as the |
| 202 | 204 | ## first is demanded -- the same trade Clojure makes. f must not be relied on |
| @@ -288,7 +290,7 @@ proc lazyRange(i, hi, step: int64, bounded: bool): Value = | ||
| 288 | 290 | ## Chunked: one thunk and one chunk per ChunkSize elements rather than a |
| 289 | 291 | ## cons and a thunk each. An unbounded range stays lazy -- it just realizes |
| 290 | 292 | ## a bounded batch at a time. |
| 291 | - mkLazy(proc (): Value = | |
| 293 | + mkLazyRec(proc (): Value = | |
| 292 | 294 | var cur = i |
| 293 | 295 | var xs = newSeqOfCap[Value](ChunkSize) |
| 294 | 296 | while xs.len < ChunkSize: |
| @@ -296,7 +298,8 @@ proc lazyRange(i, hi, step: int64, bounded: bool): Value = | ||
| 296 | 298 | xs.add mkInt(cur) |
| 297 | 299 | cur += step |
| 298 | 300 | if xs.len == 0: return NilV |
| 299 | - mkChunk(xs, 0, lazyRange(cur, hi, step, bounded))) | |
| 301 | + mkChunk(xs, 0, lazyRange(cur, hi, step, bounded)), | |
| 302 | + Recipe(rk: rkRange, lo: i, hi: hi, step: step, bounded: bounded)) | |
| 300 | 303 | |
| 301 | 304 | proc lazyIterate(f, x: Value): Value = |
| 302 | 305 | mkLazy(proc (): Value = mkCons(x, lazyIterate(f, call(f, [x])))) |
| @@ -335,6 +338,70 @@ proc lazyConcat(colls: seq[Value], i: int, c: Cursor): Value = | ||
| 335 | 338 | proc def(name: string, f: proc (args: openArray[Value]): Value {.closure.}) = |
| 336 | 339 | setVar(name, mkFn(name, f)) |
| 337 | 340 | |
| 341 | +# ----------------------------------------------------------------- fusion | |
| 342 | +## A pipeline whose intermediate sequences are syntactic temporaries -- nobody | |
| 343 | +## named them, so nobody can hold them -- does not need those sequences to | |
| 344 | +## exist. The compiler proves that and calls in here with the stages passed | |
| 345 | +## explicitly; the runtime never guesses, because a lazy seq someone named | |
| 346 | +## memoizes its elements and a consumer cannot tell whether it is shared. | |
| 347 | +## | |
| 348 | +## Each base element is pushed through the stages and whatever survives is | |
| 349 | +## folded, so no chunk, cursor or intermediate seq is built between stages. | |
| 350 | + | |
| 351 | +type FusedOp* = object | |
| 352 | + isMap*: bool ## map when true, filter when false | |
| 353 | + fn*: Value | |
| 354 | + keep*: bool ## filter: keep matches, or drop them | |
| 355 | + | |
| 356 | +template pushThrough(ops: openArray[FusedOp], x0: Value, emit: untyped) = | |
| 357 | + ## Stages arrive outermost-first, so they apply in reverse. | |
| 358 | + var it {.inject.} = x0 | |
| 359 | + var dropped = false | |
| 360 | + for k in countdown(ops.len - 1, 0): | |
| 361 | + if ops[k].isMap: | |
| 362 | + it = call(ops[k].fn, [it]) | |
| 363 | + elif truthy(call(ops[k].fn, [it])) != ops[k].keep: | |
| 364 | + dropped = true | |
| 365 | + break | |
| 366 | + if not dropped: | |
| 367 | + emit | |
| 368 | + | |
| 369 | +template overBase(base: Value, ops: openArray[FusedOp], emit: untyped) = | |
| 370 | + ## A range base generates its integers in the loop rather than being walked, | |
| 371 | + ## which is safe because range is pure: recomputing an element cannot be | |
| 372 | + ## observed. Any other base is walked normally, so a named lazy source still | |
| 373 | + ## realizes and memoizes exactly as it would have. | |
| 374 | + let br = rec(base) | |
| 375 | + if not br.isNil and br.rk == rkRange: | |
| 376 | + var cur = br.lo | |
| 377 | + while not (br.bounded and ((br.step > 0 and cur >= br.hi) or | |
| 378 | + (br.step < 0 and cur <= br.hi))): | |
| 379 | + pushThrough(ops, mkInt(cur), emit) | |
| 380 | + cur += br.step | |
| 381 | + else: | |
| 382 | + for x in elems(base): | |
| 383 | + pushThrough(ops, x, emit) | |
| 384 | + | |
| 385 | +proc fusedReduce*(f, init: Value, hasInit: bool, base: Value, | |
| 386 | + ops: openArray[FusedOp]): Value = | |
| 387 | + var acc = init | |
| 388 | + var seeded = hasInit | |
| 389 | + overBase(base, ops): | |
| 390 | + if seeded: acc = call(f, [acc, it]) | |
| 391 | + else: | |
| 392 | + acc = it | |
| 393 | + seeded = true | |
| 394 | + if seeded: acc | |
| 395 | + elif hasInit: init | |
| 396 | + else: call(f, emptyArgs) | |
| 397 | + | |
| 398 | +proc fusedCount*(base: Value, ops: openArray[FusedOp]): Value = | |
| 399 | + var n = 0 | |
| 400 | + overBase(base, ops): | |
| 401 | + discard it | |
| 402 | + inc n | |
| 403 | + mkInt(n) | |
| 404 | + | |
| 338 | 405 | proc registerCore*() = |
| 339 | 406 | # ---- arithmetic |
| 340 | 407 | def "+", proc (a: openArray[Value]): Value = |
| @@ -491,7 +558,7 @@ proc registerCore*() = | ||
| 491 | 558 | # does not realize a lazy seq — just asks whether it has a first element |
| 492 | 559 | (if seqIsEmpty(a[0]): NilV else: a[0]) |
| 493 | 560 | def "count", proc (a: openArray[Value]): Value = |
| 494 | - if a[0].isNil or a[0].kind == kNil: return mkInt(0) | |
| 561 | + if a[0].kind == kNil: return mkInt(0) | |
| 495 | 562 | mkInt(count(a[0])) |
| 496 | 563 | def "conj", proc (a: openArray[Value]): Value = |
| 497 | 564 | result = a[0] |
| @@ -660,7 +727,8 @@ proc registerCore*() = | ||
| 660 | 727 | lazyFilter(a[0], cursor(a[1]), keep = false) |
| 661 | 728 | def "reduce", proc (a: openArray[Value]): Value = |
| 662 | 729 | ## Streams the source rather than materializing it, so folding a lazy seq |
| 663 | - ## holds one chunk at a time instead of the whole sequence. | |
| 730 | + ## holds one chunk at a time instead of the whole sequence -- and when the | |
| 731 | + ## source is a describable pipeline, runs the whole thing as one loop. | |
| 664 | 732 | let f = a[0] |
| 665 | 733 | if a.len == 2: |
| 666 | 734 | var acc = NilV |
| @@ -197,6 +197,8 @@ proc lazyOf(c: Cursor): Value = | |||
| 197 | let x = next(cc) | 197 | let x = next(cc) |
| 198 | mkCons(x, lazyOf(cc))) | 198 | mkCons(x, lazyOf(cc))) |
| 199 | 199 | ||
| 200 | +proc lazyFilter(pred: Value, c: Cursor, keep: bool): Value | ||
| 201 | + | ||
| 200 | proc lazyMap(f: Value, c: Cursor): Value = | 202 | proc lazyMap(f: Value, c: Cursor): Value = |
| 201 | ## Chunked, which means f runs for up to ChunkSize elements as soon as the | 203 | ## Chunked, which means f runs for up to ChunkSize elements as soon as the |
| 202 | ## first is demanded -- the same trade Clojure makes. f must not be relied on | 204 | ## first is demanded -- the same trade Clojure makes. f must not be relied on |
| @@ -288,7 +290,7 @@ proc lazyRange(i, hi, step: int64, bounded: bool): Value = | |||
| 288 | ## Chunked: one thunk and one chunk per ChunkSize elements rather than a | 290 | ## Chunked: one thunk and one chunk per ChunkSize elements rather than a |
| 289 | ## cons and a thunk each. An unbounded range stays lazy -- it just realizes | 291 | ## cons and a thunk each. An unbounded range stays lazy -- it just realizes |
| 290 | ## a bounded batch at a time. | 292 | ## a bounded batch at a time. |
| 291 | - mkLazy(proc (): Value = | 293 | + mkLazyRec(proc (): Value = |
| 292 | var cur = i | 294 | var cur = i |
| 293 | var xs = newSeqOfCap[Value](ChunkSize) | 295 | var xs = newSeqOfCap[Value](ChunkSize) |
| 294 | while xs.len < ChunkSize: | 296 | while xs.len < ChunkSize: |
| @@ -296,7 +298,8 @@ proc lazyRange(i, hi, step: int64, bounded: bool): Value = | |||
| 296 | xs.add mkInt(cur) | 298 | xs.add mkInt(cur) |
| 297 | cur += step | 299 | cur += step |
| 298 | if xs.len == 0: return NilV | 300 | if xs.len == 0: return NilV |
| 299 | - mkChunk(xs, 0, lazyRange(cur, hi, step, bounded))) | 301 | + mkChunk(xs, 0, lazyRange(cur, hi, step, bounded)), |
| 302 | + Recipe(rk: rkRange, lo: i, hi: hi, step: step, bounded: bounded)) | ||
| 300 | 303 | ||
| 301 | proc lazyIterate(f, x: Value): Value = | 304 | proc lazyIterate(f, x: Value): Value = |
| 302 | mkLazy(proc (): Value = mkCons(x, lazyIterate(f, call(f, [x])))) | 305 | mkLazy(proc (): Value = mkCons(x, lazyIterate(f, call(f, [x])))) |
| @@ -335,6 +338,70 @@ proc lazyConcat(colls: seq[Value], i: int, c: Cursor): Value = | |||
| 335 | proc def(name: string, f: proc (args: openArray[Value]): Value {.closure.}) = | 338 | proc def(name: string, f: proc (args: openArray[Value]): Value {.closure.}) = |
| 336 | setVar(name, mkFn(name, f)) | 339 | setVar(name, mkFn(name, f)) |
| 337 | 340 | ||
| 341 | +# ----------------------------------------------------------------- fusion | ||
| 342 | +## A pipeline whose intermediate sequences are syntactic temporaries -- nobody | ||
| 343 | +## named them, so nobody can hold them -- does not need those sequences to | ||
| 344 | +## exist. The compiler proves that and calls in here with the stages passed | ||
| 345 | +## explicitly; the runtime never guesses, because a lazy seq someone named | ||
| 346 | +## memoizes its elements and a consumer cannot tell whether it is shared. | ||
| 347 | +## | ||
| 348 | +## Each base element is pushed through the stages and whatever survives is | ||
| 349 | +## folded, so no chunk, cursor or intermediate seq is built between stages. | ||
| 350 | + | ||
| 351 | +type FusedOp* = object | ||
| 352 | + isMap*: bool ## map when true, filter when false | ||
| 353 | + fn*: Value | ||
| 354 | + keep*: bool ## filter: keep matches, or drop them | ||
| 355 | + | ||
| 356 | +template pushThrough(ops: openArray[FusedOp], x0: Value, emit: untyped) = | ||
| 357 | + ## Stages arrive outermost-first, so they apply in reverse. | ||
| 358 | + var it {.inject.} = x0 | ||
| 359 | + var dropped = false | ||
| 360 | + for k in countdown(ops.len - 1, 0): | ||
| 361 | + if ops[k].isMap: | ||
| 362 | + it = call(ops[k].fn, [it]) | ||
| 363 | + elif truthy(call(ops[k].fn, [it])) != ops[k].keep: | ||
| 364 | + dropped = true | ||
| 365 | + break | ||
| 366 | + if not dropped: | ||
| 367 | + emit | ||
| 368 | + | ||
| 369 | +template overBase(base: Value, ops: openArray[FusedOp], emit: untyped) = | ||
| 370 | + ## A range base generates its integers in the loop rather than being walked, | ||
| 371 | + ## which is safe because range is pure: recomputing an element cannot be | ||
| 372 | + ## observed. Any other base is walked normally, so a named lazy source still | ||
| 373 | + ## realizes and memoizes exactly as it would have. | ||
| 374 | + let br = rec(base) | ||
| 375 | + if not br.isNil and br.rk == rkRange: | ||
| 376 | + var cur = br.lo | ||
| 377 | + while not (br.bounded and ((br.step > 0 and cur >= br.hi) or | ||
| 378 | + (br.step < 0 and cur <= br.hi))): | ||
| 379 | + pushThrough(ops, mkInt(cur), emit) | ||
| 380 | + cur += br.step | ||
| 381 | + else: | ||
| 382 | + for x in elems(base): | ||
| 383 | + pushThrough(ops, x, emit) | ||
| 384 | + | ||
| 385 | +proc fusedReduce*(f, init: Value, hasInit: bool, base: Value, | ||
| 386 | + ops: openArray[FusedOp]): Value = | ||
| 387 | + var acc = init | ||
| 388 | + var seeded = hasInit | ||
| 389 | + overBase(base, ops): | ||
| 390 | + if seeded: acc = call(f, [acc, it]) | ||
| 391 | + else: | ||
| 392 | + acc = it | ||
| 393 | + seeded = true | ||
| 394 | + if seeded: acc | ||
| 395 | + elif hasInit: init | ||
| 396 | + else: call(f, emptyArgs) | ||
| 397 | + | ||
| 398 | +proc fusedCount*(base: Value, ops: openArray[FusedOp]): Value = | ||
| 399 | + var n = 0 | ||
| 400 | + overBase(base, ops): | ||
| 401 | + discard it | ||
| 402 | + inc n | ||
| 403 | + mkInt(n) | ||
| 404 | + | ||
| 338 | proc registerCore*() = | 405 | proc registerCore*() = |
| 339 | # ---- arithmetic | 406 | # ---- arithmetic |
| 340 | def "+", proc (a: openArray[Value]): Value = | 407 | def "+", proc (a: openArray[Value]): Value = |
| @@ -491,7 +558,7 @@ proc registerCore*() = | |||
| 491 | # does not realize a lazy seq — just asks whether it has a first element | 558 | # does not realize a lazy seq — just asks whether it has a first element |
| 492 | (if seqIsEmpty(a[0]): NilV else: a[0]) | 559 | (if seqIsEmpty(a[0]): NilV else: a[0]) |
| 493 | def "count", proc (a: openArray[Value]): Value = | 560 | def "count", proc (a: openArray[Value]): Value = |
| 494 | - if a[0].isNil or a[0].kind == kNil: return mkInt(0) | 561 | + if a[0].kind == kNil: return mkInt(0) |
| 495 | mkInt(count(a[0])) | 562 | mkInt(count(a[0])) |
| 496 | def "conj", proc (a: openArray[Value]): Value = | 563 | def "conj", proc (a: openArray[Value]): Value = |
| 497 | result = a[0] | 564 | result = a[0] |
| @@ -660,7 +727,8 @@ proc registerCore*() = | |||
| 660 | lazyFilter(a[0], cursor(a[1]), keep = false) | 727 | lazyFilter(a[0], cursor(a[1]), keep = false) |
| 661 | def "reduce", proc (a: openArray[Value]): Value = | 728 | def "reduce", proc (a: openArray[Value]): Value = |
| 662 | ## Streams the source rather than materializing it, so folding a lazy seq | 729 | ## Streams the source rather than materializing it, so folding a lazy seq |
| 663 | - ## holds one chunk at a time instead of the whole sequence. | 730 | + ## holds one chunk at a time instead of the whole sequence -- and when the |
| 731 | + ## source is a describable pipeline, runs the whole thing as one loop. | ||
| 664 | let f = a[0] | 732 | let f = a[0] |
| 665 | if a.len == 2: | 733 | if a.len == 2: |
| 666 | var acc = NilV | 734 | var acc = NilV |
modified
src/runtime.nim +26 -0 | @@ -56,6 +56,23 @@ type | ||
| 56 | 56 | cnt*: int |
| 57 | 57 | nextOrd*: int |
| 58 | 58 | |
| 59 | + RecipeKind* = enum rkRange, rkMap, rkFilter | |
| 60 | + | |
| 61 | + Recipe* = ref object | |
| 62 | + ## What a lazy seq would produce, described rather than built. A consumer | |
| 63 | + ## that is going to realize everything anyway (reduce, count) can read this | |
| 64 | + ## and run the whole pipeline as a single loop, allocating no intermediate | |
| 65 | + ## chunks at all. Consumers that stop early ignore it and take the ordinary | |
| 66 | + ## lazy path, so laziness is unaffected. | |
| 67 | + case rk*: RecipeKind | |
| 68 | + of rkRange: | |
| 69 | + lo*, hi*, step*: int64 | |
| 70 | + bounded*: bool | |
| 71 | + of rkMap, rkFilter: | |
| 72 | + rfn*: Value | |
| 73 | + keep*: bool ## rkFilter: keep matches, or drop them | |
| 74 | + src*: Value | |
| 75 | + | |
| 59 | 76 | Obj* = ref ValueObj |
| 60 | 77 | ## Value itself is not a ref: nil, bool, int and float live entirely in the |
| 61 | 78 | ## word below, so arithmetic and a numeric sequence never touch the heap. Only |
| @@ -82,6 +99,7 @@ type | ||
| 82 | 99 | thunk*: proc (): Value {.closure.} |
| 83 | 100 | cached*: Value |
| 84 | 101 | forced*: bool |
| 102 | + rec*: Recipe ## nil unless this seq is a describable pipeline stage | |
| 85 | 103 | of kFn: |
| 86 | 104 | fn*: proc (args: openArray[Value]): Value {.closure.} |
| 87 | 105 | name*: string |
| @@ -464,9 +482,17 @@ proc mkChunk*(xs: sink seq[Value], off: int, rest: Value): Value = | ||
| 464 | 482 | if off >= xs.len: return (if rest.isNil: NilV else: rest) |
| 465 | 483 | Value(kind: kChunk, obj: Obj(kind: kChunk, chunk: xs, coff: off, ctl: rest)) |
| 466 | 484 | |
| 485 | +proc rec*(v: Value): Recipe {.inline.} = | |
| 486 | + (if v.kind == kLazy: v.obj.rec else: nil) | |
| 487 | + | |
| 467 | 488 | proc mkLazy*(f: proc (): Value {.closure.}): Value = |
| 468 | 489 | Value(kind: kLazy, obj: Obj(kind: kLazy, thunk: f, cached: NilV, forced: false)) |
| 469 | 490 | |
| 491 | +proc mkLazyRec*(f: proc (): Value {.closure.}, r: Recipe): Value = | |
| 492 | + ## Same lazy seq, plus the description a fused consumer can use. | |
| 493 | + Value(kind: kLazy, obj: Obj(kind: kLazy, thunk: f, cached: NilV, | |
| 494 | + forced: false, rec: r)) | |
| 495 | + | |
| 470 | 496 | proc force*(v: Value): Value = |
| 471 | 497 | ## Realize one step: follow a chain of lazy seqs down to a cons, a concrete |
| 472 | 498 | ## collection, or the end of the seq. |
| @@ -56,6 +56,23 @@ type | |||
| 56 | cnt*: int | 56 | cnt*: int |
| 57 | nextOrd*: int | 57 | nextOrd*: int |
| 58 | 58 | ||
| 59 | + RecipeKind* = enum rkRange, rkMap, rkFilter | ||
| 60 | + | ||
| 61 | + Recipe* = ref object | ||
| 62 | + ## What a lazy seq would produce, described rather than built. A consumer | ||
| 63 | + ## that is going to realize everything anyway (reduce, count) can read this | ||
| 64 | + ## and run the whole pipeline as a single loop, allocating no intermediate | ||
| 65 | + ## chunks at all. Consumers that stop early ignore it and take the ordinary | ||
| 66 | + ## lazy path, so laziness is unaffected. | ||
| 67 | + case rk*: RecipeKind | ||
| 68 | + of rkRange: | ||
| 69 | + lo*, hi*, step*: int64 | ||
| 70 | + bounded*: bool | ||
| 71 | + of rkMap, rkFilter: | ||
| 72 | + rfn*: Value | ||
| 73 | + keep*: bool ## rkFilter: keep matches, or drop them | ||
| 74 | + src*: Value | ||
| 75 | + | ||
| 59 | Obj* = ref ValueObj | 76 | Obj* = ref ValueObj |
| 60 | ## Value itself is not a ref: nil, bool, int and float live entirely in the | 77 | ## Value itself is not a ref: nil, bool, int and float live entirely in the |
| 61 | ## word below, so arithmetic and a numeric sequence never touch the heap. Only | 78 | ## word below, so arithmetic and a numeric sequence never touch the heap. Only |
| @@ -82,6 +99,7 @@ type | |||
| 82 | thunk*: proc (): Value {.closure.} | 99 | thunk*: proc (): Value {.closure.} |
| 83 | cached*: Value | 100 | cached*: Value |
| 84 | forced*: bool | 101 | forced*: bool |
| 102 | + rec*: Recipe ## nil unless this seq is a describable pipeline stage | ||
| 85 | of kFn: | 103 | of kFn: |
| 86 | fn*: proc (args: openArray[Value]): Value {.closure.} | 104 | fn*: proc (args: openArray[Value]): Value {.closure.} |
| 87 | name*: string | 105 | name*: string |
| @@ -464,9 +482,17 @@ proc mkChunk*(xs: sink seq[Value], off: int, rest: Value): Value = | |||
| 464 | if off >= xs.len: return (if rest.isNil: NilV else: rest) | 482 | if off >= xs.len: return (if rest.isNil: NilV else: rest) |
| 465 | Value(kind: kChunk, obj: Obj(kind: kChunk, chunk: xs, coff: off, ctl: rest)) | 483 | Value(kind: kChunk, obj: Obj(kind: kChunk, chunk: xs, coff: off, ctl: rest)) |
| 466 | 484 | ||
| 485 | +proc rec*(v: Value): Recipe {.inline.} = | ||
| 486 | + (if v.kind == kLazy: v.obj.rec else: nil) | ||
| 487 | + | ||
| 467 | proc mkLazy*(f: proc (): Value {.closure.}): Value = | 488 | proc mkLazy*(f: proc (): Value {.closure.}): Value = |
| 468 | Value(kind: kLazy, obj: Obj(kind: kLazy, thunk: f, cached: NilV, forced: false)) | 489 | Value(kind: kLazy, obj: Obj(kind: kLazy, thunk: f, cached: NilV, forced: false)) |
| 469 | 490 | ||
| 491 | +proc mkLazyRec*(f: proc (): Value {.closure.}, r: Recipe): Value = | ||
| 492 | + ## Same lazy seq, plus the description a fused consumer can use. | ||
| 493 | + Value(kind: kLazy, obj: Obj(kind: kLazy, thunk: f, cached: NilV, | ||
| 494 | + forced: false, rec: r)) | ||
| 495 | + | ||
| 470 | proc force*(v: Value): Value = | 496 | proc force*(v: Value): Value = |
| 471 | ## Realize one step: follow a chain of lazy seqs down to a cons, a concrete | 497 | ## Realize one step: follow a chain of lazy seqs down to a cons, a concrete |
| 472 | ## collection, or the end of the seq. | 498 | ## collection, or the end of the seq. |
added
tests/fusion.expected +16 -0 | new file mode 100644 | ||
| @@ -0,0 +1,16 @@ | ||
| 1 | +1140 | |
| 2 | +1030 | |
| 3 | +7 4 | |
| 4 | +25 | |
| 5 | +25 | |
| 6 | +0 | |
| 7 | +99 | |
| 8 | +3 | |
| 9 | +9 9 | |
| 10 | +9 | |
| 11 | +45 10 45 10 | |
| 12 | +55 10 55 10 | |
| 13 | +8 | |
| 14 | +[:f :g :p :src] | |
| 15 | +99 | |
| 16 | +[:replaced] | |
| new file mode 100644 | |||
| @@ -0,0 +1,16 @@ | |||
| 1 | +1140 | ||
| 2 | +1030 | ||
| 3 | +7 4 | ||
| 4 | +25 | ||
| 5 | +25 | ||
| 6 | +0 | ||
| 7 | +99 | ||
| 8 | +3 | ||
| 9 | +9 9 | ||
| 10 | +9 | ||
| 11 | +45 10 45 10 | ||
| 12 | +55 10 55 10 | ||
| 13 | +8 | ||
| 14 | +[:f :g :p :src] | ||
| 15 | +99 | ||
| 16 | +[:replaced] | ||