cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From yu...@apache.org
Subject cassandra git commit: Unregister FD when all validations are done
Date Tue, 12 May 2015 19:12:56 GMT
Repository: cassandra
Updated Branches:
  refs/heads/trunk 9a2cae69c -> d30614428


Unregister FD when all validations are done

patch by yukim; reviewed by jbellis for CASSANDRA-9145


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/d3061442
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/d3061442
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/d3061442

Branch: refs/heads/trunk
Commit: d306144282595be6818fa386a3fbb4aece040884
Parents: 9a2cae6
Author: Yuki Morishita <yukim@apache.org>
Authored: Tue May 12 14:12:07 2015 -0500
Committer: Yuki Morishita <yukim@apache.org>
Committed: Tue May 12 14:12:07 2015 -0500

----------------------------------------------------------------------
 CHANGES.txt                                            |  2 +-
 src/java/org/apache/cassandra/repair/RepairJob.java    |  6 ------
 .../org/apache/cassandra/repair/RepairSession.java     | 13 +++++++++++++
 3 files changed, 14 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/d3061442/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index ed62eed..e4956c9 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -93,7 +93,7 @@
  * Use unsafe mutations for most unit tests (CASSANDRA-6969)
  * Fix race condition during calculation of pending ranges (CASSANDRA-7390)
  * Fail on very large batch sizes (CASSANDRA-8011)
- * Improve concurrency of repair (CASSANDRA-6455, 8208)
+ * Improve concurrency of repair (CASSANDRA-6455, 8208, 9145)
  * Select optimal CRC32 implementation at runtime (CASSANDRA-8614)
  * Evaluate MurmurHash of Token once per query (CASSANDRA-7096)
  * Generalize progress reporting (CASSANDRA-8901)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/d3061442/src/java/org/apache/cassandra/repair/RepairJob.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/repair/RepairJob.java b/src/java/org/apache/cassandra/repair/RepairJob.java
index 850a894..754e26f 100644
--- a/src/java/org/apache/cassandra/repair/RepairJob.java
+++ b/src/java/org/apache/cassandra/repair/RepairJob.java
@@ -27,7 +27,6 @@ import org.slf4j.LoggerFactory;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.tracing.Tracing;
-import org.apache.cassandra.gms.FailureDetector;
 import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.Pair;
 
@@ -115,11 +114,6 @@ public class RepairJob extends AbstractFuture<RepairResult> implements
Runnable
         {
             public ListenableFuture<List<SyncStat>> apply(List<TreeResponse>
trees) throws Exception
             {
-                // Unregister from FailureDetector once we've completed synchronizing Merkle
trees.
-                // After this point, we rely on tcp_keepalive for individual sockets to notify
us when a connection is down.
-                // See CASSANDRA-3569
-                FailureDetector.instance.unregisterFailureDetectionEventListener(session);
-
                 InetAddress local = FBUtilities.getLocalAddress();
 
                 List<SyncTask> syncTasks = new ArrayList<>();

http://git-wip-us.apache.org/repos/asf/cassandra/blob/d3061442/src/java/org/apache/cassandra/repair/RepairSession.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/repair/RepairSession.java b/src/java/org/apache/cassandra/repair/RepairSession.java
index 70bfaa6..a2dcdd1 100644
--- a/src/java/org/apache/cassandra/repair/RepairSession.java
+++ b/src/java/org/apache/cassandra/repair/RepairSession.java
@@ -23,6 +23,7 @@ import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.*;
@@ -91,6 +92,9 @@ public class RepairSession extends AbstractFuture<RepairSessionResult>
implement
     public final Set<InetAddress> endpoints;
     private final long repairedAt;
 
+    // number of validations left to be performed
+    private final AtomicInteger validationRemaining;
+
     private final AtomicBoolean isFailed = new AtomicBoolean(false);
 
     // Each validation task waits response from replica in validating ConcurrentMap (keyed
by CF name and endpoint address)
@@ -134,6 +138,7 @@ public class RepairSession extends AbstractFuture<RepairSessionResult>
implement
         this.range = range;
         this.endpoints = endpoints;
         this.repairedAt = repairedAt;
+        this.validationRemaining = new AtomicInteger(cfnames.length);
     }
 
     public UUID getId()
@@ -176,6 +181,14 @@ public class RepairSession extends AbstractFuture<RepairSessionResult>
implement
         logger.info("[repair #{}] {}", getId(), message);
         Tracing.traceRepair(message);
         task.treeReceived(tree);
+
+        // Unregister from FailureDetector once we've completed synchronizing Merkle trees.
+        // After this point, we rely on tcp_keepalive for individual sockets to notify us
when a connection is down.
+        // See CASSANDRA-3569
+        if (validationRemaining.decrementAndGet() == 0)
+        {
+            FailureDetector.instance.unregisterFailureDetectionEventListener(this);
+        }
     }
 
     /**


Mime
View raw message