cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject svn commit: r1212088 - in /cassandra/trunk: CHANGES.txt src/java/org/apache/cassandra/db/RowMutationVerbHandler.java src/java/org/apache/cassandra/net/MessagingService.java src/java/org/apache/cassandra/service/StorageProxy.java
Date Thu, 08 Dec 2011 20:31:23 GMT
Author: jbellis
Date: Thu Dec  8 20:31:23 2011
New Revision: 1212088

URL: http://svn.apache.org/viewvc?rev=1212088&view=rev
Log:
multi-dc replication optimization supporting CL > ONE
patch by Vijay and jbellis for CASSANDRA-3577

Modified:
    cassandra/trunk/CHANGES.txt
    cassandra/trunk/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java
    cassandra/trunk/src/java/org/apache/cassandra/net/MessagingService.java
    cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java

Modified: cassandra/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/trunk/CHANGES.txt?rev=1212088&r1=1212087&r2=1212088&view=diff
==============================================================================
--- cassandra/trunk/CHANGES.txt (original)
+++ cassandra/trunk/CHANGES.txt Thu Dec  8 20:31:23 2011
@@ -1,4 +1,5 @@
 1.1-dev
+ * multi-dc replication optimization supporting CL > ONE (CASSANDRA-3577)
  * add command to stop compactions (CASSANDRA-1740, 3566)
  * multithreaded streaming (CASSANDRA-3494)
  * removed in-tree redhat spec (CASSANDRA-3567)

Modified: cassandra/trunk/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java?rev=1212088&r1=1212087&r2=1212088&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/RowMutationVerbHandler.java Thu Dec 
8 20:31:23 2011
@@ -18,18 +18,18 @@
 
 package org.apache.cassandra.db;
 
+import java.io.DataInputStream;
 import java.io.IOException;
 import java.net.InetAddress;
-import java.net.UnknownHostException;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.io.util.FastByteArrayInputStream;
+import org.apache.cassandra.net.CompactEndpointSerializationHelper;
 import org.apache.cassandra.net.IVerbHandler;
 import org.apache.cassandra.net.Message;
 import org.apache.cassandra.net.MessagingService;
-import org.apache.cassandra.utils.FBUtilities;
-
 
 public class RowMutationVerbHandler implements IVerbHandler
 {
@@ -45,7 +45,7 @@ public class RowMutationVerbHandler impl
 
             // Check if there were any forwarding headers in this message
             byte[] forwardBytes = message.getHeader(RowMutation.FORWARD_HEADER);
-            if (forwardBytes != null)
+            if (forwardBytes != null && message.getVersion() >= MessagingService.VERSION_11)
                 forwardToLocalNodes(message, forwardBytes);
 
             rm.apply();
@@ -62,32 +62,26 @@ public class RowMutationVerbHandler impl
         }
     }  
     
-    private void forwardToLocalNodes(Message message, byte[] forwardBytes) throws UnknownHostException
+    /**
+     * Older version (< 1.0) will not send this message at all, hence we don't
+     * need to check the version of the data.
+     */
+    private void forwardToLocalNodes(Message message, byte[] forwardBytes) throws IOException
     {
+        DataInputStream dis = new DataInputStream(new FastByteArrayInputStream(forwardBytes));
+        int size = dis.readInt();
+        
         // remove fwds from message to avoid infinite loop
         Message messageCopy = message.withHeaderRemoved(RowMutation.FORWARD_HEADER);
-
-        int bytesPerInetAddress = FBUtilities.getBroadcastAddress().getAddress().length;
-        assert forwardBytes.length >= bytesPerInetAddress;
-        assert forwardBytes.length % bytesPerInetAddress == 0;
-
-        int offset = 0;
-        byte[] addressBytes = new byte[bytesPerInetAddress];
-
-        // Send a message to each of the addresses on our Forward List
-        while (offset < forwardBytes.length)
+        for (int i = 0; i < size; i++)
         {
-            System.arraycopy(forwardBytes, offset, addressBytes, 0, bytesPerInetAddress);
-            InetAddress address = InetAddress.getByAddress(addressBytes);
-
+            // Send a message to each of the addresses on our Forward List
+            InetAddress address = CompactEndpointSerializationHelper.deserialize(dis);
+            String id = dis.readUTF();
             if (logger_.isDebugEnabled())
-                logger_.debug("Forwarding message to " + address);
-
-            // Send the original message to the address specified by the FORWARD_HINT
+                logger_.debug("Forwarding message to " + address + " with= ID: " + id);
             // Let the response go back to the coordinator
-            MessagingService.instance().sendOneWay(messageCopy, address);
-
-            offset += bytesPerInetAddress;
+            MessagingService.instance().sendOneWay(messageCopy, id, address);
         }
     }
 }

Modified: cassandra/trunk/src/java/org/apache/cassandra/net/MessagingService.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/net/MessagingService.java?rev=1212088&r1=1212087&r2=1212088&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/net/MessagingService.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/net/MessagingService.java Thu Dec  8 20:31:23
2011
@@ -341,8 +341,14 @@ public final class MessagingService impl
         return verbHandlers_.get(type);
     }
 
-    private void addCallback(IMessageCallback cb, String messageId, Message message, InetAddress
to, long timeout)
+    public String addCallback(IMessageCallback cb, Message message, InetAddress to)
     {
+        return addCallback(cb, message, to, DEFAULT_CALLBACK_TIMEOUT);
+    }
+    
+    public String addCallback(IMessageCallback cb, Message message, InetAddress to, long
timeout)
+    {
+        String messageId = nextId();
         CallbackInfo previous;
 
         // If HH is enabled and this is a mutation message => store the message to track
for potential hints.
@@ -352,6 +358,7 @@ public final class MessagingService impl
             previous = callbacks.put(messageId, new CallbackInfo(to, cb), timeout);
 
         assert previous == null;
+        return messageId;
     }
 
     private static AtomicInteger idGen = new AtomicInteger(0);
@@ -384,8 +391,7 @@ public final class MessagingService impl
      */
     public String sendRR(Message message, InetAddress to, IMessageCallback cb, long timeout)
     {
-        String id = nextId();
-        addCallback(cb, id, message, to, timeout);
+        String id = addCallback(cb, message, to, timeout);
         sendOneWay(message, id, to);
         return id;
     }
@@ -426,7 +432,7 @@ public final class MessagingService impl
      * @param message messages to be sent.
      * @param to endpoint to which the message needs to be sent
      */
-    private void sendOneWay(Message message, String id, InetAddress to)
+    public void sendOneWay(Message message, String id, InetAddress to)
     {
         if (logger_.isTraceEnabled())
             logger_.trace(FBUtilities.getBroadcastAddress() + " sending " + message.getVerb()
+ " to " + id + "@" + to);

Modified: cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java?rev=1212088&r1=1212087&r2=1212088&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java Thu Dec  8 20:31:23
2011
@@ -400,33 +400,41 @@ public class StorageProxy implements Sto
                 // a single message object is used for unhinted writes, so clean out any
forwards
                 // from previous loop iterations
                 message = message.withHeaderRemoved(RowMutation.FORWARD_HEADER);
+                Iterator<InetAddress> iter = messages.getValue().iterator();
+                InetAddress target = iter.next();
 
-                if (dataCenter.equals(localDataCenter))
+                // direct writes to local DC or old Cassadra versions
+                if (dataCenter.equals(localDataCenter) || Gossiper.instance.getVersion(target)
< MessagingService.VERSION_11)
                 {
-                    // direct writes to local DC or old Cassadra versions
-                    for (InetAddress destination : messages.getValue())
-                        MessagingService.instance().sendRR(message, destination, handler);
-                }
-                else
-                {
-                    // Non-local DC. First endpoint in list is the destination for this group
-                    Iterator<InetAddress> iter = messages.getValue().iterator();
-                    InetAddress target = iter.next();
-                    // Add all the other destinations of the same message as a header in
the primary message.
-                    if (iter.hasNext())
-                    {
-                        FastByteArrayOutputStream bos = new FastByteArrayOutputStream();
-                        DataOutputStream dos = new DataOutputStream(bos);
-                        while (iter.hasNext())
-                        {
-                            InetAddress destination = iter.next();
-                            dos.write(destination.getAddress());
-                        }
-                        message = message.withHeaderAdded(RowMutation.FORWARD_HEADER, bos.toByteArray());
-                    }
-                    // send the combined message + forward headers
+                    // yes, the loop and non-loop code here are the same; this is clunky
but we want to avoid
+                    // creating a second iterator since we already have a perfectly good
one
                     MessagingService.instance().sendRR(message, target, handler);
+                    while (iter.hasNext())
+                    {
+                        target = iter.next();
+                        MessagingService.instance().sendRR(message, target, handler);
+                    }
+                    continue;
+                }
+
+                // Add all the other destinations of the same message as a FORWARD_HEADER
entry
+                FastByteArrayOutputStream bos = new FastByteArrayOutputStream();
+                DataOutputStream dos = new DataOutputStream(bos);
+                dos.writeInt(messages.getValue().size() - 1);
+                while (iter.hasNext())
+                {
+                    InetAddress destination = iter.next();
+                    CompactEndpointSerializationHelper.serialize(destination, dos);
+                    String id = MessagingService.instance().addCallback(handler, message,
destination);
+                    dos.writeUTF(id);
+                    if (logger.isDebugEnabled())
+                        logger.debug("Adding FWD message to: " + destination + " with ID
" + id);
                 }
+                message = message.withHeaderAdded(RowMutation.FORWARD_HEADER, bos.toByteArray());
+                // send the combined message + forward headers
+                String id = MessagingService.instance().sendRR(message, target, handler);
+                if (logger.isDebugEnabled())
+                    logger.debug("Sending message to: " + target + " with ID " + id);
             }
         }
     }



Mime
View raw message