rwlock

Author: mhl-b

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

@@ -27,10 +27,10 @@
 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.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -260,13 +260,13 @@ public boolean trackingMaxQueueLatency() {
      * Can be extended to remember multiple past frames.
      */
     public static class FramedTimeTracker {
-        final long interval;
+        private final long interval;
         private final Supplier<Long> timeNow;
+        private final ReentrantReadWriteLock rwlock = new ReentrantReadWriteLock();
         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) {
@@ -281,6 +281,10 @@ public FramedTimeTracker(long intervalNano, Supplier<Long> timeNow) {
             this.timeNow = System::nanoTime;
         }
 
+        public long interval() {
+            return interval;
+        }
+
         /**
          * 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.
@@ -296,7 +300,10 @@ private void updateFrame0(long nowTime) {
             var now = nowTime / interval;
             var current = currentFrame.get();
             if (current < now) {
-                if (updatingFrame.compareAndSet(false, true)) {
+                rwlock.readLock().unlock();
+                rwlock.writeLock().lock();
+                current = currentFrame.get(); // make sure it didnt change during lock acquisition
+                if (current < now) {
                     var tasks = ongoingTasks.get();
                     if (current == now - 1) {
                         previousTime.set(currentTime.get());
@@ -305,12 +312,9 @@ private void updateFrame0(long nowTime) {
                     }
                     currentTime.set(tasks * interval);
                     currentFrame.set(now);
-                    updatingFrame.set(false);
-                } else {
-                    while (currentFrame.get() != now) {
-                        Thread.onSpinWait();
-                    }
                 }
+                rwlock.readLock().lock();
+                rwlock.writeLock().unlock();
             }
         }
 
@@ -319,28 +323,37 @@ private void updateFrame0(long nowTime) {
          * If task finishes sooner than end of interval {@link FramedTimeTracker#endTask()} will deduct remaining time.
          */
         public void startTask() {
+            rwlock.readLock().lock();
             var now = timeNow.get();
             updateFrame0(now);
             ongoingTasks.incrementAndGet();
             currentTime.updateAndGet((t) -> t + (currentFrame.get() + 1) * interval - now);
+            rwlock.readLock().unlock();
         }
 
         /**
          * Stop task tracking. We already assumed that task runs till end of frame, here we deduct not used time.
          */
         public void endTask() {
+            rwlock.readLock().lock();
             var now = timeNow.get();
             updateFrame0(now);
             ongoingTasks.decrementAndGet();
             currentTime.updateAndGet((t) -> t - (currentFrame.get() + 1) * interval + now);
+            rwlock.readLock().unlock();
         }
 
         /**
          * Returns previous frame total execution time.
          */
         public long previousFrameTime() {
-            updateFrame0(timeNow.get());
-            return previousTime.get();
+            try {
+                rwlock.readLock().lock();
+                updateFrame0(timeNow.get());
+                return previousTime.get();
+            } finally {
+                rwlock.readLock().unlock();
+            }
         }
     }
 }

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

@@ -44,19 +44,19 @@ public void testSingleFrameTask() {
         tracker.startTask();
         fakeTime.time += 10;
         tracker.endTask();
-        fakeTime.time += tracker.interval;
+        fakeTime.time += tracker.interval();
         assertEquals(10, tracker.previousFrameTime());
     }
 
     public void testTwoFrameTask() {
         var tracker = newTracker(100);
         var startTime = between(0, 100);
-        var taskDuration = tracker.interval;
+        var taskDuration = tracker.interval();
         fakeTime.time += startTime;
         tracker.startTask();
         fakeTime.time += taskDuration;
         tracker.endTask();
-        assertEquals(tracker.interval - startTime, tracker.previousFrameTime());
+        assertEquals(tracker.interval() - startTime, tracker.previousFrameTime());
     }
 
     public void testMultiFrameTask() {
@@ -66,16 +66,16 @@ public void testMultiFrameTask() {
         var taskDuration = between(3, 100) * interval;
         fakeTime.time += taskDuration;
         tracker.endTask();
-        assertEquals(tracker.interval, tracker.previousFrameTime());
+        assertEquals(tracker.interval(), tracker.previousFrameTime());
     }
 
     public void testOngoingTask() {
         var interval = 10;
         var tracker = newTracker(interval);
         tracker.startTask();
         for (int i = 0; i < between(10, 100); i++) {
-            fakeTime.time += tracker.interval;
-            assertEquals(tracker.interval, tracker.previousFrameTime());
+            fakeTime.time += tracker.interval();
+            assertEquals(tracker.interval(), tracker.previousFrameTime());
         }
     }