Throttling ?

Author: albertzaharovits

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

@@ -2826,29 +2826,19 @@ LiveIndexWriterConfig getCurrentIndexWriterConfig() {
     protected ElasticsearchMergeScheduler createMergeScheduler(ShardId shardId, IndexSettings indexSettings) {
         // return new EngineMergeScheduler(shardId, indexSettings);
         return new ThreadPoolMergeScheduler(shardId, indexSettings, engineConfig.getThreadPool()) {
-            private final AtomicInteger numMergesInFlight = new AtomicInteger(0);
-            private final AtomicBoolean isThrottling = new AtomicBoolean();
 
             @Override
-            public synchronized void beforeMerge(OnGoingMerge merge) {
-                int maxNumMerges = getMaxMergeCount();
-                if (numMergesInFlight.incrementAndGet() > maxNumMerges) {
-                    if (isThrottling.getAndSet(true) == false) {
-                        logger.info("now throttling indexing: numMergesInFlight={}, maxNumMerges={}", numMergesInFlight, maxNumMerges);
-                        activateThrottling();
-                    }
-                }
+            protected synchronized void activateThrottling(int numActiveMerges) {
+                InternalEngine.this.activateThrottling();
+            }
+
+            @Override
+            protected synchronized void deactivateThrottling(int numActiveMerges) {
+                InternalEngine.this.deactivateThrottling();
             }
 
             @Override
             public synchronized void afterMerge(OnGoingMerge merge) {
-                int maxNumMerges = getMaxMergeCount();
-                if (numMergesInFlight.decrementAndGet() < maxNumMerges) {
-                    if (isThrottling.getAndSet(false)) {
-                        logger.info("stop throttling indexing: numMergesInFlight={}, maxNumMerges={}", numMergesInFlight, maxNumMerges);
-                        deactivateThrottling();
-                    }
-                }
                 if (indexWriter.hasPendingMerges() == false
                     && System.nanoTime() - lastWriteNanos >= engineConfig.getFlushMergesAfter().nanos()) {
                     // NEVER do this on a merge thread since we acquire some locks blocking here and if we concurrently rollback the writer

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

@@ -40,6 +40,7 @@
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class ThreadPoolMergeScheduler extends MergeScheduler implements ElasticsearchMergeScheduler {
     /**
@@ -72,6 +73,7 @@ public class ThreadPoolMergeScheduler extends MergeScheduler implements Elastics
     private final ThreadLocal<MergeRateLimiter> onGoingMergeRateLimiter = new ThreadLocal<>();
     private final PriorityQueue<MergeTask> activeMergeTasksLocalSchedulerQueue = new PriorityQueue<>();
     private final List<MergeTask> activeMergeTasksExecutingOnLocalSchedulerList = new ArrayList<>();
+    private final AtomicBoolean isThrottling = new AtomicBoolean();
 
     public ThreadPoolMergeScheduler(ShardId shardId, IndexSettings indexSettings, ThreadPool threadPool) {
         this.config = indexSettings.getMergeSchedulerConfig();
@@ -120,6 +122,10 @@ protected void beforeMerge(OnGoingMerge merge) {}
      */
     protected void afterMerge(OnGoingMerge merge) {}
 
+    protected void activateThrottling(int numActiveMerges) {}
+
+    protected void deactivateThrottling(int numActiveMerges) {}
+
     public int getMaxMergeCount() {
         return config.getMaxMergeCount();
     }
@@ -144,13 +150,15 @@ private void submitNewMergeTask(MergeSource mergeSource, MergePolicy.OneMerge me
             activeMergeTasksLocalSchedulerQueue.add(mergeTask);
         }
         maybeExecuteNextMerge();
+        maybeActivateThrottling();
     }
 
     private void mergeDone(MergeTask mergeTask) {
         synchronized (this) {
             activeMergeTasksExecutingOnLocalSchedulerList.remove(mergeTask);
         }
         maybeExecuteNextMerge();
+        maybeDeactivateThrottling();
     }
 
     private void maybeExecuteNextMerge() {
@@ -169,6 +177,20 @@ private void maybeExecuteNextMerge() {
         executorService.execute(mergeTask);
     }
 
+    private void maybeActivateThrottling() {
+        int numActiveMerges = activeMergeTasksExecutingOnLocalSchedulerList.size();
+        if (numActiveMerges > config.getMaxMergeCount() && isThrottling.getAndSet(true) == false) {
+            activateThrottling(numActiveMerges);
+        }
+    }
+
+    private void maybeDeactivateThrottling() {
+        int numActiveMerges = activeMergeTasksExecutingOnLocalSchedulerList.size();
+        if (numActiveMerges <= config.getMaxMergeCount() && isThrottling.getAndSet(false) == true) {
+            deactivateThrottling(numActiveMerges);
+        }
+    }
+
     private static double maybeUpdateTargetMBPerSec(int poolSize) {
         if (activeThrottledMergeTasksAcrossSchedulersSet.size() < poolSize * 2 && targetMBPerSec > MIN_MERGE_MB_PER_SEC) {
             return Math.max(MIN_MERGE_MB_PER_SEC, targetMBPerSec / 1.1);