Use the keyspace aware sep queue for everything, call it parameterized (

#10)

Range scans now get parameterized with the table name (to avoid e.g.
files range scans ruining the much less frequent other ones); the
assumption is that scans on the same table are heterogeneous.

No change elsewhere.

src/java/org/apache/cassandra/concurrent/KeyspaceAwareSepQueue.java → src/java/org/apache/cassandra/concurrent/ParameterizedSepQueue.java

@@ -32,11 +32,13 @@
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.concurrent.AbstractLocalAwareExecutorService.FutureTask;
+import org.apache.cassandra.db.AbstractRangeCommand;
+import org.apache.cassandra.service.IReadCommand;
 
-public final class KeyspaceAwareSepQueue extends AbstractQueue<FutureTask<?>>
+public final class ParameterizedSepQueue extends AbstractQueue<FutureTask<?>>
 {
-    private static final Logger log = LoggerFactory.getLogger(KeyspaceAwareSepQueue.class);
-    private static final ThreadLocal<String> currentKeyspace = new ThreadLocal<>();
+    private static final Logger log = LoggerFactory.getLogger(ParameterizedSepQueue.class);
+    private static final ThreadLocal<String> queueingParameter = new ThreadLocal<>();
     @GuardedBy("this")
     private final Map<String, Queue<FutureTask<?>>> queue = new LinkedHashMap<>();
 
@@ -49,8 +51,8 @@ private Queue<FutureTask<?>> queue(String keyspace) {
 
     // If someone didn't set the current keyspace, gracefully fall back to a backup queue.
     private static String currentKeyspace() {
-        String current = currentKeyspace.get();
-        currentKeyspace.remove();
+        String current = queueingParameter.get();
+        queueingParameter.remove();
         if (current == null) {
             log.info("Saw a command where the current keyspace was not set, which indicates a bug");
             return "";
@@ -95,7 +97,16 @@ public FutureTask<?> peek()
         throw new UnsupportedOperationException();
     }
 
-    public static void setCurrentKeyspace(String keyspace) {
-        currentKeyspace.set(checkNotNull(keyspace));
+    /**
+     * Hypothesis: Gets happen on all tables, so you want to prioritize per keyspace. Scans happen on individual
+     * tables, and we'd rather indicate that gets to a specific table are expensive rather than all tables; better
+     * to take out reads to a single table than all.
+     */
+    public static void setQueueingParameter(IReadCommand read) {
+        if (read instanceof AbstractRangeCommand) {
+            queueingParameter.set(((AbstractRangeCommand) read).columnFamily);
+        } else {
+            queueingParameter.set(read.getKeyspace());
+        }
     }
 }

src/java/org/apache/cassandra/concurrent/SharedExecutorPool.java

@@ -114,7 +114,7 @@ public LocalAwareExecutorService newKeyspaceAwareExecutor(
     int maxConcurrency, int maxQueuedTasks, String jmxPath, String name)
     {
         SEPExecutor executor = new SEPExecutor(
-            this, new KeyspaceAwareSepQueue(), maxConcurrency, maxQueuedTasks, jmxPath, name);
+        this, new ParameterizedSepQueue(), maxConcurrency, maxQueuedTasks, jmxPath, name);
         executors.add(executor);
         return executor;
     }

src/java/org/apache/cassandra/concurrent/StageManager.java

@@ -46,7 +46,7 @@ public class StageManager
     {
         stages.put(Stage.MUTATION, multiThreadedLowSignalStage(Stage.MUTATION, getConcurrentWriters()));
         stages.put(Stage.COUNTER_MUTATION, multiThreadedLowSignalStage(Stage.COUNTER_MUTATION, getConcurrentCounterWriters()));
-        stages.put(Stage.READ, multiThreadedLowSignalStage(Stage.READ, getConcurrentReaders()));
+        stages.put(Stage.READ, readStage(Stage.READ, getConcurrentReaders()));
         stages.put(Stage.READ_CHEAP, readStage(Stage.READ_CHEAP, getConcurrentReaders()));
         stages.put(Stage.REQUEST_RESPONSE, multiThreadedLowSignalStage(Stage.REQUEST_RESPONSE, FBUtilities.getAvailableProcessors()));
         stages.put(Stage.INTERNAL_RESPONSE, multiThreadedStage(Stage.INTERNAL_RESPONSE, FBUtilities.getAvailableProcessors()));

src/java/org/apache/cassandra/net/MessageIn.java

@@ -28,7 +28,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.cassandra.concurrent.KeyspaceAwareSepQueue;
+import org.apache.cassandra.concurrent.ParameterizedSepQueue;
 import org.apache.cassandra.concurrent.Stage;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.ReadCommand;
@@ -106,7 +106,7 @@ public Stage getMessageType()
     {
         if (payload instanceof ReadCommand) {
             ReadCommand command = (ReadCommand) payload;
-            KeyspaceAwareSepQueue.setCurrentKeyspace(command.ksName);
+            ParameterizedSepQueue.setQueueingParameter(command);
             return command.isCheap() ? Stage.READ_CHEAP : Stage.READ;
         }
         return MessagingService.verbStages.get(verb);

src/java/org/apache/cassandra/service/AbstractReadExecutor.java

@@ -26,7 +26,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import org.apache.cassandra.concurrent.KeyspaceAwareSepQueue;
+import org.apache.cassandra.concurrent.ParameterizedSepQueue;
 import org.apache.cassandra.concurrent.Stage;
 import org.apache.cassandra.concurrent.StageManager;
 import org.apache.cassandra.config.CFMetaData.SpeculativeRetry.RetryType;
@@ -117,7 +117,7 @@ private void makeRequests(ReadCommand readCommand, Iterable<InetAddress> endpoin
         if (hasLocalEndpoint)
         {
             logger.trace("reading {} locally", readCommand.isDigestQuery() ? "digest" : "data");
-            KeyspaceAwareSepQueue.setCurrentKeyspace(command.ksName);
+            ParameterizedSepQueue.setQueueingParameter(command);
             StageManager.getStage(stage(command)).maybeExecuteImmediately(new LocalReadRunnable(command, handler));
         }
     }

src/java/org/apache/cassandra/service/StorageProxy.java

@@ -37,6 +37,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.concurrent.ParameterizedSepQueue;
 import org.apache.cassandra.concurrent.Stage;
 import org.apache.cassandra.concurrent.StageManager;
 import org.apache.cassandra.config.CFMetaData;
@@ -1849,6 +1850,7 @@ public static List<Row> getRangeSlice(AbstractRangeCommand command, ConsistencyL
                     if (filteredEndpoints.size() == 1
                         && filteredEndpoints.get(0).equals(FBUtilities.getBroadcastAddress()))
                     {
+                        ParameterizedSepQueue.setQueueingParameter(command);
                         StageManager.getStage(Stage.READ).execute(new LocalRangeSliceRunnable(nodeCmd, handler));
                     }
                     else