cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject svn commit: r1063361 - in /cassandra/branches/cassandra-0.7: ./ src/java/org/apache/cassandra/db/ src/java/org/apache/cassandra/net/ src/java/org/apache/cassandra/service/
Date Tue, 25 Jan 2011 17:09:43 GMT
Author: jbellis
Date: Tue Jan 25 17:09:43 2011
New Revision: 1063361

URL: http://svn.apache.org/viewvc?rev=1063361&view=rev
Log:
fix bugs in multi-DC replication
patch by ivancso; reviewed by jbellis for CASSANDRA-2051

Modified:
    cassandra/branches/cassandra-0.7/CHANGES.txt
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/net/Header.java
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/net/Message.java
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/DatacenterReadCallback.java
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java

Modified: cassandra/branches/cassandra-0.7/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/CHANGES.txt?rev=1063361&r1=1063360&r2=1063361&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/CHANGES.txt (original)
+++ cassandra/branches/cassandra-0.7/CHANGES.txt Tue Jan 25 17:09:43 2011
@@ -2,7 +2,7 @@
  * buffer network stack to avoid inefficient small TCP messages while avoiding
    the nagle/delayed ack problem (CASSANDRA-1896)
  * check log4j configuration for changes every 10s (CASSANDRA-1525, 1907)
- * more-efficient cross-DC replication (CASSANDRA-1530)
+ * more-efficient cross-DC replication (CASSANDRA-1530, -2051)
  * upgrade to TFastFramedTransport (CASSANDRA-1743)
  * avoid polluting page cache with commitlog or sstable writes
    and seq scan operations (CASSANDRA-1470)

Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java?rev=1063361&r1=1063360&r2=1063361&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java
(original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java
Tue Jan 25 17:09:43 2011
@@ -90,7 +90,7 @@ public class RowMutationVerbHandler impl
     private void forwardToLocalNodes(Message message, byte[] forwardBytes) throws UnknownHostException
     {
         // remove fwds from message to avoid infinite loop
-        message.setHeader(RowMutation.FORWARD_HEADER, null);
+        message.removeHeader(RowMutation.FORWARD_HEADER);
 
         int bytesPerInetAddress = FBUtilities.getLocalAddress().getAddress().length;
         assert forwardBytes.length >= bytesPerInetAddress;
@@ -110,7 +110,7 @@ public class RowMutationVerbHandler impl
 
             // Send the original message to the address specified by the FORWARD_HINT
             // Let the response go back to the coordinator
-            MessagingService.instance().sendOneWay(message, message.getFrom());
+            MessagingService.instance().sendOneWay(message, address);
 
             offset += bytesPerInetAddress;
         }

Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/net/Header.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/net/Header.java?rev=1063361&r1=1063360&r2=1063361&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/net/Header.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/net/Header.java Tue Jan
25 17:09:43 2011
@@ -97,6 +97,11 @@ public class Header
     {
         details_.put(key, value);
     }
+
+    void removeDetail(String key)
+    {
+        details_.remove(key);
+    }
 }
 
 class HeaderSerializer implements ICompactSerializer<Header>

Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/net/Message.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/net/Message.java?rev=1063361&r1=1063360&r2=1063361&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/net/Message.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/net/Message.java Tue Jan
25 17:09:43 2011
@@ -68,6 +68,11 @@ public class Message
     {
         header_.setDetail(key, value);
     }
+    
+    public void removeHeader(String key)
+    {
+        header_.removeDetail(key);
+    }
 
     public byte[] getMessageBody()
     {

Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/DatacenterReadCallback.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/DatacenterReadCallback.java?rev=1063361&r1=1063360&r2=1063361&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/DatacenterReadCallback.java
(original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/DatacenterReadCallback.java
Tue Jan 25 17:09:43 2011
@@ -27,6 +27,7 @@ import java.util.concurrent.atomic.Atomi
 
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.Table;
+import org.apache.cassandra.db.ReadResponse;
 import org.apache.cassandra.locator.IEndpointSnitch;
 import org.apache.cassandra.locator.NetworkTopologyStrategy;
 import org.apache.cassandra.net.Message;
@@ -40,7 +41,7 @@ import org.apache.cassandra.utils.FBUtil
 public class DatacenterReadCallback<T> extends ReadCallback<T>
 {
     private static final IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
-	private static final String localdc = snitch.getDatacenter(FBUtilities.getLocalAddress());
+    private static final String localdc = snitch.getDatacenter(FBUtilities.getLocalAddress());
     private AtomicInteger localResponses;
     
     public DatacenterReadCallback(IResponseResolver<T> resolver, ConsistencyLevel consistencyLevel,
String table)
@@ -54,8 +55,7 @@ public class DatacenterReadCallback<T> e
     {
         resolver.preprocess(message);
 
-        int n;
-        n = localdc.equals(snitch.getDatacenter(message.getFrom())) 
+        int n = localdc.equals(snitch.getDatacenter(message.getFrom()))
                 ? localResponses.decrementAndGet()
                 : localResponses.get();
 
@@ -66,6 +66,19 @@ public class DatacenterReadCallback<T> e
     }
     
     @Override
+    public void response(ReadResponse result)
+    {
+        ((ReadResponseResolver) resolver).injectPreProcessed(result);
+
+        int n = localResponses.decrementAndGet();
+
+        if (n == 0 && resolver.isDataPresent())
+        {
+            condition.signal();
+        }
+    }
+    
+    @Override
     public int determineBlockFor(ConsistencyLevel consistency_level, String table)
 	{
         NetworkTopologyStrategy stategy = (NetworkTopologyStrategy) Table.open(table).getReplicationStrategy();

Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java?rev=1063361&r1=1063360&r2=1063361&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
(original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
Tue Jan 25 17:09:43 2011
@@ -234,10 +234,9 @@ public class StorageProxy implements Sto
                 Message message = messages.getKey();
                 Iterator<InetAddress> iter = messages.getValue().iterator();
                 assert iter.hasNext();
-                
+
                 // First endpoint in list is the destination for this group
                 InetAddress target = iter.next();
-            
 
                 // Add all the other destinations that are bound for the same dataCenter
as a header in the primary message.
                 while (iter.hasNext())
@@ -382,7 +381,7 @@ public class StorageProxy implements Sto
             {
                 Message message = command.makeReadMessage();
                 if (logger.isDebugEnabled())
-                    logger.debug("reading digest for " + command + " from " + message.getMessageId()
+ "@" + dataPoint);
+                    logger.debug("reading data for " + command + " from " + message.getMessageId()
+ "@" + dataPoint);
                 MessagingService.instance().sendRR(message, dataPoint, handler);
             }
 



Mime
View raw message