diff --git a/CHANGES.txt b/CHANGES.txt index ff865c542a..36a244a1c4 100644 --- a/CHANGES.txt +++ b/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) diff --git a/src/java/org/apache/cassandra/locator/TokenMetadata.java b/src/java/org/apache/cassandra/locator/TokenMetadata.java index aafd7f9674..44eb925427 100644 --- a/src/java/org/apache/cassandra/locator/TokenMetadata.java +++ b/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); diff --git a/test/unit/org/apache/cassandra/locator/TokenMetadataTest.java b/test/unit/org/apache/cassandra/locator/TokenMetadataTest.java index fc8095da31..6cd5180a11 100644 --- a/test/unit/org/apache/cassandra/locator/TokenMetadataTest.java +++ b/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() {