SOLR-13350: Multithreaded search (closes #2248)

Author: Ishan Chattopadhyaya

solr/CHANGES.txt

@@ -98,7 +98,7 @@ Other Changes
 ==================  9.7.0 ==================
 New Features
 ---------------------
-(No changes)
+* SOLR-13350: Multithreaded search execution (Ishan Chattopadhyaya, Mark Miller, Christine Poerschke, David Smiley, noble)
 
 Improvements
 ---------------------

solr/bin/solr

@@ -1441,6 +1441,9 @@ if [ $# -gt 0 ]; then
   done
 fi
 
+# Setting number of threads for search
+if ! command -v nproc &> /dev/null; then echo "Couldn't determine number of CPUs, using default number of search threads"; else cpus=`nproc`; SCRIPT_SOLR_OPTS+="-Dsolr.searchThreads=$cpus"; fi
+
 # Default placement plugin
 if [[ -n "${SOLR_PLACEMENTPLUGIN_DEFAULT:-}" ]] ; then
   SCRIPT_SOLR_OPTS+=("-Dsolr.placementplugin.default=$SOLR_PLACEMENTPLUGIN_DEFAULT")

solr/core/src/java/org/apache/solr/core/CoreContainer.java

@@ -49,6 +49,7 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
@@ -175,6 +176,10 @@ public class CoreContainer {
 
   final SolrCores solrCores;
 
+  public Executor getCollectorExecutor() {
+    return collectorExecutor;
+  }
+
   public static class CoreLoadFailure {
 
     public final CoreDescriptor cd;
@@ -278,6 +283,8 @@ public JerseyAppHandlerCache getJerseyAppHandlerCache() {
 
   public final NodeRoles nodeRoles = new NodeRoles(System.getProperty(NodeRoles.NODE_ROLES_PROP));
 
+  private final ExecutorService collectorExecutor;
+
   private final ClusterSingletons clusterSingletons =
       new ClusterSingletons(
           () ->
@@ -432,6 +439,12 @@ public CoreContainer(NodeConfig config, CoresLocator locator, boolean asyncSolrC
     this.allowPaths = allowPathBuilder.build();
 
     this.allowListUrlChecker = AllowListUrlChecker.create(config);
+
+    this.collectorExecutor =
+        ExecutorUtil.newMDCAwareCachedThreadPool(
+            cfg.getIndexSearcherExecutorThreads(), // thread count
+            cfg.getIndexSearcherExecutorThreads() * 1000, // queue size
+            new SolrNamedThreadFactory("searcherCollector"));
   }
 
   @SuppressWarnings({"unchecked"})
@@ -657,6 +670,7 @@ protected CoreContainer(Object testConstructor) {
     distributedCollectionCommandRunner = Optional.empty();
     allowPaths = null;
     allowListUrlChecker = null;
+    collectorExecutor = null;
   }
 
   public static CoreContainer createAndLoad(Path solrHome) {
@@ -1248,6 +1262,7 @@ public void shutdown() {
     }
 
     ExecutorUtil.shutdownAndAwaitTermination(coreContainerAsyncTaskExecutor);
+    ExecutorUtil.shutdownAndAwaitTermination(collectorExecutor);
     ExecutorService customThreadPool =
         ExecutorUtil.newMDCAwareCachedThreadPool(new SolrNamedThreadFactory("closeThreadPool"));
 

solr/core/src/java/org/apache/solr/core/NodeConfig.java

@@ -105,6 +105,8 @@ public class NodeConfig {
 
   private final int replayUpdatesThreads;
 
+  private final int indexSearcherExecutorThreads;
+
   @Deprecated private final int transientCacheSize;
 
   private final boolean useSchemaCache;
@@ -144,6 +146,7 @@ private NodeConfig(
       CloudConfig cloudConfig,
       Integer coreLoadThreads,
       int replayUpdatesThreads,
+      int indexSearcherExecutorThreads,
       int transientCacheSize,
       boolean useSchemaCache,
       String managementPath,
@@ -183,6 +186,7 @@ private NodeConfig(
     this.cloudConfig = cloudConfig;
     this.coreLoadThreads = coreLoadThreads;
     this.replayUpdatesThreads = replayUpdatesThreads;
+    this.indexSearcherExecutorThreads = indexSearcherExecutorThreads;
     this.transientCacheSize = transientCacheSize;
     this.useSchemaCache = useSchemaCache;
     this.managementPath = managementPath;
@@ -335,6 +339,10 @@ public int getReplayUpdatesThreads() {
     return replayUpdatesThreads;
   }
 
+  public int getIndexSearcherExecutorThreads() {
+    return indexSearcherExecutorThreads;
+  }
+
   /**
    * Returns a directory, optionally a comma separated list of directories that will be added to
    * Solr's class path for searching for classes and plugins. The path is either absolute or
@@ -597,6 +605,7 @@ public static class NodeConfigBuilder {
     private CloudConfig cloudConfig;
     private int coreLoadThreads = DEFAULT_CORE_LOAD_THREADS;
     private int replayUpdatesThreads = Runtime.getRuntime().availableProcessors();
+    private int indexSearcherExecutorThreads = DEFAULT_INDEX_SEARCHER_EXECUTOR_THREADS;
     @Deprecated private int transientCacheSize = -1;
     private boolean useSchemaCache = false;
     private String managementPath;
@@ -618,6 +627,9 @@ public static class NodeConfigBuilder {
     // No:of core load threads in cloud mode is set to a default of 8
     public static final int DEFAULT_CORE_LOAD_THREADS_IN_CLOUD = 8;
 
+    public static final int DEFAULT_INDEX_SEARCHER_EXECUTOR_THREADS =
+        4;
+
     private static final String DEFAULT_CORESLOCATORCLASS =
         "org.apache.solr.core.CorePropertiesLocator";
     private static final String DEFAULT_CORESORTERCLASS = "org.apache.solr.core.CoreSorter";
@@ -755,6 +767,11 @@ public NodeConfigBuilder setReplayUpdatesThreads(int replayUpdatesThreads) {
       return this;
     }
 
+    public NodeConfigBuilder setIndexSearcherExecutorThreads(int indexSearcherExecutorThreads) {
+      this.indexSearcherExecutorThreads = indexSearcherExecutorThreads;
+      return this;
+    }
+
     // Remove in Solr 10.0
 
     @Deprecated
@@ -904,6 +921,7 @@ public NodeConfig build() {
           cloudConfig,
           coreLoadThreads,
           replayUpdatesThreads,
+          indexSearcherExecutorThreads,
           transientCacheSize,
           useSchemaCache,
           managementPath,

solr/core/src/java/org/apache/solr/core/SolrXmlConfig.java

@@ -380,6 +380,9 @@ private static NodeConfig fillSolrSection(NodeConfig.NodeConfigBuilder builder,
               case "replayUpdatesThreads":
                 builder.setReplayUpdatesThreads(it.intVal(-1));
                 break;
+              case "indexSearcherExecutorThreads":
+                builder.setIndexSearcherExecutorThreads(it.intVal(-1));
+                break;
               case "transientCacheSize":
                 log.warn("solr.xml transientCacheSize -- transient cores is deprecated");
                 builder.setTransientCacheSize(it.intVal(-1));

solr/core/src/java/org/apache/solr/handler/component/QueryComponent.java

@@ -372,10 +372,13 @@ public void process(ResponseBuilder rb) throws IOException {
       return;
     }
 
+    final boolean multiThreaded = params.getBool("multiThreaded", true);
+
     // -1 as flag if not set.
     long timeAllowed = params.getLong(CommonParams.TIME_ALLOWED, -1L);
 
     QueryCommand cmd = rb.createQueryCommand();
+    cmd.setMultiThreaded(multiThreaded);
     cmd.setTimeAllowed(timeAllowed);
     cmd.setMinExactCount(getMinExactCount(params));
     cmd.setDistribStatsDisabled(rb.isDistribStatsDisabled());