HollowSparseIntegerSet: skip the clone and CAS when set()/clear() would not change the bit

hollow/src/main/java/com/netflix/hollow/core/index/HollowSparseIntegerSet.java

@@ -349,14 +349,20 @@ void set(int i) {
                 Bucket currentBucket = buckets.get(index);
                 if (currentBucket != null) {
                     longAtIndex = currentBucket.idx;
-                    longs = currentBucket.longs.clone();
+                    longs = currentBucket.longs;
                 }
 
                 boolean isLongInitialized = (longAtIndex & bitInIndex) != 0;
                 if (isLongInitialized) {
 
                     // if a long value is set, the find the correct offset to determine which long in longs to use.
                     int offset = getOffset(longAtIndex, bitInIndex);
+
+                    // if the bit is already set there is nothing to do, skip the clone and the CAS.
+                    if ((longs[offset] & whichBitInLong) != 0) return;
+
+                    // published longs are never mutated, so copy before updating.
+                    longs = longs.clone();
                     longs[offset] |= whichBitInLong;// or preserves previous set operations in this long.
 
                 } else if (longAtIndex == 0) {
@@ -413,7 +419,7 @@ void clear(int i) {
                 Bucket currentBucket = buckets.get(index);
                 if (currentBucket == null) return;
                 long longAtIndex = currentBucket.idx;
-                long[] longs = currentBucket.longs.clone();
+                long[] longs = currentBucket.longs;
 
                 // find which bit in index will point to the long in bb
                 long whichLong = i >>> LONG_SHIFT;
@@ -426,11 +432,16 @@ void clear(int i) {
                 int offset = getOffset(longAtIndex, bitInIndex);
                 long value = longs[offset];
 
+                // if the bit is already clear there is nothing to do, skip the clone and the CAS.
+                if ((value & whichBitInLong) == 0) return;
+
                 // unset whichBitInIndex in value
                 // to clear 3rd bit (00100 whichBitInLong) in 00101(value), & with 11011 to get 00001
                 long updatedValue = value & ~whichBitInLong;
                 boolean isBucketEmpty = false;
                 if (updatedValue != 0) {
+                    // published longs are never mutated, so copy before updating.
+                    longs = longs.clone();
                     longs[offset] = updatedValue;
                 } else {
 

hollow/src/test/java/com/netflix/hollow/core/index/SparseBitSetTest.java

@@ -295,4 +295,77 @@ public void testResize() {
         Assert.assertTrue(resizedSparsedBitSet.get(4095));
         Assert.assertTrue(resizedSparsedBitSet.get(512));
     }
+
+    @Test
+    public void testSetAndClearAreIdempotent() {
+        sparseBitSet = new HollowSparseIntegerSet.SparseBitSet(8192);
+
+        // bits 0 and 1 share a long, bit 64 is in the next long of the same bucket
+        sparseBitSet.set(0);
+        sparseBitSet.set(0);
+        sparseBitSet.set(1);
+        sparseBitSet.set(64);
+        sparseBitSet.set(64);
+        Assert.assertTrue(sparseBitSet.get(0));
+        Assert.assertTrue(sparseBitSet.get(1));
+        Assert.assertTrue(sparseBitSet.get(64));
+        Assert.assertEquals(3, sparseBitSet.cardinality());
+
+        // clearing a bit that is not set, in an initialized long, leaves the others intact
+        sparseBitSet.clear(2);
+        sparseBitSet.clear(65);
+        Assert.assertEquals(3, sparseBitSet.cardinality());
+
+        sparseBitSet.clear(0);
+        sparseBitSet.clear(0);
+        Assert.assertFalse(sparseBitSet.get(0));
+        Assert.assertTrue(sparseBitSet.get(1));
+        Assert.assertTrue(sparseBitSet.get(64));
+        Assert.assertEquals(2, sparseBitSet.cardinality());
+
+        sparseBitSet.clear(1);
+        sparseBitSet.clear(64);
+        sparseBitSet.clear(64);
+        Assert.assertEquals(0, sparseBitSet.cardinality());
+        Assert.assertEquals(-1, sparseBitSet.findMaxValue());
+    }
+
+    @Test
+    public void testRepeatedSetAndClearMultipleThread() {
+        int maxValue = 8192;
+        int repetitions = 50;
+        sparseBitSet = new HollowSparseIntegerSet.SparseBitSet(maxValue);
+
+        // every thread repeatedly sets every value starting at a different offset, so threads race both to
+        // set fresh bits in shared buckets and to re-set bits that are already set.
+        SimultaneousExecutor executor = new SimultaneousExecutor(getClass(), "test");
+        int parallelism = executor.getMaximumPoolSize();
+        for (int t = 0; t < parallelism; t++) {
+            int offset = t * 37;
+            executor.execute(() -> {
+                for (int r = 0; r < repetitions; r++)
+                    for (int i = 0; i <= maxValue; i++)
+                        sparseBitSet.set((i + offset) % (maxValue + 1));
+            });
+        }
+        executor.awaitUninterruptibly();
+        Assert.assertEquals(maxValue + 1, sparseBitSet.cardinality());
+
+        // likewise for clearing every odd value.
+        executor = new SimultaneousExecutor(getClass(), "test");
+        for (int t = 0; t < parallelism; t++) {
+            int offset = t * 19;
+            int numOdd = (maxValue + 1) / 2;
+            executor.execute(() -> {
+                for (int r = 0; r < repetitions; r++)
+                    for (int k = 0; k < numOdd; k++)
+                        sparseBitSet.clear(2 * ((k + offset) % numOdd) + 1);
+            });
+        }
+        executor.awaitUninterruptibly();
+
+        Assert.assertEquals(maxValue / 2 + 1, sparseBitSet.cardinality());
+        for (int i = 0; i <= maxValue; i++)
+            Assert.assertEquals(i % 2 == 0, sparseBitSet.get(i));
+    }
 }