simons patch + 4.8.1 snapshot

Author: rmuir

pom.xml

@@ -31,7 +31,7 @@
     </parent>
 
     <properties>
-        <lucene.version>4.8.0</lucene.version>
+        <lucene.version>4.8.1</lucene.version>
         <tests.jvms>auto</tests.jvms>
         <tests.shuffle>true</tests.shuffle>
         <tests.output>onerror</tests.output>
@@ -47,6 +47,10 @@
             <id>Codehaus Snapshots</id>
             <url>http://repository.codehaus.org/</url>
         </repository>
+        <repository>
+            <id>Release Candidate</id>
+            <url>http://people.apache.org/~rmuir/fake_staging_area/lucene-solr-4.8.1-r1592647/lucene/maven/</url>
+        </repository>
     </repositories>
 
     <dependencies>

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

@@ -163,6 +163,8 @@ public class InternalEngine extends AbstractIndexShardComponent implements Engin
 
     private SegmentInfos lastCommittedSegmentInfos;
 
+    private IndexThrottle throttle;
+
     @Inject
     public InternalEngine(ShardId shardId, @IndexSettings Settings indexSettings, ThreadPool threadPool,
                           IndexSettingsService indexSettingsService, ShardIndexingService indexingService, @Nullable IndicesWarmer warmer,
@@ -257,6 +259,10 @@ public void start() throws EngineException {
             }
             try {
                 this.indexWriter = createWriter();
+                mergeScheduler.removeListener(this.throttle);
+                //nocommit find a good value for this?
+                this.throttle = new IndexThrottle(ConcurrentMergeScheduler.DEFAULT_MAX_MERGE_COUNT, indexWriter);
+                mergeScheduler.addListener(throttle);
             } catch (IOException e) {
                 throw new EngineCreationFailureException(shardId, "failed to create engine", e);
             }
@@ -373,7 +379,9 @@ public void create(Create create) throws EngineException {
             if (writer == null) {
                 throw new EngineClosedException(shardId, failedEngine);
             }
-            innerCreate(create, writer);
+            try (Releasable r = throttle.acquireThrottle()) {
+                innerCreate(create, writer);
+            }
             dirty = true;
             possibleMergeNeeded = true;
             flushNeeded = true;
@@ -462,8 +470,9 @@ public void index(Index index) throws EngineException {
             if (writer == null) {
                 throw new EngineClosedException(shardId, failedEngine);
             }
-
-            innerIndex(index, writer);
+            try (Releasable r = throttle.acquireThrottle()) {
+                innerIndex(index, writer);
+            }
             dirty = true;
             possibleMergeNeeded = true;
             flushNeeded = true;
@@ -744,7 +753,10 @@ public void flush(Flush flush) throws EngineException {
                         // to be allocated to a different node
                         currentIndexWriter().close(false);
                         indexWriter = createWriter();
-
+                        mergeScheduler.removeListener(this.throttle);
+                        //nocommit find a good value for this?
+                        this.throttle = new IndexThrottle(ConcurrentMergeScheduler.DEFAULT_MAX_MERGE_COUNT, indexWriter);
+                        mergeScheduler.addListener(throttle);
                         // commit on a just opened writer will commit even if there are no changes done to it
                         // we rely on that for the commit data translog id key
                         if (flushNeeded || flush.force()) {
@@ -1559,4 +1571,46 @@ boolean assertLockIsHeld() {
             return aBoolean != null && aBoolean.booleanValue();
         }
     }
+
+    private static final class IndexThrottle implements MergeSchedulerProvider.Listener {
+
+        private final IndexWriter writer;
+        private volatile InternalLock lock;
+        private final InternalLock lockReference = new InternalLock(new ReentrantLock());
+        private final AtomicInteger numMergesInFlight = new AtomicInteger(0);
+        private final int maxNumMerges;
+
+        private static final Releasable DUMMY = new Releasable() {
+            @Override
+            public void close() throws ElasticsearchException {}
+        };
+
+        public IndexThrottle(int maxNumMerges, IndexWriter writer) {
+            this.maxNumMerges = maxNumMerges;
+            this.writer = writer;
+        }
+
+        public Releasable acquireThrottle() {
+           final InternalLock lock = this.lock;
+           if (lock == null) {
+               return DUMMY;
+           }
+           return lock.acquire();
+        }
+
+
+        @Override
+        public void beforeMerge(OnGoingMerge merge) {
+            if (numMergesInFlight.incrementAndGet() > maxNumMerges && writer.hasPendingMerges()) {
+               lock = lockReference;
+            }
+        }
+
+        @Override
+        public void afterMerge(OnGoingMerge merge) {
+            if (numMergesInFlight.decrementAndGet() <= maxNumMerges && !writer.hasPendingMerges()) {
+                lock = null;
+            }
+        }
+    }
 }