mirror of https://github.com/apache/cassandra
r/m unnecessary IReplicationStrategy interface. r/m LoadVerbHandler and unused endpoint-related code. rename AbstractStrategy -> AbstractReplicationStrategy and getEndpoint methods.
patch by jbellis; reviewed by Sandeep Tata for CASSANDRA-393 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@809586 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis
parent
86abfe1027
commit
eb83251bda
|
|
@ -24,7 +24,7 @@ import java.util.Set;
|
|||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.dht.IPartitioner;
|
||||
import org.apache.cassandra.dht.Token;
|
||||
import org.apache.cassandra.locator.IReplicaPlacementStrategy;
|
||||
import org.apache.cassandra.locator.AbstractReplicationStrategy;
|
||||
import org.apache.cassandra.locator.TokenMetadata;
|
||||
import org.apache.cassandra.net.EndPoint;
|
||||
import org.apache.cassandra.service.Cassandra;
|
||||
|
|
@ -47,7 +47,7 @@ public class RingCache
|
|||
|
||||
private Set<String> seeds_ = new HashSet<String>();
|
||||
final private int port_=DatabaseDescriptor.getThriftPort();
|
||||
private volatile IReplicaPlacementStrategy nodePicker_;
|
||||
private volatile AbstractReplicationStrategy nodePicker_;
|
||||
final private static IPartitioner partitioner_ = DatabaseDescriptor.getPartitioner();
|
||||
|
||||
public RingCache()
|
||||
|
|
@ -84,7 +84,7 @@ public class RingCache
|
|||
Class [] parameterTypes = new Class[] { TokenMetadata.class, IPartitioner.class, int.class, int.class};
|
||||
try
|
||||
{
|
||||
nodePicker_ = (IReplicaPlacementStrategy) cls.getConstructor(parameterTypes).newInstance(tokenMetadata, partitioner_, DatabaseDescriptor.getReplicationFactor(), port_);
|
||||
nodePicker_ = (AbstractReplicationStrategy) cls.getConstructor(parameterTypes).newInstance(tokenMetadata, partitioner_, DatabaseDescriptor.getReplicationFactor(), port_);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
|
|
@ -102,6 +102,6 @@ public class RingCache
|
|||
|
||||
public EndPoint[] getEndPoint(String key)
|
||||
{
|
||||
return nodePicker_.getStorageEndPoints(partitioner_.getToken(key));
|
||||
return nodePicker_.getReadStorageEndPoints(partitioner_.getToken(key));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,67 +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.db;
|
||||
|
||||
import org.apache.cassandra.net.EndPoint;
|
||||
import org.apache.cassandra.net.IVerbHandler;
|
||||
import org.apache.cassandra.net.Message;
|
||||
import org.apache.cassandra.net.MessagingService;
|
||||
import org.apache.cassandra.service.StorageService;
|
||||
import org.apache.cassandra.utils.LogUtil;
|
||||
import org.apache.cassandra.io.DataInputBuffer;
|
||||
import org.apache.log4j.Logger;
|
||||
|
||||
public class LoadVerbHandler implements IVerbHandler
|
||||
{
|
||||
private static Logger logger_ = Logger.getLogger(LoadVerbHandler.class);
|
||||
|
||||
public void doVerb(Message message)
|
||||
{
|
||||
try
|
||||
{
|
||||
byte[] body = message.getMessageBody();
|
||||
DataInputBuffer buffer = new DataInputBuffer();
|
||||
buffer.reset(body, body.length);
|
||||
RowMutationMessage rmMsg = RowMutationMessage.serializer().deserialize(buffer);
|
||||
|
||||
EndPoint[] endpoints = StorageService.instance().getNStorageEndPoint(rmMsg.getRowMutation().key());
|
||||
|
||||
Message messageInternal = new Message(StorageService.getLocalStorageEndPoint(),
|
||||
StorageService.mutationStage_,
|
||||
StorageService.mutationVerbHandler_,
|
||||
body
|
||||
);
|
||||
|
||||
StringBuilder sb = new StringBuilder();
|
||||
for(EndPoint endPoint : endpoints)
|
||||
{
|
||||
sb.append(endPoint);
|
||||
MessagingService.getMessagingInstance().sendOneWay(messageInternal, endPoint);
|
||||
}
|
||||
if (logger_.isDebugEnabled())
|
||||
logger_.debug("Sent data to " + sb.toString());
|
||||
}
|
||||
catch ( Exception e )
|
||||
{
|
||||
if (logger_.isDebugEnabled())
|
||||
logger_.debug(LogUtil.throwableToString(e));
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -29,8 +29,7 @@ import org.apache.cassandra.net.IVerbHandler;
|
|||
import org.apache.cassandra.net.Message;
|
||||
import org.apache.cassandra.net.MessagingService;
|
||||
import org.apache.cassandra.service.StorageService;
|
||||
import org.apache.cassandra.service.UnavailableException;
|
||||
import org.apache.cassandra.utils.LogUtil;
|
||||
|
||||
import org.apache.log4j.Logger;
|
||||
|
||||
public class ReadVerbHandler implements IVerbHandler
|
||||
|
|
@ -115,7 +114,7 @@ public class ReadVerbHandler implements IVerbHandler
|
|||
|
||||
private void doReadRepair(Row row, ReadCommand readCommand)
|
||||
{
|
||||
List<EndPoint> endpoints = StorageService.instance().getNLiveStorageEndPoint(readCommand.key);
|
||||
List<EndPoint> endpoints = StorageService.instance().getLiveReadStorageEndPoints(readCommand.key);
|
||||
/* Remove the local storage endpoint from the list. */
|
||||
endpoints.remove( StorageService.getLocalStorageEndPoint() );
|
||||
|
||||
|
|
|
|||
|
|
@ -37,16 +37,16 @@ import org.apache.cassandra.net.EndPoint;
|
|||
* all abstraction that implement the IReplicaPlacementStrategy
|
||||
* interface.
|
||||
*/
|
||||
public abstract class AbstractStrategy implements IReplicaPlacementStrategy
|
||||
public abstract class AbstractReplicationStrategy
|
||||
{
|
||||
protected static final Logger logger_ = Logger.getLogger(AbstractStrategy.class);
|
||||
protected static final Logger logger_ = Logger.getLogger(AbstractReplicationStrategy.class);
|
||||
|
||||
protected TokenMetadata tokenMetadata_;
|
||||
protected IPartitioner partitioner_;
|
||||
protected int replicas_;
|
||||
protected int storagePort_;
|
||||
|
||||
AbstractStrategy(TokenMetadata tokenMetadata, IPartitioner partitioner, int replicas, int storagePort)
|
||||
AbstractReplicationStrategy(TokenMetadata tokenMetadata, IPartitioner partitioner, int replicas, int storagePort)
|
||||
{
|
||||
tokenMetadata_ = tokenMetadata;
|
||||
partitioner_ = partitioner;
|
||||
|
|
@ -54,6 +54,21 @@ public abstract class AbstractStrategy implements IReplicaPlacementStrategy
|
|||
storagePort_ = storagePort;
|
||||
}
|
||||
|
||||
public abstract EndPoint[] getWriteStorageEndPoints(Token token);
|
||||
public abstract EndPoint[] getReadStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap);
|
||||
public abstract EndPoint[] getReadStorageEndPoints(Token token);
|
||||
|
||||
/*
|
||||
* This method returns the hint map. The key is the endpoint
|
||||
* on which the data is being placed and the value is the
|
||||
* endpoint which is in the top N.
|
||||
* Get the map of top N to the live nodes currently.
|
||||
*/
|
||||
public Map<EndPoint, EndPoint> getHintedStorageEndPoints(Token token)
|
||||
{
|
||||
return getHintedMapForEndpoints(getWriteStorageEndPoints(token));
|
||||
}
|
||||
|
||||
/*
|
||||
* This method changes the ports of the endpoints from
|
||||
* the control port to the storage ports.
|
||||
|
|
@ -94,18 +109,6 @@ public abstract class AbstractStrategy implements IReplicaPlacementStrategy
|
|||
return endPoint;
|
||||
}
|
||||
|
||||
/*
|
||||
* This method returns the hint map. The key is the endpoint
|
||||
* on which the data is being placed and the value is the
|
||||
* endpoint which is in the top N.
|
||||
* Get the map of top N to the live nodes currently.
|
||||
*/
|
||||
public Map<EndPoint, EndPoint> getHintedStorageEndPoints(Token token)
|
||||
{
|
||||
EndPoint[] topN = getStorageEndPointsForWrite( token );
|
||||
return getHintedMapForEndpoints(topN);
|
||||
}
|
||||
|
||||
private Map<EndPoint, EndPoint> getHintedMapForEndpoints(EndPoint[] topN)
|
||||
{
|
||||
List<EndPoint> liveList = new ArrayList<EndPoint>();
|
||||
|
|
@ -135,6 +138,4 @@ public abstract class AbstractStrategy implements IReplicaPlacementStrategy
|
|||
}
|
||||
return map;
|
||||
}
|
||||
|
||||
public abstract EndPoint[] getStorageEndPoints(Token token);
|
||||
}
|
||||
|
|
@ -1,40 +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;
|
||||
|
||||
import java.util.Map;
|
||||
import java.math.BigInteger;
|
||||
|
||||
import org.apache.cassandra.dht.Token;
|
||||
import org.apache.cassandra.net.EndPoint;
|
||||
|
||||
|
||||
/*
|
||||
* This interface has two implementations. One which
|
||||
* does not respect rack or datacenter awareness and
|
||||
* the other which does.
|
||||
*/
|
||||
public interface IReplicaPlacementStrategy
|
||||
{
|
||||
public EndPoint[] getStorageEndPoints(Token token);
|
||||
public EndPoint[] getStorageEndPointsForWrite(Token token);
|
||||
public Map<String, EndPoint[]> getStorageEndPoints(String[] keys);
|
||||
public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap);
|
||||
public Map<EndPoint, EndPoint> getHintedStorageEndPoints(Token token);
|
||||
}
|
||||
|
|
@ -21,7 +21,6 @@ package org.apache.cassandra.locator;
|
|||
import java.net.UnknownHostException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
|
|
@ -38,14 +37,14 @@ import org.apache.cassandra.utils.LogUtil;
|
|||
* a node in a different rack in the same datacenter as
|
||||
* the primary.
|
||||
*/
|
||||
public class RackAwareStrategy extends AbstractStrategy
|
||||
public class RackAwareStrategy extends AbstractReplicationStrategy
|
||||
{
|
||||
public RackAwareStrategy(TokenMetadata tokenMetadata, IPartitioner partitioner, int replicas, int storagePort)
|
||||
{
|
||||
super(tokenMetadata, partitioner, replicas, storagePort);
|
||||
}
|
||||
|
||||
public EndPoint[] getStorageEndPoints(Token token)
|
||||
public EndPoint[] getReadStorageEndPoints(Token token)
|
||||
{
|
||||
int startIndex;
|
||||
List<EndPoint> list = new ArrayList<EndPoint>();
|
||||
|
|
@ -123,24 +122,12 @@ public class RackAwareStrategy extends AbstractStrategy
|
|||
return list.toArray(new EndPoint[list.size()]);
|
||||
}
|
||||
|
||||
public Map<String, EndPoint[]> getStorageEndPoints(String[] keys)
|
||||
{
|
||||
Map<String, EndPoint[]> results = new HashMap<String, EndPoint[]>();
|
||||
|
||||
for ( String key : keys )
|
||||
{
|
||||
results.put(key, getStorageEndPoints(partitioner_.getToken(key)));
|
||||
}
|
||||
|
||||
return results;
|
||||
}
|
||||
|
||||
public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap)
|
||||
public EndPoint[] getReadStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap)
|
||||
{
|
||||
throw new UnsupportedOperationException("This operation is not currently supported");
|
||||
}
|
||||
|
||||
public EndPoint[] getStorageEndPointsForWrite(Token token)
|
||||
public EndPoint[] getWriteStorageEndPoints(Token token)
|
||||
{
|
||||
throw new UnsupportedOperationException("Rack-aware bootstrapping not supported");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -20,7 +20,6 @@ package org.apache.cassandra.locator;
|
|||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
|
|
@ -34,19 +33,19 @@ import org.apache.cassandra.net.EndPoint;
|
|||
* returns the 3 nodes that lie right next to each other
|
||||
* on the ring.
|
||||
*/
|
||||
public class RackUnawareStrategy extends AbstractStrategy
|
||||
public class RackUnawareStrategy extends AbstractReplicationStrategy
|
||||
{
|
||||
public RackUnawareStrategy(TokenMetadata tokenMetadata, IPartitioner partitioner, int replicas, int storagePort)
|
||||
{
|
||||
super(tokenMetadata, partitioner, replicas, storagePort);
|
||||
}
|
||||
|
||||
public EndPoint[] getStorageEndPoints(Token token)
|
||||
public EndPoint[] getReadStorageEndPoints(Token token)
|
||||
{
|
||||
return getStorageEndPoints(token, tokenMetadata_.cloneTokenEndPointMap());
|
||||
return getReadStorageEndPoints(token, tokenMetadata_.cloneTokenEndPointMap());
|
||||
}
|
||||
|
||||
public EndPoint[] getStorageEndPointsForWrite(Token token)
|
||||
public EndPoint[] getWriteStorageEndPoints(Token token)
|
||||
{
|
||||
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
|
||||
Map<Token, EndPoint> bootstrapTokensToEndpointMap = tokenMetadata_.cloneBootstrapNodes();
|
||||
|
|
@ -64,7 +63,7 @@ public class RackUnawareStrategy extends AbstractStrategy
|
|||
return list.toArray(new EndPoint[list.size()]);
|
||||
}
|
||||
|
||||
public EndPoint[] getStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap)
|
||||
public EndPoint[] getReadStorageEndPoints(Token token, Map<Token, EndPoint> tokenToEndPointMap)
|
||||
{
|
||||
List<Token> tokenList = getStorageTokens(token, tokenToEndPointMap, null);
|
||||
List<EndPoint> list = new ArrayList<EndPoint>();
|
||||
|
|
@ -115,15 +114,4 @@ public class RackUnawareStrategy extends AbstractStrategy
|
|||
}
|
||||
return tokenList;
|
||||
}
|
||||
|
||||
public Map<String, EndPoint[]> getStorageEndPoints(String[] keys)
|
||||
{
|
||||
Map<String, EndPoint[]> results = new HashMap<String, EndPoint[]>();
|
||||
|
||||
for ( String key : keys )
|
||||
{
|
||||
results.put(key, getStorageEndPoints(partitioner_.getToken(key)));
|
||||
}
|
||||
return results;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -113,7 +113,7 @@ public class StorageProxy implements StorageProxyMBean
|
|||
try
|
||||
{
|
||||
// (This is the ZERO consistency level, so user doesn't care if we don't really have N destinations available.)
|
||||
Map<EndPoint, EndPoint> endpointMap = StorageService.instance().getNStorageEndPointMap(rm.key());
|
||||
Map<EndPoint, EndPoint> endpointMap = StorageService.instance().getHintedStorageEndpointMap(rm.key());
|
||||
Map<EndPoint, Message> messageMap = createWriteMessages(rm, endpointMap);
|
||||
for (Map.Entry<EndPoint, Message> entry : messageMap.entrySet())
|
||||
{
|
||||
|
|
@ -148,7 +148,7 @@ public class StorageProxy implements StorageProxyMBean
|
|||
}
|
||||
try
|
||||
{
|
||||
Map<EndPoint, EndPoint> endpointMap = StorageService.instance().getNStorageEndPointMap(rm.key());
|
||||
Map<EndPoint, EndPoint> endpointMap = StorageService.instance().getHintedStorageEndpointMap(rm.key());
|
||||
int blockFor = determineBlockFor(consistency_level);
|
||||
List<EndPoint> primaryNodes = getUnhintedNodes(endpointMap);
|
||||
if (primaryNodes.size() < blockFor) // guarantee blockFor = W live nodes.
|
||||
|
|
@ -294,7 +294,7 @@ public class StorageProxy implements StorageProxyMBean
|
|||
|
||||
for (ReadCommand command: commands)
|
||||
{
|
||||
EndPoint[] endpoints = StorageService.instance().getNStorageEndPoint(command.key);
|
||||
EndPoint[] endpoints = StorageService.instance().getReadStorageEndPoints(command.key);
|
||||
boolean foundLocal = Arrays.asList(endpoints).contains(StorageService.getLocalStorageEndPoint());
|
||||
//TODO: Throw InvalidRequest if we're in bootstrap mode?
|
||||
if (foundLocal && !StorageService.instance().isBootstrapMode())
|
||||
|
|
@ -358,7 +358,7 @@ public class StorageProxy implements StorageProxyMBean
|
|||
DatabaseDescriptor.getQuorum(),
|
||||
readResponseResolver);
|
||||
EndPoint dataPoint = StorageService.instance().findSuitableEndPoint(command.key);
|
||||
List<EndPoint> endpointList = new ArrayList<EndPoint>(Arrays.asList(StorageService.instance().getNStorageEndPoint(command.key)));
|
||||
List<EndPoint> endpointList = new ArrayList<EndPoint>(Arrays.asList(StorageService.instance().getReadStorageEndPoints(command.key)));
|
||||
/* Remove the local storage endpoint from the list. */
|
||||
endpointList.remove(dataPoint);
|
||||
EndPoint[] endPoints = new EndPoint[endpointList.size() + 1];
|
||||
|
|
@ -466,7 +466,7 @@ public class StorageProxy implements StorageProxyMBean
|
|||
{
|
||||
/* This is the primary */
|
||||
EndPoint dataPoint = StorageService.instance().findSuitableEndPoint(key);
|
||||
List<EndPoint> replicas = new ArrayList<EndPoint>( StorageService.instance().getNLiveStorageEndPoint(key) );
|
||||
List<EndPoint> replicas = new ArrayList<EndPoint>( StorageService.instance().getLiveReadStorageEndPoints(key) );
|
||||
replicas.remove(dataPoint);
|
||||
/* Get the messages to be sent index 0 is the data messages and index 1 is the digest message */
|
||||
Message[] message = messages.get(key);
|
||||
|
|
@ -506,7 +506,7 @@ public class StorageProxy implements StorageProxyMBean
|
|||
List<Row> rows = new ArrayList<Row>();
|
||||
for (ReadCommand command: commands)
|
||||
{
|
||||
List<EndPoint> endpoints = StorageService.instance().getNLiveStorageEndPoint(command.key);
|
||||
List<EndPoint> endpoints = StorageService.instance().getLiveReadStorageEndPoints(command.key);
|
||||
/* Remove the local storage endpoint from the list. */
|
||||
endpoints.remove(StorageService.getLocalStorageEndPoint());
|
||||
// TODO: throw a thrift exception if we do not have N nodes
|
||||
|
|
|
|||
|
|
@ -66,7 +66,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
/* All verb handler identifiers */
|
||||
public final static String mutationVerbHandler_ = "ROW-MUTATION-VERB-HANDLER";
|
||||
public final static String tokenVerbHandler_ = "TOKEN-VERB-HANDLER";
|
||||
public final static String loadVerbHandler_ = "LOAD-VERB-HANDLER";
|
||||
public final static String binaryVerbHandler_ = "BINARY-VERB-HANDLER";
|
||||
public final static String readRepairVerbHandler_ = "READ-REPAIR-VERB-HANDLER";
|
||||
public final static String readVerbHandler_ = "ROW-READ-VERB-HANDLER";
|
||||
|
|
@ -78,12 +77,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
public final static String bsMetadataVerbHandler_ = "BS-METADATA-VERB-HANDLER";
|
||||
public final static String rangeVerbHandler_ = "RANGE-VERB-HANDLER";
|
||||
|
||||
public static enum ConsistencyLevel
|
||||
{
|
||||
WEAK,
|
||||
STRONG
|
||||
}
|
||||
|
||||
private static StorageService instance_;
|
||||
/* Used to lock the factory for creation of StorageService instance */
|
||||
private static Lock createLock_ = new ReentrantLock();
|
||||
|
|
@ -104,7 +97,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
public static IPartitioner getPartitioner() {
|
||||
return partitioner_;
|
||||
}
|
||||
|
||||
|
||||
static
|
||||
{
|
||||
partitioner_ = DatabaseDescriptor.getPartitioner();
|
||||
|
|
@ -183,7 +176,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
/* This is the entity that tracks load information of all nodes in the cluster */
|
||||
private StorageLoadBalancer storageLoadBalancer_;
|
||||
/* We use this interface to determine where replicas need to be placed */
|
||||
private IReplicaPlacementStrategy nodePicker_;
|
||||
private AbstractReplicationStrategy nodePicker_;
|
||||
/* Are we starting this node in bootstrap mode? */
|
||||
private boolean isBootstrapMode;
|
||||
private Set<EndPoint> bootstrapSet;
|
||||
|
|
@ -241,7 +234,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
/* register the verb handlers */
|
||||
MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.tokenVerbHandler_, new TokenUpdateVerbHandler());
|
||||
MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.binaryVerbHandler_, new BinaryVerbHandler());
|
||||
MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.loadVerbHandler_, new LoadVerbHandler());
|
||||
MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.mutationVerbHandler_, new RowMutationVerbHandler());
|
||||
MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.readRepairVerbHandler_, new ReadRepairVerbHandler());
|
||||
MessagingService.getMessagingInstance().registerVerbHandlers(StorageService.readVerbHandler_, new ReadVerbHandler());
|
||||
|
|
@ -268,7 +260,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
Class [] parameterTypes = new Class[] { TokenMetadata.class, IPartitioner.class, int.class, int.class};
|
||||
try
|
||||
{
|
||||
nodePicker_ = (IReplicaPlacementStrategy) cls.getConstructor(parameterTypes).newInstance(tokenMetadata_, partitioner_, DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getStoragePort());
|
||||
nodePicker_ = (AbstractReplicationStrategy) cls.getConstructor(parameterTypes).newInstance(tokenMetadata_, partitioner_, DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getStoragePort());
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
|
|
@ -377,7 +369,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
Map<Range, List<EndPoint>> rangeToEndPointMap = new HashMap<Range, List<EndPoint>>();
|
||||
for ( Range range : ranges )
|
||||
{
|
||||
EndPoint[] endpoints = getNStorageEndPoint(range.right());
|
||||
EndPoint[] endpoints = nodePicker_.getReadStorageEndPoints(range.right());
|
||||
rangeToEndPointMap.put(range, new ArrayList<EndPoint>( Arrays.asList(endpoints) ) );
|
||||
}
|
||||
if (logger_.isDebugEnabled())
|
||||
|
|
@ -399,7 +391,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
Map<Range, List<EndPoint>> rangeToEndPointMap = new HashMap<Range, List<EndPoint>>();
|
||||
for ( Range range : ranges )
|
||||
{
|
||||
EndPoint[] endpoints = getNStorageEndPoint(range.right(), tokenToEndPointMap);
|
||||
EndPoint[] endpoints = nodePicker_.getReadStorageEndPoints(range.right(), tokenToEndPointMap);
|
||||
rangeToEndPointMap.put(range, new ArrayList<EndPoint>( Arrays.asList(endpoints) ) );
|
||||
}
|
||||
if (logger_.isDebugEnabled())
|
||||
|
|
@ -417,8 +409,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
{
|
||||
Map<EndPoint, List<Range>> endPointToRangesMap = new HashMap<EndPoint, List<Range>>();
|
||||
Map<Token, EndPoint> tokenToEndPointMap = tokenMetadata_.cloneTokenEndPointMap();
|
||||
Collection<EndPoint> mbrs = tokenToEndPointMap.values();
|
||||
for ( EndPoint mbr : mbrs )
|
||||
for (EndPoint mbr : tokenToEndPointMap.values())
|
||||
{
|
||||
endPointToRangesMap.put(mbr, getRangesForEndPoint(mbr));
|
||||
}
|
||||
|
|
@ -952,16 +943,10 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
* @param key - key for which we need to find the endpoint return value -
|
||||
* the endpoint responsible for this key
|
||||
*/
|
||||
public EndPoint[] getNStorageEndPoint(String key)
|
||||
public EndPoint[] getReadStorageEndPoints(String key)
|
||||
{
|
||||
return nodePicker_.getStorageEndPoints(partitioner_.getToken(key));
|
||||
}
|
||||
|
||||
private Map<String, EndPoint[]> getNStorageEndPoints(String[] keys)
|
||||
{
|
||||
return nodePicker_.getStorageEndPoints(keys);
|
||||
}
|
||||
|
||||
return nodePicker_.getReadStorageEndPoints(partitioner_.getToken(key));
|
||||
}
|
||||
|
||||
/**
|
||||
* This method attempts to return N endpoints that are responsible for storing the
|
||||
|
|
@ -970,10 +955,10 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
* @param key - key for which we need to find the endpoint return value -
|
||||
* the endpoint responsible for this key
|
||||
*/
|
||||
public List<EndPoint> getNLiveStorageEndPoint(String key)
|
||||
public List<EndPoint> getLiveReadStorageEndPoints(String key)
|
||||
{
|
||||
List<EndPoint> liveEps = new ArrayList<EndPoint>();
|
||||
EndPoint[] endpoints = getNStorageEndPoint(key);
|
||||
EndPoint[] endpoints = getReadStorageEndPoints(key);
|
||||
|
||||
for ( EndPoint endpoint : endpoints )
|
||||
{
|
||||
|
|
@ -991,42 +976,18 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
* @param key - key for which we need to find the endpoint return value -
|
||||
* the endpoint responsible for this key
|
||||
*/
|
||||
public Map<EndPoint, EndPoint> getNStorageEndPointMap(String key)
|
||||
public Map<EndPoint, EndPoint> getHintedStorageEndpointMap(String key)
|
||||
{
|
||||
return nodePicker_.getHintedStorageEndPoints(partitioner_.getToken(key));
|
||||
}
|
||||
|
||||
/**
|
||||
* This method returns the N endpoints that are responsible for storing the
|
||||
* specified token i.e for replication.
|
||||
*
|
||||
* @param token - position on the ring
|
||||
*/
|
||||
public EndPoint[] getNStorageEndPoint(Token token)
|
||||
{
|
||||
return nodePicker_.getStorageEndPoints(token);
|
||||
}
|
||||
|
||||
/**
|
||||
* This method returns the N endpoints that are responsible for storing the
|
||||
* specified token i.e for replication and are based on the token to endpoint
|
||||
* mapping that is passed in.
|
||||
*
|
||||
* @param token - position on the ring
|
||||
* @param tokens - w/o the following tokens in the token list
|
||||
*/
|
||||
protected EndPoint[] getNStorageEndPoint(Token token, Map<Token, EndPoint> tokenToEndPointMap)
|
||||
{
|
||||
return nodePicker_.getStorageEndPoints(token, tokenToEndPointMap);
|
||||
}
|
||||
|
||||
/**
|
||||
* This function finds the most suitable endpoint given a key.
|
||||
* It checks for locality and alive test.
|
||||
*/
|
||||
public EndPoint findSuitableEndPoint(String key) throws IOException
|
||||
{
|
||||
EndPoint[] endpoints = getNStorageEndPoint(key);
|
||||
EndPoint[] endpoints = getReadStorageEndPoints(key);
|
||||
for(EndPoint endPoint: endpoints)
|
||||
{
|
||||
if(endPoint.equals(StorageService.getLocalStorageEndPoint()))
|
||||
|
|
@ -1057,68 +1018,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
|
|||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/*
|
||||
* TODO:
|
||||
* This is used by the incomplete multiget implementation. Need to rewrite
|
||||
* this to use findSuitableEndPoint above instead of copy/paste
|
||||
*/
|
||||
public Map<String, EndPoint> findSuitableEndPoints(String[] keys) throws IOException
|
||||
{
|
||||
Map<String, EndPoint> suitableEndPoints = new HashMap<String, EndPoint>();
|
||||
Map<String, EndPoint[]> results = getNStorageEndPoints(keys);
|
||||
for ( String key : keys )
|
||||
{
|
||||
EndPoint[] endpoints = results.get(key);
|
||||
/* indicates if we have to move on to the next key */
|
||||
boolean moveOn = false;
|
||||
for(EndPoint endPoint: endpoints)
|
||||
{
|
||||
if(endPoint.equals(StorageService.getLocalStorageEndPoint()))
|
||||
{
|
||||
suitableEndPoints.put(key, endPoint);
|
||||
moveOn = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if ( moveOn )
|
||||
continue;
|
||||
|
||||
int j = 0;
|
||||
for ( ; j < endpoints.length; ++j )
|
||||
{
|
||||
if ( StorageService.instance().isInSameDataCenter(endpoints[j]) && FailureDetector.instance().isAlive(endpoints[j]) )
|
||||
{
|
||||
if (logger_.isDebugEnabled())
|
||||
logger_.debug("EndPoint " + endpoints[j] + " is in the same data center as local storage endpoint.");
|
||||
suitableEndPoints.put(key, endpoints[j]);
|
||||
moveOn = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if ( moveOn )
|
||||
continue;
|
||||
|
||||
// We have tried to be really nice but looks like there are no servers
|
||||
// in the local data center that are alive and can service this request so
|
||||
// just send it to the first alive guy and see if we get anything.
|
||||
j = 0;
|
||||
for ( ; j < endpoints.length; ++j )
|
||||
{
|
||||
if ( FailureDetector.instance().isAlive(endpoints[j]) )
|
||||
{
|
||||
if (logger_.isDebugEnabled())
|
||||
logger_.debug("EndPoint " + endpoints[j] + " is alive so get data from it.");
|
||||
suitableEndPoints.put(key, endpoints[j]);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
return suitableEndPoints;
|
||||
}
|
||||
|
||||
|
||||
Map<Token, EndPoint> getLiveEndPointMap()
|
||||
{
|
||||
return tokenMetadata_.cloneTokenEndPointMap();
|
||||
|
|
|
|||
|
|
@ -40,7 +40,7 @@ public class KeyChecker
|
|||
*/
|
||||
private static boolean checkIfProcessKey(String key)
|
||||
{
|
||||
EndPoint[] endPoints = StorageService.instance().getNStorageEndPoint(key);
|
||||
EndPoint[] endPoints = StorageService.instance().getReadStorageEndPoints(key);
|
||||
EndPoint localEndPoint = StorageService.getLocalStorageEndPoint();
|
||||
for(EndPoint endPoint : endPoints)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -40,7 +40,7 @@ public class RackUnawareStrategyTest
|
|||
{
|
||||
TokenMetadata tmd = new TokenMetadata();
|
||||
IPartitioner partitioner = new RandomPartitioner();
|
||||
IReplicaPlacementStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000);
|
||||
AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000);
|
||||
|
||||
List<Token> endPointTokens = new ArrayList<Token>();
|
||||
List<Token> keyTokens = new ArrayList<Token>();
|
||||
|
|
@ -56,7 +56,7 @@ public class RackUnawareStrategyTest
|
|||
{
|
||||
TokenMetadata tmd = new TokenMetadata();
|
||||
IPartitioner partitioner = new OrderPreservingPartitioner();
|
||||
IReplicaPlacementStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000);
|
||||
AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000);
|
||||
|
||||
List<Token> endPointTokens = new ArrayList<Token>();
|
||||
List<Token> keyTokens = new ArrayList<Token>();
|
||||
|
|
@ -69,7 +69,7 @@ public class RackUnawareStrategyTest
|
|||
|
||||
// given a list of endpoint tokens, and a set of key tokens falling between the endpoint tokens,
|
||||
// make sure that the Strategy picks the right endpoints for the keys.
|
||||
private void testGetStorageEndPoints(TokenMetadata tmd, IReplicaPlacementStrategy strategy, Token[] endPointTokens, Token[] keyTokens)
|
||||
private void testGetStorageEndPoints(TokenMetadata tmd, AbstractReplicationStrategy strategy, Token[] endPointTokens, Token[] keyTokens)
|
||||
{
|
||||
List<EndPoint> hosts = new ArrayList<EndPoint>();
|
||||
for (int i = 0; i < endPointTokens.length; i++)
|
||||
|
|
@ -81,7 +81,7 @@ public class RackUnawareStrategyTest
|
|||
|
||||
for (int i = 0; i < keyTokens.length; i++)
|
||||
{
|
||||
EndPoint[] endPoints = strategy.getStorageEndPoints(keyTokens[i]);
|
||||
EndPoint[] endPoints = strategy.getReadStorageEndPoints(keyTokens[i]);
|
||||
assertEquals(3, endPoints.length);
|
||||
for (int j = 0; j < endPoints.length; j++)
|
||||
{
|
||||
|
|
@ -95,7 +95,7 @@ public class RackUnawareStrategyTest
|
|||
{
|
||||
TokenMetadata tmd = new TokenMetadata();
|
||||
IPartitioner partitioner = new RandomPartitioner();
|
||||
IReplicaPlacementStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000);
|
||||
AbstractReplicationStrategy strategy = new RackUnawareStrategy(tmd, partitioner, 3, 7000);
|
||||
|
||||
Token[] endPointTokens = new Token[5];
|
||||
Token[] keyTokens = new Token[5];
|
||||
|
|
@ -121,7 +121,7 @@ public class RackUnawareStrategyTest
|
|||
|
||||
for (int i = 0; i < keyTokens.length; i++)
|
||||
{
|
||||
EndPoint[] endPoints = strategy.getStorageEndPointsForWrite(keyTokens[i]);
|
||||
EndPoint[] endPoints = strategy.getWriteStorageEndPoints(keyTokens[i]);
|
||||
assertTrue(endPoints.length >=3);
|
||||
List<EndPoint> endPointsList = Arrays.asList(endPoints);
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue