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

move "are we done yet" check in quorum read entirely into RRR.isDataPresent instead of partly there and partly in QRH.response 

patch by jbellis; reviewed by Stu Hood for CASSANDRA-568

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@887465 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-12-05 00:24:19 +00:00
parent 42eea2c50d
commit 6d96f53add
4 changed files with 15 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/java/org/apache/cassandra/service/ConsistencyManager.java
View File

@ -79,10 +79,9 @@ class ConsistencyManager implements Runnable
private void doReadRepair() throws IOException
{
IResponseResolver<Row> readResponseResolver = new ReadResponseResolver(table_);
/* Add the local storage endpoint to the replicas_ list */
replicas_.add(FBUtilities.getLocalAddress());
IAsyncCallback responseHandler = new DataRepairHandler(ConsistencyManager.this.replicas_.size(), readResponseResolver);
IResponseResolver<Row> readResponseResolver = new ReadResponseResolver(table_, replicas_.size());
IAsyncCallback responseHandler = new DataRepairHandler(replicas_.size(), readResponseResolver);
ReadCommand readCommand = constructReadMessage(false);
Message message = readCommand.makeReadMessage();
if (logger_.isDebugEnabled())

7
src/java/org/apache/cassandra/service/QuorumResponseHandler.java
View File

@ -36,17 +36,12 @@ public class QuorumResponseHandler<T> implements IAsyncCallback
{
protected static final Logger logger = Logger.getLogger( QuorumResponseHandler.class );
protected final SimpleCondition condition = new SimpleCondition();
private final int responseCount;
protected final List<Message> responses;
private IResponseResolver<T> responseResolver;
private final long startTime;
public QuorumResponseHandler(int responseCount, IResponseResolver<T> responseResolver)
{
assert 1 <= responseCount && responseCount <= DatabaseDescriptor.getReplicationFactor()
: "invalid response count " + responseCount;
this.responseCount = responseCount;
responses = new ArrayList<Message>(responseCount);
this.responseResolver = responseResolver;
startTime = System.currentTimeMillis();
@ -94,7 +89,7 @@ public class QuorumResponseHandler<T> implements IAsyncCallback
return;
responses.add(message);
if (responses.size() >= responseCount && responseResolver.isDataPresent(responses))
if (responseResolver.isDataPresent(responses))
{
condition.signal();
}

18
src/java/org/apache/cassandra/service/ReadResponseResolver.java
View File

@ -33,23 +33,22 @@ import java.net.InetAddress;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.utils.LogUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.log4j.Logger;
/**
* This class is used by all read functions and is called by the Quorum
* when at least a few of the servers (few is specified in Quorum)
* have sent the response . The resolve function then schedules read repair
* and resolution of read data from the various servers.
*/
public class ReadResponseResolver implements IResponseResolver<Row>
{
private static Logger logger_ = Logger.getLogger(ReadResponseResolver.class);
private final String table;
private final int responseCount;
public ReadResponseResolver(String table)
public ReadResponseResolver(String table, int responseCount)
{
assert 1 <= responseCount && responseCount <= DatabaseDescriptor.getReplicationFactor()
: "invalid response count " + responseCount;
this.responseCount = responseCount;
this.table = table;
}
@ -152,6 +151,9 @@ public class ReadResponseResolver implements IResponseResolver<Row>
public boolean isDataPresent(List<Message> responses)
{
if (responses.size() < responseCount)
return false;
boolean isDataPresent = false;
for (Message response : responses)
{

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

@ -438,7 +438,7 @@ public class StorageProxy implements StorageProxyMBean
}
if (n < DatabaseDescriptor.getQuorum())
throw new UnavailableException();
QuorumResponseHandler<Row> quorumResponseHandler = new QuorumResponseHandler<Row>(DatabaseDescriptor.getQuorum(), new ReadResponseResolver(command.table));
QuorumResponseHandler<Row> quorumResponseHandler = new QuorumResponseHandler<Row>(DatabaseDescriptor.getQuorum(), new ReadResponseResolver(command.table, DatabaseDescriptor.getQuorum()));
MessagingService.instance().sendRR(messages, endPoints, quorumResponseHandler);
quorumResponseHandlers.add(quorumResponseHandler);
commandEndPoints.add(endPoints);
@ -466,7 +466,7 @@ public class StorageProxy implements StorageProxyMBean
{
if (DatabaseDescriptor.getConsistencyCheck())
{
IResponseResolver<Row> readResponseResolverRepair = new ReadResponseResolver(command.table);
IResponseResolver<Row> readResponseResolverRepair = new ReadResponseResolver(command.table, DatabaseDescriptor.getQuorum());
QuorumResponseHandler<Row> quorumResponseHandlerRepair = new QuorumResponseHandler<Row>(
DatabaseDescriptor.getQuorum(),
readResponseResolverRepair);