tests

Author: mhl-b

benchmarks/src/main/java/org/elasticsearch/benchmark/common/util/concurrent/ThreadPoolUtilizationBenchmark.java

@@ -22,37 +22,44 @@
 import org.openjdk.jmh.annotations.OutputTimeUnit;
 import org.openjdk.jmh.annotations.Param;
 import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
 import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
 import org.openjdk.jmh.annotations.Warmup;
 import org.openjdk.jmh.infra.Blackhole;
 
-import java.util.Random;
+import java.time.Duration;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
-@Warmup(iterations = 0)
+@Warmup(iterations = 1, time = 1, timeUnit = TimeUnit.SECONDS)
 @Measurement(iterations = 1, time = 10, timeUnit = TimeUnit.SECONDS)
 @BenchmarkMode(Mode.AverageTime)
 @OutputTimeUnit(TimeUnit.MILLISECONDS)
 @Fork(1)
 @State(Scope.Thread)
 public class ThreadPoolUtilizationBenchmark {
-    private final Random random = new Random();
-    @Param({ "8", "16", "32", "64" })
+
+    @Param({ "false", "true" })
+    private boolean trackUtilization;
+
+    @Param({ "4", "8", "16" })
     private int poolSize;
-    @Param({ "10000" })
+
+    @Param({ "1000000" })
     private int tasksNum;
-    @Param({ "1" })
-    private int workerTimeMinMs;
-    @Param({ "5" })
-    private int workerTimeMaxMs;
-    private TaskExecutionTimeTrackingEsThreadPoolExecutor executor;
 
-    public EsThreadPoolExecutor newExecutor(boolean tracking) {
+    @Param({ "10" }) // 10ms is aggressive interval, it increases frame updates on FramedTimeTracker, normally we run at 30/60
+    // seconds
+    private int utilizationIntervalMs;
+
+    private EsThreadPoolExecutor executor;
+
+    private EsThreadPoolExecutor newExecutor(boolean tracking) {
         var conf = EsExecutors.TaskTrackingConfig.builder();
         if (tracking) {
-            conf.trackOngoingTasks().trackUtilization();
+            conf.trackExecutionTime(0.3).trackUtilization(Duration.ofMillis(utilizationIntervalMs));
         }
         return EsExecutors.newFixed(
             "bench",
@@ -64,34 +71,48 @@ public EsThreadPoolExecutor newExecutor(boolean tracking) {
         );
     }
 
-    private void runTasks(EsThreadPoolExecutor executor, Blackhole bh) throws InterruptedException {
-        try {
-            var completedTasks = new CountDownLatch(tasksNum);
-            for (var i = 0; i < tasksNum; i++) {
-                executor.execute(() -> {
-                    try {
-                        Thread.sleep(random.nextInt(workerTimeMinMs, workerTimeMaxMs));
-                    } catch (InterruptedException e) {
-                        throw new RuntimeException(e);
-                    } finally {
-                        completedTasks.countDown();
-                    }
-                });
+    @Setup
+    public void setup() {
+        if (trackUtilization) {
+            var exec = newExecutor(true);
+            if (exec instanceof TaskExecutionTimeTrackingEsThreadPoolExecutor trackingExecutor) {
+                if (trackingExecutor.trackingConfig().trackUtilization() == false) {
+                    throw new IllegalStateException("utilization tracking must be enabled");
+                } else {
+                    executor = trackingExecutor;
+                }
+            } else {
+                throw new IllegalStateException("must be tracking executor");
+            }
+        } else {
+            var exec = newExecutor(false);
+            if (exec instanceof TaskExecutionTimeTrackingEsThreadPoolExecutor) {
+                throw new IllegalStateException("must be non-tracking executor");
             }
-            completedTasks.await();
-        } finally {
-            executor.shutdown();
-            executor.awaitTermination(0, TimeUnit.MILLISECONDS);
+            executor = exec;
         }
     }
 
-    @Benchmark
-    public void trackingExecutor(Blackhole bh) throws InterruptedException {
-        runTasks(newExecutor(true), bh);
+    @TearDown
+    public void tearDown() throws InterruptedException {
+        executor.shutdown();
+        executor.awaitTermination(0, TimeUnit.MILLISECONDS);
     }
 
     @Benchmark
-    public void nonTrackingExecutor(Blackhole bh) throws InterruptedException {
-        runTasks(newExecutor(false), bh);
+    public void run(Blackhole bh) throws InterruptedException {
+        var completedTasks = new CountDownLatch(tasksNum);
+        for (var i = 0; i < tasksNum; i++) {
+            executor.execute(() -> {
+                // busy cycles for cpu
+                var r = 0;
+                for (var j = 0; j < 1000; j++) {
+                    r += j * 2;
+                }
+                bh.consume(r);
+                completedTasks.countDown();
+            });
+        }
+        completedTasks.await();
     }
 }

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

@@ -17,6 +17,7 @@
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.node.Node;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.AbstractExecutorService;
@@ -576,71 +577,45 @@ public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) {
         }
     }
 
-    public static class TaskTrackingConfig {
+    /**
+     * @param trackExecutionTime Whether to track execution stats
+     * @param trackUtilization enables thread-pool utilization metrics
+     * @param utilizationInterval when utilization is enabled, specifies interval for measurement
+     * @param trackOngoingTasks Whether to track ongoing task execution time, not just finished tasks
+     * @param trackMaxQueueLatency Whether to track max queue latency.
+     * @param executionTimeEwmaAlpha The alpha seed for execution time EWMA (ExponentiallyWeightedMovingAverage).
+     */
+    public record TaskTrackingConfig(
+        boolean trackExecutionTime,
+        boolean trackUtilization,
+        Duration utilizationInterval,
+        boolean trackOngoingTasks,
+        boolean trackMaxQueueLatency,
+        double executionTimeEwmaAlpha
+    ) {
+
         // This is a random starting point alpha.
         public static final double DEFAULT_EXECUTION_TIME_EWMA_ALPHA_FOR_TEST = 0.3;
-
-        private final boolean trackExecutionTime;
-        private final boolean trackUtilization;
-        private final boolean trackOngoingTasks;
-        private final boolean trackMaxQueueLatency;
-        private final double executionTimeEwmaAlpha;
+        public static final Duration DEFAULT_UTILIZATION_INTERVAL = Duration.ofSeconds(30);
 
         public static final TaskTrackingConfig DO_NOT_TRACK = new TaskTrackingConfig(
             false,
             false,
+            DEFAULT_UTILIZATION_INTERVAL,
             false,
             false,
             DEFAULT_EXECUTION_TIME_EWMA_ALPHA_FOR_TEST
         );
+
         public static final TaskTrackingConfig DEFAULT = new TaskTrackingConfig(
             true,
             true,
+            DEFAULT_UTILIZATION_INTERVAL,
             false,
             false,
             DEFAULT_EXECUTION_TIME_EWMA_ALPHA_FOR_TEST
         );
 
-        /**
-         * @param trackExecutionTime Whether to track execution stats
-         * @param trackOngoingTasks Whether to track ongoing task execution time, not just finished tasks
-         * @param trackMaxQueueLatency Whether to track max queue latency.
-         * @param executionTimeEWMAAlpha The alpha seed for execution time EWMA (ExponentiallyWeightedMovingAverage).
-         */
-        private TaskTrackingConfig(
-            boolean trackExecutionTime,
-            boolean trackUtilization,
-            boolean trackOngoingTasks,
-            boolean trackMaxQueueLatency,
-            double executionTimeEWMAAlpha
-        ) {
-            this.trackExecutionTime = trackExecutionTime;
-            this.trackUtilization = trackUtilization;
-            this.trackOngoingTasks = trackOngoingTasks;
-            this.trackMaxQueueLatency = trackMaxQueueLatency;
-            this.executionTimeEwmaAlpha = executionTimeEWMAAlpha;
-        }
-
-        public boolean trackExecutionTime() {
-            return trackExecutionTime;
-        }
-
-        public boolean trackOngoingTasks() {
-            return trackOngoingTasks;
-        }
-
-        public boolean trackMaxQueueLatency() {
-            return trackMaxQueueLatency;
-        }
-
-        public boolean TrackUtilization() {
-            return trackUtilization;
-        }
-
-        public double getExecutionTimeEwmaAlpha() {
-            return executionTimeEwmaAlpha;
-        }
-
         public static Builder builder() {
             return new Builder();
         }
@@ -651,6 +626,7 @@ public static class Builder {
             private boolean trackOngoingTasks = false;
             private boolean trackMaxQueueLatency = false;
             private double ewmaAlpha = DEFAULT_EXECUTION_TIME_EWMA_ALPHA_FOR_TEST;
+            private Duration utilizationInterval = DEFAULT_UTILIZATION_INTERVAL;
 
             public Builder() {}
 
@@ -660,8 +636,9 @@ public Builder trackExecutionTime(double alpha) {
                 return this;
             }
 
-            public Builder trackUtilization() {
+            public Builder trackUtilization(Duration interval) {
                 trackUtilization = true;
+                utilizationInterval = interval;
                 return this;
             }
 
@@ -676,7 +653,14 @@ public Builder trackMaxQueueLatency() {
             }
 
             public TaskTrackingConfig build() {
-                return new TaskTrackingConfig(trackExecutionTime, trackUtilization, trackOngoingTasks, trackMaxQueueLatency, ewmaAlpha);
+                return new TaskTrackingConfig(
+                    trackExecutionTime,
+                    trackUtilization,
+                    utilizationInterval,
+                    trackOngoingTasks,
+                    trackMaxQueueLatency,
+                    ewmaAlpha
+                );
             }
         }
     }