From 412ed394a9cfb5d1dcaaada32d626ef7266b88b5 Mon Sep 17 00:00:00 2001 From: Yogthos Date: Tue, 13 Oct 2026 16:03:02 -0400 Subject: [PATCH 1/2] Keep a sample transfer bounded when inputs notify from other threads Sample.java's monitor blocks every other notifier while a transfer reads an input, so the do/while only re-reads for notifications the read itself raised, and a surplus notification is discarded by the thread that raised it. Ebb reads outside the lock, so a notification from another thread landing mid-read made the transfer read again, and a surplus was left for whoever held the slot to pay. An input notified as fast as it can be read then kept one transfer going indefinitely. That is what timed out sample-never-derefs-one-input-twice-at-once on CI with jolt 0.8.21 and later. Thread/sleep there takes ~1.25ms for (Thread/sleep 0 n) where it used to return at once, which put the test's notifier and its reads at the same rate, and the transfer ended at a random point: up to 2.5s locally under load, past the 15s deadline on a 2-core runner. A read now marks itself in a carried context-local, so notifications from inside it (including an ap or cp input's owner fiber serving the deref) are still the read's own. Anyone else's notification during a transfer is recorded as late and becomes the mark for the next transfer, and an outside surplus waits for the slot and discards for itself. --- src/ebb/impl/sample.clj | 102 ++++++++++++++++++++++++++++++++------- test/ebb/stress_test.clj | 79 ++++++++++++++++++++++++++++++ 2 files changed, 163 insertions(+), 18 deletions(-) diff --git a/src/ebb/impl/sample.clj b/src/ebb/impl/sample.clj index c84fbea..682587c 100644 --- a/src/ebb/impl/sample.clj +++ b/src/ebb/impl/sample.clj @@ -32,8 +32,29 @@ ;; combinator gets to it -- is not mutually exclusive with `transfer`'s deref ;; of the same slot. The monitor makes those two orderly; one atom cannot, ;; because the slot's claim would have to be held across a park (ebb-8nq.33). +;; +;; What the monitor also does is make every OTHER notifier wait while a +;; transfer or a discard reads an input. Two things follow from that, and ebb +;; has to arrange both by hand, since its reads happen outside the lock: +;; +;; * A transfer reads a slot again only for a notification raised from inside +;; its own read -- the do/while. Anyone else's notification lands after the +;; transfer and marks the slot for the next one. Here it is recorded as `late` +;; and turned into that mark when the transfer lets go of the slot. Counting +;; it as the read's own let an input fed from another thread as fast as it is +;; read keep one transfer reading forever. +;; * A surplus -- a notification on a slot that is still notified -- is +;; discarded by the notifier that raised it, which is therefore slowed to the +;; pace the input can be read at. Here an outside notifier that finds the slot +;; taken waits for it rather than leaving the read to the holder, which would +;; otherwise pay for every notification raised while it paid for the last. +;; +;; "From inside the read" is the `reading` context-local, which server/call! +;; carries: an ap or cp input notifies from its owner fiber while it serves our +;; deref, and that notification is as much the read's own as a synchronous one. (ns ^:no-doc ebb.impl.sample - (:require [ebb.impl.arbiter :as arb])) + (:require [ebb.impl.arbiter :as arb] + [ebb.impl.util :as u])) (declare transfer) @@ -41,7 +62,7 @@ (deftype Process [^:unsynchronized-mutable combinator ^:unsynchronized-mutable notifier - terminator args owed busy gates inputs state + terminator args owed busy gates sampling late inputs state ^:unsynchronized-mutable done alive] clojure.lang.IFn @@ -49,6 +70,20 @@ clojure.lang.IDeref (deref [p] (transfer p))) +(def ^:private reading + "The slots whose deref this context is inside, as a set of [process index]." + (u/carried-local)) + +(defn- inside? [p i] + (let [s (u/local-get reading)] (and (some? s) (contains? s [p i])))) + +(defn- read! + "Deref slot i's input, marked as inside that read for whatever it notifies." + [p i input] + (let [was (u/local-get reading)] + (u/local-set! reading (conj (or was #{}) [p i])) + (try @input (finally (u/local-set! reading was))))) + (defn- take-slot! "Take slot i's deref rights. Answers [n nil] when we now hold it and it owes n DISCARDS, or [nil gate] when somebody else holds it -- gate opens when they @@ -84,11 +119,11 @@ [^Process p i n] (let [input (aget (.-inputs p) i)] (loop [n n] - (dotimes [_ n] (try @input (catch Throwable _ nil))) + (dotimes [_ n] (try (read! p i input) (catch Throwable _ nil))) (let [m (release-slot! p i)] (when (pos? m) (recur m)))))) -(defn dirty +(defn- notify! "Input i notified. One notification is left for the combinator to sample; any FURTHER one before that happens is a discard the input is owed, and draining it eagerly is what keeps a sampled input flowing between samples -- defer it @@ -99,21 +134,40 @@ to: `mark!` answered `already dirty, go drain` WITHOUT clearing the flag, so `resample!` could take the still-dirty slot and deref concurrently -- 2630 overlaps in 20 runs once the window was widened enough to see (ebb-8nq.33). - The slot is claimed now, and a `dirty` that loses simply leaves the count - behind for whoever holds it. It never waits: the holder's deref never depends - on us." - [^Process p i] + The slot is claimed now. A notification raised from inside the holder's own + read leaves the count behind for the holder and returns: that read may be + waiting on it. Anyone else waits for the slot when `wait?`, as they would for + the monitor, and drains what is owed themselves. + + While a transfer reads the slot, an outside notification is the next + transfer's, not this one's: it is recorded as `late`, and only a second one + is a surplus." + [^Process p i wait?] (let [args (.-args p) owed (.-owed p) + late (.-late p) + own? (inside? p i) surplus? (locking p - (if (identical? (aget args i) args) + (cond + (identical? (aget args i) args) (do (aset owed i (inc (aget owed i))) true) - (do (aset args i args) false)))] + (and (not own?) (pos? (aget (.-sampling p) i))) + (if (zero? (aget late i)) + (do (aset late i 1) false) + (do (aset owed i (inc (aget owed i))) true)) + :else (do (aset args i args) false)))] (when surplus? - (let [[n _] (take-slot! p i)] - (when n (discard! p i n))))) + (loop [] + (let [[n gate] (take-slot! p i)] + (if (some? n) + (discard! p i n) + (when (and wait? (not own?)) + (deref gate 1000 nil) + (recur))))))) nil) +(defn dirty [p i] (notify! p i true)) + (defn- rounds "Run rounds while the claim is held; answer the callback to fire outside, or nil. Both exits keep the claim, as the Java's two `break`s do." @@ -127,7 +181,7 @@ ;; cancel first, then take every transfer it owes us -- both ;; outside the lock, both able to park ((aget inputs i)) - (dirty ps i)) + (notify! ps i false)) (when (zero? (swap! (.-alive ps) dec)) (.-terminator ps))) (if (identical? (aget args sampled) args) (do (try @(aget inputs sampled) (catch Throwable _ nil)) @@ -154,7 +208,9 @@ (do (deref gate 1000 nil) (recur)) (try ;; the discards this slot is owed come first, and are dropped - (dotimes [_ n] (try @input (catch Throwable _ nil))) + (dotimes [_ n] (try (read! ps i input) (catch Throwable _ nil))) + ;; from here an outside notification is the next transfer's + (locking ps (aset (.-sampling ps) i 1)) ;; then the notification the combinator is actually here for (loop [] ;; ONE region for both questions. Asking `is it dirty?` and @@ -167,9 +223,10 @@ (do (aset args i nil) nil) [(aget args i)]))] (if (nil? r) - (let [x @input] + (let [x (read! ps i input)] ;; the deref may have notified this slot again -- the Java's - ;; do/while -- in which case that notification is ours too + ;; do/while -- in which case that notification is ours too. + ;; Only from inside it: anyone else's went to `late`. (if (locking ps (if (identical? (aget args i) args) true (do (aset args i x) false))) @@ -180,10 +237,17 @@ ;; hand the slot back however we leave, exception included, or a ;; failed transfer strands every later notification behind a claim ;; nobody holds + (locking ps + (aset (.-sampling ps) i 0) + ;; what an outsider raised meanwhile is a mark, as it is when + ;; Sample.java's blocked notifier gets the monitor back + (when (pos? (aget (.-late ps) i)) + (aset (.-late ps) i 0) + (aset args i args))) (loop [] (let [m (release-slot! ps i)] (when (pos? m) - (dotimes [_ m] (try @input (catch Throwable _ nil))) + (dotimes [_ m] (try (read! ps i input) (catch Throwable _ nil))) (recur))))))))))) (defn transfer [^Process ps] @@ -222,11 +286,13 @@ owed (int-array arity) busy (int-array arity) gates (object-array arity) + sampling (int-array arity) + late (int-array arity) inputs (object-array arity) ;; the claim starts held: Sample.java holds the monitor across the whole ;; of this construction, and the sampler's notifier must not run a round ;; against an `inputs` array that is not filled in yet. - ps (->Process c n t args owed busy gates inputs (arb/arbiter) false (atom arity)) + ps (->Process c n t args owed busy gates sampling late inputs (arb/arbiter) false (atom arity)) done #(when (zero? (swap! (.-alive ps) dec)) ((.-terminator ps)))] (loop [index 0 flow f] diff --git a/test/ebb/stress_test.clj b/test/ebb/stress_test.clj index 0851a16..40a6c55 100644 --- a/test/ebb/stress_test.clj +++ b/test/ebb/stress_test.clj @@ -352,3 +352,82 @@ (fn [v] (deliver out [:ok v])) (fn [e] (deliver out [:err e]))) (is (= :ok (first (deref out timeout-ms [:err ::timeout])))))) (is (zero? @overlaps) "a sampled input is never deref'd by two parties at once"))) + +;; A transfer of `sample` takes each notified input once, plus once more for +;; every notification that input raises from INSIDE that read -- Sample.java's +;; do/while. A notification raised by anyone else while the read is in flight +;; belongs to the next transfer: on the JVM its notifier is blocked on the +;; process monitor until this transfer is over, and then it just marks the slot. +;; +;; Ebb reads outside any lock, so that notifier is not blocked, and sample used +;; to count it as the read's own and read again. An input that moves on while it +;; is being read -- the common case for a continuous input fed from another +;; thread faster than it is sampled -- then kept one transfer reading forever. +;; With the notifier and the read at about the same rate the loop ends at a +;; random point, so the stress test above passed or timed out by luck. + +(defn- moved-on + "An input that is notified again from another thread during every read, and + waits (up to 20ms) for that notification to land before returning. One + notification per transfer, so within the flow protocol. `stop` silences it." + [reads stop] + (fn [n t] + (let [live (atom true) v (atom 0)] + (n) + (reify + clojure.lang.IFn + (invoke [_] (when (compare-and-set! live true false) (t)) nil) + clojure.lang.IDeref + (deref [_] + (swap! reads inc) + (let [x (swap! v inc) landed (promise)] + (doto (Thread. (fn [] (when (and @live (not @stop)) (n)) (deliver landed true))) + (.setDaemon true) (.start)) + (deref landed 20 nil) + x)))))) + +(deftest sample-leaves-an-outside-notification-for-the-next-transfer + (let [reads (atom 0) stop (atom false) out (promise)] + ((m/reduce conj [] (m/sample vector (moved-on reads stop) (offloaded (range 12)))) + (fn [v] (deliver out [:ok v])) (fn [e] (deliver out [:err e]))) + (let [[tag v] (deref out timeout-ms [:err ::timeout])] + (reset! stop true) + (is (= [:ok 12] [tag (when (= :ok tag) (count v))])) + ;; one read per sample, plus the cancelled input's last drain + (is (<= @reads 13) (str @reads " reads for 12 samples"))))) + +(deftest sample-makes-an-outside-notifier-discard-its-own-surplus + ;; An input that notifies without waiting to be transferred is notified while + ;; still notified, and Sample.java's `dirty` discards that surplus on the + ;; notifier's own thread, under the monitor: a notifier faster than the input + ;; can be read is slowed to the read's pace. Ebb used to hand every surplus to + ;; whoever held the slot, so a transfer holding it paid for each notification + ;; raised while it paid for the last, and never finished. + ;; + ;; This input's reads wait for two more notifications to land (or 20ms), so + ;; the notifier always outruns them. + (let [stop (atom false) out (promise) + input (fn [n t] + (let [live (atom true) notes (atom 0)] + (n) + (doto (Thread. (fn [] (loop [] + (when (and @live (not @stop)) + (swap! notes inc) + (n) + (Thread/yield) + (recur))))) + (.setDaemon true) (.start)) + (reify + clojure.lang.IFn + (invoke [_] (when (compare-and-set! live true false) (t)) nil) + clojure.lang.IDeref + (deref [_] + (let [from @notes end (+ (System/nanoTime) 20000000)] + (while (and (< @notes (+ from 2)) (< (System/nanoTime) end)) + (Thread/yield)) + from)))))] + ((m/reduce conj [] (m/sample vector input (offloaded (range 12)))) + (fn [v] (deliver out [:ok v])) (fn [e] (deliver out [:err e]))) + (let [[tag v] (deref out timeout-ms [:err ::timeout])] + (reset! stop true) + (is (= [:ok 12] [tag (when (= :ok tag) (count v))]))))) From b74c4a0b2e6e34c1505b7ceb1a7a22196c339bd5 Mon Sep 17 00:00:00 2001 From: Yogthos Date: Tue, 13 Oct 2026 16:19:35 -0400 Subject: [PATCH 2/2] Hand a sample slot to the waiting transfer With an outside notifier draining its own surplus, the slot was freed between its discards and the notifier took it straight back before the transfer waiting on the gate had woken up. On a 2-core runner the transfer never got in and the new surplus test timed out every run. A transfer that loses the slot now asks for it, and the holder hands it over on release instead of freeing it. --- src/ebb/impl/sample.clj | 40 +++++++++++++++++++++++++++------------- 1 file changed, 27 insertions(+), 13 deletions(-) diff --git a/src/ebb/impl/sample.clj b/src/ebb/impl/sample.clj index 682587c..2ce579c 100644 --- a/src/ebb/impl/sample.clj +++ b/src/ebb/impl/sample.clj @@ -62,7 +62,7 @@ (deftype Process [^:unsynchronized-mutable combinator ^:unsynchronized-mutable notifier - terminator args owed busy gates sampling late inputs state + terminator args owed busy gates sampling late wanted inputs state ^:unsynchronized-mutable done alive] clojure.lang.IFn @@ -88,8 +88,14 @@ "Take slot i's deref rights. Answers [n nil] when we now hold it and it owes n DISCARDS, or [nil gate] when somebody else holds it -- gate opens when they are done. The gate is chosen under the same lock as the claim it lost, so it - can never be one the holder has already opened." - [^Process p i] + can never be one the holder has already opened. + + A transfer asks with `want?`, and the holder then hands the slot straight to + it, opening the gate with :yours. Otherwise an outside notifier that drains + its own surplus and notifies again at once retakes the slot every time it is + let go, ahead of a transfer still waking up from the gate, and the transfer + never gets in -- which a 2-core runner showed." + [^Process p i want?] (locking p (let [busy (.-busy p)] (if (zero? (aget busy i)) @@ -98,6 +104,7 @@ (aset owed i 0) [n nil]) (let [gates (.-gates p)] + (when want? (aset (.-wanted p) i 1)) [nil (or (aget gates i) (aset gates i (promise)))]))))) (defn- release-slot! @@ -108,9 +115,11 @@ (let [owed (.-owed p) n (aget owed i)] (if (pos? n) (do (aset owed i 0) n) - (let [gates (.-gates p) g (aget gates i)] - (aset (.-busy p) i 0) - (when g (aset gates i nil) (deliver g true)) + (let [gates (.-gates p) g (aget gates i) + wanted (.-wanted p) yours? (pos? (aget wanted i))] + ;; a waiting transfer takes the slot over as it stands: busy stays set + (if yours? (aset wanted i 0) (aset (.-busy p) i 0)) + (when g (aset gates i nil) (deliver g (if yours? :yours true))) 0))))) (defn- discard! @@ -158,7 +167,7 @@ :else (do (aset args i args) false)))] (when surplus? (loop [] - (let [[n gate] (take-slot! p i)] + (let [[n gate] (take-slot! p i false)] (if (some? n) (discard! p i n) (when (and wait? (not own?)) @@ -202,10 +211,14 @@ [^Process ps i] (let [args (.-args ps) input (aget (.-inputs ps) i)] - (loop [] - (let [[n gate] (take-slot! ps i)] - (if (nil? n) - (do (deref gate 1000 nil) (recur)) + (let [n (loop [] + (let [[n gate] (take-slot! ps i true)] + (cond + (some? n) n + ;; handed over, owing nothing: the holder drained it all + (= :yours (loop [] (let [r (deref gate 1000 ::wait)] + (if (= ::wait r) (recur) r)))) 0 + :else (recur))))] (try ;; the discards this slot is owed come first, and are dropped (dotimes [_ n] (try (read! ps i input) (catch Throwable _ nil))) @@ -248,7 +261,7 @@ (let [m (release-slot! ps i)] (when (pos? m) (dotimes [_ m] (try (read! ps i input) (catch Throwable _ nil))) - (recur))))))))))) + (recur))))))))) (defn transfer [^Process ps] (let [c (.-combinator ps) @@ -288,11 +301,12 @@ gates (object-array arity) sampling (int-array arity) late (int-array arity) + wanted (int-array arity) inputs (object-array arity) ;; the claim starts held: Sample.java holds the monitor across the whole ;; of this construction, and the sampler's notifier must not run a round ;; against an `inputs` array that is not filled in yet. - ps (->Process c n t args owed busy gates sampling late inputs (arb/arbiter) false (atom arity)) + ps (->Process c n t args owed busy gates sampling late wanted inputs (arb/arbiter) false (atom arity)) done #(when (zero? (swap! (.-alive ps) dec)) ((.-terminator ps)))] (loop [index 0 flow f]