在 core.cache 中的多线程缓存竞态

Question

我们最近在 Web 上下文中部署了 core.cache，并发现存在多线程访问的缓存竞态问题。回过头来看很明显，但在文档中对竞态的引用使我以为这已经得到处理。

在 Slack 的帮助下，我这里有一个复现情况

(let [thread-count 20
      cache-atom (-> {}
                   (cache/ttl-cache-factory :ttl 120000)
                   (cache/lu-cache-factory :threshold 100)
                   (atom))
      latch (java.util.concurrent.CountDownLatch. thread-count)
      invocations-counter (atom 0)]
 (doseq [i (range thread-count)]
   (println "starting thread" i)
   (.start (Thread. (fn []
                     (cache-wrapped/lookup-or-miss cache-atom "my-key"
                                                   (fn [k]
                                                     (swap! invocations-counter inc)
                                                     (Thread/sleep 3000)
                                                     "some value"))
                     (.countDown latch)))))

 (.await latch)
 (deref invocations-counter))

我预计调用计数器会是 1，但它是 20（每个线程一次）。

Comments

根据使用 core.memoize 的建议，它按预期工作

    (let [thread-count 20
          invocations-counter (atom 0)
          expensive-function (fn [k]
                             (swap! invocations-counter inc)
                             (Thread/sleep 3000)
                             (str "value-" k))
          cache (-> {}
                        (cache/ttl-cache-factory :ttl 120000)
                          (cache/lu-cache-factory :threshold 100))
          memoized-function (memoize/memoizer expensive-function cache)
          锁存器（java.util.concurrent.CountDownLatch. 线程数）]
     （doseq [i (range 线程数)]
       （打印 "开始线程" i）
       （.start （Thread. （fn []
                           （memoized-function "my-key"）
                           （.countDown 锁存器）)))

     （.await 锁存器）
     （断言 (= 1 （deref 调用计数器）））

Answer 1

登录为https://clojure.atlassian.net/browse/CCACHE-65

Comments

谢谢，Alex！

Answer 2

lookup-or-miss 确实存在一个并发问题。在第 57 行和 67 行，延迟的解引用值将通过 swap! 插入缓存。这个延迟只能执行一次，然而 当许多线程同时出现相同的缓存未命中时，许多线程将
解引用一个值。结果，在这之前会有许多执行。

https://github.com/clojure/core.cache/blob/master/src/main/clojure/clojure/core/cache/wrapped.clj#L57
https://github.com/clojure/core.cache/blob/master/src/main/clojure/clojure/core/cache/wrapped.clj#L67

要解决这个问题，必须将包裹命名空间中的延迟存储在缓存中，并且只在检索时解引用这些延迟。

当前版本的解决方案是自己来做这个包裹。

;; clj -Sdeps '{:deps {org.clojure/core.cache {:mvn/version "1.0.225"}}}'

(require '[clojure.core.cache :as cache]
         '[clojure.core.cache.wrapped :as cache-wrapped])

(let [thread-count 20
      cache-atom (-> {}
                   (cache/ttl-cache-factory :ttl 120000)
                   (cache/lu-cache-factory :threshold 100)
                   (atom))
      latch (java.util.concurrent.CountDownLatch. thread-count)
      invocations-counter (atom 0)]
 (doseq [i (range thread-count)]
   (println "starting thread" i)
   (.start (Thread. (fn []
                     ;; Deref the outcome
                     @(cache-wrapped/lookup-or-miss cache-atom "my-key"
                                                   (fn [k]
                                                     ;; Put the action in a delay
                                                     (delay
                                                       (swap! invocations-counter inc)
                                                       (Thread/sleep 3000)
                                                       "some value")))
                     (.countDown latch)))))

 (.await latch)
 (deref invocations-counter))

Comments

我认为很难修复这个问题而不破坏现有代码，因为必须对这些新延迟进行包装/解包来通过API的每一条路径，包括种子需要包装所提供的基本中的每个值，这可能是已经组合的缓存数据结构。