From 989fd859f6feb8d0d7d24dbaee2b1ebeaa191115 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 19 Jan 2011 14:43:37 +0000 Subject: [PATCH] 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 | 1 + .../locator/DynamicEndpointSnitch.java | 18 ++++++-------- .../cassandra/locator/ILatencyPublisher.java | 24 ------------------- .../cassandra/net/MessagingService.java | 17 +++++++------ .../cassandra/net/ResponseVerbHandler.java | 18 +++----------- .../cassandra/service/StorageProxy.java | 4 +++- 6 files changed, 22 insertions(+), 60 deletions(-) delete mode 100644 src/java/org/apache/cassandra/locator/ILatencyPublisher.java diff --git a/CHANGES.txt b/CHANGES.txt index 2f0ab4c73e..b6e56826ff 100644 --- a/CHANGES.txt +++ b/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 diff --git a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java index 835164740d..67481fb825 100644 --- a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java +++ b/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 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 entry: windows.entrySet()) { diff --git a/src/java/org/apache/cassandra/locator/ILatencyPublisher.java b/src/java/org/apache/cassandra/locator/ILatencyPublisher.java deleted file mode 100644 index d1eabc2b90..0000000000 --- a/src/java/org/apache/cassandra/locator/ILatencyPublisher.java +++ /dev/null @@ -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); -} diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index 497e7ca498..eecba97a61 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/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[]; diff --git a/src/java/org/apache/cassandra/net/ResponseVerbHandler.java b/src/java/org/apache/cassandra/net/ResponseVerbHandler.java index 75e73eaf84..5880270cb2 100644 --- a/src/java/org/apache/cassandra/net/ResponseVerbHandler.java +++ b/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 subscribers = new ArrayList(); - + 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); - } } diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index e01945d982..12f6ddadb6 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -702,7 +702,8 @@ public class StorageProxy implements StorageProxyMBean static class weakReadLocalCallable implements Callable { - 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; } }