Make TokenMetadata's ring version increments atomic

patch by Caleb Rackliffe; reviewed by Andrés de la Peña and Yifan Cai for CASSANDRA-16286

CHANGES.txt

@@ -1,4 +1,5 @@
 2.2.20
+ * Make TokenMetadata's ring version increments atomic (CASSANDRA-16286)
  * Remove OpenJDK log warning (CASSANDRA-15563)
  * Fix the histogram merge of the table metrics (CASSANDRA-16259)
 

src/java/org/apache/cassandra/locator/TokenMetadata.java

@@ -26,6 +26,8 @@ import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
+import javax.annotation.concurrent.GuardedBy;
+
 import com.google.common.base.Optional;
 import com.google.common.collect.*;
 import org.apache.commons.lang3.StringUtils;
@@ -106,7 +108,8 @@ public class TokenMetadata
     };
 
     // signals replication strategies that nodes have joined or left the ring and they need to recompute ownership
-    private volatile long ringVersion = 0;
+    @GuardedBy("lock")
+    private long ringVersion = 0;
 
     public TokenMetadata()
     {
@@ -443,7 +446,7 @@ public class TokenMetadata
             }
             endpointToHostIdMap.remove(endpoint);
             sortedTokens = sortTokens();
-            invalidateCachedRings();
+            invalidateCachedRingsUnsafe();
         }
         finally
         {
@@ -463,7 +466,7 @@ public class TokenMetadata
         {
             logger.info("Updating topology for {}", endpoint);
             topology.updateEndpoint(endpoint);
-            invalidateCachedRings();
+            invalidateCachedRingsUnsafe();
         }
         finally
         {
@@ -482,7 +485,7 @@ public class TokenMetadata
         {
             logger.info("Updating topology for all endpoints that have changed");
             topology.updateEndpoints();
-            invalidateCachedRings();
+            invalidateCachedRingsUnsafe();
         }
         finally
         {
@@ -510,7 +513,7 @@ public class TokenMetadata
                 }
             }
 
-            invalidateCachedRings();
+            invalidateCachedRingsUnsafe();
         }
         finally
         {
@@ -1088,7 +1091,7 @@ public class TokenMetadata
             movingEndpoints.clear();
             sortedTokens.clear();
             topology.clear();
-            invalidateCachedRings();
+            invalidateCachedRingsUnsafe();
         }
         finally
         {
@@ -1234,10 +1237,33 @@ public class TokenMetadata
 
     public long getRingVersion()
     {
-        return ringVersion;
+        lock.readLock().lock();
+
+        try
+        {
+            return ringVersion;
+        }
+        finally
+        {
+            lock.readLock().unlock();
+        }
     }
 
     public void invalidateCachedRings()
+    {   
+        lock.writeLock().lock();
+
+        try
+        {   
+            invalidateCachedRingsUnsafe();
+        }
+        finally
+        {
+            lock.writeLock().unlock();
+        }
+    }
+    
+    private void invalidateCachedRingsUnsafe()
     {
         ringVersion++;
         cachedTokenMap.set(null);

test/unit/org/apache/cassandra/locator/TokenMetadataTest.java

@@ -22,6 +22,9 @@ import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 import com.google.common.collect.Iterators;
 import com.google.common.collect.Multimap;
@@ -67,6 +70,33 @@ public class TokenMetadataTest
             assertEquals("Mismatch at index " + i + ": " + actual, token(expected[i]), actual.get(i));
     }
 
+    /**
+     * This test is very likely (but not guaranteed) to fail if ring invalidations are ever allowed to interleave.
+     */
+    @Test
+    public void testConcurrentInvalidation() throws InterruptedException
+    {
+        long startVersion = tmd.getRingVersion();
+
+        ExecutorService pool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() + 1);
+        
+        int invalidations = 1024;
+        
+        for (int i = 0; i < invalidations; i++)
+            pool.execute(new Runnable()
+            {
+                public void run()
+                {
+                    tmd.invalidateCachedRings();
+                }
+            });
+
+        pool.shutdown();
+        
+        assertTrue(pool.awaitTermination(30, TimeUnit.SECONDS));
+        assertEquals(invalidations + startVersion, tmd.getRingVersion());
+    }
+
     @Test
     public void testRingIterator()
     {