Updated Branches: refs/heads/trunk 788f873f5 -> b20795679
reformat gms package Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b2079567 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b2079567 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b2079567 Branch: refs/heads/trunk Commit: b20795679b5bf9ff1f2fa757a9f361ddd291ee23 Parents: 1954345 Author: Jonathan Ellis <[email protected]> Authored: Sat Feb 23 10:28:48 2013 -0600 Committer: Jonathan Ellis <[email protected]> Committed: Sat Feb 23 10:28:48 2013 -0600 ---------------------------------------------------------------------- .../org/apache/cassandra/gms/EndpointState.java | 3 +- .../org/apache/cassandra/gms/FailureDetector.java | 12 +- .../org/apache/cassandra/gms/GossipDigest.java | 4 +- .../org/apache/cassandra/gms/GossipDigestAck.java | 2 +- .../org/apache/cassandra/gms/GossipDigestAck2.java | 4 +- .../cassandra/gms/GossipDigestAckVerbHandler.java | 10 +- .../org/apache/cassandra/gms/GossipDigestSyn.java | 2 +- .../cassandra/gms/GossipDigestSynVerbHandler.java | 26 ++-- src/java/org/apache/cassandra/gms/Gossiper.java | 127 ++++++++------- .../gms/IEndpointStateChangeSubscriber.java | 2 +- .../gms/IFailureDetectionEventListener.java | 3 +- .../org/apache/cassandra/gms/IFailureDetector.java | 11 +- .../apache/cassandra/gms/IFailureNotification.java | 1 + .../org/apache/cassandra/gms/TokenSerializer.java | 1 + .../org/apache/cassandra/gms/VersionedValue.java | 24 ++-- 15 files changed, 125 insertions(+), 107 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/EndpointState.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/EndpointState.java b/src/java/org/apache/cassandra/gms/EndpointState.java index 6cf7a06..fc6c1ab 100644 --- a/src/java/org/apache/cassandra/gms/EndpointState.java +++ b/src/java/org/apache/cassandra/gms/EndpointState.java @@ -25,6 +25,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; + import org.cliffc.high_scale_lib.NonBlockingHashMap; /** @@ -135,7 +136,7 @@ class EndpointStateSerializer implements IVersionedSerializer<EndpointState> EndpointState epState = new EndpointState(hbState); int appStateSize = dis.readInt(); - for ( int i = 0; i < appStateSize; ++i ) + for (int i = 0; i < appStateSize; ++i) { int key = dis.readInt(); VersionedValue value = VersionedValue.serializer.deserialize(dis, version); http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/FailureDetector.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/FailureDetector.java b/src/java/org/apache/cassandra/gms/FailureDetector.java index 2b3905a..61e2b3b 100644 --- a/src/java/org/apache/cassandra/gms/FailureDetector.java +++ b/src/java/org/apache/cassandra/gms/FailureDetector.java @@ -168,7 +168,7 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean logger.trace("reporting {}", ep); long now = System.currentTimeMillis(); ArrivalWindow heartbeatWindow = arrivalSamples.get(ep); - if ( heartbeatWindow == null ) + if (heartbeatWindow == null) { heartbeatWindow = new ArrivalWindow(SAMPLE_SIZE); arrivalSamples.put(ep, heartbeatWindow); @@ -179,7 +179,7 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean public void interpret(InetAddress ep) { ArrivalWindow hbWnd = arrivalSamples.get(ep); - if ( hbWnd == null ) + if (hbWnd == null) { return; } @@ -192,7 +192,7 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean { logger.trace("notifying listeners that {} is down", ep); logger.trace("intervals: {} mean: {}", hbWnd, hbWnd.mean()); - for ( IFailureDetectionEventListener listener : fdEvntListeners ) + for (IFailureDetectionEventListener listener : fdEvntListeners) { listener.convict(ep, phi); } @@ -229,12 +229,12 @@ public class FailureDetector implements IFailureDetector, FailureDetectorMBean Set<InetAddress> eps = arrivalSamples.keySet(); sb.append("-----------------------------------------------------------------------"); - for ( InetAddress ep : eps ) + for (InetAddress ep : eps) { ArrivalWindow hWnd = arrivalSamples.get(ep); sb.append(ep + " : "); sb.append(hWnd.toString()); - sb.append( System.getProperty("line.separator") ); + sb.append(System.getProperty("line.separator")); } sb.append("-----------------------------------------------------------------------"); return sb.toString(); @@ -270,7 +270,7 @@ class ArrivalWindow synchronized void add(double value) { double interArrivalTime; - if ( tLast > 0L ) + if (tLast > 0L) { interArrivalTime = (value - tLast); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/GossipDigest.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/GossipDigest.java b/src/java/org/apache/cassandra/gms/GossipDigest.java index 53a3b5a..ec2ce50 100644 --- a/src/java/org/apache/cassandra/gms/GossipDigest.java +++ b/src/java/org/apache/cassandra/gms/GossipDigest.java @@ -60,8 +60,8 @@ public class GossipDigest implements Comparable<GossipDigest> public int compareTo(GossipDigest gDigest) { - if ( generation != gDigest.generation ) - return ( generation - gDigest.generation ); + if (generation != gDigest.generation) + return (generation - gDigest.generation); return (maxVersion - gDigest.maxVersion); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/GossipDigestAck.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAck.java b/src/java/org/apache/cassandra/gms/GossipDigestAck.java index e0e2039..01d2f6e 100644 --- a/src/java/org/apache/cassandra/gms/GossipDigestAck.java +++ b/src/java/org/apache/cassandra/gms/GossipDigestAck.java @@ -99,7 +99,7 @@ class GossipDigestAckSerializer implements IVersionedSerializer<GossipDigestAck> size += TypeSizes.NATIVE.sizeof(ack.epStateMap.size()); for (Map.Entry<InetAddress, EndpointState> entry : ack.epStateMap.entrySet()) size += CompactEndpointSerializationHelper.serializedSize(entry.getKey()) - + EndpointState.serializer.serializedSize(entry.getValue(), version); + + EndpointState.serializer.serializedSize(entry.getValue(), version); return size; } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/GossipDigestAck2.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAck2.java b/src/java/org/apache/cassandra/gms/GossipDigestAck2.java index 5f044fb..d29bd4a 100644 --- a/src/java/org/apache/cassandra/gms/GossipDigestAck2.java +++ b/src/java/org/apache/cassandra/gms/GossipDigestAck2.java @@ -43,7 +43,7 @@ public class GossipDigestAck2 Map<InetAddress, EndpointState> getEndpointStateMap() { - return epStateMap; + return epStateMap; } } @@ -79,7 +79,7 @@ class GossipDigestAck2Serializer implements IVersionedSerializer<GossipDigestAck long size = TypeSizes.NATIVE.sizeof(ack2.epStateMap.size()); for (Map.Entry<InetAddress, EndpointState> entry : ack2.epStateMap.entrySet()) size += CompactEndpointSerializationHelper.serializedSize(entry.getKey()) - + EndpointState.serializer.serializedSize(entry.getValue(), version); + + EndpointState.serializer.serializedSize(entry.getValue(), version); return size; } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java b/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java index cdc9686..cff9afc 100644 --- a/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java +++ b/src/java/org/apache/cassandra/gms/GossipDigestAckVerbHandler.java @@ -50,7 +50,7 @@ public class GossipDigestAckVerbHandler implements IVerbHandler<GossipDigestAck> List<GossipDigest> gDigestList = gDigestAckMessage.getGossipDigestList(); Map<InetAddress, EndpointState> epStateMap = gDigestAckMessage.getEndpointStateMap(); - if ( epStateMap.size() > 0 ) + if (epStateMap.size() > 0) { /* Notify the Failure Detector */ Gossiper.instance.notifyFailureDetector(epStateMap); @@ -59,17 +59,17 @@ public class GossipDigestAckVerbHandler implements IVerbHandler<GossipDigestAck> /* Get the state required to send to this gossipee - construct GossipDigestAck2Message */ Map<InetAddress, EndpointState> deltaEpStateMap = new HashMap<InetAddress, EndpointState>(); - for( GossipDigest gDigest : gDigestList ) + for (GossipDigest gDigest : gDigestList) { InetAddress addr = gDigest.getEndpoint(); EndpointState localEpStatePtr = Gossiper.instance.getStateForVersionBiggerThan(addr, gDigest.getMaxVersion()); - if ( localEpStatePtr != null ) + if (localEpStatePtr != null) deltaEpStateMap.put(addr, localEpStatePtr); } MessageOut<GossipDigestAck2> gDigestAck2Message = new MessageOut<GossipDigestAck2>(MessagingService.Verb.GOSSIP_DIGEST_ACK2, - new GossipDigestAck2(deltaEpStateMap), - GossipDigestAck2.serializer); + new GossipDigestAck2(deltaEpStateMap), + GossipDigestAck2.serializer); if (logger.isTraceEnabled()) logger.trace("Sending a GossipDigestAck2Message to {}", from); MessagingService.instance().sendOneWay(gDigestAck2Message, from); http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/GossipDigestSyn.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/GossipDigestSyn.java b/src/java/org/apache/cassandra/gms/GossipDigestSyn.java index 24979f1..06cb3da 100644 --- a/src/java/org/apache/cassandra/gms/GossipDigestSyn.java +++ b/src/java/org/apache/cassandra/gms/GossipDigestSyn.java @@ -67,7 +67,7 @@ class GossipDigestSerializationHelper gDigests.add(GossipDigest.serializer.deserialize(dis, version)); return gDigests; } - + static int serializedSize(List<GossipDigest> digests, int version) { int size = TypeSizes.NATIVE.sizeof(digests.size()); http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java b/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java index 6e90ac4..29f802a 100644 --- a/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java +++ b/src/java/org/apache/cassandra/gms/GossipDigestSynVerbHandler.java @@ -31,7 +31,7 @@ import org.apache.cassandra.net.MessagingService; public class GossipDigestSynVerbHandler implements IVerbHandler<GossipDigestSyn> { - private static final Logger logger = LoggerFactory.getLogger( GossipDigestSynVerbHandler.class); + private static final Logger logger = LoggerFactory.getLogger(GossipDigestSynVerbHandler.class); public void doVerb(MessageIn<GossipDigestSyn> message, String id) { @@ -49,13 +49,13 @@ public class GossipDigestSynVerbHandler implements IVerbHandler<GossipDigestSyn> /* If the message is from a different cluster throw it away. */ if (!gDigestMessage.clusterId.equals(DatabaseDescriptor.getClusterName())) { - logger.warn("ClusterName mismatch from " + from + " " + gDigestMessage.clusterId + "!=" + DatabaseDescriptor.getClusterName()); + logger.warn("ClusterName mismatch from " + from + " " + gDigestMessage.clusterId + "!=" + DatabaseDescriptor.getClusterName()); return; } if (gDigestMessage.partioner != null && !gDigestMessage.partioner.equals(DatabaseDescriptor.getPartitionerName())) { - logger.warn("Partitioner mismatch from " + from + " " + gDigestMessage.partioner + "!=" + DatabaseDescriptor.getPartitionerName()); + logger.warn("Partitioner mismatch from " + from + " " + gDigestMessage.partioner + "!=" + DatabaseDescriptor.getPartitionerName()); return; } @@ -63,7 +63,7 @@ public class GossipDigestSynVerbHandler implements IVerbHandler<GossipDigestSyn> if (logger.isTraceEnabled()) { StringBuilder sb = new StringBuilder(); - for ( GossipDigest gDigest : gDigestList ) + for (GossipDigest gDigest : gDigestList) { sb.append(gDigest); sb.append(" "); @@ -78,8 +78,8 @@ public class GossipDigestSynVerbHandler implements IVerbHandler<GossipDigestSyn> Gossiper.instance.examineGossiper(gDigestList, deltaGossipDigestList, deltaEpStateMap); MessageOut<GossipDigestAck> gDigestAckMessage = new MessageOut<GossipDigestAck>(MessagingService.Verb.GOSSIP_DIGEST_ACK, - new GossipDigestAck(deltaGossipDigestList, deltaEpStateMap), - GossipDigestAck.serializer); + new GossipDigestAck(deltaGossipDigestList, deltaEpStateMap), + GossipDigestAck.serializer); if (logger.isTraceEnabled()) logger.trace("Sending a GossipDigestAckMessage to {}", from); MessagingService.instance().sendOneWay(gDigestAckMessage, from); @@ -96,7 +96,7 @@ public class GossipDigestSynVerbHandler implements IVerbHandler<GossipDigestSyn> { /* Construct a map of endpoint to GossipDigest. */ Map<InetAddress, GossipDigest> epToDigestMap = new HashMap<InetAddress, GossipDigest>(); - for ( GossipDigest gDigest : gDigestList ) + for (GossipDigest gDigest : gDigestList) { epToDigestMap.put(gDigest.getEndpoint(), gDigest); } @@ -106,13 +106,13 @@ public class GossipDigestSynVerbHandler implements IVerbHandler<GossipDigestSyn> * of the local EndpointState and the version found in the GossipDigest. */ List<GossipDigest> diffDigests = new ArrayList<GossipDigest>(gDigestList.size()); - for ( GossipDigest gDigest : gDigestList ) + for (GossipDigest gDigest : gDigestList) { InetAddress ep = gDigest.getEndpoint(); EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(ep); - int version = (epState != null) ? Gossiper.instance.getMaxEndpointStateVersion( epState ) : 0; - int diffVersion = Math.abs(version - gDigest.getMaxVersion() ); - diffDigests.add( new GossipDigest(ep, gDigest.getGeneration(), diffVersion) ); + int version = (epState != null) ? Gossiper.instance.getMaxEndpointStateVersion(epState) : 0; + int diffVersion = Math.abs(version - gDigest.getMaxVersion()); + diffDigests.add(new GossipDigest(ep, gDigest.getGeneration(), diffVersion)); } gDigestList.clear(); @@ -122,9 +122,9 @@ public class GossipDigestSynVerbHandler implements IVerbHandler<GossipDigestSyn> * Report the digests in descending order. This takes care of the endpoints * that are far behind w.r.t this local endpoint */ - for ( int i = size - 1; i >= 0; --i ) + for (int i = size - 1; i >= 0; --i) { - gDigestList.add( epToDigestMap.get(diffDigests.get(i).getEndpoint()) ); + gDigestList.add(epToDigestMap.get(diffDigests.get(i).getEndpoint())); } } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/Gossiper.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index d8ca378..e298f72 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -72,7 +72,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean private final Random random = new Random(); private final Comparator<InetAddress> inetcomparator = new Comparator<InetAddress>() { - public int compare(InetAddress addr1, InetAddress addr2) + public int compare(InetAddress addr1, InetAddress addr2) { return addr1.getHostAddress().compareTo(addr2.getHostAddress()); } @@ -117,14 +117,14 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean final List<GossipDigest> gDigests = new ArrayList<GossipDigest>(); Gossiper.instance.makeRandomGossipDigest(gDigests); - if ( gDigests.size() > 0 ) + if (gDigests.size() > 0) { GossipDigestSyn digestSynMessage = new GossipDigestSyn(DatabaseDescriptor.getClusterName(), DatabaseDescriptor.getPartitionerName(), gDigests); MessageOut<GossipDigestSyn> message = new MessageOut<GossipDigestSyn>(MessagingService.Verb.GOSSIP_DIGEST_SYN, - digestSynMessage, - GossipDigestSyn.serializer); + digestSynMessage, + GossipDigestSyn.serializer); /* Gossip to some random live member */ boolean gossipedToSeed = doGossipToLiveMember(message); @@ -165,7 +165,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean private Gossiper() { // half of QUARATINE_DELAY, to ensure justRemovedEndpoints has enough leeway to prevent re-gossip - FatClientTimeout = (long)(QUARANTINE_DELAY / 2); + FatClientTimeout = (long) (QUARANTINE_DELAY / 2); /* register with the Failure Detector for receiving Failure detector events */ FailureDetector.instance.registerFailureDetectionEventListener(this); @@ -183,6 +183,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /** * Register for interesting state changes. + * * @param subscriber module which implements the IEndpointStateChangeSubscriber */ public void register(IEndpointStateChangeSubscriber subscriber) @@ -192,6 +193,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /** * Unregister interest for state changes. + * * @param subscriber module which implements the IEndpointStateChangeSubscriber */ public void unregister(IEndpointStateChangeSubscriber subscriber) @@ -226,7 +228,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean * by the Failure Detector when it convicts an end point. * * @param endpoint end point that is convicted. - */ + */ public void convict(InetAddress endpoint, double phi) { EndpointState epState = endpointStateMap.get(endpoint); @@ -240,6 +242,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /** * Return either: the greatest heartbeat or application state + * * @param epState * @return */ @@ -247,7 +250,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean { int maxVersion = epState.getHeartBeatState().getHeartBeatVersion(); for (VersionedValue value : epState.getApplicationStateMap().values()) - maxVersion = Math.max(maxVersion, value.version); + maxVersion = Math.max(maxVersion, value.version); return maxVersion; } @@ -255,7 +258,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean * Removes the endpoint from gossip completely * * @param endpoint endpoint to be removed from the current membership. - */ + */ private void evictFromMembership(InetAddress endpoint) { unreachableEndpoints.remove(endpoint); @@ -288,6 +291,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /** * Quarantines the endpoint for QUARANTINE_DELAY + * * @param endpoint */ private void quarantineEndpoint(InetAddress endpoint) @@ -298,6 +302,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /** * Remove the Endpoint and evict immediately, to avoid gossiping about this node. * This should only be called when a token is taken over by a new IP address. + * * @param endpoint The endpoint that has been replaced */ public void replacedEndpoint(InetAddress endpoint) @@ -311,7 +316,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean * rather than just looping through the collection of live endpoints. * * @param gDigests list of Gossip Digests. - */ + */ private void makeRandomGossipDigest(List<GossipDigest> gDigests) { EndpointState epState; @@ -335,12 +340,12 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean if (logger.isTraceEnabled()) { StringBuilder sb = new StringBuilder(); - for ( GossipDigest gDigest : gDigests ) + for (GossipDigest gDigest : gDigests) { sb.append(gDigest); sb.append(" "); } - logger.trace("Gossip Digests are : " + sb.toString()); + logger.trace("Gossip Digests are : " + sb.toString()); } } @@ -348,8 +353,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean * This method will begin removing an existing endpoint from the cluster by spoofing its state * This should never be called unless this coordinator has had 'removetoken' invoked * - * @param endpoint - the endpoint being removed - * @param hostId - the ID of the host being removed + * @param endpoint - the endpoint being removed + * @param hostId - the ID of the host being removed * @param localHostId - my own host ID for replication coordination */ public void advertiseRemoving(InetAddress endpoint, UUID hostId, UUID localHostId) @@ -383,6 +388,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /** * Handles switching the endpoint's state from REMOVING_TOKEN to REMOVED_TOKEN * This should only be called after advertiseRemoving + * * @param endpoint * @param hostId */ @@ -412,7 +418,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean * It will try extremely hard to obliterate any endpoint from the ring, * even if it does not know about it. * This should only ever be called by human via JMX. - * @param address + * + * @param address * @throws UnknownHostException */ public void unsafeAssassinateEndpoint(String address) throws UnknownHostException @@ -423,7 +430,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean logger.warn("Assassinating {} via gossip", endpoint); if (epState == null) { - epState = new EndpointState(new HeartBeatState((int)((System.currentTimeMillis() + 60000) / 1000), 9999)); + epState = new EndpointState(new HeartBeatState((int) ((System.currentTimeMillis() + 60000) / 1000), 9999)); } else { @@ -480,10 +487,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /** * Returns true if the chosen target was also a seed. False otherwise * - * * @param message - * @param epSet a set of endpoint from which a random endpoint is chosen. - * @return true if the chosen endpoint is also a seed. + * @param epSet a set of endpoint from which a random endpoint is chosen. + * @return true if the chosen endpoint is also a seed. */ private boolean sendGossip(MessageOut<GossipDigestSyn> message, Set<InetAddress> epSet) { @@ -504,7 +510,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean private boolean doGossipToLiveMember(MessageOut<GossipDigestSyn> message) { int size = liveEndpoints.size(); - if ( size == 0 ) + if (size == 0) return false; return sendGossip(message, liveEndpoints); } @@ -514,12 +520,12 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean { double liveEndpointCount = liveEndpoints.size(); double unreachableEndpointCount = unreachableEndpoints.size(); - if ( unreachableEndpointCount > 0 ) + if (unreachableEndpointCount > 0) { /* based on some probability */ double prob = unreachableEndpointCount / (liveEndpointCount + 1); double randDbl = random.nextDouble(); - if ( randDbl < prob ) + if (randDbl < prob) sendGossip(message, unreachableEndpoints.keySet()); } } @@ -528,23 +534,23 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean private void doGossipToSeed(MessageOut<GossipDigestSyn> prod) { int size = seeds.size(); - if ( size > 0 ) + if (size > 0) { - if ( size == 1 && seeds.contains(FBUtilities.getBroadcastAddress()) ) + if (size == 1 && seeds.contains(FBUtilities.getBroadcastAddress())) { return; } - if ( liveEndpoints.size() == 0 ) + if (liveEndpoints.size() == 0) { sendGossip(prod, seeds); } else { /* Gossip with the seed with some probability. */ - double probability = seeds.size() / (double)( liveEndpoints.size() + unreachableEndpoints.size() ); + double probability = seeds.size() / (double) (liveEndpoints.size() + unreachableEndpoints.size()); double randDbl = random.nextDouble(); - if ( randDbl <= probability ) + if (randDbl <= probability) sendGossip(prod, seeds); } } @@ -555,14 +561,14 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean long now = System.currentTimeMillis(); Set<InetAddress> eps = endpointStateMap.keySet(); - for ( InetAddress endpoint : eps ) + for (InetAddress endpoint : eps) { - if ( endpoint.equals(FBUtilities.getBroadcastAddress()) ) + if (endpoint.equals(FBUtilities.getBroadcastAddress())) continue; FailureDetector.instance.interpret(endpoint); EndpointState epState = endpointStateMap.get(endpoint); - if ( epState != null ) + if (epState != null) { long duration = now - epState.getUpdateTimestamp(); @@ -578,7 +584,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean // check for dead state removal long expireTime = getExpireTimeForEndpoint(endpoint); if (!epState.isAlive() && (now > expireTime) - && (!StorageService.instance.getTokenMetadata().isMember(endpoint))) + && (!StorageService.instance.getTokenMetadata().isMember(endpoint))) { if (logger.isDebugEnabled()) { @@ -624,7 +630,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean { if (MessagingService.instance().knowsVersion(endpoint) && MessagingService.instance().getVersion(endpoint) >= MessagingService.VERSION_12) return true; - else if (getEndpointStateForEndpoint(endpoint).getApplicationState(ApplicationState.NET_VERSION) != null && Integer.parseInt(getEndpointStateForEndpoint(endpoint).getApplicationState(ApplicationState.NET_VERSION).value) >= MessagingService.VERSION_12) + else if (getEndpointStateForEndpoint(endpoint).getApplicationState(ApplicationState.NET_VERSION) != null && Integer.parseInt(getEndpointStateForEndpoint(endpoint).getApplicationState(ApplicationState.NET_VERSION).value) >= MessagingService.VERSION_12) return true; return false; } @@ -646,7 +652,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean EndpointState epState = endpointStateMap.get(forEndpoint); EndpointState reqdEndpointState = null; - if ( epState != null ) + if (epState != null) { /* * Here we try to include the Heart Beat state only if it is @@ -657,7 +663,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean * heart beat and throw it away on the receiver if it is redundant. */ int localHbVersion = epState.getHeartBeatState().getHeartBeatVersion(); - if ( localHbVersion > version ) + if (localHbVersion > version) { reqdEndpointState = new EndpointState(epState.getHeartBeatState()); if (logger.isTraceEnabled()) @@ -667,9 +673,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean for (Entry<ApplicationState, VersionedValue> entry : epState.getApplicationStateMap().entrySet()) { VersionedValue value = entry.getValue(); - if ( value.version > version ) + if (value.version > version) { - if ( reqdEndpointState == null ) + if (reqdEndpointState == null) { reqdEndpointState = new EndpointState(epState.getHeartBeatState()); } @@ -683,7 +689,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean return reqdEndpointState; } - /** determine which endpoint started up earlier */ + /** + * determine which endpoint started up earlier + */ public int compareEndpointStartup(InetAddress addr1, InetAddress addr2) { EndpointState ep1 = getEndpointStateForEndpoint(addr1); @@ -707,12 +715,12 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean * If the local endpoint state exists then report to the FD only * if the versions workout. */ - if ( localEndpointState != null ) + if (localEndpointState != null) { IFailureDetector fd = FailureDetector.instance; int localGeneration = localEndpointState.getHeartBeatState().getGeneration(); int remoteGeneration = remoteEndpointState.getHeartBeatState().getGeneration(); - if ( remoteGeneration > localGeneration ) + if (remoteGeneration > localGeneration) { localEndpointState.updateTimestamp(); // this node was dead and the generation changed, this indicates a reboot, or possibly a takeover @@ -726,11 +734,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean return; } - if ( remoteGeneration == localGeneration ) + if (remoteGeneration == localGeneration) { int localVersion = getMaxEndpointStateVersion(localEndpointState); int remoteVersion = remoteEndpointState.getHeartBeatState().getHeartBeatVersion(); - if ( remoteVersion > localVersion ) + if (remoteVersion > localVersion) { localEndpointState.updateTimestamp(); // just a version change, report to the fd @@ -775,7 +783,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /** * This method is called whenever there is a "big" change in ep state (a generation change for a known node). * - * @param ep endpoint + * @param ep endpoint * @param epState EndpointState for the endpoint */ private void handleMajorStateChange(InetAddress ep, EndpointState epState) @@ -827,7 +835,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean for (Entry<InetAddress, EndpointState> entry : epStateMap.entrySet()) { InetAddress ep = entry.getKey(); - if ( ep.equals(FBUtilities.getBroadcastAddress())) + if (ep.equals(FBUtilities.getBroadcastAddress())) continue; if (justRemovedEndpoints.containsKey(ep)) { @@ -842,7 +850,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean If state does not exist just add it. If it does then add it if the remote generation is greater. If there is a generation tie, attempt to break it by heartbeat version. */ - if ( localEpStatePtr != null ) + if (localEpStatePtr != null) { int localGeneration = localEpStatePtr.getHeartBeatState().getGeneration(); int remoteGeneration = remoteState.getHeartBeatState().getGeneration(); @@ -856,18 +864,18 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean // major state change will handle the update by inserting the remote state directly handleMajorStateChange(ep, remoteState); } - else if ( remoteGeneration == localGeneration ) // generation has not changed, apply new states + else if (remoteGeneration == localGeneration) // generation has not changed, apply new states { /* find maximum state */ int localMaxVersion = getMaxEndpointStateVersion(localEpStatePtr); int remoteMaxVersion = getMaxEndpointStateVersion(remoteState); - if ( remoteMaxVersion > localMaxVersion ) + if (remoteMaxVersion > localMaxVersion) { // apply states, but do not notify since there is no major change applyNewStates(ep, localEpStatePtr, remoteState); } else if (logger.isTraceEnabled()) - logger.trace("Ignoring remote version " + remoteMaxVersion + " <= " + localMaxVersion + " for " + ep); + logger.trace("Ignoring remote version " + remoteMaxVersion + " <= " + localMaxVersion + " for " + ep); if (!localEpStatePtr.isAlive() && !isDeadState(localEpStatePtr)) // unless of course, it was dead markAlive(ep, localEpStatePtr); } @@ -923,7 +931,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean private void requestAll(GossipDigest gDigest, List<GossipDigest> deltaGossipDigestList, int remoteGeneration) { /* We are here since we have no data for this endpoint locally so request everthing. */ - deltaGossipDigestList.add( new GossipDigest(gDigest.getEndpoint(), remoteGeneration, 0) ); + deltaGossipDigestList.add(new GossipDigest(gDigest.getEndpoint(), remoteGeneration, 0)); if (logger.isTraceEnabled()) logger.trace("requestAll for " + gDigest.getEndpoint()); } @@ -931,8 +939,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean /* Send all the data with version greater than maxRemoteVersion */ private void sendAll(GossipDigest gDigest, Map<InetAddress, EndpointState> deltaEpStateMap, int maxRemoteVersion) { - EndpointState localEpStatePtr = getStateForVersionBiggerThan(gDigest.getEndpoint(), maxRemoteVersion) ; - if ( localEpStatePtr != null ) + EndpointState localEpStatePtr = getStateForVersionBiggerThan(gDigest.getEndpoint(), maxRemoteVersion); + if (localEpStatePtr != null) deltaEpStateMap.put(gDigest.getEndpoint(), localEpStatePtr); } @@ -942,7 +950,7 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean */ void examineGossiper(List<GossipDigest> gDigestList, List<GossipDigest> deltaGossipDigestList, Map<InetAddress, EndpointState> deltaEpStateMap) { - for ( GossipDigest gDigest : gDigestList ) + for (GossipDigest gDigest : gDigestList) { int remoteGeneration = gDigest.getGeneration(); int maxRemoteVersion = gDigest.getMaxVersion(); @@ -953,25 +961,25 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean then we follow the "if" path of the logic. If we have absolutely nothing for this endpoint we need to request all the data for this endpoint. */ - if ( epStatePtr != null ) + if (epStatePtr != null) { int localGeneration = epStatePtr.getHeartBeatState().getGeneration(); /* get the max version of all keys in the state associated with this endpoint */ int maxLocalVersion = getMaxEndpointStateVersion(epStatePtr); - if ( remoteGeneration == localGeneration && maxRemoteVersion == maxLocalVersion ) + if (remoteGeneration == localGeneration && maxRemoteVersion == maxLocalVersion) continue; - if ( remoteGeneration > localGeneration ) + if (remoteGeneration > localGeneration) { /* we request everything from the gossiper */ requestAll(gDigest, deltaGossipDigestList, remoteGeneration); } - else if ( remoteGeneration < localGeneration ) + else if (remoteGeneration < localGeneration) { /* send all data with generation = localgeneration and version > 0 */ sendAll(gDigest, deltaEpStateMap, 0); } - else if ( remoteGeneration == localGeneration ) + else if (remoteGeneration == localGeneration) { /* If the max remote version is greater then we request the remote endpoint send us all the data @@ -980,11 +988,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean If the max remote version is lesser, then we send all the data we have locally for this endpoint with version greater than the max remote version. */ - if ( maxRemoteVersion > maxLocalVersion ) + if (maxRemoteVersion > maxLocalVersion) { - deltaGossipDigestList.add( new GossipDigest(gDigest.getEndpoint(), remoteGeneration, maxLocalVersion) ); + deltaGossipDigestList.add(new GossipDigest(gDigest.getEndpoint(), remoteGeneration, maxLocalVersion)); } - else if ( maxRemoteVersion < maxLocalVersion ) + else if (maxRemoteVersion < maxLocalVersion) { /* send all data with generation = localgeneration and version > maxRemoteVersion */ sendAll(gDigest, deltaEpStateMap, maxRemoteVersion); @@ -1133,7 +1141,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean expireTimeEndpointMap.put(endpoint, expireTime); } - public static long computeExpireTime() { + public static long computeExpireTime() + { return System.currentTimeMillis() + Gossiper.aVeryLongTime; } http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java b/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java index 2abee01..dd7ee2c 100644 --- a/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java +++ b/src/java/org/apache/cassandra/gms/IEndpointStateChangeSubscriber.java @@ -34,7 +34,7 @@ public interface IEndpointStateChangeSubscriber * for specified endpoint * * @param endpoint endpoint for which the state change occurred. - * @param epState state that actually changed for the above endpoint. + * @param epState state that actually changed for the above endpoint. */ public void onJoin(InetAddress endpoint, EndpointState epState); http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/IFailureDetectionEventListener.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/IFailureDetectionEventListener.java b/src/java/org/apache/cassandra/gms/IFailureDetectionEventListener.java index 25b90e4..8b274b6 100644 --- a/src/java/org/apache/cassandra/gms/IFailureDetectionEventListener.java +++ b/src/java/org/apache/cassandra/gms/IFailureDetectionEventListener.java @@ -29,7 +29,8 @@ public interface IFailureDetectionEventListener { /** * Convict the specified endpoint. - * @param ep endpoint to be convicted + * + * @param ep endpoint to be convicted * @param phi the value of phi with with ep was convicted */ public void convict(InetAddress ep, double phi); http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/IFailureDetector.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/IFailureDetector.java b/src/java/org/apache/cassandra/gms/IFailureDetector.java index 444c100..82124be 100644 --- a/src/java/org/apache/cassandra/gms/IFailureDetector.java +++ b/src/java/org/apache/cassandra/gms/IFailureDetector.java @@ -39,6 +39,7 @@ public interface IFailureDetector /** * Clear any existing interval timings for this endpoint + * * @param ep */ public void clear(InetAddress ep); @@ -47,18 +48,18 @@ public interface IFailureDetector * This method is invoked by any entity wanting to interrogate the status of an endpoint. * In our case it would be the Gossiper. The Failure Detector will then calculate Phi and * deem an endpoint as suspicious or alive as explained in the Hayashibara paper. - * + * <p/> * param ep endpoint for which we interpret the inter arrival times. - */ + */ public void interpret(InetAddress ep); /** * This method is invoked by the receiver of the heartbeat. In our case it would be * the Gossiper. Gossiper inform the Failure Detector on receipt of a heartbeat. The * FailureDetector will then sample the arrival time as explained in the paper. - * + * <p/> * param ep endpoint being reported. - */ + */ public void report(InetAddress ep); /** @@ -73,12 +74,14 @@ public interface IFailureDetector /** * Register interest for Failure Detector events. + * * @param listener implementation of an application provided IFailureDetectionEventListener */ public void registerFailureDetectionEventListener(IFailureDetectionEventListener listener); /** * Un-register interest for Failure Detector events. + * * @param listener implementation of an application provided IFailureDetectionEventListener */ public void unregisterFailureDetectionEventListener(IFailureDetectionEventListener listener); http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/IFailureNotification.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/IFailureNotification.java b/src/java/org/apache/cassandra/gms/IFailureNotification.java index 84195e6..7e1c955 100644 --- a/src/java/org/apache/cassandra/gms/IFailureNotification.java +++ b/src/java/org/apache/cassandra/gms/IFailureNotification.java @@ -22,5 +22,6 @@ import java.net.InetAddress; public interface IFailureNotification { public void convict(InetAddress ep); + public void revive(InetAddress ep); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/TokenSerializer.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/TokenSerializer.java b/src/java/org/apache/cassandra/gms/TokenSerializer.java index cc7f177..5d01aff 100644 --- a/src/java/org/apache/cassandra/gms/TokenSerializer.java +++ b/src/java/org/apache/cassandra/gms/TokenSerializer.java @@ -20,6 +20,7 @@ package org.apache.cassandra.gms; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.Token; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; http://git-wip-us.apache.org/repos/asf/cassandra/blob/b2079567/src/java/org/apache/cassandra/gms/VersionedValue.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/VersionedValue.java b/src/java/org/apache/cassandra/gms/VersionedValue.java index 5841c4b..4f1d275 100644 --- a/src/java/org/apache/cassandra/gms/VersionedValue.java +++ b/src/java/org/apache/cassandra/gms/VersionedValue.java @@ -24,6 +24,7 @@ import java.util.Collection; import java.util.UUID; import com.google.common.collect.Iterables; + import static com.google.common.base.Charsets.ISO_8859_1; import org.apache.cassandra.db.TypeSizes; @@ -32,6 +33,7 @@ import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.utils.FBUtilities; + import org.apache.commons.lang.StringUtils; @@ -40,11 +42,11 @@ import org.apache.commons.lang.StringUtils; * application wants to make available to the rest of the nodes in the cluster. * Whenever a piece of state needs to be disseminated to the rest of cluster wrap * the state in an instance of <i>ApplicationState</i> and add it to the Gossiper. - * + * <p/> * e.g. if we want to disseminate load information for node A do the following: - * - * ApplicationState loadState = new ApplicationState(<string representation of load>); - * Gossiper.instance.addApplicationState("LOAD STATE", loadState); + * <p/> + * ApplicationState loadState = new ApplicationState(<string representation of load>); + * Gossiper.instance.addApplicationState("LOAD STATE", loadState); */ public class VersionedValue implements Comparable<VersionedValue> @@ -54,7 +56,7 @@ public class VersionedValue implements Comparable<VersionedValue> // this must be a char that cannot be present in any token public final static char DELIMITER = ','; - public final static String DELIMITER_STR = new String(new char[] { DELIMITER }); + public final static String DELIMITER_STR = new String(new char[]{ DELIMITER }); // values for ApplicationState.STATUS public final static String STATUS_BOOTSTRAPPING = "BOOT"; @@ -98,7 +100,7 @@ public class VersionedValue implements Comparable<VersionedValue> return "Value(" + value + "," + version + ")"; } - private static String versionString(String...args) + private static String versionString(String... args) { return StringUtils.join(args, VersionedValue.DELIMITER); } @@ -142,14 +144,14 @@ public class VersionedValue implements Comparable<VersionedValue> public VersionedValue leaving(Collection<Token> tokens) { return new VersionedValue(versionString(VersionedValue.STATUS_LEAVING, - makeTokenString(tokens))); + makeTokenString(tokens))); } public VersionedValue left(Collection<Token> tokens, long expireTime) { return new VersionedValue(versionString(VersionedValue.STATUS_LEFT, - Long.toString(expireTime), - makeTokenString(tokens))); + Long.toString(expireTime), + makeTokenString(tokens))); } public VersionedValue moving(Token token) @@ -269,8 +271,8 @@ public class VersionedValue implements Comparable<VersionedValue> // three component 'left' was adopted starting from Cassandra 1.0 // previous versions have '<type>:<token>' format outValue = (version < MessagingService.VERSION_10) - ? versionString(pieces[0], pieces[2]) - : versionString(pieces[0], pieces[2], pieces[1]); + ? versionString(pieces[0], pieces[2]) + : versionString(pieces[0], pieces[2], pieces[1]); } if ((type.equals(REMOVAL_COORDINATOR)) || (type.equals(REMOVING_TOKEN)) || (type.equals(REMOVED_TOKEN)))
