diff --git a/src/ebb/impl/sample.clj b/src/ebb/impl/sample.clj index c84fbea..2ce579c 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 wanted inputs state ^:unsynchronized-mutable done alive] clojure.lang.IFn @@ -49,12 +70,32 @@ 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)) @@ -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! @@ -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! @@ -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 @@ -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." @@ -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)) @@ -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 @@ -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))) @@ -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) @@ -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] 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))])))))