LUCENE-4158: leave stall control on each wakeup call to prevent full flush starvation due to queued flushes

git-svn-id: https://svn.apache.org/repos/asf/lucene/dev/trunk@1353433 13f79535-47bb-0310-9956-ffa450edef68

lucene/core/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java

@@ -633,7 +633,13 @@ final class DocumentsWriterFlushControl  {
    * checked out DWPT are available
    */
   void waitIfStalled() {
-      stallControl.waitIfStalled();
+    if (documentsWriter.infoStream.isEnabled("DWFC")) {
+      documentsWriter.infoStream.message("DWFC",
+          "waitIfStalled: numFlushesPending: " + flushQueue.size()
+              + " netBytes: " + netBytes() + " flushBytes: " + flushBytes()
+              + " fullFlush: " + fullFlush);
+    }
+    stallControl.waitIfStalled();
   }
 
   /**

lucene/core/src/java/org/apache/lucene/index/DocumentsWriterStallControl.java

@@ -67,17 +67,16 @@ final class DocumentsWriterStallControl {
   void waitIfStalled() {
     if (stalled) {
       synchronized (this) {
-        boolean hasWaited = false;
+        if (stalled) { // react on the first wakeup call!
-        while (stalled) {
+          // don't loop here, higher level logic will re-stall!
           try {
-            assert hasWaited || incWaiters();
+            assert incWaiters();
-            assert (hasWaited = true);
             wait();
+            assert  decrWaiters();
           } catch (InterruptedException e) {
             throw new ThreadInterruptedException(e);
           }
         }
-        assert !hasWaited || decrWaiters();
       }
     }
   }

lucene/core/src/test/org/apache/lucene/index/TestDocumentsWriterStallControl.java

@@ -63,7 +63,6 @@ public class TestDocumentsWriterStallControl extends LuceneTestCase {
     
     Thread[] stallThreads = new Thread[atLeast(3)];
     for (int i = 0; i < stallThreads.length; i++) {
-      final int threadId = i;
       final int stallProbability = 1 +random().nextInt(10);
       stallThreads[i] = new Thread() {
         public void run() {