Review

Author: albertzaharovits

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

@@ -87,8 +87,10 @@ public class ThreadPoolMergeExecutorService {
     private ThreadPoolMergeExecutorService(ThreadPool threadPool) {
         this.executorService = threadPool.executor(ThreadPool.Names.MERGE);
         this.maxConcurrentMerges = threadPool.info(ThreadPool.Names.MERGE).getMax();
+        // the intent here is to throttle down whenever we submit a task and no other task is running
         this.concurrentMergesFloorLimitForThrottling = 2;
         this.concurrentMergesCeilLimitForThrottling = maxConcurrentMerges * 2;
+        assert concurrentMergesFloorLimitForThrottling <= concurrentMergesCeilLimitForThrottling;
     }
 
     boolean submitMergeTask(MergeTask mergeTask) {
@@ -233,7 +235,7 @@ private static long newTargetIORateBytesPerSec(
                 // increase target IO rate by 20% (capped)
                 newTargetIORateBytesPerSec = Math.min(
                     MAX_IO_RATE.getBytes(),
-                    currentTargetIORateBytesPerSec + currentTargetIORateBytesPerSec / 20L
+                    currentTargetIORateBytesPerSec + currentTargetIORateBytesPerSec / 5L
                 );
             } else {
                 newTargetIORateBytesPerSec = currentTargetIORateBytesPerSec;
@@ -295,14 +297,4 @@ long getTargetIORateBytesPerSec() {
     int getMaxConcurrentMerges() {
         return maxConcurrentMerges;
     }
-
-    // exposed for tests
-    int getConcurrentMergesFloorLimitForThrottling() {
-        return concurrentMergesFloorLimitForThrottling;
-    }
-
-    // exposed for tests
-    int getConcurrentMergesCeilLimitForThrottling() {
-        return concurrentMergesCeilLimitForThrottling;
-    }
 }

server/src/test/java/org/elasticsearch/index/engine/ThreadPoolMergeExecutorServiceTests.java

@@ -219,12 +219,10 @@ public void testTargetIORateChangesWhenSubmittingMergeTasks() throws Exception {
                     }
                     long newIORate = threadPoolMergeExecutorService.getTargetIORateBytesPerSec();
                     if (supportsIOThrottling) {
-                        if (submittedIOThrottledMergeTasks.get() < threadPoolMergeExecutorService
-                            .getConcurrentMergesFloorLimitForThrottling()) {
-                            // assert the IO rate decreases, with a floor limit, when there are few merge tasks enqueued
+                        if (submittedIOThrottledMergeTasks.get() < 2) {
+                            // assert the IO rate decreases, with a floor limit, when there is just a single merge task running
                             assertThat(newIORate, either(is(MIN_IO_RATE.getBytes())).or(lessThan(currentIORate)));
-                        } else if (submittedIOThrottledMergeTasks.get() > threadPoolMergeExecutorService
-                            .getConcurrentMergesCeilLimitForThrottling()) {
+                        } else if (submittedIOThrottledMergeTasks.get() > threadPoolMergeExecutorService.getMaxConcurrentMerges() * 2) {
                                 // assert the IO rate increases, with a ceiling limit, when there are many merge tasks enqueued
                                 assertThat(newIORate, either(is(MAX_IO_RATE.getBytes())).or(greaterThan(currentIORate)));
                             } else {
@@ -375,11 +373,10 @@ private void testIORateAdjustedForSubmittedTasks(
                 initialTasksCounter--;
                 threadPoolMergeExecutorService.submitMergeTask(mergeTask);
                 long newTargetIORateLimit = threadPoolMergeExecutorService.getTargetIORateBytesPerSec();
-                if (currentlySubmittedMergeTaskCount.get() < threadPoolMergeExecutorService.getConcurrentMergesFloorLimitForThrottling()) {
+                if (currentlySubmittedMergeTaskCount.get() < 2) {
                     // assert the IO rate decreases, with a floor limit, when there are few merge tasks enqueued
                     assertThat(newTargetIORateLimit, either(is(MIN_IO_RATE.getBytes())).or(lessThan(targetIORateLimit.get())));
-                } else if (currentlySubmittedMergeTaskCount.get() > threadPoolMergeExecutorService
-                    .getConcurrentMergesCeilLimitForThrottling()) {
+                } else if (currentlySubmittedMergeTaskCount.get() > threadPoolMergeExecutorService.getMaxConcurrentMerges() * 2) {
                         // assert the IO rate increases, with a ceiling limit, when there are many merge tasks enqueued
                         assertThat(newTargetIORateLimit, either(is(MAX_IO_RATE.getBytes())).or(greaterThan(targetIORateLimit.get())));
                     } else {