pause indexing and race condition diags

Author: ankikuma

muted-tests.yml

@@ -444,9 +444,9 @@ tests:
 - class: org.elasticsearch.xpack.esql.qa.mixed.EsqlClientYamlIT
   method: test {p0=esql/120_profile/avg 8.14 or after}
   issue: https://github.com/elastic/elasticsearch/issues/127879
-- class: org.elasticsearch.indices.stats.IndexStatsIT
-  method: testThrottleStats
-  issue: https://github.com/elastic/elasticsearch/issues/126359
+#- class: org.elasticsearch.indices.stats.IndexStatsIT
+#  method: testThrottleStats
+#  issue: https://github.com/elastic/elasticsearch/issues/126359
 - class: org.elasticsearch.search.vectors.IVFKnnFloatVectorQueryTests
   method: testRandomWithFilter
   issue: https://github.com/elastic/elasticsearch/issues/127963

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

@@ -504,6 +504,17 @@ long getThrottleTimeInMillis() {
             return throttleTimeMillisMetric.count() + TimeValue.nsecToMSec(currentThrottleNS);
         }
 
+        void pauseThrottle() {
+            if (isThrottled()) {
+                throttlingLock.pauseThrottle();
+            }
+            // TODO: calculate pause time
+        }
+
+        void unpauseThrottle() {
+            throttlingLock.unPauseThrottle();
+        }
+
         boolean isThrottled() {
             return lock != NOOP_LOCK;
         }
@@ -577,9 +588,13 @@ public Condition newCondition() {
     protected static final class PauseLock implements Lock {
         private final Semaphore semaphore = new Semaphore(Integer.MAX_VALUE);
         private final int allowThreads;
+        // If the lock is currently throttled, we might need to pause throttling
+        // to let some threads pass when another task is trying to acquire indexing permits
+        private volatile AtomicBoolean throttlePaused;
 
         public PauseLock(int allowThreads) {
             this.allowThreads = allowThreads;
+            throttlePaused = new AtomicBoolean(false);
         }
 
         public void lock() {
@@ -612,14 +627,34 @@ public Condition newCondition() {
         }
 
         public void throttle() {
-            assert semaphore.availablePermits() == Integer.MAX_VALUE;
+            if(throttlePaused.get()) {
+                return;
+            }
+            assert semaphore.availablePermits() == Integer.MAX_VALUE : "Available permits should be " + Integer.MAX_VALUE
+                + " but is " + semaphore.availablePermits();
             semaphore.acquireUninterruptibly(Integer.MAX_VALUE - allowThreads);
         }
 
         public void unthrottle() {
-            assert semaphore.availablePermits() <= allowThreads;
+            unPauseThrottle();
+            assert semaphore.availablePermits() <= allowThreads : "Available permits should be no greater than " + allowThreads
+                + " but is " + semaphore.availablePermits();
             semaphore.release(Integer.MAX_VALUE - allowThreads);
         }
+
+        // Pause throttling by allowing 1 thread to pass
+        public void pauseThrottle() {
+            // we only need to pause throttling if throttling allows no threads to pass i.e. allowThreads is 0
+            if((allowThreads == 0) && throttlePaused.getAndSet(true) == false) {
+                semaphore.release(1);
+            }
+        }
+
+        public void unPauseThrottle() {
+            if(throttlePaused.getAndSet(false)) {
+                semaphore.acquireUninterruptibly(1);
+            }
+        }
     }
 
     /**
@@ -2259,6 +2294,18 @@ public interface Warmer {
      */
     public abstract void deactivateThrottling();
 
+    /**
+     * If indexing is throttled to the point where it is paused completely,
+     * another task trying to get indexing permits might want to pause throttling
+     * by letting one thread pass at a time so that it does not get starved.
+     */
+    public abstract void pauseThrottling();
+
+    /**
+     * Reverses a previous {@link #pauseThrottling} call.
+     */
+    public abstract void unPauseThrottling();
+
     /**
      * This method replays translog to restore the Lucene index which might be reverted previously.
      * This ensures that all acknowledged writes are restored correctly when this engine is promoted.

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

@@ -2827,6 +2827,7 @@ public void activateThrottling() {
         assert count >= 1 : "invalid post-increment throttleRequestCount=" + count;
         if (count == 1) {
             throttle.activate();
+            logger.info("Activated throttling");
         }
     }
 
@@ -2836,9 +2837,20 @@ public void deactivateThrottling() {
         assert count >= 0 : "invalid post-decrement throttleRequestCount=" + count;
         if (count == 0) {
             throttle.deactivate();
+            logger.info("Deactivated throttling");
         }
     }
 
+    @Override
+    public void pauseThrottling() {
+        throttle.pauseThrottle();
+    }
+
+    @Override
+    public void unPauseThrottling() {
+        throttle.unpauseThrottle();
+    }
+
     @Override
     public boolean isThrottled() {
         return throttle.isThrottled();
@@ -2920,7 +2932,9 @@ protected synchronized void enableIndexingThrottling(int numRunningMerges, int n
                 numQueuedMerges,
                 configuredMaxMergeCount
             );
-            InternalEngine.this.activateThrottling();
+            // System.out.println("Activate throttling enableIndexingThrottling");
+                InternalEngine.this.activateThrottling();
+
         }
 
         @Override
@@ -2931,7 +2945,9 @@ protected synchronized void disableIndexingThrottling(int numRunningMerges, int
                 numQueuedMerges,
                 configuredMaxMergeCount
             );
+            // System.out.println("DeActivate throttling disableIndexingThrottling");
             InternalEngine.this.deactivateThrottling();
+
         }
 
         @Override
@@ -2956,21 +2972,28 @@ private final class EngineConcurrentMergeScheduler extends ElasticsearchConcurre
         @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();
+            synchronized (isThrottling) {
+                if (numMergesInFlight.incrementAndGet() > maxNumMerges) {
+                    if (isThrottling.getAndSet(true) == false) {
+                        logger.info("now throttling indexing: numMergesInFlight={}, maxNumMerges={}", numMergesInFlight, maxNumMerges);
+                        // System.out.println("Activate throttling beforeMerge");
+                        activateThrottling();
+                    }
                 }
             }
         }
 
         @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();
+            synchronized (isThrottling) {
+                if (numMergesInFlight.decrementAndGet() <= maxNumMerges) {
+                    if (isThrottling.getAndSet(false)) {
+                        logger.info("stop throttling indexing: numMergesInFlight={}, maxNumMerges={}", numMergesInFlight, maxNumMerges);
+                        // System.out.println("DeActivate throttling afterMerge");
+                        deactivateThrottling();
+                    }
+
                 }
             }
             maybeFlushAfterMerge(merge);

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

@@ -507,6 +507,12 @@ public void activateThrottling() {}
     @Override
     public void deactivateThrottling() {}
 
+    @Override
+    public void pauseThrottling() {}
+
+    @Override
+    public void unPauseThrottling() {}
+
     @Override
     public void trimUnreferencedTranslogFiles() {}
 

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

@@ -189,13 +189,14 @@ MergeTask newMergeTask(MergeSource mergeSource, MergePolicy.OneMerge merge, Merg
         );
     }
 
-    private void checkMergeTaskThrottling() {
+    private synchronized void checkMergeTaskThrottling() {
         long submittedMergesCount = submittedMergeTaskCount.get();
         long doneMergesCount = doneMergeTaskCount.get();
         int runningMergesCount = runningMergeTasks.size();
         int configuredMaxMergeCount = config.getMaxMergeCount();
         // both currently running and enqueued merge tasks are considered "active" for throttling purposes
         int activeMerges = (int) (submittedMergesCount - doneMergesCount);
+
         if (activeMerges > configuredMaxMergeCount
             // only throttle indexing if disk IO is un-throttled, and we still can't keep up with the merge load
             && threadPoolMergeExecutorService.usingMaxTargetIORateBytesPerSec()

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -2752,6 +2752,19 @@ public void deactivateThrottling() {
         }
     }
 
+    public void pauseThrottling() {
+        Engine engine = getEngineOrNull();
+        final boolean throttled;
+        if (engine == null) {
+            throttled = false;
+        } else {
+            throttled = engine.isThrottled();
+        }
+        if (throttled) {
+            engine.pauseThrottling();
+        }
+    }
+
     private void handleRefreshException(Exception e) {
         if (e instanceof AlreadyClosedException) {
             // ignore

server/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

@@ -85,8 +85,13 @@ public void blockOperations(
         final long timeout,
         final TimeUnit timeUnit,
         final Executor executor
+        //@Nullable IndexShard indexShard
     ) {
         delayOperations();
+        // If indexing is paused on the shard, unpause it so that any currently waiting task can
+        // go ahead and release the indexing permit it holds.
+        //if(indexShard && indexShard)
+
         waitUntilBlocked(ActionListener.assertOnce(onAcquired), timeout, timeUnit, executor);
     }