You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2012/08/16 22:43:12 UTC
[2/2] git commit: revert CASSANDRA-1337 comprising commits ef23335,
f17fbac, 9cf915f.
revert CASSANDRA-1337
comprising commits ef23335, f17fbac, 9cf915f.
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/29fed1f1
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/29fed1f1
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/29fed1f1
Branch: refs/heads/trunk
Commit: 29fed1f18188cfcd71c817db394c1087e0698dbd
Parents: fe784f5
Author: Jonathan Ellis <jb...@apache.org>
Authored: Thu Aug 16 15:42:30 2012 -0500
Committer: Jonathan Ellis <jb...@apache.org>
Committed: Thu Aug 16 15:42:30 2012 -0500
----------------------------------------------------------------------
.../org/apache/cassandra/service/StorageProxy.java | 60 ++++----------
1 files changed, 17 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/29fed1f1/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 8d0e0b3..9d55739 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -853,23 +853,6 @@ public class StorageProxy implements StorageProxyMBean
int columnsCount = 0;
rows = new ArrayList<Row>();
List<AbstractBounds<RowPosition>> ranges = getRestrictedRanges(command.range);
-
- // get the cardinality of this index based on row count
- // use this info to decide how many scans to do in parallel
- Table table = Table.open(command.keyspace);
- long estimatedKeysPerRange = table.getColumnFamilyStore(command.column_family)
- .estimateKeys() / table.getReplicationStrategy().getReplicationFactor();
-
- int concurrencyFactor = (int) (command.maxResults / (estimatedKeysPerRange + 1));
- if (concurrencyFactor <= 0 || command.maxIsColumns)
- concurrencyFactor = 1;
- else if (concurrencyFactor > ranges.size())
- concurrencyFactor = ranges.size();
-
- // parallel scan handlers
- List<ReadCallback<RangeSliceReply, Iterable<Row>>> scanHandlers = new ArrayList<ReadCallback<RangeSliceReply, Iterable<Row>>>(concurrencyFactor);
-
- int parallelHandlers = concurrencyFactor;
for (AbstractBounds<RowPosition> range : ranges)
{
RangeSliceCommand nodeCmd = new RangeSliceCommand(command.keyspace,
@@ -904,7 +887,6 @@ public class StorageProxy implements StorageProxyMBean
{
throw new AssertionError(e);
}
- parallelHandlers--;
}
else
{
@@ -921,36 +903,28 @@ public class StorageProxy implements StorageProxyMBean
logger.debug("reading " + nodeCmd + " from " + endpoint);
}
- scanHandlers.add(handler);
-
- if (scanHandlers.size() >= parallelHandlers)
+ try
{
- for (ReadCallback<RangeSliceReply, Iterable<Row>> scanHandler : scanHandlers)
+ for (Row row : handler.get())
{
- try
- {
- for (Row row : scanHandler.get())
- {
- rows.add(row);
- columnsCount += row.getLiveColumnCount();
- logger.debug("range slices read {}", row.key);
- }
- FBUtilities.waitOnFutures(resolver.repairResults, DatabaseDescriptor.getRangeRpcTimeout());
- }
- catch (TimeoutException ex)
- {
- if (logger.isDebugEnabled())
- logger.debug("Range slice timeout: {}", ex.toString());
- throw ex;
- }
- catch (DigestMismatchException e)
- {
- throw new AssertionError(e); // no digests in range slices yet
- }
+ rows.add(row);
+ columnsCount += row.getLiveColumnCount();
+ logger.debug("range slices read {}", row.key);
}
- scanHandlers.clear(); //go back for more
+ FBUtilities.waitOnFutures(resolver.repairResults, DatabaseDescriptor.getWriteRpcTimeout());
+ }
+ catch (TimeoutException ex)
+ {
+ if (logger.isDebugEnabled())
+ logger.debug("Range slice timeout: {}", ex.toString());
+ throw ex;
+ }
+ catch (DigestMismatchException e)
+ {
+ throw new AssertionError(e); // no digests in range slices yet
}
}
+
// if we're done, great, otherwise, move to the next range
int count = nodeCmd.maxIsColumns ? columnsCount : rows.size();
if (count >= nodeCmd.maxResults)