clojure - 一个人可以监控 STM 的争用级别吗？

Question

有什么方法可以查询 Clojure 的 STM 事务是否正在重试，以及以什么速率重试？

Answer 1

您可以观察history countref 的 ，这将表明它存在争用：

user=> (def my-ref (ref 0 :min-history 1))
#'user/my-ref
user=> (ref-history-count my-ref)
0
user=> (dosync (alter my-ref inc))
1
user=> (ref-history-count my-ref)
1

历史计数并不直接代表争用。相反，它表示为服务并发读取而维护的过去值的数量。

历史的大小受min和max值的限制。默认情况下，它们分别是0和10，但您可以在创建时更改它们ref（见上文）。由于默认min-history情况0下，您通常不会看到ref-history-count返回非零值，除非在 ref 上存在争用。

history count在此处查看更多讨论： https ://groups.google.com/forum/?fromgroups#!topic/clojure/n_MKCoa870o

我认为目前没有任何方法clojure.core可以观察 STM 交易的速率。你当然可以做类似于@Chouser 在他的历史压力测试中所做的事情：

(dosync
    (swap! try-count inc)
    ...)

即在事务中增加一个计数器。每次尝试事务时都会发生增量。如果try-count大于1，则重试事务。

Answer 2

通过引入命名的 dosync 块和提交计数（命名的 dosync 成功的时间），可以很容易地跟踪线程重试给定事务的时间。

(def ^{:doc "ThreadLocal<Map<TxName, Map<CommitNumber, TriesCount>>>"}
  local-tries (let [l (ThreadLocal.)]
                (.set l {})
                l))

(def ^{:doc "Map<TxName, Int>"}
  commit-number (ref {}))

(def history ^{:doc "Map<ThreadId, Map<TxName, Map<CommitNumber, TriesCount>>>"}
  (atom {}))

(defn report [_ thread-id tries]
  (swap! history assoc thread-id tries))

(def reporter (agent nil))

(defmacro dosync [tx-name & body]
  `(clojure.core/dosync
    (let [cno# (@commit-number ~tx-name 0)
          tries# (update-in (.get local-tries) [~tx-name] update-in [cno#] (fnil inc 0))]
      (.set local-tries tries#)
      (send reporter report (.getId (Thread/currentThread)) tries#))
    ~@body
    (alter commit-number update-in [~tx-name] (fnil inc 0))))

鉴于以下示例...

(def foo (ref {}))

(def bar (ref {}))

(defn x []
  (dosync :x ;; `:x`: the tx-name.
          (let [r (rand-int 2)]
            (alter foo assoc r (rand))
            (Thread/sleep (rand-int 400))
            (alter bar assoc (rand-int 2) (@foo r)))))

(dotimes [i 4]
  (future
   (dotimes [i 10]
     (x))))

...@history评估为：

;; {thread-id {tx-name {commit-number tries-count}}}
{40 {:x {3 1, 2 4, 1 3, 0 1}}, 39 {:x {2 1, 1 3, 0 1}}, ...}

Answer 3

这种额外的实现方式要简单得多。

;; {thread-id retries-of-latest-tx}
(def tries (atom {}))

;; The max amount of tries any thread has performed
(def max-tries (atom 0))

(def ninc (fnil inc 0))

(def reporter (agent nil))

(defn report [_ tid]
  (swap! max-tries #(max % (get @tries tid 0)))
  (swap! tries update-in [tid] (constantly 0)))

(defmacro dosync [& body]
  `(clojure.core/dosync
    (swap! tries update-in [(.getId (Thread/currentThread))] ninc)
    (commute commit-id inc)
    (send reporter report (.getId (Thread/currentThread)))
    ~@body))