diff --git a/CHANGES.txt b/CHANGES.txt index 878b3dd2db..8c64d012f0 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -16,6 +16,7 @@ * Fixed multiple single-node SAI query bugs relating to static columns (CASSANDRA-20338) * Upgrade com.datastax.cassandra:cassandra-driver-core:3.11.5 to org.apache.cassandra:cassandra-driver-core:3.12.1 (CASSANDRA-17231) Merged from 4.0: + * Gossip doesn't converge due to race condition when updating EndpointStates multiple fields (CASSANDRA-20659) * Honor MAX_PARALLEL_TRANSFERS correctly (CASSANDRA-20532) * Updating a column with a new TTL but same expiration time is non-deterministic and causes repair mismatches. (CASSANDRA-20561) * Avoid computing prepared statement size for unprepared batches (CASSANDRA-20556) diff --git a/src/java/org/apache/cassandra/gms/EndpointState.java b/src/java/org/apache/cassandra/gms/EndpointState.java index 49847a3c71..7955fd664d 100644 --- a/src/java/org/apache/cassandra/gms/EndpointState.java +++ b/src/java/org/apache/cassandra/gms/EndpointState.java @@ -26,6 +26,7 @@ import javax.annotation.Nullable; import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Function; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -52,8 +53,19 @@ public class EndpointState public final static IVersionedSerializer serializer = new EndpointStateSerializer(); public final static IVersionedSerializer nullableSerializer = NullableSerializer.wrap(serializer); - private volatile HeartBeatState hbState; - private final AtomicReference> applicationState; + private static class View + { + final HeartBeatState hbState; + final Map applicationState; + + private View(HeartBeatState hbState, Map applicationState) + { + this.hbState = hbState; + this.applicationState = applicationState; + } + } + + private final AtomicReference ref; /* fields below do not get serialized */ private volatile long updateTimestamp; @@ -61,18 +73,20 @@ public class EndpointState public EndpointState(HeartBeatState initialHbState) { - this(initialHbState, new EnumMap(ApplicationState.class)); + this(initialHbState, new EnumMap<>(ApplicationState.class)); } public EndpointState(EndpointState other) { - this(new HeartBeatState(other.hbState), new EnumMap<>(other.applicationState.get())); + ref = new AtomicReference<>(other.ref.get()); + updateTimestamp = nanoTime(); + isAlive = true; } - EndpointState(HeartBeatState initialHbState, Map states) + @VisibleForTesting + public EndpointState(HeartBeatState initialHbState, Map states) { - hbState = initialHbState; - applicationState = new AtomicReference>(new EnumMap<>(states)); + ref = new AtomicReference<>(new View(initialHbState, new EnumMap<>(states))); updateTimestamp = nanoTime(); isAlive = true; } @@ -80,28 +94,58 @@ public class EndpointState @VisibleForTesting public HeartBeatState getHeartBeatState() { - return hbState; + return ref.get().hbState; } - void setHeartBeatState(HeartBeatState newHbState) + public void updateHeartBeat() { - updateTimestamp(); - hbState = newHbState; + updateHeartBeat(HeartBeatState::updateHeartBeat); + } + + public void forceNewerGenerationUnsafe() + { + updateHeartBeat(HeartBeatState::forceNewerGenerationUnsafe); + } + + @VisibleForTesting + public void forceHighestPossibleVersionUnsafe() + { + updateHeartBeat(HeartBeatState::forceHighestPossibleVersionUnsafe); + } + + void unsafeSetEmptyHeartBeatState() + { + updateHeartBeat(ignore -> HeartBeatState.empty()); + } + + private void updateHeartBeat(Function fn) + { + HeartBeatState previous = null; + HeartBeatState update = null; + while (true) + { + View view = ref.get(); + if (previous == null || view.hbState != previous) // if this races with updating states then can avoid bumping versions + update = fn.apply(view.hbState); + if (ref.compareAndSet(view, new View(update, view.applicationState))) + return; + previous = view.hbState; + } } public VersionedValue getApplicationState(ApplicationState key) { - return applicationState.get().get(key); + return ref.get().applicationState.get(key); } public boolean containsApplicationState(ApplicationState key) { - return applicationState.get().containsKey(key); + return ref.get().applicationState.containsKey(key); } public Set> states() { - return applicationState.get().entrySet(); + return ref.get().applicationState.entrySet(); } public void addApplicationState(ApplicationState key, VersionedValue value) @@ -115,17 +159,27 @@ public class EndpointState } public void addApplicationStates(Set> values) + { + addApplicationStates(values, null); + } + + public void addApplicationStates(Set> values, @Nullable HeartBeatState hbState) { while (true) { - Map orig = applicationState.get(); + View view = this.ref.get(); + Map orig = view.applicationState; Map copy = new EnumMap<>(orig); for (Map.Entry value : values) copy.put(value.getKey(), value.getValue()); - if (applicationState.compareAndSet(orig, copy)) + if (this.ref.compareAndSet(view, new View(hbState == null ? view.hbState : hbState, copy))) + { + if (hbState != null) + updateTimestamp(); return; + } } } @@ -133,18 +187,19 @@ public class EndpointState { while (hasLegacyFields()) { - Map orig = applicationState.get(); + View view = ref.get(); + Map orig = view.applicationState; Map updatedStates = filterMajorVersion3LegacyApplicationStates(orig); // avoid updating if no state is removed if (orig.size() == updatedStates.size() - || applicationState.compareAndSet(orig, updatedStates)) + || ref.compareAndSet(view, new View(view.hbState, updatedStates))) return; } } private boolean hasLegacyFields() { - Set statesPresent = applicationState.get().keySet(); + Set statesPresent = ref.get().applicationState.keySet(); if (statesPresent.isEmpty()) return false; return (statesPresent.contains(ApplicationState.STATUS) && statesPresent.contains(ApplicationState.STATUS_WITH_PORT)) @@ -209,7 +264,7 @@ public class EndpointState public boolean isStateEmpty() { - return applicationState.get().isEmpty(); + return ref.get().applicationState.isEmpty(); } /** @@ -217,14 +272,15 @@ public class EndpointState */ public boolean isEmptyWithoutStatus() { - Map state = applicationState.get(); + View view = ref.get(); + Map state = view.applicationState; boolean hasStatus = state.containsKey(ApplicationState.STATUS_WITH_PORT) || state.containsKey(ApplicationState.STATUS); - return hbState.isEmpty() && !hasStatus + return view.hbState.isEmpty() && !hasStatus // In the very specific case where hbState.isEmpty and STATUS is missing, this is known to be safe to "fake" // the data, as this happens when the gossip state isn't coming from the node but instead from a peer who // restarted and is missing the node's state. // - // When hbState is not empty, then the node gossiped an empty STATUS; this happens during bootstrap and it's not + // When hbState is not empty, then the node gossiped an empty STATUS; this happens during bootstrap, and it's not // possible to tell if this is ok or not (we can't really tell if the node is dead or having networking issues). // For these cases allow an external actor to verify and inform Cassandra that it is safe - this is done by // updating the LOOSE_DEF_OF_EMPTY_ENABLED field. @@ -279,7 +335,8 @@ public class EndpointState public String toString() { - return "EndpointState: HeartBeatState = " + hbState + ", AppStateMap = " + applicationState.get(); + View view = ref.get(); + return "EndpointState: HeartBeatState = " + view.hbState + ", AppStateMap = " + view.applicationState; } public boolean isSupersededBy(EndpointState that) diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index 2e045296df..3d4bc95cac 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -341,7 +341,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, taskLock.lock(); /* Update the local heartbeat counter. */ - endpointStateMap.get(getBroadcastAddressAndPort()).getHeartBeatState().updateHeartBeat(); + endpointStateMap.get(getBroadcastAddressAndPort()).updateHeartBeat(); if (logger.isTraceEnabled()) logger.trace("My heartbeat is now {}", endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()).getHeartBeatState().getHeartBeatVersion()); final List gDigests = new ArrayList<>(); @@ -638,7 +638,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, epState.addApplicationState(ApplicationState.STATUS_WITH_PORT, shutdown); epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.shutdown(true)); epState.addApplicationState(ApplicationState.RPC_READY, StorageService.instance.valueFactory.rpcReady(false)); - epState.getHeartBeatState().forceHighestPossibleVersionUnsafe(); + epState.forceHighestPossibleVersionUnsafe(); markDead(endpoint, epState); FailureDetector.instance.forceConviction(endpoint); GossiperDiagnostics.markedAsShutdown(this, endpoint); @@ -664,7 +664,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, VersionedValue shutdown = remoteState.getApplicationState(ApplicationState.STATUS_WITH_PORT); if (shutdown == null) throw new AssertionError("Remote shutdown sent but missing STATUS_WITH_PORT; " + remoteState); - remoteState.getHeartBeatState().forceHighestPossibleVersionUnsafe(); + remoteState.forceHighestPossibleVersionUnsafe(); endpointStateMap.put(endpoint, remoteState); markDead(endpoint, remoteState); FailureDetector.instance.forceConviction(endpoint); @@ -859,7 +859,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, // update the other node's generation to mimic it as if it had changed it itself logger.info("Advertising removal for {}", endpoint); epState.updateTimestamp(); // make sure we don't evict it too soon - epState.getHeartBeatState().forceNewerGenerationUnsafe(); + epState.forceNewerGenerationUnsafe(); Map states = new EnumMap<>(ApplicationState.class); states.put(ApplicationState.STATUS_WITH_PORT, StorageService.instance.valueFactory.removingNonlocal(hostId)); states.put(ApplicationState.STATUS, StorageService.instance.valueFactory.removingNonlocal(hostId)); @@ -879,7 +879,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, { EndpointState epState = endpointStateMap.get(endpoint); epState.updateTimestamp(); // make sure we don't evict it too soon - epState.getHeartBeatState().forceNewerGenerationUnsafe(); + epState.forceNewerGenerationUnsafe(); long expireTime = computeExpireTime(); epState.addApplicationState(ApplicationState.STATUS_WITH_PORT, StorageService.instance.valueFactory.removedNonlocal(hostId, expireTime)); epState.addApplicationState(ApplicationState.STATUS, StorageService.instance.valueFactory.removedNonlocal(hostId, expireTime)); @@ -930,7 +930,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, else if (newState.getHeartBeatState().getHeartBeatVersion() != heartbeat) throw new RuntimeException("Endpoint still alive: " + endpoint + " heartbeat changed while trying to assassinate it"); epState.updateTimestamp(); // make sure we don't evict it too soon - epState.getHeartBeatState().forceNewerGenerationUnsafe(); + epState.forceNewerGenerationUnsafe(); } Collection tokens = null; @@ -1755,15 +1755,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, // don't assert here, since if the node restarts the version will go back to zero int oldVersion = localState.getHeartBeatState().getHeartBeatVersion(); - localState.setHeartBeatState(remoteState.getHeartBeatState()); - if (logger.isTraceEnabled()) - logger.trace("Updating heartbeat state version to {} from {} for {} ...", localState.getHeartBeatState().getHeartBeatVersion(), oldVersion, addr); - - Set> remoteStates = remoteState.states(); - assert remoteState.getHeartBeatState().getGeneration() == localState.getHeartBeatState().getGeneration(); - - - Set> updatedStates = remoteStates.stream().filter(entry -> { + Set> updatedStates = remoteState.states().stream().filter(entry -> { // filter out the states that are already up to date (has the same or higher version) VersionedValue local = localState.getApplicationState(entry.getKey()); return (local == null || local.version < entry.getValue().version); @@ -1776,7 +1768,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, logger.trace("Updating {} state version to {} for {}", entry.getKey().toString(), entry.getValue().version, addr); } } - localState.addApplicationStates(updatedStates); + localState.addApplicationStates(updatedStates, remoteState.getHeartBeatState()); + if (logger.isTraceEnabled()) + logger.trace("Updating heartbeat state version to {} from {} for {} ...", localState.getHeartBeatState().getHeartBeatVersion(), oldVersion, addr); // get rid of legacy fields once the cluster is not in mixed mode if (!hasMajorVersion3Nodes) @@ -2177,7 +2171,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, public void forceNewerGeneration() { EndpointState epstate = endpointStateMap.get(getBroadcastAddressAndPort()); - epstate.getHeartBeatState().forceNewerGenerationUnsafe(); + epstate.forceNewerGenerationUnsafe(); } @@ -2198,7 +2192,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, if (epState != null) { logger.debug("not replacing a previous epState for {}, but reusing it: {}", ep, epState); - epState.setHeartBeatState(HeartBeatState.empty()); + epState.unsafeSetEmptyHeartBeatState(); } else { @@ -2670,7 +2664,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean, EndpointState epState = endpointStateMap.get(getBroadcastAddressAndPort()); if (epState != null) { - epState.getHeartBeatState().updateHeartBeat(); + epState.updateHeartBeat(); if (logger.isTraceEnabled()) logger.trace("My heartbeat is now {}", epState.getHeartBeatState().getHeartBeatVersion()); } diff --git a/src/java/org/apache/cassandra/gms/HeartBeatState.java b/src/java/org/apache/cassandra/gms/HeartBeatState.java index 3f633cb0fa..374d346a0a 100644 --- a/src/java/org/apache/cassandra/gms/HeartBeatState.java +++ b/src/java/org/apache/cassandra/gms/HeartBeatState.java @@ -19,8 +19,6 @@ package org.apache.cassandra.gms; import java.io.*; -import com.google.common.annotations.VisibleForTesting; - import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; @@ -29,15 +27,14 @@ import org.apache.cassandra.io.util.DataOutputPlus; /** * HeartBeat State associated with any given endpoint. */ - public class HeartBeatState { public static final int EMPTY_VERSION = -1; public static final IVersionedSerializer serializer = new HeartBeatStateSerializer(); - private volatile int generation; - private volatile int version; + private final int generation; + private final int version; HeartBeatState(int gen) { @@ -75,9 +72,9 @@ public class HeartBeatState return generation; } - void updateHeartBeat() + HeartBeatState updateHeartBeat() { - version = VersionGenerator.getNextVersion(); + return new HeartBeatState(generation, VersionGenerator.getNextVersion()); } public int getHeartBeatVersion() @@ -85,15 +82,14 @@ public class HeartBeatState return version; } - void forceNewerGenerationUnsafe() + HeartBeatState forceNewerGenerationUnsafe() { - generation += 1; + return new HeartBeatState(generation + 1, version); } - @VisibleForTesting - public void forceHighestPossibleVersionUnsafe() + HeartBeatState forceHighestPossibleVersionUnsafe() { - version = Integer.MAX_VALUE; + return new HeartBeatState(generation, Integer.MAX_VALUE); } public String toString() diff --git a/test/distributed/org/apache/cassandra/distributed/impl/UnsafeGossipHelper.java b/test/distributed/org/apache/cassandra/distributed/impl/UnsafeGossipHelper.java index 7cf4c167f7..8b5c01b02b 100644 --- a/test/distributed/org/apache/cassandra/distributed/impl/UnsafeGossipHelper.java +++ b/test/distributed/org/apache/cassandra/distributed/impl/UnsafeGossipHelper.java @@ -269,7 +269,7 @@ public class UnsafeGossipHelper EndpointState state = Gossiper.instance.getEndpointStateForEndpoint(getByAddress(address)); VersionedValue status = new VersionedValue.VersionedValueFactory(partitioner).shutdown(true); state.addApplicationState(ApplicationState.STATUS, status); - state.getHeartBeatState().forceHighestPossibleVersionUnsafe(); + state.forceHighestPossibleVersionUnsafe(); StorageService.instance.onChange(getByAddress(address), ApplicationState.STATUS, status); }); }; diff --git a/test/unit/org/apache/cassandra/gms/GossiperTest.java b/test/unit/org/apache/cassandra/gms/GossiperTest.java index f5f5dfb997..2922d1c35f 100644 --- a/test/unit/org/apache/cassandra/gms/GossiperTest.java +++ b/test/unit/org/apache/cassandra/gms/GossiperTest.java @@ -321,7 +321,7 @@ public class GossiperTest proposedRemoteState = new EndpointState(proposedRemoteHeartBeat); // Bump the heartbeat version and use the same TOKENS state - proposedRemoteHeartBeat.updateHeartBeat(); + proposedRemoteState.updateHeartBeat(); proposedRemoteState.addApplicationState(ApplicationState.TOKENS, tokensValue); // The following state change should only update heartbeat without updating the TOKENS state diff --git a/test/unit/org/apache/cassandra/gms/SerializationsTest.java b/test/unit/org/apache/cassandra/gms/SerializationsTest.java index 0422ac0017..e23fdf4928 100644 --- a/test/unit/org/apache/cassandra/gms/SerializationsTest.java +++ b/test/unit/org/apache/cassandra/gms/SerializationsTest.java @@ -130,7 +130,7 @@ public class SerializationsTest extends AbstractSerializationsTester private static List Digests = new ArrayList(); { - HeartbeatSt.updateHeartBeat(); + EndpointSt.updateHeartBeat(); EndpointSt.addApplicationState(ApplicationState.LOAD, vv0); EndpointSt.addApplicationState(ApplicationState.STATUS_WITH_PORT, vv1); for (int i = 0; i < 100; i++)