Merge pull request #63 from jolt-lang/pmap

core: lazy realization shared across walks (once-only effects); pmap family

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)))
 

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))

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))))))

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...]}"

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 ")"))
 

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)
 
 # ============================================================

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))"])

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))"])