AtomicPut.patch

Index: jctools-core/src/main/java/org/jctools/maps/NonBlockingHashMap.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- jctools-core/src/main/java/org/jctools/maps/NonBlockingHashMap.java	(revision a2191c1e0243c8cd4690990d6b250b26a3ca8bfa)
+++ jctools-core/src/main/java/org/jctools/maps/NonBlockingHashMap.java	(revision )
@@ -650,134 +650,130 @@
     int reprobe_cnt=0;
     Object K=null, V=null;
     Object[] newkvs=null;
-    while( true ) {             // Spin till we get a Key slot
-      V = val(kvs,idx);         // Get old value (before volatile read below!)
-      K = key(kvs,idx);         // Get current key
-      if( K == null ) {         // Slot is free?
-        // Found an empty Key slot - which means this Key has never been in
-        // this table.  No need to put a Tombstone - the Key is not here!
-        if( putval == TOMBSTONE ) return putval; // Not-now & never-been in this table
-        if( expVal == MATCH_ANY ) return null;   // Will not match, even after K inserts
-        // Claim the null key-slot
-        if( CAS_key(kvs,idx, null, key ) ) { // Claim slot for Key
-          chm._slots.add(1);      // Raise key-slots-used count
-          hashes[idx] = fullhash; // Memoize fullhash
-          break;                  // Got it!
-        }
-        // CAS to claim the key-slot failed.
-        //
-        // This re-read of the Key points out an annoying short-coming of Java
-        // CAS.  Most hardware CAS's report back the existing value - so that
-        // if you fail you have a *witness* - the value which caused the CAS to
-        // fail.  The Java API turns this into a boolean destroying the
-        // witness.  Re-reading does not recover the witness because another
-        // thread can write over the memory after the CAS.  Hence we can be in
-        // the unfortunate situation of having a CAS fail *for cause* but
-        // having that cause removed by a later store.  This turns a
-        // non-spurious-failure CAS (such as Azul has) into one that can
-        // apparently spuriously fail - and we avoid apparent spurious failure
-        // by not allowing Keys to ever change.
+    while ( true ) {              // Spin till we insert a value
+      while( true ) {             // Spin till we get a Key slot
+        V = val(kvs,idx);         // Get old value (before volatile read below!)
+        K = key(kvs,idx);         // Get current key
+        if( K == null ) {         // Slot is free?
+          // Found an empty Key slot - which means this Key has never been in
+          // this table.  No need to put a Tombstone - the Key is not here!
+          if( putval == TOMBSTONE ) return putval; // Not-now & never-been in this table
+          if( expVal == MATCH_ANY ) return null;   // Will not match, even after K inserts
+          // Claim the null key-slot
+          if( CAS_key(kvs,idx, null, key ) ) { // Claim slot for Key
+            chm._slots.add(1);      // Raise key-slots-used count
+            hashes[idx] = fullhash; // Memoize fullhash
+            break;                  // Got it!
+          }
+          // CAS to claim the key-slot failed.
+          //
+          // This re-read of the Key points out an annoying short-coming of Java
+          // CAS.  Most hardware CAS's report back the existing value - so that
+          // if you fail you have a *witness* - the value which caused the CAS to
+          // fail.  The Java API turns this into a boolean destroying the
+          // witness.  Re-reading does not recover the witness because another
+          // thread can write over the memory after the CAS.  Hence we can be in
+          // the unfortunate situation of having a CAS fail *for cause* but
+          // having that cause removed by a later store.  This turns a
+          // non-spurious-failure CAS (such as Azul has) into one that can
+          // apparently spuriously fail - and we avoid apparent spurious failure
+          // by not allowing Keys to ever change.
 
-        // Volatile read, to force loads of K to retry despite JIT, otherwise
-        // it is legal to e.g. haul the load of "K = key(kvs,idx);" outside of
-        // this loop (since failed CAS ops have no memory ordering semantics).
-        int dummy = DUMMY_VOLATILE;
-        continue;
-      }
-      // Key slot was not null, there exists a Key here
+          // Volatile read, to force loads of K to retry despite JIT, otherwise
+          // it is legal to e.g. haul the load of "K = key(kvs,idx);" outside of
+          // this loop (since failed CAS ops have no memory ordering semantics).
+          int dummy = DUMMY_VOLATILE;
+          continue;
+        }
+        // Key slot was not null, there exists a Key here
 
-      // We need a volatile-read here to preserve happens-before semantics on
-      // newly inserted Keys.  If the Key body was written just before inserting
-      // into the table a Key-compare here might read the uninitialized Key body.
-      // Annoyingly this means we have to volatile-read before EACH key compare.
-      newkvs = chm._newkvs;     // VOLATILE READ before key compare
+        // We need a volatile-read here to preserve happens-before semantics on
+        // newly inserted Keys.  If the Key body was written just before inserting
+        // into the table a Key-compare here might read the uninitialized Key body.
+        // Annoyingly this means we have to volatile-read before EACH key compare.
+        newkvs = chm._newkvs;     // VOLATILE READ before key compare
 
-      if( keyeq(K,key,hashes,idx,fullhash) )
-        break;                  // Got it!
+        if( keyeq(K,key,hashes,idx,fullhash) )
+          break;                  // Got it!
 
-      // get and put must have the same key lookup logic!  Lest 'get' give
-      // up looking too soon.
-      //topmap._reprobes.add(1);
-      if( ++reprobe_cnt >= reprobe_limit(len) || // too many probes or
-          K == TOMBSTONE ) { // found a TOMBSTONE key, means no more keys
-        // We simply must have a new table to do a 'put'.  At this point a
-        // 'get' will also go to the new table (if any).  We do not need
-        // to claim a key slot (indeed, we cannot find a free one to claim!).
-        newkvs = chm.resize(topmap,kvs);
-        if( expVal != null ) topmap.help_copy(newkvs); // help along an existing copy
-        return putIfMatch(topmap,newkvs,key,putval,expVal);
-      }
+        // get and put must have the same key lookup logic!  Lest 'get' give
+        // up looking too soon.
+        //topmap._reprobes.add(1);
+        if( ++reprobe_cnt >= reprobe_limit(len) || // too many probes or
+            K == TOMBSTONE ) { // found a TOMBSTONE key, means no more keys
+          // We simply must have a new table to do a 'put'.  At this point a
+          // 'get' will also go to the new table (if any).  We do not need
+          // to claim a key slot (indeed, we cannot find a free one to claim!).
+          newkvs = chm.resize(topmap,kvs);
+          if( expVal != null ) topmap.help_copy(newkvs); // help along an existing copy
+          return putIfMatch(topmap,newkvs,key,putval,expVal);
+        }
 
-      idx = (idx+1)&(len-1); // Reprobe!
-    } // End of spinning till we get a Key slot
+        idx = (idx+1)&(len-1); // Reprobe!
+      } // End of spinning till we get a Key slot
 
-    // ---
-    // Found the proper Key slot, now update the matching Value slot.  We
-    // never put a null, so Value slots monotonically move from null to
-    // not-null (deleted Values use Tombstone).  Thus if 'V' is null we
-    // fail this fast cutout and fall into the check for table-full.
-    if( putval == V ) return V; // Fast cutout for no-change
+      // ---
+      // Found the proper Key slot, now update the matching Value slot.  We
+      // never put a null, so Value slots monotonically move from null to
+      // not-null (deleted Values use Tombstone).  Thus if 'V' is null we
+      // fail this fast cutout and fall into the check for table-full.
+      if( putval == V ) return V; // Fast cutout for no-change
 
-    // See if we want to move to a new table (to avoid high average re-probe
-    // counts).  We only check on the initial set of a Value from null to
-    // not-null (i.e., once per key-insert).  Of course we got a 'free' check
-    // of newkvs once per key-compare (not really free, but paid-for by the
-    // time we get here).
-    if( newkvs == null &&       // New table-copy already spotted?
-        // Once per fresh key-insert check the hard way
-        ((V == null && chm.tableFull(reprobe_cnt,len)) ||
-         // Or we found a Prime, but the JMM allowed reordering such that we
-         // did not spot the new table (very rare race here: the writing
-         // thread did a CAS of _newkvs then a store of a Prime.  This thread
-         // reads the Prime, then reads _newkvs - but the read of Prime was so
-         // delayed (or the read of _newkvs was so accelerated) that they
-         // swapped and we still read a null _newkvs.  The resize call below
-         // will do a CAS on _newkvs forcing the read.
-         V instanceof Prime) )
-      newkvs = chm.resize(topmap,kvs); // Force the new table copy to start
-    // See if we are moving to a new table.
-    // If so, copy our slot and retry in the new table.
-    if( newkvs != null )
-      return putIfMatch(topmap,chm.copy_slot_and_check(topmap,kvs,idx,expVal),key,putval,expVal);
+      // See if we want to move to a new table (to avoid high average re-probe
+      // counts).  We only check on the initial set of a Value from null to
+      // not-null (i.e., once per key-insert).  Of course we got a 'free' check
+      // of newkvs once per key-compare (not really free, but paid-for by the
+      // time we get here).
+      if( newkvs == null &&       // New table-copy already spotted?
+          // Once per fresh key-insert check the hard way
+          ((V == null && chm.tableFull(reprobe_cnt,len)) ||
+           // Or we found a Prime, but the JMM allowed reordering such that we
+           // did not spot the new table (very rare race here: the writing
+           // thread did a CAS of _newkvs then a store of a Prime.  This thread
+           // reads the Prime, then reads _newkvs - but the read of Prime was so
+           // delayed (or the read of _newkvs was so accelerated) that they
+           // swapped and we still read a null _newkvs.  The resize call below
+           // will do a CAS on _newkvs forcing the read.
+           V instanceof Prime) )
+        newkvs = chm.resize(topmap,kvs); // Force the new table copy to start
+      // See if we are moving to a new table.
+      // If so, copy our slot and retry in the new table.
+      if( newkvs != null )
+        return putIfMatch(topmap,chm.copy_slot_and_check(topmap,kvs,idx,expVal),key,putval,expVal);
 
-    // ---
-    // We are finally prepared to update the existing table
-    assert !(V instanceof Prime);
+      // ---
+      // We are finally prepared to update the existing table
+      assert !(V instanceof Prime);
 
-    // Must match old, and we do not?  Then bail out now.  Note that either V
-    // or expVal might be TOMBSTONE.  Also V can be null, if we've never
-    // inserted a value before.  expVal can be null if we are called from
-    // copy_slot.
+      // Must match old, and we do not?  Then bail out now.  Note that either V
+      // or expVal might be TOMBSTONE.  Also V can be null, if we've never
+      // inserted a value before.  expVal can be null if we are called from
+      // copy_slot.
 
-    if( expVal != NO_MATCH_OLD && // Do we care about expected-Value at all?
-        V != expVal &&            // No instant match already?
-        (expVal != MATCH_ANY || V == TOMBSTONE || V == null) &&
-        !(V==null && expVal == TOMBSTONE) &&    // Match on null/TOMBSTONE combo
-        (expVal == null || !expVal.equals(V)) ) // Expensive equals check at the last
-      return V;                                 // Do not update!
+      if( expVal != NO_MATCH_OLD && // Do we care about expected-Value at all?
+          V != expVal &&            // No instant match already?
+          (expVal != MATCH_ANY || V == TOMBSTONE || V == null) &&
+          !(V==null && expVal == TOMBSTONE) &&    // Match on null/TOMBSTONE combo
+          (expVal == null || !expVal.equals(V)) ) // Expensive equals check at the last
+        return V;                                 // Do not update!
 
-    // Actually change the Value in the Key,Value pair
-    if( CAS_val(kvs, idx, V, putval ) ) {
-      // CAS succeeded - we did the update!
-      // Both normal put's and table-copy calls putIfMatch, but table-copy
-      // does not (effectively) increase the number of live k/v pairs.
-      if( expVal != null ) {
-        // Adjust sizes - a striped counter
-        if(  (V == null || V == TOMBSTONE) && putval != TOMBSTONE ) chm._size.add( 1);
-        if( !(V == null || V == TOMBSTONE) && putval == TOMBSTONE ) chm._size.add(-1);
-      }
-    } else {                    // Else CAS failed
-      V = val(kvs,idx);         // Get new value
-      // If a Prime'd value got installed, we need to re-run the put on the
-      // new table.  Otherwise we lost the CAS to another racing put.
-      // Simply retry from the start.
-      if( V instanceof Prime )
-        return putIfMatch(topmap,chm.copy_slot_and_check(topmap,kvs,idx,expVal),key,putval,expVal);
-    }
-    // Win or lose the CAS, we are done.  If we won then we know the update
-    // happened as expected.  If we lost, it means "we won but another thread
-    // immediately stomped our update with no chance of a reader reading".
-    return (V==null && expVal!=null) ? TOMBSTONE : V;
+      // Actually change the Value in the Key,Value pair
+      if( CAS_val(kvs, idx, V, putval ) ) {
+        // CAS succeeded - we did the update!
+        // Both normal put's and table-copy calls putIfMatch, but table-copy
+        // does not (effectively) increase the number of live k/v pairs.
+        if( expVal != null ) {
+          // Adjust sizes - a striped counter
+          if(  (V == null || V == TOMBSTONE) && putval != TOMBSTONE ) chm._size.add( 1);
+          if( !(V == null || V == TOMBSTONE) && putval == TOMBSTONE ) chm._size.add(-1);
+        }
+      } else {                    // Else CAS failed
+        // we're racing with other writes to same key, repeat from the top to
+        // maintain getAndSet semantics.
+        continue;
+      }
+      return (V==null && expVal!=null) ? TOMBSTONE : V;
+    }
   }
 
   // --- help_copy ---------------------------------------------------------
Index: jctools-core/src/test/java/org/jctools/maps/KeyAtomicityTest.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- jctools-core/src/test/java/org/jctools/maps/KeyAtomicityTest.java	(revision )
+++ jctools-core/src/test/java/org/jctools/maps/KeyAtomicityTest.java	(revision )
@@ -0,0 +1,103 @@
+package org.jctools.maps;
+
+import org.junit.Test;
+
+import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.Assert.assertEquals;
+
+public class KeyAtomicityTest
+{
+    static final int THREAD_SEGMENT = 1000000;
+
+    @Test
+    public void putReturnValuesAreDistinct() throws Exception
+    {
+        Map<String, Long> map = new NonBlockingHashMap<>();
+        map.put("K", -1l);
+        int processors = Runtime.getRuntime().availableProcessors();
+        CountDownLatch ready = new CountDownLatch(processors);
+        CountDownLatch start = new CountDownLatch(1);
+        CountDownLatch done = new CountDownLatch(processors);
+        AtomicBoolean keepRunning = new AtomicBoolean(true);
+        PutKey[] putKeys = new PutKey[processors];
+        for (int i = 0; i < processors; i++)
+        {
+            putKeys[i] = new PutKey(map, "K", keepRunning, ready, start, done, i * THREAD_SEGMENT);
+            Thread t = new Thread(putKeys[i]);
+            t.setName("Putty McPutkey-"+i);
+            t.start();
+        }
+        ready.await();
+        start.countDown();
+        Thread.sleep(1000);
+        keepRunning.set(false);
+        done.await();
+        Set<Long> values = new HashSet((int)(processors*THREAD_SEGMENT));
+        long totalKeys = 0;
+        for (PutKey putKey : putKeys)
+        {
+            values.addAll(putKey.values);
+            totalKeys += putKey.endIndex - putKey.startIndex;
+        }
+        assertEquals(totalKeys, values.size());
+    }
+
+    static class PutKey implements Runnable
+    {
+        final Map<String, Long> map;
+        final String key;
+        final AtomicBoolean keepRunning;
+        final CountDownLatch ready;
+        final CountDownLatch start;
+        final CountDownLatch done;
+        final int startIndex;
+        int endIndex;
+
+        List<Long> values = new ArrayList<>(THREAD_SEGMENT);
+
+        PutKey(
+            Map<String, Long> map,
+            String key,
+            AtomicBoolean keepRunning, CountDownLatch ready,
+            CountDownLatch start,
+            CountDownLatch done,
+            int startIndex)
+        {
+            this.map = map;
+            this.key = key;
+            this.keepRunning = keepRunning;
+            this.ready = ready;
+            this.start = start;
+            this.done = done;
+            this.startIndex = startIndex;
+            assert startIndex >= 0 && startIndex + THREAD_SEGMENT > 0;
+        }
+
+        @Override
+        public void run()
+        {
+            ready.countDown();
+            try
+            {
+                start.await();
+            }
+            catch (InterruptedException e)
+            {
+                e.printStackTrace();
+                return;
+            }
+            long limit = startIndex + THREAD_SEGMENT;
+            long v = startIndex;
+            String k = key;
+            for (; v < limit && keepRunning.get(); v++)
+            {
+                values.add(map.put(k, v));
+            }
+            endIndex = (int) v;
+            done.countDown();
+        }
+    }
+}