svn commit: r1188740 - in /cassandra/branches/cassandra-0.8: CHANGES.txt src/java/org/apache/cassandra/service/AntiEntropyService.java

[sl...@apache.org]

Author: slebresne
Date: Tue Oct 25 15:49:54 2011
New Revision: 1188740

URL: http://svn.apache.org/viewvc?rev=1188740&view=rev
Log:
Fix race in AntiEntropyService
patch by slebresne; reviewed by jbellis for CASSANDRA-3400

Modified:
    cassandra/branches/cassandra-0.8/CHANGES.txt
    cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/service/AntiEntropyService.java

Modified: cassandra/branches/cassandra-0.8/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/CHANGES.txt?rev=1188740&r1=1188739&r2=1188740&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/CHANGES.txt (original)
+++ cassandra/branches/cassandra-0.8/CHANGES.txt Tue Oct 25 15:49:54 2011
@@ -32,6 +32,7 @@
  * prevent nodes that failed to join from hanging around forever 
    (CASSANDRA-3351)
  * remove incorrect optimization from slice read path (CASSANDRA-3390)
+ * Fix race in AntiEntropyService (CASSANDRA-3400)
 
 
 0.8.7

Modified: cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/service/AntiEntropyService.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/service/AntiEntropyService.java?rev=1188740&r1=1188739&r2=1188740&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/service/AntiEntropyService.java (original)
+++ cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/service/AntiEntropyService.java Tue Oct 25 15:49:54 2011
@@ -766,6 +766,7 @@ public class AntiEntropyService
             private final Set<InetAddress> requestedEndpoints = new HashSet<InetAddress>();
             private final Map<InetAddress, MerkleTree> trees = new HashMap<InetAddress, MerkleTree>();
             private final Set<InetAddress> syncJobs = new HashSet<InetAddress>();
+            private final Condition requestsSent = new SimpleCondition();
 
             public RepairJob(String cfname)
             {
@@ -783,6 +784,8 @@ public class AntiEntropyService
                 // send requests to all nodes
                 for (InetAddress endpoint : requestedEndpoints)
                     AntiEntropyService.instance.request(getName(), endpoint, range, tablename, cfname);
+
+                requestsSent.signalAll();
             }
 
             /**
@@ -791,6 +794,16 @@ public class AntiEntropyService
              */
             public synchronized int addTree(TreeRequest request, MerkleTree tree)
             {
+                // Wait for all request to have been performed (see #3400)
+                try
+                {
+                    requestsSent.await();
+                }
+                catch (InterruptedException e)
+                {
+                    throw new AssertionError("Interrupted while waiting for requests to be sent");
+                }
+
                 assert request.cf.right.equals(cfname);
                 trees.put(request.endpoint, tree);
                 requestedEndpoints.remove(request.endpoint);