cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject [1/4] git commit: send dc-local replicas inline instead of putting them in a Map patch by jbellis; reviewed by dbrosius for CASSANDRA-5538
Date Sat, 01 Jun 2013 23:30:10 GMT
Updated Branches:
  refs/heads/trunk fa7d87de8 -> 3d1cfd1c9


send dc-local replicas inline instead of putting them in a Map
patch by jbellis; reviewed by dbrosius for CASSANDRA-5538


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

Branch: refs/heads/trunk
Commit: d0dc597261b0b9949ecd26150cd30ef89ecc7a21
Parents: be0523a
Author: Jonathan Ellis <jbellis@apache.org>
Authored: Thu May 30 20:09:30 2013 -0500
Committer: Jonathan Ellis <jbellis@apache.org>
Committed: Thu May 30 20:09:30 2013 -0500

----------------------------------------------------------------------
 .../org/apache/cassandra/service/StorageProxy.java |   68 +++++++-------
 1 files changed, 34 insertions(+), 34 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/d0dc5972/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 7e288ab..3da923b 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -603,7 +603,9 @@ public class StorageProxy implements StorageProxyMBean
         }
         else
         {
-            sendMessagesToOneDC(rm.createMessage(), endpoints, true, handler);
+            MessageOut<RowMutation> message = rm.createMessage();
+            for (InetAddress target : endpoints)
+                MessagingService.instance().sendRR(message, target, handler);
         }
     }
 
@@ -749,8 +751,10 @@ public class StorageProxy implements StorageProxyMBean
                                              ConsistencyLevel consistency_level)
     throws OverloadedException
     {
-        // replicas grouped by datacenter
+        // extra-datacenter replicas, grouped by dc
         Map<String, Collection<InetAddress>> dcGroups = null;
+        // only need to create a Message for non-local writes
+        MessageOut<RowMutation> message = null;
 
         for (InetAddress destination : targets)
         {
@@ -774,17 +778,28 @@ public class StorageProxy implements StorageProxyMBean
                 else
                 {
                     // belongs on a different server
+                    if (message == null)
+                        message = rm.createMessage();
+
                     String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(destination);
-                    Collection<InetAddress> messages = (dcGroups != null) ? dcGroups.get(dc)
: null;
-                    if (messages == null)
+                    // direct writes to local DC or old Cassandra versions
+                    // (1.1 knows how to forward old-style String message IDs; updated to
int in 2.0)
+                    if (localDataCenter.equals(dc) || MessagingService.instance().getVersion(destination)
< MessagingService.VERSION_20)
                     {
-                        messages = new ArrayList<InetAddress>(3); // most DCs will
have <= 3 replicas
-                        if (dcGroups == null)
-                            dcGroups = new HashMap<String, Collection<InetAddress>>();
-                        dcGroups.put(dc, messages);
+                        MessagingService.instance().sendRR(message, destination, responseHandler);
+                    }
+                    else
+                    {
+                        Collection<InetAddress> messages = (dcGroups != null) ? dcGroups.get(dc)
: null;
+                        if (messages == null)
+                        {
+                            messages = new ArrayList<InetAddress>(3); // most DCs will
have <= 3 replicas
+                            if (dcGroups == null)
+                                dcGroups = new HashMap<String, Collection<InetAddress>>();
+                            dcGroups.put(dc, messages);
+                        }
+                        messages.add(destination);
                     }
-
-                    messages.add(destination);
                 }
             }
             else
@@ -799,16 +814,16 @@ public class StorageProxy implements StorageProxyMBean
 
         if (dcGroups != null)
         {
-            MessageOut<RowMutation> message = rm.createMessage();
             // for each datacenter, send the message to one node to relay the write to other
replicas
-            for (Map.Entry<String, Collection<InetAddress>> entry: dcGroups.entrySet())
+            if (message == null)
+                message = rm.createMessage();
+
+            for (Collection<InetAddress> dcTargets : dcGroups.values())
             {
-                boolean isLocalDC = entry.getKey().equals(localDataCenter);
-                Collection<InetAddress> dcTargets = entry.getValue();
-                // a single message object is used for unhinted writes, so clean out any
forwards
-                // from previous loop iterations
+                // clean out any forwards from previous loop iterations
                 message = message.withHeaderRemoved(RowMutation.FORWARD_TO);
-                sendMessagesToOneDC(message, dcTargets, isLocalDC, responseHandler);
+
+                sendMessagesToNonlocalDC(message, dcTargets, responseHandler);
             }
         }
     }
@@ -860,27 +875,12 @@ public class StorageProxy implements StorageProxyMBean
         totalHints.incrementAndGet();
     }
 
-    private static void sendMessagesToOneDC(MessageOut message, Collection<InetAddress>
targets, boolean localDC, AbstractWriteResponseHandler handler)
+    private static void sendMessagesToNonlocalDC(MessageOut message, Collection<InetAddress>
targets, AbstractWriteResponseHandler handler)
     {
         Iterator<InetAddress> iter = targets.iterator();
         InetAddress target = iter.next();
 
-        // direct writes to local DC or old Cassandra versions
-        // (1.1 knows how to forward old-style String message IDs; updated to int in 2.0)
-        if (localDC || MessagingService.instance().getVersion(target) < MessagingService.VERSION_20)
-        {
-            // 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);
-            }
-            return;
-        }
-
-        // Add all the other destinations of the same message as a FORWARD_HEADER entry
+        // Add the other destinations of the same message as a FORWARD_HEADER entry
         DataOutputBuffer out = new DataOutputBuffer();
         try
         {


Mime
View raw message