Gossip doesn't converge due to race condition when updating EndpointStates multiple fields

patch by David Capwell, Matt Byrd; reviewed by Blake Eggleston, Brandon Williams for CASSANDRA-20659

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0.18
+ * Gossip doesn't converge due to race condition when updating EndpointStates multiple fields (CASSANDRA-20659)
  * Handle sstable metadata stats file getting a new mtime after compaction has finished (CASSANDRA-18119)
  * 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)

src/java/org/apache/cassandra/gms/EndpointState.java

@@ -24,6 +24,9 @@ import java.util.stream.Collectors;
 
 import javax.annotation.Nullable;
 
+import com.google.common.annotations.VisibleForTesting;
+
+import com.google.common.base.Function;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -37,16 +40,25 @@ import org.apache.cassandra.utils.CassandraVersion;
  * This abstraction represents both the HeartBeatState and the ApplicationState in an EndpointState
  * instance. Any state for a given endpoint can be retrieved from this instance.
  */
-
-
 public class EndpointState
 {
     protected static final Logger logger = LoggerFactory.getLogger(EndpointState.class);
 
     public final static IVersionedSerializer<EndpointState> serializer = new EndpointStateSerializer();
 
-    private volatile HeartBeatState hbState;
+    private static class View
-    private final AtomicReference<Map<ApplicationState, VersionedValue>> applicationState;
+    {
+        final HeartBeatState hbState;
+        final Map<ApplicationState, VersionedValue> applicationState;
+
+        private View(HeartBeatState hbState, Map<ApplicationState, VersionedValue> applicationState)
+        {
+            this.hbState = hbState;
+            this.applicationState = applicationState;
+        }
+    }
+
+    private final AtomicReference<View> ref;
 
     /* fields below do not get serialized */
     private volatile long updateTimestamp;
@@ -54,46 +66,79 @@ public class EndpointState
 
     public EndpointState(HeartBeatState initialHbState)
     {
-        this(initialHbState, new EnumMap<ApplicationState, VersionedValue>(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());
-    }
-
-    EndpointState(HeartBeatState initialHbState, Map<ApplicationState, VersionedValue> states)
-    {
-        hbState = initialHbState;
-        applicationState = new AtomicReference<Map<ApplicationState, VersionedValue>>(new EnumMap<>(states));
         updateTimestamp = System.nanoTime();
         isAlive = true;
     }
 
-    HeartBeatState getHeartBeatState()
+    @VisibleForTesting
+    public EndpointState(HeartBeatState initialHbState, Map<ApplicationState, VersionedValue> states)
     {
-        return hbState;
+        ref = new AtomicReference<>(new View(initialHbState, new EnumMap<>(states)));
+        updateTimestamp = System.nanoTime();
+        isAlive = true;
     }
 
-    void setHeartBeatState(HeartBeatState newHbState)
+    @VisibleForTesting
+    public HeartBeatState getHeartBeatState()
     {
-        updateTimestamp();
+        return ref.get().hbState;
-        hbState = newHbState;
+    }
+
+    public void updateHeartBeat()
+    {
+        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<HeartBeatState, HeartBeatState> 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<Map.Entry<ApplicationState, VersionedValue>> states()
     {
-        return applicationState.get().entrySet();
+        return ref.get().applicationState.entrySet();
     }
 
     public void addApplicationState(ApplicationState key, VersionedValue value)
@@ -107,17 +152,27 @@ public class EndpointState
     }
 
     public void addApplicationStates(Set<Map.Entry<ApplicationState, VersionedValue>> values)
+    {
+        addApplicationStates(values, null);
+    }
+
+    public void addApplicationStates(Set<Map.Entry<ApplicationState, VersionedValue>> values, @Nullable HeartBeatState hbState)
     {
         while (true)
         {
-            Map<ApplicationState, VersionedValue> orig = applicationState.get();
+            View view = this.ref.get();
+            Map<ApplicationState, VersionedValue> orig = view.applicationState;
             Map<ApplicationState, VersionedValue> copy = new EnumMap<>(orig);
 
             for (Map.Entry<ApplicationState, VersionedValue> 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;
+            }
         }
     }
 
@@ -125,18 +180,19 @@ public class EndpointState
     {
         while (hasLegacyFields())
         {
-            Map<ApplicationState, VersionedValue> orig = applicationState.get();
+            View view = ref.get();
+            Map<ApplicationState, VersionedValue> orig = view.applicationState;
             Map<ApplicationState, VersionedValue> 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<ApplicationState> statesPresent = applicationState.get().keySet();
+        Set<ApplicationState> statesPresent = ref.get().applicationState.keySet();
         if (statesPresent.isEmpty())
             return false;
         return (statesPresent.contains(ApplicationState.STATUS) && statesPresent.contains(ApplicationState.STATUS_WITH_PORT))
@@ -193,7 +249,7 @@ public class EndpointState
 
     public boolean isStateEmpty()
     {
-        return applicationState.get().isEmpty();
+        return ref.get().applicationState.isEmpty();
     }
 
     /**
@@ -201,8 +257,10 @@ public class EndpointState
      */
     public boolean isEmptyWithoutStatus()
     {
-        Map<ApplicationState, VersionedValue> state = applicationState.get();
+        View view = ref.get();
-        return hbState.isEmpty() && !(state.containsKey(ApplicationState.STATUS_WITH_PORT) || state.containsKey(ApplicationState.STATUS));
+        Map<ApplicationState, VersionedValue> state = view.applicationState;
+        boolean hasStatus = state.containsKey(ApplicationState.STATUS_WITH_PORT) || state.containsKey(ApplicationState.STATUS);
+        return view.hbState.isEmpty() && !hasStatus;
     }
 
     public boolean isRpcReady()
@@ -253,7 +311,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;
     }
 }
 

src/java/org/apache/cassandra/gms/Gossiper.java

@@ -310,7 +310,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
                 taskLock.lock();
 
                 /* Update the local heartbeat counter. */
-                endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()).getHeartBeatState().updateHeartBeat();
+                endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()).updateHeartBeat();
                 if (logger.isTraceEnabled())
                     logger.trace("My heartbeat is now {}", endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort()).getHeartBeatState().getHeartBeatVersion());
                 final List<GossipDigest> gDigests = new ArrayList<>();
@@ -598,7 +598,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);
@@ -778,7 +778,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<ApplicationState, VersionedValue> 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));
@@ -798,7 +798,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));
@@ -849,7 +849,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<Token> tokens = null;
@@ -1580,15 +1580,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());
+        Set<Entry<ApplicationState, VersionedValue>> updatedStates = remoteState.states().stream().filter(entry -> {
-        if (logger.isTraceEnabled())
-            logger.trace("Updating heartbeat state version to {} from {} for {} ...", localState.getHeartBeatState().getHeartBeatVersion(), oldVersion, addr);
-
-        Set<Entry<ApplicationState, VersionedValue>> remoteStates = remoteState.states();
-        assert remoteState.getHeartBeatState().getGeneration() == localState.getHeartBeatState().getGeneration();
-
-
-        Set<Entry<ApplicationState, VersionedValue>> updatedStates = remoteStates.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);
@@ -1601,7 +1593,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 (!hasMajorVersion3OrUnknownNodes())
@@ -1983,7 +1977,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
     public void forceNewerGeneration()
     {
         EndpointState epstate = endpointStateMap.get(FBUtilities.getBroadcastAddressAndPort());
-        epstate.getHeartBeatState().forceNewerGenerationUnsafe();
+        epstate.forceNewerGenerationUnsafe();
     }
 
 
@@ -2004,7 +1998,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
         {

src/java/org/apache/cassandra/gms/HeartBeatState.java

@@ -33,8 +33,8 @@ public class HeartBeatState
 
     public static final IVersionedSerializer<HeartBeatState> serializer = new HeartBeatStateSerializer();
 
-    private volatile int generation;
+    private final int generation;
-    private volatile int version;
+    private final int version;
 
     HeartBeatState(int gen)
     {
@@ -67,29 +67,29 @@ public class HeartBeatState
         return version == EMPTY_VERSION;
     }
 
-    int getGeneration()
+    public int getGeneration()
     {
         return generation;
     }
 
-    void updateHeartBeat()
+    HeartBeatState updateHeartBeat()
     {
-        version = VersionGenerator.getNextVersion();
+        return new HeartBeatState(generation, VersionGenerator.getNextVersion());
     }
 
-    int getHeartBeatVersion()
+    public int getHeartBeatVersion()
     {
         return version;
     }
 
-    void forceNewerGenerationUnsafe()
+    HeartBeatState forceNewerGenerationUnsafe()
     {
-        generation += 1;
+        return new HeartBeatState(generation + 1, version);
     }
 
-    void forceHighestPossibleVersionUnsafe()
+    HeartBeatState forceHighestPossibleVersionUnsafe()
     {
-        version = Integer.MAX_VALUE;
+        return new HeartBeatState(generation, Integer.MAX_VALUE);
     }
 
     public String toString()

test/unit/org/apache/cassandra/gms/GossiperTest.java

@@ -304,7 +304,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

test/unit/org/apache/cassandra/gms/SerializationsTest.java

@@ -130,7 +130,7 @@ public class SerializationsTest extends AbstractSerializationsTester
         private static List<GossipDigest> Digests = new ArrayList<GossipDigest>();
 
         {
-            HeartbeatSt.updateHeartBeat();
+            EndpointSt.updateHeartBeat();
             EndpointSt.addApplicationState(ApplicationState.LOAD, vv0);
             EndpointSt.addApplicationState(ApplicationState.STATUS_WITH_PORT, vv1);
             for (int i = 0; i < 100; i++)