From 4a1a9e3aec5adf9b9cd62ab486c72d5434ecba3e Mon Sep 17 00:00:00 2001 From: Yogthos Date: Wed, 10 Jun 2026 19:14:45 -0400 Subject: [PATCH] core: lazy realization is shared across walks (once-only effects); pmap family MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every walk over a lazy seq created FRESH wrapper tables around the shared rest-thunks (ls-rest, ls-seq/ls-count, realize-for-iteration, the printers, reduce — each had its own make-lazy-seq loop), so independent walks re-ran the thunks: side effects duplicated, and a doall'd seq of futures was re-spawned serially by the deref walk. Every walker now goes through ls-rest-cached, which memoizes the rest wrapper on its node — thunks run exactly once, as in Clojure. Costs ~10% on walk-heavy benches (the per-node cache get/put — Clojure's LazySeq pays the same); net still -9% vs the pre-linear-walks baseline. Three regression rows pin once-only effects and value stability across walks. On top of that: pmap/pcalls/pvalues (jolt-oeu) over the real-thread futures — spawn-all-then-deref (the once-only fix is what makes the doall actually mean that), snapshot semantics documented, multi-coll arity via the canonical vector-zip. System/currentTimeMillis + nanoTime land as System statics (the realtime clock — os/time is whole seconds, which quantized every elapsed measurement to 1000ms). Seven pmap rows incl. a generous- margin parallelism check (4 x 200ms sleeps under 700ms after warmup). --- jolt-core/clojure/core/30-macros.clj | 4 ++++ jolt-core/clojure/core/40-lazy.clj | 14 ++++++++++++++ src/jolt/core.janet | 8 ++++---- src/jolt/evaluator.janet | 14 ++++++++++++-- src/jolt/main.janet | 2 +- src/jolt/phm.janet | 22 +++++++++++++++++----- test/spec/futures-spec.janet | 13 +++++++++++++ test/spec/lazy-seqs-spec.janet | 13 +++++++++++++ 8 files changed, 78 insertions(+), 12 deletions(-) diff --git a/jolt-core/clojure/core/30-macros.clj b/jolt-core/clojure/core/30-macros.clj index 9c20ba1..635ac01 100644 --- a/jolt-core/clojure/core/30-macros.clj +++ b/jolt-core/clojure/core/30-macros.clj @@ -191,6 +191,10 @@ (let [msg (if message message (str "Assert failed: " (pr-str x)))] `(when-not ~x (throw (ex-info ~msg {}))))) +;; (pvalues e1 e2 ...) — each expression evaluated in parallel (pcalls). +(defmacro pvalues [& exprs] + `(pcalls ~@(map (fn [e] `(fn [] ~e)) exprs))) + (defmacro delay [& body] `(make-delay (fn [] ~@body))) diff --git a/jolt-core/clojure/core/40-lazy.clj b/jolt-core/clojure/core/40-lazy.clj index aa1a904..878b13d 100644 --- a/jolt-core/clojure/core/40-lazy.clj +++ b/jolt-core/clojure/core/40-lazy.clj @@ -175,3 +175,17 @@ (lazy-seq (when-let [s (seq coll)] (cons (first s) (take-nth n (drop n s))))))) + +;; --- pmap family (jolt-oeu): parallel map over real-thread futures ---------- +;; Each element's work runs on its own OS thread with SNAPSHOT semantics +;; (futures marshal captured state — pure fns only, mutations don't propagate +;; back). All futures are spawned up front (doall), then derefed in order: +;; coarse-grained work only, as with Clojure's pmap. + +(defn pmap + ([f coll] + (map deref (doall (map (fn [x] (future (f x))) coll)))) + ([f coll & colls] + (pmap (fn [xs] (apply f xs)) (apply map vector coll colls)))) + +(defn pcalls [& fns] (pmap (fn [f] (f)) fns)) diff --git a/src/jolt/core.janet b/src/jolt/core.janet index b5a39ac..5a3cd6b 100644 --- a/src/jolt/core.janet +++ b/src/jolt/core.janet @@ -144,7 +144,7 @@ (do (array/push items (in cell 0)) (let [rt (in cell 1)] - (if (nil? rt) (set go false) (set cur (make-lazy-seq rt)))))))) + (if (nil? rt) (set go false) (set cur (ls-rest-cached cur rt)))))))) items) c)) @@ -256,7 +256,7 @@ (set go false) (if (pred (in cell 0)) (let [rt (in cell 1)] - (if (nil? rt) (set go false) (set cur (make-lazy-seq rt)))) + (if (nil? rt) (set go false) (set cur (ls-rest-cached cur rt)))) (set result false))))) result) (do @@ -870,7 +870,7 @@ (if (core-reduced? acc) (do (set acc (acc :val)) (set go false)) (let [rt (in cell 1)] - (if (nil? rt) (set go false) (set cur (make-lazy-seq rt)))))))))) + (if (nil? rt) (set go false) (set cur (ls-rest-cached cur rt)))))))))) (do (var stop false) (each x (if (set? coll) (phs-seq coll) (realize-for-iteration coll)) @@ -1098,7 +1098,7 @@ (f) (let [rt (in cell 1)] (if (nil? rt) (in cell 0) - (reduce-with-reduced f (in cell 0) (make-lazy-seq rt)))))) + (reduce-with-reduced f (in cell 0) (ls-rest-cached coll rt)))))) (let [c (if (set? coll) (phs-seq coll) (realize-for-iteration coll))] (if (= 0 (length c)) (f) (reduce-with-reduced f (in c 0) (array/slice c 1)))))) diff --git a/src/jolt/evaluator.janet b/src/jolt/evaluator.janet index 8f790d0..a43a8e1 100644 --- a/src/jolt/evaluator.janet +++ b/src/jolt/evaluator.janet @@ -181,7 +181,7 @@ (set go false) (do (array/push items (in cell 0)) (let [rt (in cell 1)] - (if (nil? rt) (set go false) (set cur (make-lazy-seq rt)))))))) + (if (nil? rt) (set go false) (set cur (ls-rest-cached cur rt)))))))) items) val)))) @@ -425,6 +425,13 @@ {"sleep" (fn [ms] (ev/sleep (/ ms 1000)) nil) "yield" (fn [] (ev/sleep 0) nil)}) +# System statics (wall/monotonic clocks — what portable timing code uses). +(def- system-statics + # realtime clock (sub-ms float epoch seconds) — os/time is whole seconds, + # which quantized every elapsed-time measurement to 1000ms. + {"currentTimeMillis" (fn [] (math/floor (* 1000 (os/clock :realtime)))) + "nanoTime" (fn [] (math/floor (* 1e9 (os/clock :monotonic))))}) + (defn- resolve-sym [ctx bindings sym-s] (let [name (sym-s :name) ns (sym-s :ns)] @@ -434,6 +441,9 @@ (if (= ns "Thread") (let [v (get thread-statics name)] (if (nil? v) (error (string "Unsupported Thread member: Thread/" name)) v)) + (if (= ns "System") + (let [v (get system-statics name)] + (if (nil? v) (error (string "Unsupported System member: System/" name)) v)) (if (not (nil? ns)) (let [current-ns (ctx-find-ns ctx (ctx-current-ns ctx)) aliased-ns (or (ns-alias-lookup current-ns ns) (ns-import-lookup current-ns ns)) @@ -483,7 +493,7 @@ # No implicit Janet fallback (Stage 3): an unresolved # Clojure symbol is an error. Host access is the explicit # janet/ prefix above. - (error (string "Unable to resolve symbol: " name))))))))))))))) + (error (string "Unable to resolve symbol: " name)))))))))))))))) (defn- parse-arg-names "Parse a parameter vector, handling & rest args. Returns {:fixed [names...] :rest name-or-nil :all [names...]}" diff --git a/src/jolt/main.janet b/src/jolt/main.janet index a120b59..43cb54d 100644 --- a/src/jolt/main.janet +++ b/src/jolt/main.janet @@ -84,7 +84,7 @@ (write-value (in cell 0) buf) (++ i) (let [rt (in cell 1)] - (if (nil? rt) (set go false) (set cur (make-lazy-seq rt)))))))) + (if (nil? rt) (set go false) (set cur (ls-rest-cached cur rt)))))))) (when (and go (>= i 1000)) (push-str buf " ...")) (push-str buf ")")) diff --git a/src/jolt/phm.janet b/src/jolt/phm.janet index b7b22a5..1307fc6 100644 --- a/src/jolt/phm.janet +++ b/src/jolt/phm.janet @@ -208,11 +208,25 @@ (let [cell (realize-ls ls)] (if (or (nil? cell) (= :jolt/pending cell) (= 0 (length cell))) nil (in cell 0)))) +# The memoized rest wrapper for a node whose cell yielded rest-thunk rt. +# EVERY walk must go through this (not a fresh make-lazy-seq) or independent +# walks re-run the shared thunks and side effects duplicate. +(defn ls-rest-cached [ls rt] + (or (get ls :rest-ls) + (let [w (make-lazy-seq rt)] + (put ls :rest-ls w) + w))) + (defn ls-rest [ls] (let [cell (realize-ls ls)] (if (or (nil? cell) (= 0 (length cell))) nil (let [rt (in cell 1)] - (if (nil? rt) nil (make-lazy-seq rt)))))) + (if (nil? rt) nil + # Memoized wrapper (see ls-rest-cached): a fresh table per call gave + # every independent walk its own realization state, so the shared + # rest-thunks re-ran — duplicating side effects (a doall'd seq of + # futures re-spawned them on the deref walk, serializing pmap). + (ls-rest-cached ls rt)))))) (defn ls-seq [ls] (var result @[]) @@ -221,8 +235,7 @@ (let [cell (realize-ls cur)] (if (nil? cell) (break)) (array/push result (in cell 0)) - (let [rt (in cell 1)] - (set cur (if (nil? rt) nil (make-lazy-seq rt)))))) + (set cur (ls-rest cur)))) (if (= 0 (length result)) nil result)) (defn ls-count [ls] @@ -232,8 +245,7 @@ (let [cell (realize-ls cur)] (if (nil? cell) (break)) (++ cnt) - (let [rt (in cell 1)] - (set cur (if (nil? rt) nil (make-lazy-seq rt)))))) + (set cur (ls-rest cur)))) cnt) # ============================================================ diff --git a/test/spec/futures-spec.janet b/test/spec/futures-spec.janet index a394f64..82e5996 100644 --- a/test/spec/futures-spec.janet +++ b/test/spec/futures-spec.janet @@ -49,3 +49,16 @@ # The future's own return value still reflects the swap on its copy. ["future sees its own mutation" "1" "(let [a (atom 0)] (deref (future (swap! a inc))))"]) + +# pmap/pcalls/pvalues over the real-thread futures: spawn-all-then-deref, so +# independent work overlaps; snapshot semantics (pure fns only). The wall-time +# row uses generous margins (4 x 150ms in parallel vs 600ms serial). +(defspec "clojure.core / pmap family" + ["pmap values in order" "[2 3 4]" "(vec (pmap inc [1 2 3]))"] + ["pmap multi-coll" "[5 7 9]" "(vec (pmap + [1 2 3] [4 5 6]))"] + ["pmap empty" "()" "(pmap inc [])"] + ["pmap is parallel" "true" + "(do (deref (future :warmup)) (let [t0 (System/currentTimeMillis)] (dorun (pmap (fn [_] (Thread/sleep 200)) [1 2 3 4])) (< (- (System/currentTimeMillis) t0) 700)))"] + ["pcalls" "[1 2]" "(vec (pcalls (fn [] 1) (fn [] 2)))"] + ["pvalues" "[3 7]" "(vec (pvalues (+ 1 2) (+ 3 4)))"] + ["snapshot semantics" "0" "(let [a (atom 0)] (dorun (pmap (fn [_] (swap! a inc)) [1 2])) (deref a))"]) diff --git a/test/spec/lazy-seqs-spec.janet b/test/spec/lazy-seqs-spec.janet index cdb4491..0c9c99d 100644 --- a/test/spec/lazy-seqs-spec.janet +++ b/test/spec/lazy-seqs-spec.janet @@ -29,3 +29,16 @@ ["unrealized" "false" "(realized? (lazy-seq (cons 1 nil)))"] ["realized after" "true" "(let [s (lazy-seq (cons 1 nil))] (first s) (realized? s))"] ["body runs once" "1" "(let [c (atom 0) s (lazy-seq (do (swap! c inc) [1 2 3]))] (seq s) (seq s) @c)"]) + +# Independent walks over the SAME lazy seq share realization: each node's rest +# wrapper is memoized (ls-rest-cached), so the shared rest-thunks run exactly +# once. Pre-fix, every walk after the first re-ran the thunks — duplicating +# side effects, and a doall'd seq of futures re-spawned them serially on the +# deref walk (which is how it surfaced, via pmap). +(defspec "lazy-seq / realization is shared across walks" + ["effects run once across three walks" "3" + "(let [a (atom 0) s (map (fn [x] (swap! a inc) x) [1 2 3])] (doall s) (dorun s) (vec s) (deref a))"] + ["values stable across walks" "true" + "(let [s (map inc [1 2 3])] (= (vec s) (vec s) [2 3 4]))"] + ["filter effects once" "4" + "(let [a (atom 0) s (filter (fn [x] (swap! a inc) (odd? x)) [1 2 3 4])] (dorun s) (count s) (deref a))"])