non-locking counting

Author: mhl-b

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

@@ -30,30 +30,20 @@
 @Threads(12)
 @Warmup(iterations = 3, time = 200, timeUnit = TimeUnit.MILLISECONDS)
 @Measurement(iterations = 5, time = 600, timeUnit = TimeUnit.MILLISECONDS)
-@BenchmarkMode(Mode.AverageTime)
+@BenchmarkMode(Mode.SampleTime)
 @OutputTimeUnit(TimeUnit.MICROSECONDS)
 @State(Scope.Benchmark)
 @Fork(1)
 public class ThreadPoolUtilizationBenchmark {
 
-    @Param({ "0", "10000", "100000" })
+    @Param({ "10000" })
     private int callIntervalTicks;
 
     /**
      * This makes very little difference, all the overhead is in the synchronization
      */
     @Param({ "10" })
     private int utilizationIntervalMs;
-
-    @State(Scope.Thread)
-    public static class TaskState {
-        boolean running = false;
-
-        boolean shouldStart() {
-            return (running = running == false);
-        }
-    }
-
     private TaskExecutionTimeTrackingEsThreadPoolExecutor.FramedTimeTracker timeTracker;
 
     @Setup
@@ -69,32 +59,20 @@ public void baseline() {
         Blackhole.consumeCPU(callIntervalTicks);
     }
 
-    @Group("ReadAndWrite")
+    @Group("StartAndEnd")
     @Benchmark
     public void startAndStopTasks(TaskState state) {
+        timeTracker.startTask();
         Blackhole.consumeCPU(callIntervalTicks);
-        if (state.shouldStart()) {
-            timeTracker.startTask();
-        } else {
-            timeTracker.endTask();
-        }
+        timeTracker.endTask();
     }
 
-    @Benchmark
-    @Group("ReadAndWrite")
-    public void readPrevious(Blackhole blackhole) {
-        Blackhole.consumeCPU(callIntervalTicks);
-        blackhole.consume(timeTracker.previousFrameTime());
-    }
+    @State(Scope.Thread)
+    public static class TaskState {
+        boolean running = false;
 
-    @Benchmark
-    @Group("JustWrite")
-    public void startAndStopTasksOnly(TaskState state) {
-        Blackhole.consumeCPU(callIntervalTicks);
-        if (state.shouldStart()) {
-            timeTracker.startTask();
-        } else {
-            timeTracker.endTask();
+        boolean shouldStart() {
+            return (running = running == false);
         }
     }
 }

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

@@ -27,6 +27,8 @@
 import java.util.concurrent.RejectedExecutionHandler;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAccumulator;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.function.Function;
@@ -260,10 +262,11 @@ public boolean trackingMaxQueueLatency() {
     public static class FramedTimeTracker {
         final long interval;
         private final Supplier<Long> timeNow;
-        long ongoingTasks;
-        long currentFrame;
-        long currentTime;
-        long previousTime;
+        private final AtomicLong ongoingTasks = new AtomicLong();
+        private final AtomicLong currentFrame = new AtomicLong();
+        private final AtomicLong currentTime = new AtomicLong();
+        private final AtomicLong previousTime = new AtomicLong();
+        private final AtomicBoolean updatingFrame = new AtomicBoolean();
 
         // for testing
         public FramedTimeTracker(long intervalNano, Supplier<Long> timeNow) {
@@ -278,10 +281,6 @@ public FramedTimeTracker(long intervalNano, Supplier<Long> timeNow) {
             this.timeNow = System::nanoTime;
         }
 
-        public synchronized void updateFrame() {
-            updateFrame0(timeNow.get());
-        }
-
         /**
          * Update frames to current time. There are no guaranties that it will be invoked frequently.
          * For example when there are no tasks and no requests for previousFrameTime.
@@ -295,44 +294,53 @@ public synchronized void updateFrame() {
          */
         private void updateFrame0(long nowTime) {
             var now = nowTime / interval;
-            if (currentFrame < now) {
-                if (currentFrame == now - 1) {
-                    previousTime = currentTime; //
+            var current = currentFrame.get();
+            if (current < now) {
+                if (updatingFrame.compareAndSet(false, true)) {
+                    var tasks = ongoingTasks.get();
+                    if (current == now - 1) {
+                        previousTime.set(currentTime.get());
+                    } else {
+                        previousTime.set(tasks * interval);
+                    }
+                    currentTime.set(tasks * interval);
+                    currentFrame.set(now);
+                    updatingFrame.set(false);
                 } else {
-                    previousTime = ongoingTasks * interval;
+                    while (currentFrame.get() != now) {
+                        Thread.onSpinWait();
+                    }
                 }
-                currentTime = ongoingTasks * interval;
-                currentFrame = now;
             }
         }
 
         /**
          * 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.
          */
-        public synchronized void startTask() {
+        public void startTask() {
             var now = timeNow.get();
             updateFrame0(now);
-            currentTime += (currentFrame + 1) * interval - now;
-            ++ongoingTasks;
+            ongoingTasks.incrementAndGet();
+            currentTime.updateAndGet((t) -> t + (currentFrame.get() + 1) * interval - now);
         }
 
         /**
          * Stop task tracking. We already assumed that task runs till end of frame, here we deduct not used time.
          */
-        public synchronized void endTask() {
+        public void endTask() {
             var now = timeNow.get();
             updateFrame0(now);
-            currentTime -= (currentFrame + 1) * interval - now;
-            --ongoingTasks;
+            ongoingTasks.decrementAndGet();
+            currentTime.updateAndGet((t) -> t - (currentFrame.get() + 1) * interval + now);
         }
 
         /**
          * Returns previous frame total execution time.
          */
-        public synchronized long previousFrameTime() {
+        public long previousFrameTime() {
             updateFrame0(timeNow.get());
-            return previousTime;
+            return previousTime.get();
         }
     }
 }

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

@@ -32,7 +32,7 @@ public void setup() {
 
     public void testNoTasks() {
         var tracker = newTracker(1);
-        tracker.updateFrame();
+        tracker.previousFrameTime();
         assertEquals(0, tracker.previousFrameTime());
         fakeTime.time += between(1, 100);
         assertEquals(0, tracker.previousFrameTime());

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

@@ -244,7 +244,7 @@ public void testUtilization() throws InterruptedException {
 
         final Runnable waitTillNextFrame = () -> {
             var now = System.nanoTime();
-            var waitTillNext = (now / interval.getNano() + 1) * interval.getNano() - now;
+            var waitTillNext = (now / interval.toNanos() + 1) * interval.toNanos() - now;
             trySleep.accept(Duration.ofNanos(waitTillNext));
         };
 
@@ -274,7 +274,7 @@ public void testUtilization() throws InterruptedException {
             waitTillNextFrame.run();
 
             DoubleStream.of(0.1, 0.3, 0.5, 3.).forEach(loadFactor -> {
-                var sleepTime = (long) (interval.getNano() * loadFactor);
+                var sleepTime = (long) (interval.toNanos() * loadFactor);
                 executor.submit(sleepTaskFn.apply(Duration.ofNanos(sleepTime)));
             });