targetIORateBytesPerSec

Author: albertzaharovits

server/src/main/java/org/elasticsearch/index/engine/ThreadPoolMergeQueue.java

@@ -10,6 +10,7 @@
 package org.elasticsearch.index.engine;
 
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.engine.ThreadPoolMergeScheduler.MergeTask;
@@ -19,30 +20,32 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.PriorityBlockingQueue;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
 public class ThreadPoolMergeQueue {
     /**
-     * Floor for IO write rate limit (we will never go any lower than this)
+     * Floor for IO write rate limit of individual merge tasks (we will never go any lower than this)
      */
-    private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+    private static final ByteSizeValue MIN_IO_RATE = ByteSizeValue.ofMb(5L);
     /**
-     * Ceiling for IO write rate limit (we will never go any higher than this)
+     * Ceiling for IO write rate limit of individual merge tasks (we will never go any higher than this)
      */
-    private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+    private static final ByteSizeValue MAX_IO_RATE = ByteSizeValue.ofMb(10240L);
     /**
-     * Initial value for IO write rate limit when doAutoIOThrottle is true
+     * Initial value for IO write rate limit of individual merge tasks when doAutoIOThrottle is true
      */
-    private static final double START_MB_PER_SEC = 20.0;
+    private static final ByteSizeValue START_IO_RATE = ByteSizeValue.ofMb(20L);
     private final AtomicInteger activeIOThrottledMergeTasksCount = new AtomicInteger();
     private final PriorityBlockingQueue<MergeTask> queuedMergeTasks = new PriorityBlockingQueue<>();
     // the set of all merge tasks currently being executed by merge threads from the pool,
     // in order to be able to update the IO throttle rate of merge tasks also after they have started (while executing)
     private final Set<MergeTask> currentlyRunningMergeTasks = ConcurrentCollections.newConcurrentSet();
     /**
-     * Current IO write throttle rate, for all merge, across all merge schedulers (shards) on the node
+     * Current IO write throttle rate, in bytes per sec, that's in effect for all currently running merge tasks,
+     * across all {@link ThreadPoolMergeScheduler}s that use this instance of the queue.
      */
-    private final AtomicReference<Double> targetMBPerSec = new AtomicReference<>(START_MB_PER_SEC);
+    private final AtomicLong targetIORateBytesPerSec = new AtomicLong(START_IO_RATE.getBytes());
     private final ExecutorService executorService;
     private final int maxConcurrentMerges;
 
@@ -110,7 +113,7 @@ private void runMergeTask(MergeTask mergeTask) {
         assert added : "starting merge task [" + mergeTask + "] registered as already running";
         try {
             if (mergeTask.supportsIOThrottling()) {
-                mergeTask.setIORateLimit(targetMBPerSec.get());
+                mergeTask.setIORateLimit(ByteSizeValue.ofBytes(targetIORateBytesPerSec.get()).getMbFrac());
             }
             mergeTask.run();
         } finally {
@@ -124,35 +127,48 @@ private void runMergeTask(MergeTask mergeTask) {
     }
 
     private void maybeUpdateTargetMBPerSec() {
-        double currentTargetMBPerSec = targetMBPerSec.get();
-        double newTargetMBPerSec = newTargetMBPerSec(currentTargetMBPerSec, activeIOThrottledMergeTasksCount.get(), maxConcurrentMerges);
-        if (currentTargetMBPerSec != newTargetMBPerSec && targetMBPerSec.compareAndSet(currentTargetMBPerSec, newTargetMBPerSec)) {
+        long currentTargetIORateBytesPerSec = targetIORateBytesPerSec.get();
+        long newTargetIORateBytesPerSec = newTargetIORateBytesPerSec(
+            currentTargetIORateBytesPerSec,
+            activeIOThrottledMergeTasksCount.get(),
+            maxConcurrentMerges
+        );
+        if (currentTargetIORateBytesPerSec != newTargetIORateBytesPerSec
+            && targetIORateBytesPerSec.compareAndSet(currentTargetIORateBytesPerSec, newTargetIORateBytesPerSec)) {
             // it's OK to have this method update merge tasks concurrently, with different targetMBPerSec values,
             // as it's not important that all merge tasks are throttled to the same IO rate at all time.
             // For performance reasons, we don't synchronize the updates to targetMBPerSec values with the update of running merges.
             currentlyRunningMergeTasks.forEach(mergeTask -> {
                 if (mergeTask.supportsIOThrottling()) {
-                    mergeTask.setIORateLimit(newTargetMBPerSec);
+                    mergeTask.setIORateLimit(ByteSizeValue.ofBytes(newTargetIORateBytesPerSec).getMbFrac());
                 }
             });
         }
     }
 
-    private static double newTargetMBPerSec(double currentTargetMBPerSec, int activeTasksCount, int maxConcurrentMerges) {
-        final double newTargetMBPerSec;
-        if (activeTasksCount < maxConcurrentMerges * 2 && currentTargetMBPerSec > MIN_MERGE_MB_PER_SEC) {
-            newTargetMBPerSec = Math.max(MIN_MERGE_MB_PER_SEC, currentTargetMBPerSec / 1.1);
-        } else if (activeTasksCount > maxConcurrentMerges * 4 && currentTargetMBPerSec < MAX_MERGE_MB_PER_SEC) {
-            newTargetMBPerSec = Math.min(MAX_MERGE_MB_PER_SEC, currentTargetMBPerSec * 1.1);
+    private static long newTargetIORateBytesPerSec(long currentTargetIORateBytesPerSec, int activeTasksCount, int maxConcurrentMerges) {
+        final long newTargetIORateBytesPerSec;
+        if (activeTasksCount < maxConcurrentMerges * 2 && currentTargetIORateBytesPerSec > MIN_IO_RATE.getBytes()) {
+            // decrease target IO rate by 10% (capped)
+            newTargetIORateBytesPerSec = Math.max(
+                MIN_IO_RATE.getBytes(),
+                currentTargetIORateBytesPerSec - currentTargetIORateBytesPerSec / 10L
+            );
+        } else if (activeTasksCount > maxConcurrentMerges * 4 && currentTargetIORateBytesPerSec < MAX_IO_RATE.getBytes()) {
+            // increase target IO rate by 10% (capped)
+            newTargetIORateBytesPerSec = Math.min(
+                MAX_IO_RATE.getBytes(),
+                currentTargetIORateBytesPerSec + currentTargetIORateBytesPerSec / 10L
+            );
         } else {
-            newTargetMBPerSec = currentTargetMBPerSec;
+            newTargetIORateBytesPerSec = currentTargetIORateBytesPerSec;
         }
-        return newTargetMBPerSec;
+        return newTargetIORateBytesPerSec;
     }
 
     // exposed for stats
     double getTargetMBPerSec() {
-        return targetMBPerSec.get();
+        return ByteSizeValue.ofBytes(targetIORateBytesPerSec.get()).getMbFrac();
     }
 
     public boolean isEmpty() {