You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2013/11/13 16:38:51 UTC
[1/3] git commit: Improve Gossip tolerance of abnormally slow tasks
patch by jbellis; reviewed by jasobrown for CASSANDRA-6338
Updated Branches:
refs/heads/cassandra-2.0 91f976ce1 -> b45715665
refs/heads/trunk fa86ce88c -> 0619da2ae
Improve Gossip tolerance of abnormally slow tasks
patch by jbellis; reviewed by jasobrown for CASSANDRA-6338
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b4571566
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b4571566
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b4571566
Branch: refs/heads/cassandra-2.0
Commit: b45715665845f7d2b597adfc0fc9c055b271e8d0
Parents: 91f976c
Author: Jonathan Ellis <jb...@apache.org>
Authored: Wed Nov 13 09:37:45 2013 -0600
Committer: Jonathan Ellis <jb...@apache.org>
Committed: Wed Nov 13 09:37:45 2013 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
src/java/org/apache/cassandra/gms/Gossiper.java | 29 ++++++++++++++++++--
.../cassandra/net/MessageDeliveryTask.java | 10 +++++++
3 files changed, 38 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b4571566/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 3305739..e4f1862 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.0.3
+ * Improve Gossip tolerance of abnormally slow tasks (CASSANDRA-6338)
* Fix trying to hint timed out counter writes (CASSANDRA-6322)
* Allow restoring specific columnfamilies from archived CL (CASSANDRA-4809)
* Avoid flushing compaction_history after each operation (CASSANDRA-6287)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b4571566/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 8237fa3..e3079c0 100644
--- a/src/java/org/apache/cassandra/gms/Gossiper.java
+++ b/src/java/org/apache/cassandra/gms/Gossiper.java
@@ -34,6 +34,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.DebuggableScheduledThreadPoolExecutor;
+import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.net.IAsyncCallback;
@@ -113,6 +116,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
private boolean inShadowRound = false;
+ private volatile long lastProcessedMessageAt = System.currentTimeMillis();
+
private class GossipTask implements Runnable
{
public void run()
@@ -162,8 +167,6 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
if (!gossipedToSeed || liveEndpoints.size() < seeds.size())
doGossipToSeed(message);
- if (logger.isTraceEnabled())
- logger.trace("Performing status check ...");
doStatusCheck();
}
}
@@ -193,6 +196,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
}
}
+ public void setLastProcessedMessageAt(long timeInMillis)
+ {
+ this.lastProcessedMessageAt = timeInMillis;
+ }
+
protected void checkSeedContact(InetAddress ep)
{
if (!seedContacted && seeds.contains(ep))
@@ -586,9 +594,26 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
private void doStatusCheck()
{
+ if (logger.isTraceEnabled())
+ logger.trace("Performing status check ...");
+
long now = System.currentTimeMillis();
long nowNano = System.nanoTime();
+ long pending = ((JMXEnabledThreadPoolExecutor) StageManager.getStage(Stage.GOSSIP)).getPendingTasks();
+ if (pending > 0 && lastProcessedMessageAt < now - 1000)
+ {
+ // if some new messages just arrived, give the executor some time to work on them
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ // still behind? something's broke
+ if (lastProcessedMessageAt < now - 1000)
+ {
+ logger.warn("Gossip stage has {} pending tasks; skipping status check (no nodes will be marked down)", pending);
+ return;
+ }
+ }
+
Set<InetAddress> eps = endpointStateMap.keySet();
for (InetAddress endpoint : eps)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b4571566/src/java/org/apache/cassandra/net/MessageDeliveryTask.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/MessageDeliveryTask.java b/src/java/org/apache/cassandra/net/MessageDeliveryTask.java
index d7645b6..e49b93c 100644
--- a/src/java/org/apache/cassandra/net/MessageDeliveryTask.java
+++ b/src/java/org/apache/cassandra/net/MessageDeliveryTask.java
@@ -17,9 +17,13 @@
*/
package org.apache.cassandra.net;
+import java.util.EnumSet;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.gms.Gossiper;
+
public class MessageDeliveryTask implements Runnable
{
private static final Logger logger = LoggerFactory.getLogger(MessageDeliveryTask.class);
@@ -54,5 +58,11 @@ public class MessageDeliveryTask implements Runnable
}
verbHandler.doVerb(message, id);
+ if (GOSSIP_VERBS.contains(message.verb))
+ Gossiper.instance.setLastProcessedMessageAt(constructionTime);
}
+
+ EnumSet<MessagingService.Verb> GOSSIP_VERBS = EnumSet.of(MessagingService.Verb.GOSSIP_DIGEST_ACK,
+ MessagingService.Verb.GOSSIP_DIGEST_ACK2,
+ MessagingService.Verb.GOSSIP_DIGEST_SYN);
}
[2/3] git commit: Improve Gossip tolerance of abnormally slow tasks
patch by jbellis; reviewed by jasobrown for CASSANDRA-6338
Posted by jb...@apache.org.
Improve Gossip tolerance of abnormally slow tasks
patch by jbellis; reviewed by jasobrown for CASSANDRA-6338
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/b4571566
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/b4571566
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/b4571566
Branch: refs/heads/trunk
Commit: b45715665845f7d2b597adfc0fc9c055b271e8d0
Parents: 91f976c
Author: Jonathan Ellis <jb...@apache.org>
Authored: Wed Nov 13 09:37:45 2013 -0600
Committer: Jonathan Ellis <jb...@apache.org>
Committed: Wed Nov 13 09:37:45 2013 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
src/java/org/apache/cassandra/gms/Gossiper.java | 29 ++++++++++++++++++--
.../cassandra/net/MessageDeliveryTask.java | 10 +++++++
3 files changed, 38 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b4571566/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 3305739..e4f1862 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.0.3
+ * Improve Gossip tolerance of abnormally slow tasks (CASSANDRA-6338)
* Fix trying to hint timed out counter writes (CASSANDRA-6322)
* Allow restoring specific columnfamilies from archived CL (CASSANDRA-4809)
* Avoid flushing compaction_history after each operation (CASSANDRA-6287)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b4571566/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 8237fa3..e3079c0 100644
--- a/src/java/org/apache/cassandra/gms/Gossiper.java
+++ b/src/java/org/apache/cassandra/gms/Gossiper.java
@@ -34,6 +34,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.DebuggableScheduledThreadPoolExecutor;
+import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.net.IAsyncCallback;
@@ -113,6 +116,8 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
private boolean inShadowRound = false;
+ private volatile long lastProcessedMessageAt = System.currentTimeMillis();
+
private class GossipTask implements Runnable
{
public void run()
@@ -162,8 +167,6 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
if (!gossipedToSeed || liveEndpoints.size() < seeds.size())
doGossipToSeed(message);
- if (logger.isTraceEnabled())
- logger.trace("Performing status check ...");
doStatusCheck();
}
}
@@ -193,6 +196,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
}
}
+ public void setLastProcessedMessageAt(long timeInMillis)
+ {
+ this.lastProcessedMessageAt = timeInMillis;
+ }
+
protected void checkSeedContact(InetAddress ep)
{
if (!seedContacted && seeds.contains(ep))
@@ -586,9 +594,26 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
private void doStatusCheck()
{
+ if (logger.isTraceEnabled())
+ logger.trace("Performing status check ...");
+
long now = System.currentTimeMillis();
long nowNano = System.nanoTime();
+ long pending = ((JMXEnabledThreadPoolExecutor) StageManager.getStage(Stage.GOSSIP)).getPendingTasks();
+ if (pending > 0 && lastProcessedMessageAt < now - 1000)
+ {
+ // if some new messages just arrived, give the executor some time to work on them
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
+
+ // still behind? something's broke
+ if (lastProcessedMessageAt < now - 1000)
+ {
+ logger.warn("Gossip stage has {} pending tasks; skipping status check (no nodes will be marked down)", pending);
+ return;
+ }
+ }
+
Set<InetAddress> eps = endpointStateMap.keySet();
for (InetAddress endpoint : eps)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/b4571566/src/java/org/apache/cassandra/net/MessageDeliveryTask.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/MessageDeliveryTask.java b/src/java/org/apache/cassandra/net/MessageDeliveryTask.java
index d7645b6..e49b93c 100644
--- a/src/java/org/apache/cassandra/net/MessageDeliveryTask.java
+++ b/src/java/org/apache/cassandra/net/MessageDeliveryTask.java
@@ -17,9 +17,13 @@
*/
package org.apache.cassandra.net;
+import java.util.EnumSet;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.gms.Gossiper;
+
public class MessageDeliveryTask implements Runnable
{
private static final Logger logger = LoggerFactory.getLogger(MessageDeliveryTask.class);
@@ -54,5 +58,11 @@ public class MessageDeliveryTask implements Runnable
}
verbHandler.doVerb(message, id);
+ if (GOSSIP_VERBS.contains(message.verb))
+ Gossiper.instance.setLastProcessedMessageAt(constructionTime);
}
+
+ EnumSet<MessagingService.Verb> GOSSIP_VERBS = EnumSet.of(MessagingService.Verb.GOSSIP_DIGEST_ACK,
+ MessagingService.Verb.GOSSIP_DIGEST_ACK2,
+ MessagingService.Verb.GOSSIP_DIGEST_SYN);
}
[3/3] git commit: Merge branch 'cassandra-2.0' into trunk
Posted by jb...@apache.org.
Merge branch 'cassandra-2.0' into trunk
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/0619da2a
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/0619da2a
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/0619da2a
Branch: refs/heads/trunk
Commit: 0619da2aeab0be2da0b4db47872e0dd7c33c2457
Parents: fa86ce8 b457156
Author: Jonathan Ellis <jb...@apache.org>
Authored: Wed Nov 13 09:38:02 2013 -0600
Committer: Jonathan Ellis <jb...@apache.org>
Committed: Wed Nov 13 09:38:02 2013 -0600
----------------------------------------------------------------------
CHANGES.txt | 1 +
src/java/org/apache/cassandra/gms/Gossiper.java | 29 ++++++++++++++++++--
.../cassandra/net/MessageDeliveryTask.java | 10 +++++++
3 files changed, 38 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/0619da2a/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index 727c97d,e4f1862..6213580
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,17 -1,5 +1,18 @@@
+2.1
+ * Parallelize fetching rows for low-cardinality indexes (CASSANDRA-1337)
+ * change logging from log4j to logback (CASSANDRA-5883)
+ * switch to LZ4 compression for internode communication (CASSANDRA-5887)
+ * Stop using Thrift-generated Index* classes internally (CASSANDRA-5971)
+ * Remove 1.2 network compatibility code (CASSANDRA-5960)
+ * Remove leveled json manifest migration code (CASSANDRA-5996)
+ * Remove CFDefinition (CASSANDRA-6253)
+ * Use AtomicIntegerFieldUpdater in RefCountedMemory (CASSANDRA-6278)
+ * User-defined types for CQL3 (CASSANDRA-5590)
+ * Use of o.a.c.metrics in nodetool (CASSANDRA-5871)
+
+
2.0.3
+ * Improve Gossip tolerance of abnormally slow tasks (CASSANDRA-6338)
* Fix trying to hint timed out counter writes (CASSANDRA-6322)
* Allow restoring specific columnfamilies from archived CL (CASSANDRA-4809)
* Avoid flushing compaction_history after each operation (CASSANDRA-6287)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/0619da2a/src/java/org/apache/cassandra/gms/Gossiper.java
----------------------------------------------------------------------