commit

Author: ankikuma

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

@@ -3235,6 +3235,16 @@ public void noopUpdate() {
         internalIndexingStats.noopUpdate();
     }
 
+    /**
+     * Increment relevant stats when indexing buffers are written to disk using indexing threads,
+     * in order to apply back-pressure on indexing.
+     * @param took  time it took to write the index buffers for this shard
+     * @see org.elasticsearch.indices.IndexingMemoryController
+     */
+    public void writeIndexBuffersOnIndexThreads(long took) {
+        internalIndexingStats.writeIndexBuffers(took);
+    }
+
     public void maybeCheckIndex() {
         recoveryState.setStage(RecoveryState.Stage.VERIFY_INDEX);
         if (Booleans.isTrue(checkIndexOnStartup) || "checksum".equals(checkIndexOnStartup)) {

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

@@ -153,6 +153,17 @@ void noopUpdate() {
         totalStats.noopUpdates.inc();
     }
 
+    /**
+     * Increment relevant stats when indexing buffers are written to disk using indexing threads,
+     * in order to apply back-pressure on indexing.
+     * @param took   time taken to write buffers
+     * @see org.elasticsearch.indices.IndexingMemoryController
+     */
+    void writeIndexBuffers(long took) {
+        totalStats.indexMetric.inc(took);
+        totalStats.recentIndexMetric.addIncrement(took, relativeTimeInNanosSupplier.getAsLong());
+    }
+
     static class StatsHolder {
         private final MeanMetric indexMetric = new MeanMetric(); // Used for the count and total 'took' time (in ns) of index operations
         private final ExponentiallyWeightedMovingRate recentIndexMetric; // An EWMR of the total 'took' time of index operations (in ns)

server/src/main/java/org/elasticsearch/indices/IndexingMemoryController.java

@@ -205,11 +205,16 @@ protected void enqueueWriteIndexingBuffer(IndexShard shard) {
      */
     private boolean writePendingIndexingBuffers() {
         boolean wrotePendingIndexingBuffer = false;
+        long startTime;
+        long took;
         for (IndexShard shard = pendingWriteIndexingBufferQueue.pollFirst(); shard != null; shard = pendingWriteIndexingBufferQueue
             .pollFirst()) {
             // Remove the shard from the set first, so that multiple threads can run writeIndexingBuffer concurrently on the same shard.
             pendingWriteIndexingBufferSet.remove(shard);
+            startTime = System.nanoTime();
             shard.writeIndexingBuffer();
+            took = System.nanoTime() - startTime;
+            shard.writeIndexBuffersOnIndexThreads(took);
             wrotePendingIndexingBuffer = true;
         }
         return wrotePendingIndexingBuffer;
@@ -258,6 +263,7 @@ private void postOperation(ShardId shardId, Engine.Operation operation, Engine.R
         // be reclaimed rapidly. This has the downside of increasing the latency of _bulk requests though. Lucene does the same thing in
         // DocumentsWriter#postUpdate, flushing a segment because the size limit on the RAM buffer was reached happens on the call to
         // IndexWriter#addDocument.
+
         while (writePendingIndexingBuffers()) {
             // If we just wrote segments, then run the checker again if not already running to check if we released enough memory.
             if (statusChecker.tryRun() == false) {