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

support ConsistencyLevel.ALL on read. patch by jbellis; reviewed by goffinet for CASSANDRA-584 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@888707 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-12-09 05:49:03 +00:00
parent e3ec998433
commit d267cf77bd
2 changed files with 8 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -19,6 +19,7 @@
* increase failure conviction threshold, resulting in less nodes * increase failure conviction threshold, resulting in less nodes
incorrectly (and temporarily) marked as down (CASSANDRA-610) incorrectly (and temporarily) marked as down (CASSANDRA-610)
* respect memtable thresholds during log replay (CASSANDRA-609) * respect memtable thresholds during log replay (CASSANDRA-609)
* support ConsistencyLevel.ALL on read (CASSANDRA-584)
0.5.0 beta 0.5.0 beta

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

@ -407,6 +407,7 @@ public class StorageProxy implements StorageProxyMBean
List<InetAddress[]> commandEndPoints = new ArrayList<InetAddress[]>(); List<InetAddress[]> commandEndPoints = new ArrayList<InetAddress[]>();
List<Row> rows = new ArrayList<Row>(); List<Row> rows = new ArrayList<Row>();
int responseCount = determineBlockFor(DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getReplicationFactor(), consistency_level);
int commandIndex = 0; int commandIndex = 0;
for (ReadCommand command: commands) for (ReadCommand command: commands)
@ -419,28 +420,24 @@ public class StorageProxy implements StorageProxyMBean
Message messageDigestOnly = readMessageDigestOnly.makeReadMessage(); Message messageDigestOnly = readMessageDigestOnly.makeReadMessage();
InetAddress dataPoint = StorageService.instance().findSuitableEndPoint(command.key); InetAddress dataPoint = StorageService.instance().findSuitableEndPoint(command.key);
List<InetAddress> endpointList = StorageService.instance().getNaturalEndpoints(command.key); List<InetAddress> endpointList = StorageService.instance().getLiveNaturalEndpoints(command.key);
if (endpointList.size() < responseCount)
throw new UnavailableException();
InetAddress[] endPoints = new InetAddress[endpointList.size()]; InetAddress[] endPoints = new InetAddress[endpointList.size()];
Message messages[] = new Message[endpointList.size()]; Message messages[] = new Message[endpointList.size()];
/* // data-request message is sent to dataPoint, the node that will actually get
* data-request message is sent to dataPoint, the node that will actually get // the data for us. The other replicas are only sent a digest query.
* the data for us. The other replicas are only sent a digest query.
*/
int n = 0; int n = 0;
for (InetAddress endpoint : endpointList) for (InetAddress endpoint : endpointList)
{ {
if (!FailureDetector.instance().isAlive(endpoint))
continue;
Message m = endpoint.equals(dataPoint) ? message : messageDigestOnly; Message m = endpoint.equals(dataPoint) ? message : messageDigestOnly;
endPoints[n] = endpoint; endPoints[n] = endpoint;
messages[n++] = m; messages[n++] = m;
if (logger.isDebugEnabled()) if (logger.isDebugEnabled())
logger.debug("strongread reading " + (m == message ? "data" : "digest") + " for " + command + " from " + m.getMessageId() + "@" + endpoint); logger.debug("strongread reading " + (m == message ? "data" : "digest") + " for " + command + " from " + m.getMessageId() + "@" + endpoint);
} }
if (n < DatabaseDescriptor.getQuorum()) QuorumResponseHandler<Row> quorumResponseHandler = new QuorumResponseHandler<Row>(DatabaseDescriptor.getQuorum(), new ReadResponseResolver(command.table, responseCount));
throw new UnavailableException();
QuorumResponseHandler<Row> quorumResponseHandler = new QuorumResponseHandler<Row>(DatabaseDescriptor.getQuorum(), new ReadResponseResolver(command.table, DatabaseDescriptor.getQuorum()));
MessagingService.instance().sendRR(messages, endPoints, quorumResponseHandler); MessagingService.instance().sendRR(messages, endPoints, quorumResponseHandler);
quorumResponseHandlers.add(quorumResponseHandler); quorumResponseHandlers.add(quorumResponseHandler);
commandEndPoints.add(endPoints); commandEndPoints.add(endPoints);