Merge pull request #2 from nicktindall/framed-thread-pool-utilization_bm

Framed thread pool utilization benchmark hacking

Author: mhl-b

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

@@ -9,110 +9,92 @@
 
 package org.elasticsearch.benchmark.common.util.concurrent;
 
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.util.concurrent.EsExecutors;
-import org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor;
 import org.elasticsearch.common.util.concurrent.TaskExecutionTimeTrackingEsThreadPoolExecutor;
-import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
 import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Group;
 import org.openjdk.jmh.annotations.Measurement;
 import org.openjdk.jmh.annotations.Mode;
 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.Threads;
 import org.openjdk.jmh.annotations.Warmup;
 import org.openjdk.jmh.infra.Blackhole;
 
-import java.time.Duration;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
-@Warmup(iterations = 1, time = 1, timeUnit = TimeUnit.SECONDS)
-@Measurement(iterations = 1, time = 10, timeUnit = TimeUnit.SECONDS)
+@Threads(12)
+@Warmup(iterations = 3, time = 200, timeUnit = TimeUnit.MILLISECONDS)
+@Measurement(iterations = 5, time = 600, timeUnit = TimeUnit.MILLISECONDS)
 @BenchmarkMode(Mode.AverageTime)
-@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@State(Scope.Benchmark)
 @Fork(1)
-@State(Scope.Thread)
 public class ThreadPoolUtilizationBenchmark {
 
-    @Param({ "false", "true" })
-    private boolean trackUtilization;
+    @Param({ "0", "10000", "100000" })
+    private int callIntervalTicks;
 
-    @Param({ "4", "8", "16" })
-    private int poolSize;
-
-    @Param({ "1000000" })
-    private int tasksNum;
-
-    @Param({ "10" }) // 10ms is aggressive interval, it increases frame updates on FramedTimeTracker, normally we run at 30/60
-    // seconds
+    /**
+     * This makes very little difference, all the overhead is in the synchronization
+     */
+    @Param({ "10" })
     private int utilizationIntervalMs;
 
-    private EsThreadPoolExecutor executor;
+    @State(Scope.Thread)
+    public static class TaskState {
+        boolean running = false;
 
-    private EsThreadPoolExecutor newExecutor(boolean tracking) {
-        var conf = EsExecutors.TaskTrackingConfig.builder();
-        if (tracking) {
-            conf.trackExecutionTime(0.3).trackUtilization(Duration.ofMillis(utilizationIntervalMs));
+        boolean shouldStart() {
+            return (running = running == false);
         }
-        return EsExecutors.newFixed(
-            "bench",
-            poolSize,
-            tasksNum,
-            Executors.defaultThreadFactory(),
-            new ThreadContext(Settings.EMPTY),
-            conf.build()
-        );
     }
 
+    private TaskExecutionTimeTrackingEsThreadPoolExecutor.FramedTimeTracker timeTracker;
+
     @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");
-            }
+        timeTracker = new TaskExecutionTimeTrackingEsThreadPoolExecutor.FramedTimeTracker(
+            TimeUnit.MILLISECONDS.toNanos(utilizationIntervalMs),
+            System::nanoTime
+        );
+    }
+
+    @Benchmark
+    public void baseline() {
+        Blackhole.consumeCPU(callIntervalTicks);
+    }
+
+    @Group("ReadAndWrite")
+    @Benchmark
+    public void startAndStopTasks(TaskState state) {
+        Blackhole.consumeCPU(callIntervalTicks);
+        if (state.shouldStart()) {
+            timeTracker.startTask();
         } else {
-            var exec = newExecutor(false);
-            if (exec instanceof TaskExecutionTimeTrackingEsThreadPoolExecutor) {
-                throw new IllegalStateException("must be non-tracking executor");
-            }
-            executor = exec;
+            timeTracker.endTask();
         }
     }
 
-    @TearDown
-    public void tearDown() throws InterruptedException {
-        executor.shutdown();
-        executor.awaitTermination(0, TimeUnit.MILLISECONDS);
+    @Benchmark
+    @Group("ReadAndWrite")
+    public void readPrevious(Blackhole blackhole) {
+        Blackhole.consumeCPU(callIntervalTicks);
+        blackhole.consume(timeTracker.previousFrameTime());
     }
 
     @Benchmark
-    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();
-            });
+    @Group("JustWrite")
+    public void startAndStopTasksOnly(TaskState state) {
+        Blackhole.consumeCPU(callIntervalTicks);
+        if (state.shouldStart()) {
+            timeTracker.startTask();
+        } else {
+            timeTracker.endTask();
         }
-        completedTasks.await();
     }
 }

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

@@ -257,7 +257,7 @@ public boolean trackingMaxQueueLatency() {
      *
      * Can be extended to remember multiple past frames.
      */
-    static class FramedTimeTracker {
+    public static class FramedTimeTracker {
         final long interval;
         private final Supplier<Long> timeNow;
         long ongoingTasks;
@@ -266,7 +266,7 @@ static class FramedTimeTracker {
         long previousTime;
 
         // for testing
-        FramedTimeTracker(long intervalNano, Supplier<Long> timeNow) {
+        public FramedTimeTracker(long intervalNano, Supplier<Long> timeNow) {
             assert intervalNano > 0;
             this.interval = intervalNano;
             this.timeNow = timeNow;
@@ -278,7 +278,7 @@ static class FramedTimeTracker {
             this.timeNow = System::nanoTime;
         }
 
-        synchronized void updateFrame() {
+        public synchronized void updateFrame() {
             updateFrame0(timeNow.get());
         }
 
@@ -310,7 +310,7 @@ private void updateFrame0(long nowTime) {
          * Start tracking new task, assume that task runs indefinitely, or at least till end of frame.
          * If task finishes sooner than end of interval {@link FramedTimeTracker#endTask()} will deduct remaining time.
          */
-        synchronized void startTask() {
+        public synchronized void startTask() {
             var now = timeNow.get();
             updateFrame0(now);
             currentTime += (currentFrame + 1) * interval - now;
@@ -320,7 +320,7 @@ synchronized void startTask() {
         /**
          * Stop task tracking. We already assumed that task runs till end of frame, here we deduct not used time.
          */
-        synchronized void endTask() {
+        public synchronized void endTask() {
             var now = timeNow.get();
             updateFrame0(now);
             currentTime -= (currentFrame + 1) * interval - now;
@@ -330,7 +330,7 @@ synchronized void endTask() {
         /**
          * Returns previous frame total execution time.
          */
-        synchronized long previousFrameTime() {
+        public synchronized long previousFrameTime() {
             updateFrame0(timeNow.get());
             return previousTime;
         }