diff --git a/CHANGES.txt b/CHANGES.txt index e5a1a8b8b7..9f3ae2e98b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -106,6 +106,7 @@ Merged from 3.11: * Update Jackson from 2.9.10 to 2.12.5 (CASSANDRA-16851) * Make assassinate more resilient to missing tokens (CASSANDRA-16847) Merged from 3.0: + * Avoid race in AbstractReplicationStrategy endpoint caching (CASSANDRA-16673) * Fix abort when window resizing during cqlsh COPY (CASSANDRA-15230) * Fix slow keycache load which blocks startup for tables with many sstables (CASSANDRA-14898) * Fix rare NPE caused by batchlog replay / node decomission races (CASSANDRA-17049) diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index e6db054e07..0b7c378232 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -23,6 +23,7 @@ import java.lang.reflect.Method; import java.util.Collection; import java.util.Collections; import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; import com.google.common.base.Preconditions; @@ -54,10 +55,7 @@ public abstract class AbstractReplicationStrategy public final Map configOptions; protected final String keyspaceName; private final TokenMetadata tokenMetadata; - - // track when the token range changes, signaling we need to invalidate our endpoint cache - private volatile long lastInvalidatedVersion = 0; - + private final ReplicaCache replicas = new ReplicaCache<>(); public IEndpointSnitch snitch; protected AbstractReplicationStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map configOptions) @@ -70,26 +68,9 @@ public abstract class AbstractReplicationStrategy this.keyspaceName = keyspaceName; } - private final Map cachedReplicas = new NonBlockingHashMap<>(); - - public EndpointsForRange getCachedReplicas(Token t) + public EndpointsForRange getCachedReplicas(long ringVersion, Token t) { - long lastVersion = tokenMetadata.getRingVersion(); - - if (lastVersion > lastInvalidatedVersion) - { - synchronized (this) - { - if (lastVersion > lastInvalidatedVersion) - { - logger.trace("clearing cached endpoints"); - cachedReplicas.clear(); - lastInvalidatedVersion = lastVersion; - } - } - } - - return cachedReplicas.get(t); + return replicas.get(ringVersion, t); } /** @@ -107,15 +88,16 @@ public abstract class AbstractReplicationStrategy public EndpointsForRange getNaturalReplicas(RingPosition searchPosition) { Token searchToken = searchPosition.getToken(); + long currentRingVersion = tokenMetadata.getRingVersion(); Token keyToken = TokenMetadata.firstToken(tokenMetadata.sortedTokens(), searchToken); - EndpointsForRange endpoints = getCachedReplicas(keyToken); + EndpointsForRange endpoints = getCachedReplicas(currentRingVersion, keyToken); if (endpoints == null) { TokenMetadata tm = tokenMetadata.cachedOnlyTokenMap(); // if our cache got invalidated, it's possible there is a new token to account for too keyToken = TokenMetadata.firstToken(tm.sortedTokens(), searchToken); endpoints = calculateNaturalReplicas(searchToken, tm); - cachedReplicas.put(keyToken, endpoints); + replicas.put(tm.getRingVersion(), keyToken, endpoints); } return endpoints; @@ -463,4 +445,64 @@ public abstract class AbstractReplicationStrategy throw new ConfigurationException(String.format("Unrecognized strategy option {%s} passed to %s for keyspace %s", key, getClass().getSimpleName(), keyspaceName)); } } + + static class ReplicaCache + { + private final AtomicReference> cachedReplicas = new AtomicReference<>(new ReplicaHolder<>(0, 4)); + + V get(long ringVersion, K keyToken) + { + ReplicaHolder replicaHolder = maybeClearAndGet(ringVersion); + if (replicaHolder == null) + return null; + + return replicaHolder.replicas.get(keyToken); + } + + void put(long ringVersion, K keyToken, V endpoints) + { + ReplicaHolder current = maybeClearAndGet(ringVersion); + if (current != null) + { + // if we have the same ringVersion, but already know about the keyToken the endpoints should be the same + current.replicas.putIfAbsent(keyToken, endpoints); + } + } + + ReplicaHolder maybeClearAndGet(long ringVersion) + { + ReplicaHolder current = cachedReplicas.get(); + if (ringVersion == current.ringVersion) + return current; + else if (ringVersion < current.ringVersion) // things have already moved on + return null; + + // If ring version has changed, create a fresh replica holder and try to replace the current one. + // This may race with other threads that have the same new ring version and one will win and the loosers + // will be garbage collected + ReplicaHolder cleaned = new ReplicaHolder<>(ringVersion, current.replicas.size()); + cachedReplicas.compareAndSet(current, cleaned); + + // A new ring version may have come along while making the new holder, so re-check the + // reference and return the ring version if the same, otherwise return null as there is no point + // in using it. + current = cachedReplicas.get(); + if (ringVersion == current.ringVersion) + return current; + else + return null; + } + } + + static class ReplicaHolder + { + private final long ringVersion; + private final NonBlockingHashMap replicas; + + ReplicaHolder(long ringVersion, int expectedEntries) + { + this.ringVersion = ringVersion; + this.replicas = new NonBlockingHashMap<>(expectedEntries); + } + } } diff --git a/src/java/org/apache/cassandra/locator/TokenMetadata.java b/src/java/org/apache/cassandra/locator/TokenMetadata.java index 202bd6a9d3..0e60a1f2a9 100644 --- a/src/java/org/apache/cassandra/locator/TokenMetadata.java +++ b/src/java/org/apache/cassandra/locator/TokenMetadata.java @@ -131,12 +131,18 @@ public class TokenMetadata } private TokenMetadata(BiMultiValMap tokenToEndpointMap, BiMap endpointsMap, Topology topology, IPartitioner partitioner) + { + this(tokenToEndpointMap, endpointsMap, topology, partitioner, 0); + } + + private TokenMetadata(BiMultiValMap tokenToEndpointMap, BiMap endpointsMap, Topology topology, IPartitioner partitioner, long ringVersion) { this.tokenToEndpointMap = tokenToEndpointMap; this.topology = topology; this.partitioner = partitioner; endpointToHostIdMap = endpointsMap; sortedTokens = sortTokens(); + this.ringVersion = ringVersion; } /** @@ -675,7 +681,8 @@ public class TokenMetadata return new TokenMetadata(SortedBiMultiValMap.create(tokenToEndpointMap), HashBiMap.create(endpointToHostIdMap), topology, - partitioner); + partitioner, + ringVersion); } finally { diff --git a/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java b/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java index c3e92f2fe4..2dc7303c23 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ring/BootstrapTest.java @@ -56,7 +56,7 @@ public class BootstrapTest extends TestBaseImpl .withConfig(config -> config.with(NETWORK, GOSSIP)) .start()) { - populate(cluster,0, 100); + populate(cluster, 0, 100); IInstanceConfig config = cluster.newInstanceConfig(); IInvokableInstance newInstance = cluster.bootstrap(config); @@ -95,7 +95,7 @@ public class BootstrapTest extends TestBaseImpl cluster.forEach(statusToBootstrap(newInstance)); - populate(cluster,0, 100); + populate(cluster, 0, 100); Assert.assertEquals(100, newInstance.executeInternal("SELECT *FROM " + KEYSPACE + ".tbl").length); } @@ -113,7 +113,7 @@ public class BootstrapTest extends TestBaseImpl .withConfig(config -> config.with(NETWORK, GOSSIP)) .start()) { - populate(cluster,0, 100); + populate(cluster, 0, 100); bootstrapAndJoinNode(cluster); for (Map.Entry e : count(cluster).entrySet()) diff --git a/test/distributed/org/apache/cassandra/distributed/test/ring/ReadsDuringBootstrapTest.java b/test/distributed/org/apache/cassandra/distributed/test/ring/ReadsDuringBootstrapTest.java new file mode 100644 index 0000000000..932dbdc46f --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/ring/ReadsDuringBootstrapTest.java @@ -0,0 +1,115 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test.ring; + +import java.io.IOException; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Test; + +import net.bytebuddy.ByteBuddy; +import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; +import net.bytebuddy.implementation.MethodDelegation; +import net.bytebuddy.implementation.bind.annotation.FieldValue; +import net.bytebuddy.implementation.bind.annotation.SuperCall; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.TokenSupplier; +import org.apache.cassandra.distributed.shared.NetworkTopology; +import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.locator.AbstractReplicationStrategy; +import org.apache.cassandra.locator.EndpointsForRange; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; + +public class ReadsDuringBootstrapTest extends TestBaseImpl +{ + + @Test + public void readsDuringBootstrapTest() throws IOException, ExecutionException, InterruptedException, TimeoutException + { + int originalNodeCount = 3; + int expandedNodeCount = originalNodeCount + 1; + ExecutorService es = Executors.newSingleThreadExecutor(); + try (Cluster cluster = builder().withNodes(originalNodeCount) + .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(expandedNodeCount)) + .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(expandedNodeCount, "dc0", "rack0")) + .withConfig(config -> config.with(NETWORK, GOSSIP) + .set("read_request_timeout_in_ms", Integer.MAX_VALUE) + .set("request_timeout_in_ms", Integer.MAX_VALUE)) + .withInstanceInitializer(BB::install) + .start()) + { + String query = withKeyspace("SELECT * FROM %s.tbl WHERE id = ?"); + cluster.schemaChange(withKeyspace("CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 2};")); + cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (id int PRIMARY KEY)")); + cluster.get(1).runOnInstance(() -> BB.block.set(true)); + Future read = es.submit(() -> cluster.coordinator(1).execute(query, ConsistencyLevel.QUORUM, 3)); + long mark = cluster.get(1).logs().mark(); + bootstrapAndJoinNode(cluster); + cluster.get(1).logs().watchFor(mark, "New node /127.0.0.4"); + cluster.get(1).runOnInstance(() -> BB.block.set(false)); + // populate cache + for (int i = 0; i < 10; i++) + cluster.coordinator(1).execute(query, ConsistencyLevel.QUORUM, i); + cluster.get(1).runOnInstance(() -> BB.latch.countDown()); + read.get(); + } + finally + { + es.shutdown(); + } + } + + public static class BB + { + public static final AtomicBoolean block = new AtomicBoolean(); + public static final CountDownLatch latch = new CountDownLatch(1); + + private static void install(ClassLoader cl, Integer instanceId) + { + if (instanceId != 1) + return; + new ByteBuddy().rebase(AbstractReplicationStrategy.class) + .method(named("getCachedReplicas")) + .intercept(MethodDelegation.to(BB.class)) + .make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + } + + public static EndpointsForRange getCachedReplicas(long ringVersion, Token t, + @FieldValue("keyspaceName") String keyspaceName, + @SuperCall Callable zuper) throws Exception + { + if (keyspaceName.equals(KEYSPACE) && block.get()) + latch.await(); + return zuper.call(); + } + } +} diff --git a/test/unit/org/apache/cassandra/locator/AbstractReplicationStrategyTest.java b/test/unit/org/apache/cassandra/locator/AbstractReplicationStrategyTest.java new file mode 100644 index 0000000000..62206ce8bb --- /dev/null +++ b/test/unit/org/apache/cassandra/locator/AbstractReplicationStrategyTest.java @@ -0,0 +1,45 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.locator; + +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +public class AbstractReplicationStrategyTest +{ + @Test + public void testReplicaCache() + { + AbstractReplicationStrategy.ReplicaCache cache = new AbstractReplicationStrategy.ReplicaCache<>(); + + cache.put(10, 1, 1); + assertEquals(1, (int)cache.get(10, 1)); + assertNull(cache.get(9,1)); // get with old ringversion, return null to force a recalculation + assertNull(cache.get(11,1)); // newer ringVersion - cache gets cleared + assertNull(cache.get(10,1)); // and make sure the map got cleared + + cache.put(11, 1, 100); + cache.put(10, 1, 99); + assertEquals(100, (int)cache.get(11, 1)); + assertNull(cache.get(12, 55)); + assertNull(cache.get(11, 1)); + } +}