add latency information fromlocal reads to DynamicSnitch

patch by brandonwilliams and jbellis for CASSANDRA-2004

git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.6@1060828 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -2,6 +2,7 @@
  * buffer network stack to avoid inefficient small TCP messages while avoiding
    the nagle/delayed ack problem (CASSANDRA-1896)
  * fix race condition in MessagingService.targets (CASSANDRA-1959)
+ * add latency information from local reads to DynamicSnitch (CASSANDRA-2004)
 
 
 0.6.9

src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java

@@ -38,10 +38,11 @@ import org.apache.cassandra.utils.FBUtilities;
  */
 public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILatencySubscriber, DynamicEndpointSnitchMBean
 {
-    private static int UPDATES_PER_INTERVAL = 10000;
-    private static int UPDATE_INTERVAL_IN_MS = 100;
-    private static int RESET_INTERVAL_IN_MS = 60000 * 10;
-    private static int WINDOW_SIZE = 100;
+    private static final int UPDATES_PER_INTERVAL = 10000;
+    private static final int UPDATE_INTERVAL_IN_MS = 100;
+    private static final int RESET_INTERVAL_IN_MS = 60000 * 10;
+    private static final int WINDOW_SIZE = 100;
+
     private boolean registered = false;
 
     private ConcurrentHashMap<InetAddress, Double> scores = new ConcurrentHashMap();
@@ -151,13 +152,8 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements ILa
     {
         if (!registered)
         {
-       	    ILatencyPublisher handler = (ILatencyPublisher)MessagingService.instance.getVerbHandler(StorageService.Verb.READ_RESPONSE);
-            if (handler != null)
-            {
-                handler.register(this);
-                registered = true;
-            }
-
+            MessagingService.instance.register(this);
+            registered = true;
         }
         for (Map.Entry<InetAddress, AdaptiveLatencyTracker> entry: windows.entrySet())
         {

src/java/org/apache/cassandra/locator/ILatencyPublisher.java

@@ -1,24 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.locator;
-
-public interface ILatencyPublisher
-{
-    public void register(ILatencySubscriber subcriber);
-}

src/java/org/apache/cassandra/net/MessagingService.java

@@ -35,16 +35,12 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import com.google.common.base.Function;
-import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.Multimap;
-import com.google.common.collect.Multimaps;
 import org.apache.log4j.Logger;
 
 import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
 import org.apache.cassandra.concurrent.StageManager;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.io.util.DataOutputBuffer;
-import org.apache.cassandra.locator.ILatencyPublisher;
 import org.apache.cassandra.locator.ILatencySubscriber;
 import org.apache.cassandra.net.io.SerializerType;
 import org.apache.cassandra.net.sink.SinkManager;
@@ -56,7 +52,7 @@ import org.apache.cassandra.utils.SimpleCondition;
 import org.cliffc.high_scale_lib.NonBlockingHashMap;
 import org.cliffc.high_scale_lib.NonBlockingHashSet;
 
-public class MessagingService implements ILatencyPublisher
+public class MessagingService
 {
     private static int version_ = 1;
     //TODO: make this parameter dynamic somehow.  Not sure if config is appropriate.
@@ -116,10 +112,7 @@ public class MessagingService implements ILatencyPublisher
                     return null;
 
                 for (InetAddress address : addresses)
-                {
-                    for (ILatencySubscriber subscriber : subscribers)
-                        subscriber.receiveTiming(address, (double) DatabaseDescriptor.getRpcTimeout());
-                }
+                    addLatency(address, (double) DatabaseDescriptor.getRpcTimeout());
 
                 return null;
             }
@@ -140,6 +133,12 @@ public class MessagingService implements ILatencyPublisher
         timer.schedule(logDropped, LOG_DROPPED_INTERVAL_IN_MS, LOG_DROPPED_INTERVAL_IN_MS);
     }
 
+    public void addLatency(InetAddress address, double latency)
+    {
+        for (ILatencySubscriber subscriber : subscribers)
+            subscriber.receiveTiming(address, latency);
+    }
+
     public byte[] hash(String type, byte data[])
     {
         byte result[];

src/java/org/apache/cassandra/net/ResponseVerbHandler.java

@@ -18,19 +18,12 @@
 
 package org.apache.cassandra.net;
 
-import java.util.ArrayList;
-import java.util.List;
-
 import org.apache.log4j.Logger;
 
-import org.apache.cassandra.locator.ILatencyPublisher;
-import org.apache.cassandra.locator.ILatencySubscriber;
-
-public class ResponseVerbHandler implements IVerbHandler, ILatencyPublisher
+public class ResponseVerbHandler implements IVerbHandler
 {
     private static final Logger logger_ = Logger.getLogger( ResponseVerbHandler.class );
-    private List<ILatencySubscriber>  subscribers = new ArrayList<ILatencySubscriber>();
-    
+
     public void doVerb(Message message)
     {     
         String messageId = message.getMessageId();
@@ -41,8 +34,7 @@ public class ResponseVerbHandler implements IVerbHandler, ILatencyPublisher
             return;
 
         // if cb is not null, then age will be valid
-        for (ILatencySubscriber subscriber : subscribers)
-            subscriber.receiveTiming(message.getFrom(), age);
+        MessagingService.instance.addLatency(message.getFrom(), age);
 
         if (cb instanceof IAsyncCallback)
         {
@@ -58,8 +50,4 @@ public class ResponseVerbHandler implements IVerbHandler, ILatencyPublisher
         }
     }
 
-    public void register(ILatencySubscriber subscriber)
-    {
-        subscribers.add(subscriber);
-    }
 }

src/java/org/apache/cassandra/service/StorageProxy.java

@@ -702,7 +702,8 @@ public class StorageProxy implements StorageProxyMBean
 
     static class weakReadLocalCallable implements Callable<Object>
     {
-        private ReadCommand command;
+        private final ReadCommand command;
+        private final long start = System.currentTimeMillis();
 
         weakReadLocalCallable(ReadCommand command)
         {
@@ -718,6 +719,7 @@ public class StorageProxy implements StorageProxyMBean
             Row row = command.getRow(table);
             StorageService.instance.doConsistencyCheck(row, command, FBUtilities.getLocalAddress());
 
+            MessagingService.instance.addLatency(FBUtilities.getLocalAddress(), System.currentTimeMillis() - start);
             return row;
         }
     }