do sync clock using atoms

sync.clj

(ns a)

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(defn clj-mk-ref
  [data]
  (ref {:version 0
        :data data}))

(defn clj-sync-write
  [r1 r2 f1 f2]
  (dosync
    (let [new-r1 (-> @r1
                   (update-in [:version] inc)
                   (update-in [:data] f1))
          new-r2 (-> @r2
                   (update-in [:version] inc)
                   (update-in [:data] f2))]
      (ref-set r1 new-r1)
      (ref-set r2 new-r2))))

(defn clj-sync-read
  [r1 r2]
  (dosync
    (ensure r1)
    (ensure r2)
    [(:data @r1) (:data @r2)]))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(defn my-mk-ref
  [data]
  (atom {:version 0
         :data data}))

(defn my-sync-write
  [r1 r2 f1 f2]
  (loop []
    (let [r1-v (:version @r1)
          r2-v (:version @r2)
          new-r1 (-> @r1
                   (update-in [:version] inc)
                   (update-in [:data] f1))
          new-r2 (-> @r2
                   (update-in [:version] inc)
                   (update-in [:data] f2))]
      (when-not (locking r1
                  (locking r2
                    (if (and
                          (= r1-v (:version @r1))
                          (= r2-v (:version @r2)))
                      (do
                        (reset! r1 new-r1)
                        (reset! r2 new-r2)
                        true)
                      false)))
        (recur)))))

(defn my-sync-read
  [r1 r2]
  (locking r1
    (locking r2
      [(:data @r1) (:data @r2)])))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(def rnd (java.util.Random.))

(defn chng
  [data f & args]
  (reduce (fn [d _]
            (apply update-in d [(.nextInt rnd (count d))] f args))
    data
    (range 20)))

(defmacro start-thread
  [name & body]
  `(let [f# (fn [] ~@body)
         th# (Thread. f# ~name)]
     (.start th#)
     th#))

(defn writer-thread-fn
  [sync-write-fn r1 r2 secs id]
  (start-thread (str "writer" id)
    (let [end-time (+ (System/currentTimeMillis)
                     (* 1000 secs))]
      (while (< (System/currentTimeMillis) end-time)
        (sync-write-fn r1 r2 #(chng % + 10) #(chng % - 10))
        (Thread/sleep 10)))
    (.println System/out (str "Done " (Thread/currentThread)))))

(defn reader-thread-fn
  [sync-read-fn r1 r2 secs id]
  (start-thread (str "reader" id)
    (let [end-time (+ (System/currentTimeMillis)
                     (* 1000 secs))]
      (while (< (System/currentTimeMillis) end-time)
        (let [sum (reduce + (mapcat identity (sync-read-fn r1 r2)))]
          (if-not (= 9900 sum)
            (.println System/err (str "sum=" sum " wrong in " (Thread/currentThread)))))
        (Thread/sleep 100)))
    (.println System/out (str "Done " (Thread/currentThread)))))

(defn test-sync
  [mk-ref-fn sync-write-fn sync-read-fn secs numr numw]
  (let [r1 (mk-ref-fn (vec (range 100)))
        r2 (mk-ref-fn (vec (range 100)))
        writer-threads (doall (map (partial writer-thread-fn sync-write-fn r1 r2 secs) (range numw)))
        reader-threads (doall (map (partial reader-thread-fn sync-read-fn r1 r2 secs) (range numr)))]
    [r1 r2 reader-threads writer-threads]))

(def test-my-sync (partial test-sync my-mk-ref my-sync-write my-sync-read))

(def test-clj-sync (partial test-sync clj-mk-ref clj-sync-write clj-sync-read))

; (def k (test-my-sync 10 5 70))  : [1 5 70 => 3975] [10 5 70 => 46733] [100 5 70 => 467767]
; (def k (test-clj-sync 10 5 70)) : [1 5 70 => 2145] [10 5 70 => 26049] [100 5 70 => 258859]