natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

rename [Datacenter]QuorumResponseHandler -> [Datacenter]ReadCallback 

patch by jbellis

git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1055320 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-01-05 07:00:33 +00:00
parent e547733845
commit dcb61125e6
5 changed files with 26 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
View File

@ -223,15 +223,6 @@ public abstract class AbstractReplicationStrategy
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();

6
src/java/org/apache/cassandra/service/DatacenterQuorumResponseHandler.java → src/java/org/apache/cassandra/service/DatacenterReadCallback.java
View File

@ -37,15 +37,15 @@ import org.apache.cassandra.utils.FBUtilities;
/**
* Datacenter Quorum response handler blocks for a quorum of responses from the local DC
*/
public class DatacenterQuorumResponseHandler<T> extends QuorumResponseHandler<T>
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 DatacenterQuorumResponseHandler(IResponseResolver<T> responseResolver, ConsistencyLevel consistencyLevel, String table)
public DatacenterReadCallback(IResponseResolver<T> resolver, ConsistencyLevel consistencyLevel, String table)
{
super(responseResolver, consistencyLevel, table);
super(resolver, consistencyLevel, table);
localResponses = new AtomicInteger(blockfor);
}

8
src/java/org/apache/cassandra/service/QuorumResponseHandler.java → src/java/org/apache/cassandra/service/ReadCallback.java
View File

@ -36,9 +36,9 @@ import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.SimpleCondition;
public class QuorumResponseHandler<T> implements IAsyncCallback
public class ReadCallback<T> implements IAsyncCallback
{
protected static final Logger logger = LoggerFactory.getLogger( QuorumResponseHandler.class );
protected static final Logger logger = LoggerFactory.getLogger( ReadCallback.class );
public final IResponseResolver<T> resolver;
protected final SimpleCondition condition = new SimpleCondition();
@ -48,13 +48,13 @@ public class QuorumResponseHandler<T> implements IAsyncCallback
/**
* Constructor when response count has to be calculated and blocked for.
*/
public QuorumResponseHandler(IResponseResolver<T> resolver, ConsistencyLevel consistencyLevel, String table)
public ReadCallback(IResponseResolver<T> resolver, ConsistencyLevel consistencyLevel, String table)
{
this.blockfor = determineBlockFor(consistencyLevel, table);
this.resolver = resolver;
this.startTime = System.currentTimeMillis();
logger.debug("QuorumResponseHandler blocking for {} responses", blockfor);
logger.debug("ReadCallback blocking for {} responses", blockfor);
}
public T get() throws TimeoutException, DigestMismatchException, IOException

27
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -328,7 +328,7 @@ public class StorageProxy implements StorageProxyMBean
*/
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 StorageProxyMBean
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 StorageProxyMBean
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 StorageProxyMBean
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 StorageProxyMBean
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 StorageProxyMBean
// 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 StorageProxyMBean
// 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())

2
test/unit/org/apache/cassandra/service/ConsistencyLevelTest.java
View File

@ -96,7 +96,7 @@ public class ConsistencyLevelTest extends CleanupHelper
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;