Track total task execution time in TaskExecutionTimeTrackingEsThreadPoolExecutor (#96944)

Track total task execution time in EWMATrackingEsThreadPoolExecutor and
rename it.

Relates ES-6249

Author: pxsalehi

server/src/main/java/org/elasticsearch/common/util/concurrent/EsExecutors.java

@@ -122,16 +122,16 @@ public static EsThreadPoolExecutor newFixed(
         int queueCapacity,
         ThreadFactory threadFactory,
         ThreadContext contextHolder,
-        boolean trackEWMA
+        boolean trackExecutionTime
     ) {
         BlockingQueue<Runnable> queue;
         if (queueCapacity < 0) {
             queue = ConcurrentCollections.newBlockingQueue();
         } else {
             queue = new SizeBlockingQueue<>(ConcurrentCollections.<Runnable>newBlockingQueue(), queueCapacity);
         }
-        if (trackEWMA) {
-            return new EWMATrackingEsThreadPoolExecutor(
+        if (trackExecutionTime) {
+            return new TaskExecutionTimeTrackingEsThreadPoolExecutor(
                 name,
                 size,
                 size,

server/src/main/java/org/elasticsearch/common/util/concurrent/EWMATrackingEsThreadPoolExecutor.java renamed to server/src/main/java/org/elasticsearch/common/util/concurrent/TaskExecutionTimeTrackingEsThreadPoolExecutor.java

@@ -15,20 +15,22 @@
 import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
 import java.util.function.Function;
 
 /**
- * An extension to thread pool executor, which tracks the exponentially weighted moving average of the task execution time.
+ * An extension to thread pool executor, which tracks statistics for the task execution time.
  */
-public final class EWMATrackingEsThreadPoolExecutor extends EsThreadPoolExecutor {
+public final class TaskExecutionTimeTrackingEsThreadPoolExecutor extends EsThreadPoolExecutor {
 
     // This is a random starting point alpha. TODO: revisit this with actual testing and/or make it configurable
     public static double EWMA_ALPHA = 0.3;
 
     private final Function<Runnable, WrappedRunnable> runnableWrapper;
     private final ExponentiallyWeightedMovingAverage executionEWMA;
+    private final LongAdder totalExecutionTime = new LongAdder();
 
-    EWMATrackingEsThreadPoolExecutor(
+    TaskExecutionTimeTrackingEsThreadPoolExecutor(
         String name,
         int corePoolSize,
         int maximumPoolSize,
@@ -67,6 +69,13 @@ public double getTaskExecutionEWMA() {
         return executionEWMA.getAverage();
     }
 
+    /**
+     * Returns the total time (in nanoseconds) spend executing tasks in this executor.
+     */
+    public long getTotalTaskExecutionTime() {
+        return totalExecutionTime.sum();
+    }
+
     /**
      * Returns the current queue size (operations that are queued)
      */
@@ -93,12 +102,18 @@ protected void afterExecute(Runnable r, Throwable t) {
         if (taskExecutionNanos != -1) {
             // taskExecutionNanos may be -1 if the task threw an exception
             executionEWMA.addValue(taskExecutionNanos);
+            totalExecutionTime.add(taskExecutionNanos);
         }
     }
 
     @Override
     protected void appendThreadPoolExecutorDetails(StringBuilder sb) {
-        sb.append("task execution EWMA = ").append(TimeValue.timeValueNanos((long) executionEWMA.getAverage())).append(", ");
+        sb.append("task execution EWMA = ")
+            .append(TimeValue.timeValueNanos((long) executionEWMA.getAverage()))
+            .append(", ")
+            .append("total task execution time = ")
+            .append(TimeValue.timeValueNanos(getTotalTaskExecutionTime()))
+            .append(", ");
     }
 
 }

server/src/main/java/org/elasticsearch/search/query/QueryPhase.java

@@ -30,8 +30,8 @@
 import org.elasticsearch.common.lucene.MinimumScoreCollector;
 import org.elasticsearch.common.lucene.search.FilteredCollector;
 import org.elasticsearch.common.lucene.search.TopDocsAndMaxScore;
-import org.elasticsearch.common.util.concurrent.EWMATrackingEsThreadPoolExecutor;
 import org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor;
+import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingEsThreadPoolExecutor;
 import org.elasticsearch.lucene.queries.SearchAfterSortedDocQuery;
 import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.SearchContextSourcePrinter;
@@ -268,10 +268,10 @@ static void addCollectorsAndSearch(SearchContext searchContext) throws QueryPhas
                 searchWithCollectorManager(searchContext, searcher, query, collectorManager, timeoutRunnable != null);
                 queryResult.topDocs(topDocsFactory.topDocsAndMaxScore(), topDocsFactory.sortValueFormats);
                 ExecutorService executor = searchContext.indexShard().getThreadPool().executor(ThreadPool.Names.SEARCH);
-                assert executor instanceof EWMATrackingEsThreadPoolExecutor
+                assert executor instanceof TaskExecutionTimeTrackingEsThreadPoolExecutor
                     || (executor instanceof EsThreadPoolExecutor == false /* in case thread pool is mocked out in tests */)
                     : "SEARCH threadpool should have an executor that exposes EWMA metrics, but is of type " + executor.getClass();
-                if (executor instanceof EWMATrackingEsThreadPoolExecutor rExecutor) {
+                if (executor instanceof TaskExecutionTimeTrackingEsThreadPoolExecutor rExecutor) {
                     queryResult.nodeQueueSize(rExecutor.getCurrentQueueSize());
                     queryResult.serviceTimeEWMA((long) rExecutor.getTaskExecutionEWMA());
                 }

server/src/main/java/org/elasticsearch/threadpool/FixedExecutorBuilder.java

@@ -28,7 +28,7 @@ public final class FixedExecutorBuilder extends ExecutorBuilder<FixedExecutorBui
 
     private final Setting<Integer> sizeSetting;
     private final Setting<Integer> queueSizeSetting;
-    private final boolean trackEWMA;
+    private final boolean trackExecutionTime;
 
     /**
      * Construct a fixed executor builder; the settings will have the key prefix "thread_pool." followed by the executor name.
@@ -37,10 +37,16 @@ public final class FixedExecutorBuilder extends ExecutorBuilder<FixedExecutorBui
      * @param name      the name of the executor
      * @param size      the fixed number of threads
      * @param queueSize the size of the backing queue, -1 for unbounded
-     * @param trackEWMA whether to track the exponentially weighted moving average of the task execution time
+     * @param trackExecutionTime whether to track statics about task execution time
      */
-    FixedExecutorBuilder(final Settings settings, final String name, final int size, final int queueSize, final boolean trackEWMA) {
-        this(settings, name, size, queueSize, "thread_pool." + name, trackEWMA);
+    FixedExecutorBuilder(
+        final Settings settings,
+        final String name,
+        final int size,
+        final int queueSize,
+        final boolean trackExecutionTime
+    ) {
+        this(settings, name, size, queueSize, "thread_pool." + name, trackExecutionTime);
     }
 
     /**
@@ -51,15 +57,15 @@ public final class FixedExecutorBuilder extends ExecutorBuilder<FixedExecutorBui
      * @param size      the fixed number of threads
      * @param queueSize the size of the backing queue, -1 for unbounded
      * @param prefix    the prefix for the settings keys
-     * @param trackEWMA whether to track the exponentially weighted moving average of the task execution time
+     * @param trackExecutionTime whether to track statics about task execution time
      */
     public FixedExecutorBuilder(
         final Settings settings,
         final String name,
         final int size,
         final int queueSize,
         final String prefix,
-        final boolean trackEWMA
+        final boolean trackExecutionTime
     ) {
         super(name);
         final String sizeKey = settingsKey(prefix, "size");
@@ -71,7 +77,7 @@ public FixedExecutorBuilder(
         );
         final String queueSizeKey = settingsKey(prefix, "queue_size");
         this.queueSizeSetting = Setting.intSetting(queueSizeKey, queueSize, Setting.Property.NodeScope);
-        this.trackEWMA = trackEWMA;
+        this.trackExecutionTime = trackExecutionTime;
     }
 
     @Override
@@ -98,7 +104,7 @@ ThreadPool.ExecutorHolder build(final FixedExecutorSettings settings, final Thre
             queueSize,
             threadFactory,
             threadContext,
-            trackEWMA
+            trackExecutionTime
         );
         final ThreadPool.Info info = new ThreadPool.Info(
             name(),

server/src/test/java/org/elasticsearch/common/util/concurrent/EsExecutorsTests.java

@@ -308,7 +308,7 @@ public String toString() {
                 assertThat(
                     message,
                     either(containsString("on EsThreadPoolExecutor[name = " + getName())).or(
-                        containsString("on EWMATrackingEsThreadPoolExecutor[name = " + getName())
+                        containsString("on TaskExecutionTimeTrackingEsThreadPoolExecutor[name = " + getName())
                     )
                 );
                 assertThat(message, containsString("queue capacity = " + queue));
@@ -353,7 +353,7 @@ public String toString() {
             assertThat(
                 message,
                 either(containsString("on EsThreadPoolExecutor[name = " + getName())).or(
-                    containsString("on EWMATrackingEsThreadPoolExecutor[name = " + getName())
+                    containsString("on TaskExecutionTimeTrackingEsThreadPoolExecutor[name = " + getName())
                 )
             );
             assertThat(message, containsString("queue capacity = " + queue));

server/src/test/java/org/elasticsearch/common/util/concurrent/EWMATrackingEsThreadPoolExecutorTests.java renamed to server/src/test/java/org/elasticsearch/common/util/concurrent/TaskExecutionTimeTrackingEsThreadPoolExecutorTests.java

@@ -20,19 +20,19 @@
  * Tests for the automatic queue resizing of the {@code QueueResizingEsThreadPoolExecutorTests}
  * based on the time taken for each event.
  */
-public class EWMATrackingEsThreadPoolExecutorTests extends ESTestCase {
+public class TaskExecutionTimeTrackingEsThreadPoolExecutorTests extends ESTestCase {
 
     public void testExecutionEWMACalculation() throws Exception {
         ThreadContext context = new ThreadContext(Settings.EMPTY);
 
-        EWMATrackingEsThreadPoolExecutor executor = new EWMATrackingEsThreadPoolExecutor(
+        TaskExecutionTimeTrackingEsThreadPoolExecutor executor = new TaskExecutionTimeTrackingEsThreadPoolExecutor(
             "test-threadpool",
             1,
             1,
             1000,
             TimeUnit.MILLISECONDS,
             ConcurrentCollections.newBlockingQueue(),
-            fastWrapper(),
+            settableWrapper(TimeUnit.NANOSECONDS.toNanos(100)),
             EsExecutors.daemonThreadFactory("queuetest"),
             new EsAbortPolicy(),
             context
@@ -41,16 +41,33 @@ public void testExecutionEWMACalculation() throws Exception {
         logger.info("--> executor: {}", executor);
 
         assertThat((long) executor.getTaskExecutionEWMA(), equalTo(0L));
+        assertThat(executor.getTotalTaskExecutionTime(), equalTo(0L));
+        // Using the settableWrapper each task would take 100ns
         executeTask(executor, 1);
-        assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(30L)); });
+        assertBusy(() -> {
+            assertThat((long) executor.getTaskExecutionEWMA(), equalTo(30L));
+            assertThat(executor.getTotalTaskExecutionTime(), equalTo(100L));
+        });
         executeTask(executor, 1);
-        assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(51L)); });
+        assertBusy(() -> {
+            assertThat((long) executor.getTaskExecutionEWMA(), equalTo(51L));
+            assertThat(executor.getTotalTaskExecutionTime(), equalTo(200L));
+        });
         executeTask(executor, 1);
-        assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(65L)); });
+        assertBusy(() -> {
+            assertThat((long) executor.getTaskExecutionEWMA(), equalTo(65L));
+            assertThat(executor.getTotalTaskExecutionTime(), equalTo(300L));
+        });
         executeTask(executor, 1);
-        assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(75L)); });
+        assertBusy(() -> {
+            assertThat((long) executor.getTaskExecutionEWMA(), equalTo(75L));
+            assertThat(executor.getTotalTaskExecutionTime(), equalTo(400L));
+        });
         executeTask(executor, 1);
-        assertBusy(() -> { assertThat((long) executor.getTaskExecutionEWMA(), equalTo(83L)); });
+        assertBusy(() -> {
+            assertThat((long) executor.getTaskExecutionEWMA(), equalTo(83L));
+            assertThat(executor.getTotalTaskExecutionTime(), equalTo(500L));
+        });
 
         executor.shutdown();
         executor.awaitTermination(10, TimeUnit.SECONDS);
@@ -59,7 +76,7 @@ public void testExecutionEWMACalculation() throws Exception {
     /** Use a runnable wrapper that simulates a task with unknown failures. */
     public void testExceptionThrowingTask() throws Exception {
         ThreadContext context = new ThreadContext(Settings.EMPTY);
-        EWMATrackingEsThreadPoolExecutor executor = new EWMATrackingEsThreadPoolExecutor(
+        TaskExecutionTimeTrackingEsThreadPoolExecutor executor = new TaskExecutionTimeTrackingEsThreadPoolExecutor(
             "test-threadpool",
             1,
             1,
@@ -74,14 +91,23 @@ public void testExceptionThrowingTask() throws Exception {
         executor.prestartAllCoreThreads();
         logger.info("--> executor: {}", executor);
 
+        // Using the exceptionalWrapper each task's execution time is -1 to simulate unknown failures/rejections.
         assertThat((long) executor.getTaskExecutionEWMA(), equalTo(0L));
-        executeTask(executor, 1);
+        int taskCount = randomIntBetween(1, 100);
+        executeTask(executor, taskCount);
+        assertBusy(() -> assertThat(executor.getCompletedTaskCount(), equalTo((long) taskCount)));
+        assertThat((long) executor.getTaskExecutionEWMA(), equalTo(0L));
+        assertThat(executor.getTotalTaskExecutionTime(), equalTo(0L));
         executor.shutdown();
         executor.awaitTermination(10, TimeUnit.SECONDS);
     }
 
-    private Function<Runnable, WrappedRunnable> fastWrapper() {
-        return (runnable) -> new SettableTimedRunnable(TimeUnit.NANOSECONDS.toNanos(100), false);
+    /**
+     * The returned function outputs a WrappedRunnabled that simulates the case
+     * where {@link TimedRunnable#getTotalExecutionNanos()} always returns {@code timeTakenNanos}.
+     */
+    private Function<Runnable, WrappedRunnable> settableWrapper(long timeTakenNanos) {
+        return (runnable) -> new SettableTimedRunnable(timeTakenNanos, false);
     }
 
     /**
@@ -94,7 +120,7 @@ private Function<Runnable, WrappedRunnable> exceptionalWrapper() {
     }
 
     /** Execute a blank task {@code times} times for the executor */
-    private void executeTask(EWMATrackingEsThreadPoolExecutor executor, int times) {
+    private void executeTask(TaskExecutionTimeTrackingEsThreadPoolExecutor executor, int times) {
         logger.info("--> executing a task [{}] times", times);
         for (int i = 0; i < times; i++) {
             executor.execute(() -> {});

server/src/test/java/org/elasticsearch/threadpool/ThreadPoolTests.java

@@ -14,9 +14,9 @@
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
-import org.elasticsearch.common.util.concurrent.EWMATrackingEsThreadPoolExecutor;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.FutureUtils;
+import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingEsThreadPoolExecutor;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.MockLogAppender;
@@ -351,10 +351,10 @@ public void testGetMaxSnapshotCores() {
         assertThat(getMaxSnapshotThreadPoolSize(allocatedProcessors, ByteSizeValue.ofGb(4)), equalTo(10));
     }
 
-    public void testWriteThreadPoolUsesEWMATrackingEsThreadPoolExecutor() {
+    public void testWriteThreadPoolUsesTaskExecutionTimeTrackingEsThreadPoolExecutor() {
         final ThreadPool threadPool = new TestThreadPool("test", Settings.EMPTY);
         try {
-            assertThat(threadPool.executor(ThreadPool.Names.WRITE), instanceOf(EWMATrackingEsThreadPoolExecutor.class));
+            assertThat(threadPool.executor(ThreadPool.Names.WRITE), instanceOf(TaskExecutionTimeTrackingEsThreadPoolExecutor.class));
         } finally {
             assertTrue(terminate(threadPool));
         }