Set DWPT inactive on IW close

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

lucene/src/java/org/apache/lucene/index/DocumentsWriter.java

@@ -586,4 +586,20 @@ final class DocumentsWriter {
       return (!isSegmentFlush || segment != null);  
     }
   }
+  
+  // use by IW during close to assert all DWPT are inactive after final flush
+  boolean assertNoActiveDWPT() {
+    Iterator<ThreadState> activePerThreadsIterator = perThreadPool.getAllPerThreadsIterator();
+    while(activePerThreadsIterator.hasNext()) {
+      ThreadState next = activePerThreadsIterator.next();
+      next.lock();
+      try {
+        assert !next.isActive();
+      } finally  {
+        next.unlock();
+      }
+    }
+    return true;
+  }
+ 
 }

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

@@ -304,6 +304,7 @@ public final class DocumentsWriterFlushControl {
   synchronized void setClosed() {
     // set by DW to signal that we should not release new DWPT after close
     this.closed = true;
+    perThreadPool.deactivateUnreleasedStates();
   }
 
   /**
@@ -386,8 +387,12 @@ public final class DocumentsWriterFlushControl {
             toFlush.add(flushingDWPT);
           }
         } else {
-          // get the new delete queue from DW
-          next.perThread.initialize();
+          if (closed) {
+            next.resetWriter(null); // make this state inactive
+          } else {
+            // get the new delete queue from DW
+            next.perThread.initialize();
+          }
         }
       } finally {
         next.unlock();
@@ -522,5 +527,4 @@ public final class DocumentsWriterFlushControl {
   boolean anyStalledThreads() {
     return stallControl.anyStalledThreads();
   }
- 
 }

lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThreadPool.java

@@ -193,6 +193,21 @@ public abstract class DocumentsWriterPerThreadPool {
     return null;
   }
   
+  /**
+   * Deactivate all unreleased threadstates 
+   */
+  protected synchronized void deactivateUnreleasedStates() {
+    for (int i = numThreadStatesActive; i < perThreads.length; i++) {
+      final ThreadState threadState = perThreads[i];
+      threadState.lock();
+      try {
+        threadState.resetWriter(null);
+      } finally {
+        threadState.unlock();
+      }
+    }
+  }
+  
   protected DocumentsWriterPerThread replaceForFlush(ThreadState threadState, boolean closed) {
     assert threadState.isHeldByCurrentThread();
     final DocumentsWriterPerThread dwpt = threadState.perThread;

lucene/src/java/org/apache/lucene/index/IndexWriter.java

@@ -1073,7 +1073,8 @@ public class IndexWriter implements Closeable {
 
       if (infoStream != null)
         message("at close: " + segString());
-
+      // used by assert below
+      final DocumentsWriter oldWriter = docWriter;
       synchronized(this) {
         readerPool.close();
         docWriter = null;
@@ -1087,6 +1088,7 @@ public class IndexWriter implements Closeable {
       synchronized(this) {
         closed = true;
       }
+      assert oldWriter.assertNoActiveDWPT();
     } catch (OutOfMemoryError oom) {
       handleOOM(oom, "closeInternal");
     } finally {
@@ -1101,6 +1103,8 @@ public class IndexWriter implements Closeable {
     }
   }
   
+ 
+
   /** Returns the Directory used by this index. */
   public Directory getDirectory() {
     // Pass false because the flush during closing calls getDirectory