core: lazy realization is shared across walks (once-only effects); pmap family
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).
This commit is contained in:
parent
599c0468f7
commit
4a1a9e3aec
8 changed files with 78 additions and 12 deletions
|
|
@ -191,6 +191,10 @@
|
||||||
(let [msg (if message message (str "Assert failed: " (pr-str x)))]
|
(let [msg (if message message (str "Assert failed: " (pr-str x)))]
|
||||||
`(when-not ~x (throw (ex-info ~msg {})))))
|
`(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]
|
(defmacro delay [& body]
|
||||||
`(make-delay (fn [] ~@body)))
|
`(make-delay (fn [] ~@body)))
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -175,3 +175,17 @@
|
||||||
(lazy-seq
|
(lazy-seq
|
||||||
(when-let [s (seq coll)]
|
(when-let [s (seq coll)]
|
||||||
(cons (first s) (take-nth n (drop n s)))))))
|
(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))
|
||||||
|
|
|
||||||
|
|
@ -144,7 +144,7 @@
|
||||||
(do
|
(do
|
||||||
(array/push items (in cell 0))
|
(array/push items (in cell 0))
|
||||||
(let [rt (in cell 1)]
|
(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)
|
items)
|
||||||
c))
|
c))
|
||||||
|
|
||||||
|
|
@ -256,7 +256,7 @@
|
||||||
(set go false)
|
(set go false)
|
||||||
(if (pred (in cell 0))
|
(if (pred (in cell 0))
|
||||||
(let [rt (in cell 1)]
|
(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)))))
|
(set result false)))))
|
||||||
result)
|
result)
|
||||||
(do
|
(do
|
||||||
|
|
@ -870,7 +870,7 @@
|
||||||
(if (core-reduced? acc)
|
(if (core-reduced? acc)
|
||||||
(do (set acc (acc :val)) (set go false))
|
(do (set acc (acc :val)) (set go false))
|
||||||
(let [rt (in cell 1)]
|
(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
|
(do
|
||||||
(var stop false)
|
(var stop false)
|
||||||
(each x (if (set? coll) (phs-seq coll) (realize-for-iteration coll))
|
(each x (if (set? coll) (phs-seq coll) (realize-for-iteration coll))
|
||||||
|
|
@ -1098,7 +1098,7 @@
|
||||||
(f)
|
(f)
|
||||||
(let [rt (in cell 1)]
|
(let [rt (in cell 1)]
|
||||||
(if (nil? rt) (in cell 0)
|
(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))]
|
(let [c (if (set? coll) (phs-seq coll) (realize-for-iteration coll))]
|
||||||
(if (= 0 (length c)) (f)
|
(if (= 0 (length c)) (f)
|
||||||
(reduce-with-reduced f (in c 0) (array/slice c 1))))))
|
(reduce-with-reduced f (in c 0) (array/slice c 1))))))
|
||||||
|
|
|
||||||
|
|
@ -181,7 +181,7 @@
|
||||||
(set go false)
|
(set go false)
|
||||||
(do (array/push items (in cell 0))
|
(do (array/push items (in cell 0))
|
||||||
(let [rt (in cell 1)]
|
(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)
|
items)
|
||||||
val))))
|
val))))
|
||||||
|
|
||||||
|
|
@ -425,6 +425,13 @@
|
||||||
{"sleep" (fn [ms] (ev/sleep (/ ms 1000)) nil)
|
{"sleep" (fn [ms] (ev/sleep (/ ms 1000)) nil)
|
||||||
"yield" (fn [] (ev/sleep 0) 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
|
(defn- resolve-sym
|
||||||
[ctx bindings sym-s]
|
[ctx bindings sym-s]
|
||||||
(let [name (sym-s :name) ns (sym-s :ns)]
|
(let [name (sym-s :name) ns (sym-s :ns)]
|
||||||
|
|
@ -434,6 +441,9 @@
|
||||||
(if (= ns "Thread")
|
(if (= ns "Thread")
|
||||||
(let [v (get thread-statics name)]
|
(let [v (get thread-statics name)]
|
||||||
(if (nil? v) (error (string "Unsupported Thread member: Thread/" name)) v))
|
(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))
|
(if (not (nil? ns))
|
||||||
(let [current-ns (ctx-find-ns ctx (ctx-current-ns ctx))
|
(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))
|
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
|
# No implicit Janet fallback (Stage 3): an unresolved
|
||||||
# Clojure symbol is an error. Host access is the explicit
|
# Clojure symbol is an error. Host access is the explicit
|
||||||
# janet/ prefix above.
|
# janet/ prefix above.
|
||||||
(error (string "Unable to resolve symbol: " name)))))))))))))))
|
(error (string "Unable to resolve symbol: " name))))))))))))))))
|
||||||
(defn- parse-arg-names
|
(defn- parse-arg-names
|
||||||
"Parse a parameter vector, handling & rest args.
|
"Parse a parameter vector, handling & rest args.
|
||||||
Returns {:fixed [names...] :rest name-or-nil :all [names...]}"
|
Returns {:fixed [names...] :rest name-or-nil :all [names...]}"
|
||||||
|
|
|
||||||
|
|
@ -84,7 +84,7 @@
|
||||||
(write-value (in cell 0) buf)
|
(write-value (in cell 0) buf)
|
||||||
(++ i)
|
(++ i)
|
||||||
(let [rt (in cell 1)]
|
(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 " ..."))
|
(when (and go (>= i 1000)) (push-str buf " ..."))
|
||||||
(push-str buf ")"))
|
(push-str buf ")"))
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -208,11 +208,25 @@
|
||||||
(let [cell (realize-ls ls)]
|
(let [cell (realize-ls ls)]
|
||||||
(if (or (nil? cell) (= :jolt/pending cell) (= 0 (length cell))) nil (in cell 0))))
|
(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]
|
(defn ls-rest [ls]
|
||||||
(let [cell (realize-ls ls)]
|
(let [cell (realize-ls ls)]
|
||||||
(if (or (nil? cell) (= 0 (length cell))) nil
|
(if (or (nil? cell) (= 0 (length cell))) nil
|
||||||
(let [rt (in cell 1)]
|
(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]
|
(defn ls-seq [ls]
|
||||||
(var result @[])
|
(var result @[])
|
||||||
|
|
@ -221,8 +235,7 @@
|
||||||
(let [cell (realize-ls cur)]
|
(let [cell (realize-ls cur)]
|
||||||
(if (nil? cell) (break))
|
(if (nil? cell) (break))
|
||||||
(array/push result (in cell 0))
|
(array/push result (in cell 0))
|
||||||
(let [rt (in cell 1)]
|
(set cur (ls-rest cur))))
|
||||||
(set cur (if (nil? rt) nil (make-lazy-seq rt))))))
|
|
||||||
(if (= 0 (length result)) nil result))
|
(if (= 0 (length result)) nil result))
|
||||||
|
|
||||||
(defn ls-count [ls]
|
(defn ls-count [ls]
|
||||||
|
|
@ -232,8 +245,7 @@
|
||||||
(let [cell (realize-ls cur)]
|
(let [cell (realize-ls cur)]
|
||||||
(if (nil? cell) (break))
|
(if (nil? cell) (break))
|
||||||
(++ cnt)
|
(++ cnt)
|
||||||
(let [rt (in cell 1)]
|
(set cur (ls-rest cur))))
|
||||||
(set cur (if (nil? rt) nil (make-lazy-seq rt))))))
|
|
||||||
cnt)
|
cnt)
|
||||||
|
|
||||||
# ============================================================
|
# ============================================================
|
||||||
|
|
|
||||||
|
|
@ -49,3 +49,16 @@
|
||||||
# The future's own return value still reflects the swap on its copy.
|
# The future's own return value still reflects the swap on its copy.
|
||||||
["future sees its own mutation"
|
["future sees its own mutation"
|
||||||
"1" "(let [a (atom 0)] (deref (future (swap! a inc))))"])
|
"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))"])
|
||||||
|
|
|
||||||
|
|
@ -29,3 +29,16 @@
|
||||||
["unrealized" "false" "(realized? (lazy-seq (cons 1 nil)))"]
|
["unrealized" "false" "(realized? (lazy-seq (cons 1 nil)))"]
|
||||||
["realized after" "true" "(let [s (lazy-seq (cons 1 nil))] (first s) (realized? s))"]
|
["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)"])
|
["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))"])
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue