Skip to content

Keep a sample transfer bounded when inputs notify from other threads #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yogthos merged 2 commits into from
Oct 13, 2026
Merged

Keep a sample transfer bounded when inputs notify from other threads #14

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
412ed39
Keep a sample transfer bounded when inputs notify from other threads
yogthos Oct 13, 2026
b74c4a0
Hand a sample slot to the waiting transfer
yogthos Oct 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
136 changes: 108 additions & 28 deletions src/ebb/impl/sample.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,29 +32,70 @@
;; 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)

(deftype Failed [e])

(deftype Process [^:unsynchronized-mutable combinator
^:unsynchronized-mutable notifier
terminator args owed busy gates inputs state
terminator args owed busy gates sampling late wanted inputs state
^:unsynchronized-mutable done
alive]
clojure.lang.IFn
(invoke [_] (dotimes [i (alength inputs)] ((aget inputs i))) nil)
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
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))
Expand All @@ -63,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!
Expand All @@ -73,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!
Expand All @@ -84,11 +128,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
Expand All @@ -99,21 +143,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 false)]
(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."
Expand All @@ -127,7 +190,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))
Expand All @@ -148,13 +211,19 @@
[^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 @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
Expand All @@ -167,9 +236,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)))
Expand All @@ -180,11 +250,18 @@
;; 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)))
(recur)))))))))))
(dotimes [_ m] (try (read! ps i input) (catch Throwable _ nil)))
(recur)))))))))

(defn transfer [^Process ps]
(let [c (.-combinator ps)
Expand Down Expand Up @@ -222,11 +299,14 @@
owed (int-array arity)
busy (int-array arity)
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 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]
Expand Down
79 changes: 79 additions & 0 deletions test/ebb/stress_test.clj
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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))])))))
Loading