Docs, drop skip method argument, test

Author: andreidan

server/src/main/java/org/elasticsearch/action/search/OnlinePrewarmingService.java

@@ -17,16 +17,12 @@
  * that are more speculative.
  */
 public interface OnlinePrewarmingService {
-    OnlinePrewarmingService NOOP = (indexShard, skipPrewarmingCondition) -> {};
+    OnlinePrewarmingService NOOP = indexShard -> {};
 
     /**
      * Prewarms resources (typically segments) for the given index shard.
      *
      * @param indexShard the index shard for which resources should be prewarmed
-     * @param skipPrewarming a flag indicating whether prewarming should be skipped.
-     *                       Callers should  decide if certain prewarming calls
-     *                       should be skipped and indicate this decision via this
-     *                       flag.
      */
-    void prewarm(IndexShard indexShard, boolean skipPrewarming);
+    void prewarm(IndexShard indexShard);
 }

server/src/main/java/org/elasticsearch/search/SearchService.java

@@ -290,7 +290,7 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
     // the system has a chance to catch up and prewarming doesn't take over the network bandwidth
     public static final Setting<Integer> PREWARMING_THRESHOLD_THREADPOOL_SIZE_FACTOR_POOL_SIZE = Setting.intSetting(
         "search.online_prewarming_threshold_poolsize_factor",
-        10,
+        10, // we will only execute online prewarming if there are less than 10 queued up items/ search thread
         0, // 0 would mean we only execute online prewarming if there's no queuing in the search tp
         Setting.Property.NodeScope
     );
@@ -719,11 +719,9 @@ private <T extends RefCounted> void ensureAfterSeqNoRefreshed(
             if (waitForCheckpoint <= UNASSIGNED_SEQ_NO) {
                 runAsync(executor, executable, listener);
                 // we successfully submitted the async task to the search pool so let's prewarm the shard
-                onlinePrewarmingService.prewarm(
-                    shard,
-                    executor instanceof ThreadPoolExecutor tpe
-                        && ((tpe.getMaximumPoolSize() * prewarmingMaxPoolFactorThreshold) < tpe.getQueue().size())
-                );
+                if (isExecutorQueuedBeyondPrewarmingFactor(executor, prewarmingMaxPoolFactorThreshold) == false) {
+                    onlinePrewarmingService.prewarm(shard);
+                }
                 return;
             }
             if (shard.indexSettings().getRefreshInterval().getMillis() <= 0) {
@@ -801,11 +799,9 @@ private void searchReady() {
                         }
                         runAsync(executor, executable, listener);
                         // we successfully submitted the async task to the search pool so let's prewarm the shard
-                        onlinePrewarmingService.prewarm(
-                            shard,
-                            executor instanceof ThreadPoolExecutor tpe
-                                && ((tpe.getMaximumPoolSize() * prewarmingMaxPoolFactorThreshold) < tpe.getQueue().size())
-                        );
+                        if (isExecutorQueuedBeyondPrewarmingFactor(executor, prewarmingMaxPoolFactorThreshold) == false) {
+                            onlinePrewarmingService.prewarm(shard);
+                        }
                     }
                 }
             });
@@ -814,6 +810,28 @@ private void searchReady() {
         }
     }
 
+    /**
+     * Checks if the executor is queued beyond the prewarming factor threshold, relative to the
+     * number of threads in the pool.
+     * This is used to determine if we should prewarm the shard - i.e. if the executor doesn't
+     * contain queued tasks beyond the prewarming factor threshold X max pool size.
+     *
+     * @param searchOperationsExecutor the executor that executes the search operations
+     * @param prewarmingMaxPoolFactorThreshold maximum number of queued up items / thread in the search pool
+     */
+    // visible for testing
+    static boolean isExecutorQueuedBeyondPrewarmingFactor(Executor searchOperationsExecutor, int prewarmingMaxPoolFactorThreshold) {
+        if (searchOperationsExecutor instanceof ThreadPoolExecutor tpe) {
+            return (tpe.getMaximumPoolSize() * prewarmingMaxPoolFactorThreshold) < tpe.getQueue().size();
+        } else {
+            logger.trace(
+                "received executor [{}] that we can't inspect for queueing. allowing online prewarming for all searches",
+                searchOperationsExecutor
+            );
+            return false;
+        }
+    }
+
     private IndexShard getShard(ShardSearchRequest request) {
         final ShardSearchContextId contextId = request.readerId();
         if (contextId != null && sessionId.equals(contextId.getSessionId())) {
@@ -995,11 +1013,9 @@ public void executeQueryPhase(
             }
         }, wrapFailureListener(listener, readerContext, markAsUsed));
         // we successfully submitted the async task to the search pool so let's prewarm the shard
-        onlinePrewarmingService.prewarm(
-            readerContext.indexShard(),
-            executor instanceof ThreadPoolExecutor tpe
-                && ((tpe.getMaximumPoolSize() * prewarmingMaxPoolFactorThreshold) < tpe.getQueue().size())
-        );
+        if (isExecutorQueuedBeyondPrewarmingFactor(executor, prewarmingMaxPoolFactorThreshold) == false) {
+            onlinePrewarmingService.prewarm(readerContext.indexShard());
+        }
     }
 
     /**
@@ -1066,11 +1082,9 @@ public void executeQueryPhase(
                 }
             }, wrapFailureListener(l, readerContext, markAsUsed));
             // we successfully submitted the async task to the search pool so let's prewarm the shard
-            onlinePrewarmingService.prewarm(
-                readerContext.indexShard(),
-                executor instanceof ThreadPoolExecutor tpe
-                    && ((tpe.getMaximumPoolSize() * prewarmingMaxPoolFactorThreshold) < tpe.getQueue().size())
-            );
+            if (isExecutorQueuedBeyondPrewarmingFactor(executor, prewarmingMaxPoolFactorThreshold) == false) {
+                onlinePrewarmingService.prewarm(readerContext.indexShard());
+            }
         }));
     }
 

server/src/test/java/org/elasticsearch/search/SearchServiceTests.java

@@ -53,16 +53,25 @@
 import org.elasticsearch.search.sort.BucketedSort;
 import org.elasticsearch.search.sort.MinAndMax;
 import org.elasticsearch.search.sort.SortOrder;
+import org.elasticsearch.telemetry.metric.MeterRegistry;
 import org.elasticsearch.test.MockLog;
+import org.elasticsearch.threadpool.DefaultBuiltInExecutorBuilders;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
 
 import java.io.IOException;
 import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiFunction;
 import java.util.function.Predicate;
 
 import static org.elasticsearch.common.Strings.format;
+import static org.elasticsearch.common.util.concurrent.EsExecutors.DIRECT_EXECUTOR_SERVICE;
+import static org.elasticsearch.search.SearchService.isExecutorQueuedBeyondPrewarmingFactor;
 import static org.elasticsearch.search.SearchService.maybeWrapListenerForStackTrace;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.Matchers.not;
@@ -231,6 +240,83 @@ public void onFailure(Exception e) {
         }
     }
 
+    public void testIsExecutorQueuedBeyondPrewarmingFactor() throws InterruptedException {
+        {
+            final String threadPoolName = randomFrom(
+                ThreadPool.THREAD_POOL_TYPES.entrySet()
+                    .stream()
+                    .filter(t -> t.getValue().equals(ThreadPool.ThreadPoolType.FIXED))
+                    .map(Map.Entry::getKey)
+                    .sorted()
+                    .toList()
+            );
+            final int size = 6;
+            final int queueSize = size * 100;
+
+            ThreadPool threadPool = null;
+            final Settings nodeSettings = Settings.builder()
+                .put("node.name", "testPrewarmingBasedOnQueuedItems")
+                .put("thread_pool." + threadPoolName + ".size", size)
+                .put("thread_pool." + threadPoolName + ".queue_size", queueSize)
+                .build();
+            final CountDownLatch blockThreadPoolToQueueItems = new CountDownLatch(1);
+
+            try {
+                threadPool = new ThreadPool(nodeSettings, MeterRegistry.NOOP, new DefaultBuiltInExecutorBuilders());
+                ExecutorService executor = threadPool.executor(threadPoolName);
+
+                // these tasks will consume the thread pool causing further
+                // submissions to queue
+                final CountDownLatch occupyAllThreads = new CountDownLatch(size);
+                for (int i = 0; i < size; i++) {
+                    executor.execute(() -> {
+                        try {
+                            occupyAllThreads.countDown();
+                            blockThreadPoolToQueueItems.await();
+                        } catch (InterruptedException e) {
+                            fail(e.toString());
+                        }
+                    });
+                }
+
+                // wait for all threads to have an active task in their hands
+                occupyAllThreads.await();
+
+                // now on to the fun stuff, let's queue up items - 2 queued items
+                // for every thread in the pool (plus one more for one thread)
+                for (int i = 0; i < 13; i++) {
+                    executor.execute(() -> {});
+                }
+
+                // 13 queued up items
+                assertThat(isExecutorQueuedBeyondPrewarmingFactor(executor, 10), is(false));
+                assertThat(isExecutorQueuedBeyondPrewarmingFactor(executor, 2), is(true));
+
+                // let's get us at the 10 factor (6 threads * 10 + 1= 61 queued up items - at which point we should indicate
+                // prewarming should not happen)
+                for (int i = 0; i < 48; i++) {
+                    executor.execute(() -> {});
+                }
+
+                // 61 queued up items
+                assertThat(isExecutorQueuedBeyondPrewarmingFactor(executor, 10), is(true));
+                blockThreadPoolToQueueItems.countDown();
+            } catch (AssertionError e) {
+                // terminate more gracefully if there's an assertion error above
+                blockThreadPoolToQueueItems.countDown();
+                throw e;
+            } finally {
+                ThreadPool.terminate(threadPool, 10, TimeUnit.SECONDS);
+            }
+        }
+
+        {
+            // executors that are not ThreadPoolExecutor (i.e. no stats available) are always
+            // allowing prewarming
+            assertThat(isExecutorQueuedBeyondPrewarmingFactor(DIRECT_EXECUTOR_SERVICE, 2), is(false));
+        }
+    }
+
     private void doTestCanMatch(
         SearchRequest searchRequest,
         SortField sortField,