diff --git a/CHANGES.txt b/CHANGES.txt index 37ddf2823b..da8ceaa57e 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,6 @@ 3.0.25: +Merged from 2.2: + * Make TokenMetadata's ring version increments atomic (CASSANDRA-16286) 3.0.24: diff --git a/src/java/org/apache/cassandra/locator/TokenMetadata.java b/src/java/org/apache/cassandra/locator/TokenMetadata.java index 3978eebe01..da5f5e384b 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.annotations.VisibleForTesting; import com.google.common.collect.*; import org.apache.commons.lang3.StringUtils; @@ -110,7 +112,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() { @@ -476,7 +479,7 @@ public class TokenMetadata } endpointToHostIdMap.remove(endpoint); sortedTokens = sortTokens(); - invalidateCachedRings(); + invalidateCachedRingsUnsafe(); } finally { @@ -496,7 +499,7 @@ public class TokenMetadata { logger.info("Updating topology for {}", endpoint); topology = topology.unbuild().updateEndpoint(endpoint).build(); - invalidateCachedRings(); + invalidateCachedRingsUnsafe(); return topology; } finally @@ -516,7 +519,7 @@ public class TokenMetadata { logger.info("Updating topology for all endpoints that have changed"); topology = topology.unbuild().updateEndpoints().build(); - invalidateCachedRings(); + invalidateCachedRingsUnsafe(); return topology; } finally @@ -545,7 +548,7 @@ public class TokenMetadata } } - invalidateCachedRings(); + invalidateCachedRingsUnsafe(); } finally { @@ -1134,7 +1137,7 @@ public class TokenMetadata movingEndpoints.clear(); sortedTokens.clear(); topology = Topology.empty(); - invalidateCachedRings(); + invalidateCachedRingsUnsafe(); } finally { @@ -1280,10 +1283,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 dab7082fc4..0ec8eeb7b2 100644 --- a/test/unit/org/apache/cassandra/locator/TokenMetadataTest.java +++ b/test/unit/org/apache/cassandra/locator/TokenMetadataTest.java @@ -21,6 +21,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.ImmutableMultimap; import com.google.common.collect.Iterators; @@ -68,6 +71,27 @@ 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(() -> tmd.invalidateCachedRings()); + + pool.shutdown(); + + assertTrue(pool.awaitTermination(30, TimeUnit.SECONDS)); + assertEquals(invalidations + startVersion, tmd.getRingVersion()); + } + @Test public void testRingIterator() {