cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject svn commit: r1055320 - in /cassandra/branches/cassandra-0.7: src/java/org/apache/cassandra/locator/ src/java/org/apache/cassandra/service/ test/unit/org/apache/cassandra/service/
Date Wed, 05 Jan 2011 07:00:34 GMT
Author: jbellis
Date: Wed Jan  5 07:00:33 2011
New Revision: 1055320

URL: http://svn.apache.org/viewvc?rev=1055320&view=rev
Log:
rename [Datacenter]QuorumResponseHandler -> [Datacenter]ReadCallback
patch by jbellis

Added:
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/DatacenterReadCallback.java
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/ReadCallback.java
Removed:
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/QuorumResponseHandler.java
Modified:
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
    cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
    cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java

Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java?rev=1055320&r1=1055319&r2=1055320&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
(original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
Wed Jan  5 07:00:33 2011
@@ -223,15 +223,6 @@ public abstract class AbstractReplicatio
         return getAddressRanges(temp).get(pendingAddress);
     }
 
-    public QuorumResponseHandler getQuorumResponseHandler(IResponseResolver responseResolver,
ConsistencyLevel consistencyLevel)
-    {
-        if (consistencyLevel.equals(ConsistencyLevel.LOCAL_QUORUM) || consistencyLevel.equals(ConsistencyLevel.EACH_QUORUM))
-        {
-            return new DatacenterQuorumResponseHandler(responseResolver, consistencyLevel,
table);
-        }
-        return new QuorumResponseHandler(responseResolver, consistencyLevel, table);
-    }
-
     public void invalidateCachedTokenEndpointValues()
     {
         clearEndpointCache();

Added: 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=1055320&view=auto
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/DatacenterReadCallback.java
(added)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/DatacenterReadCallback.java
Wed Jan  5 07:00:33 2011
@@ -0,0 +1,88 @@
+package org.apache.cassandra.service;
+/*
+ * 
+ * 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.
+ * 
+ */
+
+
+import java.net.InetAddress;
+import java.util.Collection;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.Table;
+import org.apache.cassandra.locator.IEndpointSnitch;
+import org.apache.cassandra.locator.NetworkTopologyStrategy;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.thrift.ConsistencyLevel;
+import org.apache.cassandra.thrift.UnavailableException;
+import org.apache.cassandra.utils.FBUtilities;
+
+/**
+ * Datacenter Quorum response handler blocks for a quorum of responses from the local DC
+ */
+public class DatacenterReadCallback<T> extends ReadCallback<T>
+{
+    private static final IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
+	private static final String localdc = snitch.getDatacenter(FBUtilities.getLocalAddress());
+    private AtomicInteger localResponses;
+    
+    public DatacenterReadCallback(IResponseResolver<T> resolver, ConsistencyLevel consistencyLevel,
String table)
+    {
+        super(resolver, consistencyLevel, table);
+        localResponses = new AtomicInteger(blockfor);
+    }
+    
+    @Override
+    public void response(Message message)
+    {
+        resolver.preprocess(message);
+
+        int n;
+        n = localdc.equals(snitch.getDatacenter(message.getFrom())) 
+                ? localResponses.decrementAndGet()
+                : localResponses.get();
+
+        if (n == 0 && resolver.isDataPresent())
+        {
+            condition.signal();
+        }
+    }
+    
+    @Override
+    public int determineBlockFor(ConsistencyLevel consistency_level, String table)
+	{
+        NetworkTopologyStrategy stategy = (NetworkTopologyStrategy) Table.open(table).getReplicationStrategy();
+		return (stategy.getReplicationFactor(localdc) / 2) + 1;
+	}
+
+    @Override
+    public void assureSufficientLiveNodes(Collection<InetAddress> endpoints) throws
UnavailableException
+    {
+        int localEndpoints = 0;
+        for (InetAddress endpoint : endpoints)
+        {
+            if (localdc.equals(snitch.getDatacenter(endpoint)))
+                localEndpoints++;
+        }
+        
+        if(localEndpoints < blockfor)
+            throw new UnavailableException();
+    }
+}

Added: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/ReadCallback.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/ReadCallback.java?rev=1055320&view=auto
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/ReadCallback.java
(added)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/ReadCallback.java
Wed Jan  5 07:00:33 2011
@@ -0,0 +1,122 @@
+/**
+ * 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.service;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.util.Collection;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.Table;
+import org.apache.cassandra.net.IAsyncCallback;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.thrift.ConsistencyLevel;
+import org.apache.cassandra.thrift.UnavailableException;
+import org.apache.cassandra.utils.SimpleCondition;
+
+public class ReadCallback<T> implements IAsyncCallback
+{
+    protected static final Logger logger = LoggerFactory.getLogger( ReadCallback.class );
+
+    public final IResponseResolver<T> resolver;
+    protected final SimpleCondition condition = new SimpleCondition();
+    private final long startTime;
+    protected final int blockfor;
+    
+    /**
+     * Constructor when response count has to be calculated and blocked for.
+     */
+    public ReadCallback(IResponseResolver<T> resolver, ConsistencyLevel consistencyLevel,
String table)
+    {
+        this.blockfor = determineBlockFor(consistencyLevel, table);
+        this.resolver = resolver;
+        this.startTime = System.currentTimeMillis();
+
+        logger.debug("ReadCallback blocking for {} responses", blockfor);
+    }
+    
+    public T get() throws TimeoutException, DigestMismatchException, IOException
+    {
+        long timeout = DatabaseDescriptor.getRpcTimeout() - (System.currentTimeMillis() -
startTime);
+        boolean success;
+        try
+        {
+            success = condition.await(timeout, TimeUnit.MILLISECONDS);
+        }
+        catch (InterruptedException ex)
+        {
+            throw new AssertionError(ex);
+        }
+
+        if (!success)
+        {
+            StringBuilder sb = new StringBuilder("");
+            for (Message message : resolver.getMessages())
+                sb.append(message.getFrom()).append(", ");
+            throw new TimeoutException("Operation timed out - received only " + resolver.getMessageCount()
+ " responses from " + sb.toString() + " .");
+        }
+
+        return blockfor == 1 ? resolver.getData() : resolver.resolve();
+    }
+
+    public void close()
+    {
+        for (Message response : resolver.getMessages())
+        {
+            MessagingService.removeRegisteredCallback(response.getMessageId());
+        }
+    }
+    
+    public void response(Message message)
+    {
+        resolver.preprocess(message);
+        if (resolver.getMessageCount() < blockfor)
+            return;
+        if (resolver.isDataPresent())
+            condition.signal();
+    }
+    
+    public int determineBlockFor(ConsistencyLevel consistencyLevel, String table)
+    {
+        switch (consistencyLevel)
+        {
+            case ONE:
+            case ANY:
+                return 1;
+            case QUORUM:
+                return (Table.open(table).getReplicationStrategy().getReplicationFactor()
/ 2) + 1;
+            case ALL:
+                return Table.open(table).getReplicationStrategy().getReplicationFactor();
+            default:
+                throw new UnsupportedOperationException("invalid consistency level: " + consistencyLevel);
+        }
+    }
+
+    public void assureSufficientLiveNodes(Collection<InetAddress> endpoints) throws
UnavailableException
+    {
+        if (endpoints.size() < blockfor)
+            throw new UnavailableException();
+    }
+}

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=1055320&r1=1055319&r2=1055320&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
Wed Jan  5 07:00:33 2011
@@ -328,7 +328,7 @@ public class StorageProxy implements Sto
      */
     private static List<Row> fetchRows(List<ReadCommand> commands, ConsistencyLevel
consistency_level) throws IOException, UnavailableException, TimeoutException
     {
-        List<QuorumResponseHandler<Row>> quorumResponseHandlers = new ArrayList<QuorumResponseHandler<Row>>();
+        List<ReadCallback<Row>> readCallbacks = new ArrayList<ReadCallback<Row>>();
         List<List<InetAddress>> commandEndpoints = new ArrayList<List<InetAddress>>();
         List<Row> rows = new ArrayList<Row>();
         Set<ReadCommand> repairs = new HashSet<ReadCommand>();
@@ -347,7 +347,7 @@ public class StorageProxy implements Sto
 
             AbstractReplicationStrategy rs = Table.open(command.table).getReplicationStrategy();
             ReadResponseResolver resolver = new ReadResponseResolver(command.table, command.key);
-            QuorumResponseHandler<Row> handler = rs.getQuorumResponseHandler(resolver,
consistency_level);
+            ReadCallback<Row> handler = getReadCallback(resolver, command.table, consistency_level);
             handler.assureSufficientLiveNodes(endpoints);
 
             int targets;
@@ -374,7 +374,7 @@ public class StorageProxy implements Sto
                     logger.debug("reading " + (m == message ? "data" : "digest") + " for
" + command + " from " + m.getMessageId() + "@" + endpoint);
             }
             MessagingService.instance().sendRR(messages, endpoints, handler);
-            quorumResponseHandlers.add(handler);
+            readCallbacks.add(handler);
             commandEndpoints.add(endpoints);
         }
 
@@ -382,22 +382,22 @@ public class StorageProxy implements Sto
         List<RepairCallback<Row>> repairResponseHandlers = null;
         for (int i = 0; i < commands.size(); i++)
         {
-            QuorumResponseHandler<Row> quorumResponseHandler = quorumResponseHandlers.get(i);
+            ReadCallback<Row> readCallback = readCallbacks.get(i);
             Row row;
             ReadCommand command = commands.get(i);
             List<InetAddress> endpoints = commandEndpoints.get(i);
             try
             {
                 long startTime2 = System.currentTimeMillis();
-                row = quorumResponseHandler.get();
+                row = readCallback.get();
                 if (row != null)
                     rows.add(row);
 
                 if (logger.isDebugEnabled())
-                    logger.debug("quorumResponseHandler: " + (System.currentTimeMillis()
- startTime2) + " ms.");
+                    logger.debug("Read: " + (System.currentTimeMillis() - startTime2) + "
ms.");
 
                 if (repairs.contains(command))
-                    repairExecutor.schedule(new RepairRunner(quorumResponseHandler.resolver,
command, endpoints), DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
+                    repairExecutor.schedule(new RepairRunner(readCallback.resolver, command,
endpoints), DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS);
             }
             catch (DigestMismatchException ex)
             {
@@ -431,6 +431,15 @@ public class StorageProxy implements Sto
         return rows;
     }
 
+    static <T> ReadCallback<T> getReadCallback(IResponseResolver<T> resolver,
String table, ConsistencyLevel consistencyLevel)
+    {
+        if (consistencyLevel.equals(ConsistencyLevel.LOCAL_QUORUM) || consistencyLevel.equals(ConsistencyLevel.EACH_QUORUM))
+        {
+            return new DatacenterReadCallback(resolver, consistencyLevel, table);
+        }
+        return new ReadCallback(resolver, consistencyLevel, table);
+    }
+
     // TODO repair resolver shouldn't take consistencylevel (it should repair exactly as
many as it receives replies for)
     private static RepairCallback<Row> repair(ReadCommand command, List<InetAddress>
endpoints)
     throws IOException
@@ -492,7 +501,7 @@ public class StorageProxy implements Sto
                     // collect replies and resolve according to consistency level
                     RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(command.keyspace,
liveEndpoints);
                     AbstractReplicationStrategy rs = Table.open(command.keyspace).getReplicationStrategy();
-                    QuorumResponseHandler<List<Row>> handler = rs.getQuorumResponseHandler(resolver,
consistency_level);
+                    ReadCallback<List<Row>> handler = getReadCallback(resolver,
command.keyspace, consistency_level);
                     // TODO bail early if live endpoints can't satisfy requested consistency
level
                     for (InetAddress endpoint : liveEndpoints) 
                     {
@@ -741,7 +750,7 @@ public class StorageProxy implements Sto
             // collect replies and resolve according to consistency level
             RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(keyspace,
liveEndpoints);
             AbstractReplicationStrategy rs = Table.open(keyspace).getReplicationStrategy();
-            QuorumResponseHandler<List<Row>> handler = rs.getQuorumResponseHandler(resolver,
consistency_level);
+            ReadCallback<List<Row>> handler = getReadCallback(resolver, keyspace,
consistency_level);
             
             // bail early if live endpoints can't satisfy requested consistency level
             if(handler.blockfor > liveEndpoints.size())

Modified: cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java?rev=1055320&r1=1055319&r2=1055320&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java
(original)
+++ cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java
Wed Jan  5 07:00:33 2011
@@ -96,7 +96,7 @@ public class ConsistencyLevelTest extend
 
                     IWriteResponseHandler writeHandler = strategy.getWriteResponseHandler(hosts,
hintedNodes, c);
 
-                    QuorumResponseHandler<Row> readHandler = strategy.getQuorumResponseHandler(new
ReadResponseResolver(table, ByteBufferUtil.bytes("foo")), c);
+                    ReadCallback<Row> readHandler = StorageProxy.getReadCallback(new
ReadResponseResolver(table, ByteBufferUtil.bytes("foo")), table, c);
 
                     boolean isWriteUnavailable = false;
                     boolean isReadUnavailable = false;



Mime
View raw message