atom watchers are not atomic with respect to reset!

Question

It is possible that two threads calling `reset!` on an atom can interleave, causing the corresponding watches to be called with the same old value but different new values. This contradicts the guarantee that atoms update atomically.


(defn reset-test []
  (let [my-atom (atom :start
                      :validator (fn [x] (Thread/sleep 100) true))
        watch-results (atom [])]
    (add-watch my-atom :watcher (fn [k a o n] (swap! watch-results conj [o n])))
  
    (future (reset! my-atom :next))
    (future (reset! my-atom :next))
    (Thread/sleep 500)
    @watch-results))

(reset-test)


Yields [[:start :next] [:start :next]]. Similar behavior can be observed when mixing reset! and swap!.

h2. Expected behavior

Under atomic circumstances, (reset-test) should yield [[:start :next] [:next :next]]. This would "serialize" the resets and give more accurate information to the watches. This is the same behavior one would achieve by using (swap! my-atom (constantly :next)).


(defn swap-test []
  (let [my-atom (atom :start
                      :validator (fn [x] (Thread/sleep 100) true))
        watch-results (atom [])]
    (add-watch my-atom :watcher (fn [k a o n] (swap! watch-results conj [o n])))
  
    (future (swap! my-atom (constantly :next)))
    (future (swap! my-atom (constantly :next)))
    (Thread/sleep 500)
    @watch-results))

(swap-test)


Yields [[:start :next] [:next :next]]. The principle of least surprise suggests that these two functions should yield similar output.
 
h3. Alternative expected behavior

It could be that atoms and reset! do not guarantee serialized updates with respect to calls to watches. In this case, it would be prudent to note this in the docstring for atom.

h2. Analysis

The code for Atom.reset non-atomically reads and sets the internal AtomicReference. This allows for multiple threads to interleave the gets and sets, resulting in holding a stale value when notifying watches. Note that this should not affect the new value, just the old value.

h2. Approach

Inside Atom.reset(), validation should happen first, then a loop calling compareAndSet on the internal state (similar to how it is implemented in swap()) should run until compareAndSet returns true. Note that this is still faster than the swap! constantly pattern shown above, since it only validates once and the tighter loop should have fewer interleavings. But it has the same watch behavior.


public Object reset(Object newval){
    validate(newval);
    for(;;)
        {
            Object oldval = state.get();
            if(state.compareAndSet(oldval, newval))
                {
                    notifyWatches(oldval, newval);
                    return newval;
                }
        }
}

Answer 1

Comment made by: ericnormand

I've made a test just to back up the timing claims I made above. If you run the file timingtest.clj, it will run code with both reset! and swap! constantly, with a validator that sleeps for 10ms. In both cases, it will print out the number of uniques (should be equal to number of reset!s, in this case 1000) and the time (using clojure.core/time). The timing numbers are relative to the machine, so should not be taken as absolutes. Instead, the ratio between them is what's important.

Run with: java -cp clojure-1.7.0-master-SNAPSHOT.jar clojure.main timingtest.clj

Results

Existing implementation:

"Elapsed time: 1265.228 msecs" Uniques with reset!: 140 "Elapsed time: 11609.686 msecs" Uniques with swap!: 1000 "Elapsed time: 7010.132 msecs" Uniques with swap! and reset!: 628

Note that the behaviors differ: swap! serializes the watchers, reset! does not (1. of uniques).

Suggested implementation:

"Elapsed time: 1268.778 msecs" Uniques with reset!: 1000 "Elapsed time: 11716.678 msecs" Uniques with swap!: 1000 "Elapsed time: 7015.994 msecs" Uniques with swap! and reset!: 1000

Same tests being run. This time, they both serialize watchers. Also, the timing has not changed significantly.

Answer 2

Comment made by: ericnormand

Adding atom-reset-atomic-watch-2015-06-30.patch. Includes test and implementation.

Answer 3

Reference: https://clojure.atlassian.net/browse/CLJ-1770 (reported by ericnormand)