Skip to content

Commit e14f4ac

Browse files
seancorfieldseancorfield
committed
Another run at CCACHE-65
Signed-off-by: Sean Corfield <sean@corfield.org>
1 parent fd7fc11 commit e14f4ac

File tree

3 files changed

+124
-36
lines changed

3 files changed

+124
-36
lines changed

‎.rebel_readline_history‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
1771730881107:(require '[clojure.core.cache :as cache] '[clojure.core.cache.wrapped :as cache-wrapped])
2+
1771730883786:(let [thread-count 20\n cache-atom (-> {}\n (cache/ttl-cache-factory :ttl 120000)\n (cache/lu-cache-factory :threshold 100)\n (atom))\n latch (java.util.concurrent.CountDownLatch. thread-count)\n invocations-counter (atom 0)]\n (doseq [i (range thread-count)]\n (println "starting thread" i)\n (.start (Thread. (fn []\n (cache-wrapped/lookup-or-miss cache-atom "my-key"\n (fn [k]\n (swap! invocations-counter inc)\n (Thread/sleep 3000)\n "some value"))\n (.countDown latch)))))\n (.await latch)\n (deref invocations-counter))
3+
1771874817202:(force (delay ((fn [] (throw (Exception. "bad"))))))
4+
1771874830891:(def x (force (delay ((fn [] (throw (Exception. "bad")))))))
5+
1771874832688:x
6+
1771875362757:(def p (promise))
7+
1771875366556:(realized? p)
8+
1771875376098:(deliver p 42)
9+
1771875377638:(realized? p)
10+
1771875417454:(force p)
11+
1771875420076:p
12+
1771875423512:@p
13+
1771879609917:(def x (delay (throw (Exception. "bad"))))
14+
1771879610748:x
15+
1771879620704:(force x)
16+
1771879622323:x
17+
1771879802447:(def x (with-meta (delay (throw (Exception. "bad"))) {::core.cache.added true}))
18+
1771885580329:(source force)

‎src/main/clojure/clojure/core/cache/wrapped.clj‎

Lines changed: 54 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -22,18 +22,52 @@
2222

2323
(set! *warn-on-reflection* true)
2424

25+
(def ^{:private true} default-wrapper-fn #(%1 %2))
26+
27+
;; Similar to clojure.lang.Delay, but will not memoize an exception and will
28+
;; instead retry.
29+
;; fun - the function, never nil
30+
;; available? - indicates a memoized value is available, volatile for visibility
31+
;; value - the value (if available) - volatile for visibility
32+
(deftype RetryingDelay [fun ^:volatile-mutable available? ^:volatile-mutable value]
33+
clojure.lang.IDeref
34+
(deref [this]
35+
;; first check (safe with volatile flag)
36+
(if available?
37+
value
38+
(locking fun
39+
;; second check (race condition with locking)
40+
(if available?
41+
value
42+
;; fun may throw - will retry on next deref
43+
(let [v (fun)]
44+
;; this ordering is important - MUST set value before setting available?
45+
;; or you have a race with the first check above
46+
(set! value v)
47+
(set! available? true)
48+
v)))))
49+
clojure.lang.IPending
50+
(isRealized [this]
51+
available?))
52+
53+
(defn- d-lay [fun]
54+
(->RetryingDelay fun false nil))
55+
56+
(defn- r-force [maybe-d-lay]
57+
(if (instance? RetryingDelay maybe-d-lay)
58+
(deref maybe-d-lay)
59+
maybe-d-lay))
60+
2561
(defn lookup
2662
"Retrieve the value associated with `e` if it exists, else `nil` in
2763
the 2-arg case. Retrieve the value associated with `e` if it exists,
2864
else `not-found` in the 3-arg case.
2965
3066
Reads from the current version of the atom."
3167
([cache-atom e]
32-
(force (c/lookup @cache-atom e)))
68+
(r-force (c/lookup @cache-atom e)))
3369
([cache-atom e not-found]
34-
(force (c/lookup @cache-atom e not-found))))
35-
36-
(def ^{:private true} default-wrapper-fn #(%1 %2))
70+
(r-force (c/lookup @cache-atom e not-found))))
3771

3872
(defn lookup-or-miss
3973
"Retrieve the value associated with `e` if it exists, else compute the
@@ -48,25 +82,24 @@
4882
([cache-atom e value-fn]
4983
(lookup-or-miss cache-atom e default-wrapper-fn value-fn))
5084
([cache-atom e wrap-fn value-fn]
51-
(let [d-new-value (delay (wrap-fn value-fn e))]
52-
(loop [n 0
53-
v (force (c/lookup (swap! cache-atom
54-
c/through-cache
55-
e
56-
default-wrapper-fn
57-
(fn [_] d-new-value))
58-
e
59-
::expired))]
85+
(let [d-new-value (d-lay #(wrap-fn value-fn e))
86+
hit-or-miss
87+
(fn []
88+
(try
89+
(r-force (c/lookup (swap! cache-atom
90+
c/through-cache
91+
e
92+
default-wrapper-fn
93+
(fn [_] d-new-value))
94+
e
95+
::expired))
96+
(catch Throwable t
97+
(swap! cache-atom c/evict e)
98+
(throw t))))]
99+
(loop [n 0 v (hit-or-miss)]
60100
(when (< n 10)
61101
(if (= ::expired v)
62-
(recur (inc n)
63-
(force (c/lookup (swap! cache-atom
64-
c/through-cache
65-
e
66-
default-wrapper-fn
67-
(fn [_] d-new-value))
68-
e
69-
::expired)))
102+
(recur (inc n) (hit-or-miss))
70103
v))))))
71104

72105
(defn has?

‎src/test/clojure/clojure/core/cache/wrapped_test.clj‎

Lines changed: 52 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -43,33 +43,70 @@
4343
start (System/currentTimeMillis)]
4444
(loop [n 0]
4545
(if-not (c/lookup-or-miss cache :a (constantly 42))
46-
(do
47-
(is false (str "Failure on call " n)))
48-
(if (< n limit)
49-
(recur (+ 1 n)))))
46+
(is false (str "Failure on call " n))
47+
(when (< n limit)
48+
(recur (inc n)))))
5049
(println "ttl test completed" limit "calls in"
5150
(- (System/currentTimeMillis) start) "ms")))
5251

5352
(deftest cache-stampede
54-
(let [thread-count 100
53+
(let [thread-count 1000
5554
cache-atom (-> {}
5655
(cache/ttl-cache-factory :ttl 120000)
5756
(cache/lu-cache-factory :threshold 100)
5857
(atom))
5958
latch (java.util.concurrent.CountDownLatch. thread-count)
6059
invocations-counter (atom 0)
61-
values (atom [])]
62-
(dotimes [_ thread-count]
63-
(.start (Thread. (fn []
64-
(swap! values conj
65-
(c/lookup-or-miss cache-atom "my-key"
66-
(fn [_]
67-
(swap! invocations-counter inc)
68-
(Thread/sleep 3000)
69-
"some value")))
70-
(.countDown latch)))))
60+
freds (atom [])
61+
values (atom [])
62+
exes (atom 0)]
63+
(dotimes [n thread-count]
64+
(if (<= n (rand-int thread-count))
65+
(swap! freds conj
66+
(Thread.
67+
(fn []
68+
(try
69+
(swap! values conj
70+
(c/lookup-or-miss cache-atom "my-key"
71+
(fn [_]
72+
(throw (Exception. "Bad")))))
73+
(catch Exception e
74+
(swap! exes inc)
75+
(is (= "Bad" (.getMessage e)))))
76+
(.countDown latch))))
77+
(swap! freds conj
78+
(Thread.
79+
(fn []
80+
(try
81+
(swap! values conj
82+
(c/lookup-or-miss cache-atom "my-key"
83+
(fn [_]
84+
(swap! invocations-counter inc)
85+
(Thread/sleep 3000)
86+
"some value")))
87+
(catch Exception _
88+
(is false "Unexpected cached exception")))
89+
(.countDown latch))))))
90+
91+
(run! #(.start ^Thread %) @freds)
7192

7293
(.await latch)
94+
(println "cache-stampede test completed with" @invocations-counter
95+
"successful invocation(s)\n\tand" (count @values)
96+
"successful lookups in" thread-count
97+
"threads with" @exes "exceptions")
7398
(is (= 1 (deref invocations-counter)))
99+
(is (= (- thread-count @exes) (count @values)))
74100
(doseq [v @values]
75101
(is (= "some value" v)))))
102+
103+
(deftest cache-thrown
104+
(let [c (c/basic-cache-factory {})]
105+
(is (thrown? Exception
106+
(c/lookup-or-miss c :a
107+
(fn [_]
108+
(throw (Exception. "bad"))))))
109+
;; cache should not have :a in it
110+
(is (= nil (c/lookup c :a)))
111+
;; and the cache should still be empty
112+
(is (= 0 (c/size c)))))

0 commit comments

Comments
 (0)