Index: src/test/org/apache/lucene/index/TestIndexWriter.java
===================================================================
--- src/test/org/apache/lucene/index/TestIndexWriter.java	(revision 688090)
+++ src/test/org/apache/lucene/index/TestIndexWriter.java	(working copy)
@@ -22,6 +22,7 @@
 import java.io.File;
 import java.util.Arrays;
 import java.util.ArrayList;
+import java.util.List;
 import java.util.Random;
 
 import org.apache.lucene.util.LuceneTestCase;
@@ -125,7 +126,7 @@
         writer.close();
     }
 
-    private void addDoc(IndexWriter writer) throws IOException
+    private static void addDoc(IndexWriter writer) throws IOException
     {
         Document doc = new Document();
         doc.add(new Field("content", "aaa", Field.Store.NO, Field.Index.TOKENIZED));
@@ -437,6 +438,10 @@
                        (dir.getMaxUsedSizeInBytes()-startDiskUsage) < 2*(startDiskUsage + inputDiskUsage));
           }
 
+          // Make sure we don't hit disk full during close below:
+          dir.setMaxSizeInBytes(0);
+          dir.setRandomIOExceptionRate(0.0, 0);
+
           writer.close();
 
           // Wait for all BG threads to finish else
@@ -2338,6 +2343,9 @@
       IndexWriter writer = new IndexWriter(dir, new WhitespaceAnalyzer(), IndexWriter.MaxFieldLength.LIMITED);
       ConcurrentMergeScheduler cms = new ConcurrentMergeScheduler();
 
+      // We expect AlreadyClosedException
+      cms.setSuppressExceptions();
+
       writer.setMergeScheduler(cms);
       writer.setMaxBufferedDocs(10);
       writer.setMergeFactor(4);
@@ -2891,7 +2899,7 @@
       writer.setMergeScheduler(new SerialMergeScheduler());
       writer.setMergePolicy(new LogDocMergePolicy());
 
-      Directory[] indexDirs = { dir};
+      Directory[] indexDirs = {new MockRAMDirectory(dir)};
       writer.addIndexes(indexDirs);
       writer.close();
     }
@@ -3732,6 +3740,278 @@
     dir.close();
   }
 
+  private abstract static class RunAddIndexesThreads {
+
+    Directory dir, dir2;
+    final static int NUM_INIT_DOCS = 17;
+    IndexWriter writer2;
+    final List failures = new ArrayList();
+    volatile boolean didClose;
+    final IndexReader[] readers;
+    final int NUM_COPY;
+    final static int NUM_THREADS = 5;
+    final Thread[] threads = new Thread[NUM_THREADS];
+    final ConcurrentMergeScheduler cms;
+
+    public RunAddIndexesThreads(int numCopy) throws Throwable {
+      NUM_COPY = numCopy;
+      dir = new MockRAMDirectory();
+      IndexWriter writer = new IndexWriter(dir, false, new WhitespaceAnalyzer(), IndexWriter.MaxFieldLength.LIMITED);
+      writer.setMaxBufferedDocs(2);
+      for (int i = 0; i < NUM_INIT_DOCS; i++)
+        addDoc(writer);
+      writer.close();
+
+      dir2 = new MockRAMDirectory();
+      writer2 = new IndexWriter(dir2, false, new WhitespaceAnalyzer(), IndexWriter.MaxFieldLength.LIMITED);
+      cms = (ConcurrentMergeScheduler) writer2.getMergeScheduler();
+
+      readers = new IndexReader[NUM_COPY];
+      for(int i=0;i<NUM_COPY;i++)
+        readers[i] = IndexReader.open(dir);
+    }
+
+    void launchThreads(final int numIter) {
+
+      for(int i=0;i<NUM_THREADS;i++) {
+        threads[i] = new Thread() {
+            public void run() {
+              try {
+
+                final Directory[] dirs = new Directory[NUM_COPY];
+                for(int k=0;k<NUM_COPY;k++)
+                  dirs[k] = new MockRAMDirectory(dir);
+
+                int j=0;
+
+                while(true) {
+                  // System.out.println(Thread.currentThread().getName() + ": iter j=" + j);
+                  if (numIter > 0 && j == numIter)
+                    break;
+                  doBody(j++, dirs);
+                }
+              } catch (Throwable t) {
+                handle(t);
+              }
+            }
+          };
+      }
+
+      for(int i=0;i<NUM_THREADS;i++)
+        threads[i].start();
+    }
+
+    void joinThreads() {
+      for(int i=0;i<NUM_THREADS;i++)
+        try {
+          threads[i].join();
+        } catch (InterruptedException ie) {
+          Thread.currentThread().interrupt();
+        }
+    }
+
+    void close(boolean doWait) throws Throwable {
+      didClose = true;
+      writer2.close(doWait);
+    }
+
+    void closeDir() throws Throwable {
+      for(int i=0;i<NUM_COPY;i++)
+        readers[i].close();
+      dir2.close();
+    }
+
+    abstract void doBody(int j, Directory[] dirs) throws Throwable;
+    abstract void handle(Throwable t);
+  }
+
+  private class CommitAndAddIndexes extends RunAddIndexesThreads {
+    public CommitAndAddIndexes(int numCopy) throws Throwable {
+      super(numCopy);
+    }
+
+    void handle(Throwable t) {
+      t.printStackTrace(System.out);
+      synchronized(failures) {
+        failures.add(t);
+      }
+    }
+
+    void doBody(int j, Directory[] dirs) throws Throwable {
+      switch(j%4) {
+      case 0:
+        writer2.addIndexes(dirs);
+        break;
+      case 1:
+        writer2.addIndexesNoOptimize(dirs);
+        break;
+      case 2:
+        writer2.addIndexes(readers);
+        break;
+      case 3:
+        writer2.commit();
+      }
+    }
+  }
+
+  // LUCENE-1335: test simultaneous addIndexes & commits
+  // from multiple threads
+  public void testAddIndexesWithThreads() throws Throwable {
+
+    final int NUM_ITER = 12;
+    final int NUM_COPY = 3;
+    CommitAndAddIndexes c = new CommitAndAddIndexes(NUM_COPY);
+    c.launchThreads(NUM_ITER);
+
+    for(int i=0;i<100;i++)
+      addDoc(c.writer2);
+
+    c.joinThreads();
+
+    assertEquals(100+NUM_COPY*(3*NUM_ITER/4)*c.NUM_THREADS*c.NUM_INIT_DOCS, c.writer2.numDocs());
+
+    c.close(true);
+
+    assertTrue(c.failures.size() == 0);
+
+    _TestUtil.checkIndex(c.dir2);
+
+    IndexReader reader = IndexReader.open(c.dir2);
+    assertEquals(100+NUM_COPY*(3*NUM_ITER/4)*c.NUM_THREADS*c.NUM_INIT_DOCS, reader.numDocs());
+    reader.close();
+
+    c.closeDir();
+  }
+
+  private class CommitAndAddIndexes2 extends CommitAndAddIndexes {
+    public CommitAndAddIndexes2(int numCopy) throws Throwable {
+      super(numCopy);
+    }
+
+    void handle(Throwable t) {
+      if (!(t instanceof AlreadyClosedException) && !(t instanceof NullPointerException)) {
+        t.printStackTrace(System.out);
+        synchronized(failures) {
+          failures.add(t);
+        }
+      }
+    }
+  }
+
+  // LUCENE-1335: test simultaneous addIndexes & close
+  public void testAddIndexesWithClose() throws Throwable {
+    final int NUM_COPY = 3;
+    CommitAndAddIndexes2 c = new CommitAndAddIndexes2(NUM_COPY);
+    //c.writer2.setInfoStream(System.out);
+    c.launchThreads(-1);
+
+    // Close w/o first stopping/joining the threads
+    c.close(true);
+    //c.writer2.close();
+
+    c.joinThreads();
+
+    _TestUtil.checkIndex(c.dir2);
+
+    c.closeDir();
+
+    assertTrue(c.failures.size() == 0);
+  }
+
+  private class CommitAndAddIndexes3 extends RunAddIndexesThreads {
+    public CommitAndAddIndexes3(int numCopy) throws Throwable {
+      super(numCopy);
+    }
+
+    void doBody(int j, Directory[] dirs) throws Throwable {
+      switch(j%5) {
+      case 0:
+        writer2.addIndexes(dirs);
+        break;
+      case 1:
+        writer2.addIndexesNoOptimize(dirs);
+        break;
+      case 2:
+        writer2.addIndexes(readers);
+        break;
+      case 3:
+        writer2.optimize();
+      case 4:
+        writer2.commit();
+      }
+    }
+
+    void handle(Throwable t) {
+      boolean report = true;
+
+      if (t instanceof AlreadyClosedException || t instanceof MergePolicy.MergeAbortedException || t instanceof NullPointerException) {
+        report = !didClose;
+      } else if (t instanceof IOException)  {
+        Throwable t2 = t.getCause();
+        if (t2 instanceof MergePolicy.MergeAbortedException) {
+          report = !didClose;
+        }
+      }
+      if (report) {
+        t.printStackTrace(System.out);
+        synchronized(failures) {
+          failures.add(t);
+        }
+      }
+    }
+  }
+
+  // LUCENE-1335: test simultaneous addIndexes & close
+  public void testAddIndexesWithCloseNoWait() throws Throwable {
+
+    final int NUM_COPY = 50;
+    CommitAndAddIndexes3 c = new CommitAndAddIndexes3(NUM_COPY);
+    c.launchThreads(-1);
+
+    try {
+      Thread.sleep(500);
+    } catch (InterruptedException ie) {
+      Thread.currentThread().interrupt();
+    }
+
+    // Close w/o first stopping/joining the threads
+    c.close(false);
+
+    c.joinThreads();
+
+    _TestUtil.checkIndex(c.dir2);
+
+    c.closeDir();
+
+    assertTrue(c.failures.size() == 0);
+  }
+
+  // LUCENE-1335: test simultaneous addIndexes & close
+  public void testAddIndexesWithRollback() throws Throwable {
+    
+    final int NUM_COPY = 50;
+    CommitAndAddIndexes3 c = new CommitAndAddIndexes3(NUM_COPY);
+    c.launchThreads(-1);
+
+    try {
+      Thread.sleep(500);
+    } catch (InterruptedException ie) {
+      Thread.currentThread().interrupt();
+    }
+
+    // Close w/o first stopping/joining the threads
+    c.didClose = true;
+    c.writer2.rollback();
+
+    c.joinThreads();
+
+    _TestUtil.checkIndex(c.dir2);
+
+    c.closeDir();
+
+    assertTrue(c.failures.size() == 0);
+  }
+
   // LUCENE-1347
   public class MockIndexWriter4 extends IndexWriter {
 
Index: src/test/org/apache/lucene/index/TestAddIndexesNoOptimize.java
===================================================================
--- src/test/org/apache/lucene/index/TestAddIndexesNoOptimize.java	(revision 688090)
+++ src/test/org/apache/lucene/index/TestAddIndexesNoOptimize.java	(working copy)
@@ -340,7 +340,7 @@
     writer.setMaxBufferedDocs(10);
     writer.setMergeFactor(4);
 
-    writer.addIndexesNoOptimize(new Directory[] { aux, aux });
+    writer.addIndexesNoOptimize(new Directory[] { aux, new RAMDirectory(aux) });
     assertEquals(1060, writer.docCount());
     assertEquals(1000, writer.getDocCount(0));
     writer.close();
@@ -369,7 +369,7 @@
     writer.setMaxBufferedDocs(4);
     writer.setMergeFactor(4);
 
-    writer.addIndexesNoOptimize(new Directory[] { aux, aux });
+    writer.addIndexesNoOptimize(new Directory[] { aux, new RAMDirectory(aux) });
     assertEquals(1020, writer.docCount());
     assertEquals(1000, writer.getDocCount(0));
     writer.close();
Index: src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java
===================================================================
--- src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java	(revision 688090)
+++ src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java	(working copy)
@@ -187,24 +187,15 @@
 
         message("  consider merge " + merge.segString(dir));
       
-        if (merge.isExternal) {
-          message("    merge involves segments from an external directory; now run in foreground");
-        } else {
-          assert mergeThreadCount() < maxThreadCount;
+        assert mergeThreadCount() < maxThreadCount;
 
-          // OK to spawn a new merge thread to handle this
-          // merge:
-          final MergeThread merger = getMergeThread(writer, merge);
-          mergeThreads.add(merger);
-          message("    launch new thread [" + merger.getName() + "]");
-          merger.start();
-          continue;
-        }
+        // OK to spawn a new merge thread to handle this
+        // merge:
+        final MergeThread merger = getMergeThread(writer, merge);
+        mergeThreads.add(merger);
+        message("    launch new thread [" + merger.getName() + "]");
+        merger.start();
       }
-
-      // This merge involves segments outside our index
-      // Directory so we must merge in foreground
-      doMerge(merge);
     }
   }
 
Index: src/java/org/apache/lucene/index/SegmentInfos.java
===================================================================
--- src/java/org/apache/lucene/index/SegmentInfos.java	(revision 688090)
+++ src/java/org/apache/lucene/index/SegmentInfos.java	(working copy)
@@ -30,8 +30,7 @@
 import java.util.Vector;
 
 final class SegmentInfos extends Vector {
-  
-  
+
   /** The file format version, a negative number. */
   /* Works since counter, the old 1st entry, is always >= 0 */
   public static final int FORMAT = -1;
@@ -826,4 +825,19 @@
     prepareCommit(dir);
     finishCommit(dir);
   }
+
+  synchronized String segString(Directory directory) {
+    StringBuffer buffer = new StringBuffer();
+    final int count = size();
+    for(int i = 0; i < count; i++) {
+      if (i > 0) {
+        buffer.append(' ');
+      }
+      final SegmentInfo info = info(i);
+      buffer.append(info.segString(directory));
+      if (info.dir != directory)
+        buffer.append("**");
+    }
+    return buffer.toString();
+  }
 }
Index: src/java/org/apache/lucene/index/IndexWriter.java
===================================================================
--- src/java/org/apache/lucene/index/IndexWriter.java	(revision 688090)
+++ src/java/org/apache/lucene/index/IndexWriter.java	(working copy)
@@ -300,7 +300,7 @@
 
   private Similarity similarity = Similarity.getDefault(); // how to normalize
 
-  private volatile long changeCount; // increments every a change is completed
+  private volatile long changeCount; // increments every time a change is completed
   private long lastCommitChangeCount; // last changeCount that was committed
 
   private SegmentInfos rollbackSegmentInfos;      // segmentInfos we will fallback to if the commit fails
@@ -345,18 +345,58 @@
   private int flushDeletesCount;
   private double maxSyncPauseSeconds = DEFAULT_MAX_SYNC_PAUSE_SECONDS;
 
+  // Used to only allow one addIndexes to proceed at once
+  // TODO: use ReadWriteLock once we are on 5.0
+  private int readCount;                          // count of how many threads are holding read lock
+  private Thread writeThread;                     // non-null if any thread holds write lock
+    
+  synchronized void acquireWrite() {
+    while(writeThread != null || readCount > 0)
+      doWait();
+
+    // We could have been closed while we were waiting:
+    ensureOpen();
+
+    writeThread = Thread.currentThread();
+  }
+
+  synchronized void releaseWrite() {
+    assert Thread.currentThread() == writeThread;
+    writeThread = null;
+    notifyAll();
+  }
+
+  synchronized void acquireRead() {
+    final Thread current = Thread.currentThread();
+    while(writeThread != null && writeThread != current)
+      doWait();
+
+    readCount++;
+  }
+
+  synchronized void releaseRead() {
+    readCount--;
+    assert readCount >= 0;
+    if (0 == readCount)
+      notifyAll();
+  }
+
   /**
    * Used internally to throw an {@link
    * AlreadyClosedException} if this IndexWriter has been
    * closed.
    * @throws AlreadyClosedException if this IndexWriter is
    */
-  protected final void ensureOpen() throws AlreadyClosedException {
-    if (closed) {
+  protected synchronized final void ensureOpen(boolean includePendingClose) throws AlreadyClosedException {
+    if (closed || (includePendingClose && closing)) {
       throw new AlreadyClosedException("this IndexWriter is closed");
     }
   }
 
+  protected synchronized final void ensureOpen() throws AlreadyClosedException {
+    ensureOpen(true);
+  }
+
   /**
    * Prints a message to the infoStream (if non-null),
    * prefixed with the identifying information for this
@@ -469,7 +509,8 @@
    * @see #setTermIndexInterval(int)
    */
   public int getTermIndexInterval() {
-    ensureOpen();
+    // We pass false because this method is called by SegmentMerger while we are in the process of closing
+    ensureOpen(false);
     return termIndexInterval;
   }
 
@@ -1127,7 +1168,7 @@
       }
 
       this.autoCommit = autoCommit;
-      setRollbackSegmentInfos();
+      setRollbackSegmentInfos(segmentInfos);
 
       docWriter = new DocumentsWriter(directory, this);
       docWriter.setInfoStream(infoStream);
@@ -1153,8 +1194,9 @@
     }
   }
 
-  private void setRollbackSegmentInfos() {
-    rollbackSegmentInfos = (SegmentInfos) segmentInfos.clone();
+  private synchronized void setRollbackSegmentInfos(SegmentInfos infos) {
+    rollbackSegmentInfos = (SegmentInfos) infos.clone();
+    assert !hasExternalSegments(rollbackSegmentInfos);
     rollbackSegments = new HashMap();
     final int size = rollbackSegmentInfos.size();
     for(int i=0;i<size;i++)
@@ -1189,7 +1231,7 @@
   /**
    * Expert: set the merge scheduler used by this writer.
    */
-  public void setMergeScheduler(MergeScheduler mergeScheduler) throws CorruptIndexException, IOException {
+  synchronized public void setMergeScheduler(MergeScheduler mergeScheduler) throws CorruptIndexException, IOException {
     ensureOpen();
     if (mergeScheduler == null)
       throw new NullPointerException("MergeScheduler must be non-null");
@@ -1645,10 +1687,7 @@
           // Another thread is presently trying to close;
           // wait until it finishes one way (closes
           // successfully) or another (fails to close)
-          try {
-            wait();
-          } catch (InterruptedException ie) {
-          }
+          doWait();
         }
       } else
         return false;
@@ -1656,6 +1695,9 @@
   }
 
   private void closeInternal(boolean waitForMerges) throws CorruptIndexException, IOException {
+
+    docWriter.pauseAllThreads();
+
     try {
       if (infoStream != null)
         message("now flush at close");
@@ -1707,8 +1749,12 @@
       synchronized(this) {
         closing = false;
         notifyAll();
-        if (!closed && infoStream != null)
-          message("hit exception while closing");
+        if (!closed) {
+          if (docWriter != null)
+            docWriter.resumeAllThreads();
+          if (infoStream != null)
+            message("hit exception while closing");
+        }
       }
     }
   }
@@ -1792,8 +1838,9 @@
   }
 
   /** Returns the Directory used by this index. */
-  public Directory getDirectory() {
-    ensureOpen();
+  public Directory getDirectory() {     
+    // Pass false because the flush during closing calls getDirectory
+    ensureOpen(false);
     return directory;
   }
 
@@ -2287,11 +2334,7 @@
     if (doWait) {
       synchronized(this) {
         while(optimizeMergesPending()) {
-          try {
-            wait();
-          } catch (InterruptedException ie) {
-            Thread.currentThread().interrupt();
-          }
+          doWait();
 
           if (mergeExceptions.size() > 0) {
             // Forward any exceptions in background merge
@@ -2310,6 +2353,12 @@
           }
         }
       }
+
+      // If close is called while we are still
+      // running, throw an exception so the calling
+      // thread will know the optimize did not
+      // complete
+      ensureOpen();
     }
 
     // NOTE: in the ConcurrentMergeScheduler case, when
@@ -2364,6 +2413,9 @@
         boolean running = true;
         while(running) {
 
+          // Check each merge that MergePolicy asked us to
+          // do, to see if any of them are still running and
+          // if any of them have hit an exception.
           running = false;
           for(int i=0;i<numMerges;i++) {
             final MergePolicy.OneMerge merge = (MergePolicy.OneMerge) spec.merges.get(i);
@@ -2377,13 +2429,9 @@
             }
           }
 
-          if (running) {
-            try {
-              wait();
-            } catch (InterruptedException ie) {
-              Thread.currentThread().interrupt();
-            }
-          }
+          // If any of our merges are still running, wait:
+          if (running)
+            doWait();
         }
       }
     }
@@ -2481,6 +2529,28 @@
     }
   }
 
+  /** Like getNextMerge() except only returns a merge if it's
+   *  external. */
+  private synchronized MergePolicy.OneMerge getNextExternalMerge() {
+    if (pendingMerges.size() == 0)
+      return null;
+    else {
+      Iterator it = pendingMerges.iterator();
+      while(it.hasNext()) {
+        // Advance the merge from pending to running
+        MergePolicy.OneMerge merge = (MergePolicy.OneMerge) it.next();
+        if (merge.isExternal) {
+          it.remove();
+          runningMerges.add(merge);
+          return merge;
+        }
+      }
+
+      // All existing merges do not involve external segments
+      return null;
+    }
+  }
+
   /*
    * Begin a transaction.  During a transaction, any segment
    * merges that happen (or ram segments flushed) will not
@@ -2494,33 +2564,66 @@
    * within the transactions, so they must be flushed before the
    * transaction is started.
    */
-  private synchronized void startTransaction() throws IOException {
+  private synchronized void startTransaction(boolean haveWriteLock) throws IOException {
 
-    if (infoStream != null)
-      message("now start transaction");
+    boolean success = false;
+    try {
+      if (infoStream != null)
+        message("now start transaction");
 
-    assert docWriter.getNumBufferedDeleteTerms() == 0 :
-           "calling startTransaction with buffered delete terms not supported";
-    assert docWriter.getNumDocsInRAM() == 0 :
-           "calling startTransaction with buffered documents not supported";
+      assert docWriter.getNumBufferedDeleteTerms() == 0 :
+      "calling startTransaction with buffered delete terms not supported: numBufferedDeleteTerms=" + docWriter.getNumBufferedDeleteTerms();
+      assert docWriter.getNumDocsInRAM() == 0 :
+      "calling startTransaction with buffered documents not supported: numDocsInRAM=" + docWriter.getNumDocsInRAM();
 
-    localRollbackSegmentInfos = (SegmentInfos) segmentInfos.clone();
-    localAutoCommit = autoCommit;
-    localFlushedDocCount = docWriter.getFlushedDocCount();
+      ensureOpen();
 
-    if (localAutoCommit) {
+      // If a transaction is trying to roll back (because
+      // addIndexes hit an exception) then wait here until
+      // that's done:
+      synchronized(this) {
+        while(stopMerges)
+          doWait();
+      }
+      success = true;
+    } finally {
+      // Releaes the write lock if our caller held it, on
+      // hitting an exception
+      if (!success && haveWriteLock)
+        releaseWrite();
+    }
 
-      if (infoStream != null)
-        message("flush at startTransaction");
+    if (!haveWriteLock)
+      acquireWrite();
 
-      flush(true, false, false);
+    success = false;
+    try {
+      localRollbackSegmentInfos = (SegmentInfos) segmentInfos.clone();
 
-      // Turn off auto-commit during our local transaction:
-      autoCommit = false;
-    } else
-      // We must "protect" our files at this point from
-      // deletion in case we need to rollback:
-      deleter.incRef(segmentInfos, false);
+      assert !hasExternalSegments(segmentInfos);
+
+      localAutoCommit = autoCommit;
+      localFlushedDocCount = docWriter.getFlushedDocCount();
+
+      if (localAutoCommit) {
+
+        if (infoStream != null)
+          message("flush at startTransaction");
+
+        flush(true, false, false);
+
+        // Turn off auto-commit during our local transaction:
+        autoCommit = false;
+      } else
+        // We must "protect" our files at this point from
+        // deletion in case we need to rollback:
+        deleter.incRef(segmentInfos, false);
+
+      success = true;
+    } finally {
+      if (!success)
+        finishAddIndexes();
+    }
   }
 
   /*
@@ -2536,6 +2639,12 @@
     autoCommit = localAutoCommit;
     docWriter.setFlushedDocCount(localFlushedDocCount);
 
+    // Must finish merges before rolling back segmentInfos
+    // so merges don't hit exceptions on trying to commit
+    // themselves, don't get files deleted out from under
+    // them, etc:
+    finishMerges(false);
+
     // Keep the same segmentInfos instance but replace all
     // of its SegmentInfo instances.  This is so the next
     // attempt to commit using this instance of IndexWriter
@@ -2544,6 +2653,11 @@
     segmentInfos.addAll(localRollbackSegmentInfos);
     localRollbackSegmentInfos = null;
 
+    // This must come after we rollback segmentInfos, so
+    // that if a commit() kicks off it does not see the
+    // segmentInfos with external segments
+    finishAddIndexes();
+
     // Ask deleter to locate unreferenced files we had
     // created & remove them:
     deleter.checkpoint(segmentInfos, false);
@@ -2552,9 +2666,16 @@
       // Remove the incRef we did in startTransaction:
       deleter.decRef(segmentInfos);
 
+    // Also ask deleter to remove any newly created files
+    // that were never incref'd; this "garbage" is created
+    // when a merge kicks off but aborts part way through
+    // before it had a chance to incRef the files it had
+    // partially created
     deleter.refresh();
-    finishMerges(false);
-    stopMerges = false;
+    
+    notifyAll();
+
+    assert !hasExternalSegments();
   }
 
   /*
@@ -2590,6 +2711,10 @@
       deleter.decRef(localRollbackSegmentInfos);
 
     localRollbackSegmentInfos = null;
+
+    assert !hasExternalSegments();
+
+    finishAddIndexes();
   }
 
   /**
@@ -2626,6 +2751,8 @@
 
     boolean success = false;
 
+    docWriter.pauseAllThreads();
+
     try {
       finishMerges(false);
 
@@ -2651,6 +2778,8 @@
         // once").
         segmentInfos.clear();
         segmentInfos.addAll(rollbackSegmentInfos);
+
+        assert !hasExternalSegments();
         
         docWriter.abort();
 
@@ -2671,6 +2800,7 @@
     } finally {
       synchronized(this) {
         if (!success) {
+          docWriter.resumeAllThreads();
           closing = false;
           notifyAll();
           if (infoStream != null)
@@ -2706,6 +2836,12 @@
         merge.abort();
       }
 
+      // Ensure any running addIndexes finishes.  It's fine
+      // if a new one attempts to start because its merges
+      // will quickly see the stopMerges == true and abort.
+      acquireRead();
+      releaseRead();
+
       // These merges periodically check whether they have
       // been aborted, and stop if so.  We wait here to make
       // sure they all stop.  It should not take very long
@@ -2714,25 +2850,27 @@
       while(runningMerges.size() > 0) {
         if (infoStream != null)
           message("now wait for " + runningMerges.size() + " running merge to abort");
-        try {
-          wait();
-        } catch (InterruptedException ie) {
-          Thread.currentThread().interrupt();
-        }
+        doWait();
       }
 
+      stopMerges = false;
+      notifyAll();
+
       assert 0 == mergingSegments.size();
 
       if (infoStream != null)
         message("all running merges have aborted");
 
     } else {
-      while(pendingMerges.size() > 0 || runningMerges.size() > 0) {
-        try {
-          wait();
-        } catch (InterruptedException ie) {
-        }
-      }
+      // Ensure any running addIndexes finishes.  It's fine
+      // if a new one attempts to start because from our
+      // caller above the call will see that we are in the
+      // process of closing, and will throw an
+      // AlreadyClosedException.
+      acquireRead();
+      releaseRead();
+      while(pendingMerges.size() > 0 || runningMerges.size() > 0)
+        doWait();
       assert 0 == mergingSegments.size();
     }
   }
@@ -2747,6 +2885,31 @@
     deleter.checkpoint(segmentInfos, false);
   }
 
+  private void finishAddIndexes() {
+    releaseWrite();
+  }
+
+  private void blockAddIndexes(boolean includePendingClose) {
+
+    acquireRead();
+
+    boolean success = false;
+    try {
+
+      // Make sure we are still open since we could have
+      // waited quite a while for last addIndexes to finish
+      ensureOpen(includePendingClose);
+      success = true;
+    } finally {
+      if (!success)
+        releaseRead();
+    }
+  }
+
+  private void resumeAddIndexes() {
+    releaseRead();
+  }
+
   /** Merges all segments from an array of indexes into this index.
    *
    * <p>This may be used to parallelize batch indexing.  A large document
@@ -2807,6 +2970,8 @@
     throws CorruptIndexException, IOException {
 
     ensureOpen();
+    
+    noDupDirs(dirs);
 
     // Do not allow add docs or deletes while we are running:
     docWriter.pauseAllThreads();
@@ -2819,18 +2984,20 @@
 
       boolean success = false;
 
-      startTransaction();
+      startTransaction(false);
 
       try {
 
         int docCount = 0;
         synchronized(this) {
+          ensureOpen();
           for (int i = 0; i < dirs.length; i++) {
             SegmentInfos sis = new SegmentInfos();	  // read infos from dir
             sis.read(dirs[i]);
             for (int j = 0; j < sis.size(); j++) {
               final SegmentInfo info = sis.info(j);
               docCount += info.docCount;
+              assert !segmentInfos.contains(info);
               segmentInfos.addElement(info);	  // add each info
             }
           }
@@ -2862,6 +3029,17 @@
     mergeGen++;
   }
 
+  private void noDupDirs(Directory[] dirs) {
+    HashSet dups = new HashSet();
+    for(int i=0;i<dirs.length;i++) {
+      if (dups.contains(dirs[i]))
+        throw new IllegalArgumentException("Directory " + dirs[i] + " appears more than once");
+      if (dirs[i] == directory)
+        throw new IllegalArgumentException("Cannot add directory to itself");
+      dups.add(dirs[i]);
+    }
+  }
+
   /**
    * Merges all segments from an array of indexes into this index.
    * <p>
@@ -2895,6 +3073,8 @@
 
     ensureOpen();
 
+    noDupDirs(dirs);
+
     // Do not allow add docs or deletes while we are running:
     docWriter.pauseAllThreads();
 
@@ -2905,12 +3085,14 @@
 
       boolean success = false;
 
-      startTransaction();
+      startTransaction(false);
 
       try {
 
         int docCount = 0;
         synchronized(this) {
+          ensureOpen();
+
           for (int i = 0; i < dirs.length; i++) {
             if (directory == dirs[i]) {
               // cannot add this index: segments may be deleted in merge before added
@@ -2921,6 +3103,7 @@
             sis.read(dirs[i]);
             for (int j = 0; j < sis.size(); j++) {
               SegmentInfo info = sis.info(j);
+              assert !segmentInfos.contains(info): "dup info dir=" + info.dir + " name=" + info.name;
               docCount += info.docCount;
               segmentInfos.addElement(info); // add each info
             }
@@ -2932,13 +3115,17 @@
 
         maybeMerge();
 
+        ensureOpen();
+
         // If after merging there remain segments in the index
         // that are in a different directory, just copy these
         // over into our index.  This is necessary (before
         // finishing the transaction) to avoid leaving the
         // index in an unusable (inconsistent) state.
-        copyExternalSegments();
+        resolveExternalSegments();
 
+        ensureOpen();
+
         success = true;
 
       } finally {
@@ -2956,47 +3143,85 @@
     }
   }
 
+  private boolean hasExternalSegments() {
+    return hasExternalSegments(segmentInfos);
+  }
+
+  private boolean hasExternalSegments(SegmentInfos infos) {
+    final int numSegments = infos.size();
+    for(int i=0;i<numSegments;i++)
+      if (infos.info(i).dir != directory)
+        return true;
+    return false;
+  }
+
   /* If any of our segments are using a directory != ours
-   * then copy them over.  Currently this is only used by
+   * then we have to either copy them over one by one, merge
+   * them (if merge policy has chosen to) or wait until
+   * currently running merges (in the background) complete.
+   * We don't return until the SegmentInfos has no more
+   * external segments.  Currently this is only used by
    * addIndexesNoOptimize(). */
-  private void copyExternalSegments() throws CorruptIndexException, IOException {
+  private void resolveExternalSegments() throws CorruptIndexException, IOException {
 
     boolean any = false;
 
-    while(true) {
+    boolean done = false;
+
+    while(!done) {
       SegmentInfo info = null;
       MergePolicy.OneMerge merge = null;
       synchronized(this) {
+
+        if (stopMerges)
+          throw new MergePolicy.MergeAbortedException("rollback() was called or addIndexes* hit an unhandled exception");
+
         final int numSegments = segmentInfos.size();
+
+        done = true;
         for(int i=0;i<numSegments;i++) {
           info = segmentInfos.info(i);
           if (info.dir != directory) {
-            merge = new MergePolicy.OneMerge(segmentInfos.range(i, 1+i), info.getUseCompoundFile());
-            break;
+            done = false;
+            final MergePolicy.OneMerge newMerge = new MergePolicy.OneMerge(segmentInfos.range(i, 1+i), info.getUseCompoundFile());
+
+            // Returns true if no running merge conflicts
+            // with this one (and, records this merge as
+            // pending), ie, this segment is not currently
+            // being merged:
+            if (registerMerge(newMerge)) {
+              merge = newMerge;
+
+              // If this segment is not currently being
+              // merged, then advance it to running & run
+              // the merge ourself (below):
+              pendingMerges.remove(merge);
+              runningMerges.add(merge);
+              break;
+            }
           }
         }
+
+        if (!done && merge == null)
+          // We are not yet done (external segments still
+          // exist in segmentInfos), yet, all such segments
+          // are currently "covered" by a pending or running
+          // merge.  We now try to grab any pending merge
+          // that involves external segments:
+          merge = getNextExternalMerge();
+
+        if (!done && merge == null)
+          // We are not yet done, and, all external segments
+          // fall under merges that the merge scheduler is
+          // currently running.  So, we now wait and check
+          // back to see if the merge has completed.
+          doWait();
       }
 
       if (merge != null) {
-        if (registerMerge(merge)) {
-          pendingMerges.remove(merge);
-          runningMerges.add(merge);
-          any = true;
-          merge(merge);
-        } else
-          // This means there is a bug in the
-          // MergeScheduler.  MergeSchedulers in general are
-          // not allowed to run a merge involving segments
-          // external to this IndexWriter's directory in the
-          // background because this would put the index
-          // into an inconsistent state (where segmentInfos
-          // has been written with such external segments
-          // that an IndexReader would fail to load).
-          throw new MergePolicy.MergeException("segment \"" + info.name + " exists in external directory yet the MergeScheduler executed the merge in a separate thread",
-                                               directory);
-      } else
-        // No more external segments
-        break;
+        any = true;
+        merge(merge);
+      }
     }
 
     if (any)
@@ -3034,31 +3259,54 @@
     // Do not allow add docs or deletes while we are running:
     docWriter.pauseAllThreads();
 
+    // We must pre-acquire the write lock here (and not in
+    // startTransaction below) so that no other addIndexes
+    // is allowed to start up after we have flushed &
+    // optimized but before we then start our transaction.
+    // This is because the merging below requires that only
+    // one segment is present in the index:
+    acquireWrite();
+
     try {
-      optimize();					  // start with zero or 1 seg
 
-      final String mergedName = newSegmentName();
-      SegmentMerger merger = new SegmentMerger(this, mergedName, null);
+      boolean success = false;
+      SegmentInfo info = null;
+      String mergedName = null;
+      SegmentMerger merger = null;
 
-      SegmentInfo info;
+      try {
+        flush(true, false, true);
+        optimize();					  // start with zero or 1 seg
+        success = true;
+      } finally {
+        // Take care to release the write lock if we hit an
+        // exception before starting the transaction
+        if (!success)
+          releaseWrite();
+      }
 
-      IndexReader sReader = null;
+      // true means we already have write lock; if this call
+      // hits an exception it will release the write lock:
+      startTransaction(true);
+
       try {
+        mergedName = newSegmentName();
+        merger = new SegmentMerger(this, mergedName, null);
+
+        IndexReader sReader = null;
         synchronized(this) {
-          if (segmentInfos.size() == 1){ // add existing index, if any
+          if (segmentInfos.size() == 1) { // add existing index, if any
             sReader = SegmentReader.get(segmentInfos.info(0));
-            merger.add(sReader);
           }
         }
 
-        for (int i = 0; i < readers.length; i++)      // add new indexes
-          merger.add(readers[i]);
+        try {
+          if (sReader != null)
+            merger.add(sReader);
 
-        boolean success = false;
+          for (int i = 0; i < readers.length; i++)      // add new indexes
+            merger.add(readers[i]);
 
-        startTransaction();
-
-        try {
           int docCount = merger.merge();                // merge 'em
 
           if(sReader != null) {
@@ -3079,43 +3327,61 @@
           success = true;
 
         } finally {
-          if (!success) {
-            if (infoStream != null)
-              message("hit exception in addIndexes during merge");
-
-            rollbackTransaction();
-          } else {
-            commitTransaction();
+          if (sReader != null) {
+            sReader.close();
           }
         }
       } finally {
-        if (sReader != null) {
-          sReader.close();
+        if (!success) {
+          if (infoStream != null)
+            message("hit exception in addIndexes during merge");
+          rollbackTransaction();
+        } else {
+          commitTransaction();
         }
       }
     
       if (mergePolicy instanceof LogMergePolicy && getUseCompoundFile()) {
 
-        boolean success = false;
+        List files = null;
 
-        startTransaction();
+        synchronized(this) {
+          // Must incRef our files so that if another thread
+          // is running merge/optimize, it doesn't delete our
+          // segment's files before we have a change to
+          // finish making the compound file.
+          if (segmentInfos.contains(info)) {
+            files = info.files();
+            deleter.incRef(files);
+          }
+        }
 
-        try {
-          merger.createCompoundFile(mergedName + ".cfs");
-          synchronized(this) {
-            info.setUseCompoundFile(true);
-          }
+        if (files != null) {
+
+          success = false;
+
+          startTransaction(false);
+
+          try {
+            merger.createCompoundFile(mergedName + ".cfs");
+            synchronized(this) {
+              info.setUseCompoundFile(true);
+            }
           
-          success = true;
+            success = true;
           
-        } finally {
-          if (!success) {
-            if (infoStream != null)
-              message("hit exception building compound file in addIndexes during merge");
+          } finally {
 
-            rollbackTransaction();
-          } else {
-            commitTransaction();
+            deleter.decRef(files);
+
+            if (!success) {
+              if (infoStream != null)
+                message("hit exception building compound file in addIndexes during merge");
+
+              rollbackTransaction();
+            } else {
+              commitTransaction();
+            }
           }
         }
       }
@@ -3164,6 +3430,7 @@
    * will internally call prepareCommit.
    */
   public final void prepareCommit() throws CorruptIndexException, IOException {
+    ensureOpen();
     prepareCommit(false);
   }
 
@@ -3213,17 +3480,41 @@
    * loss it may still lose data.  Lucene cannot guarantee
    * consistency on such devices.  </p>
    */
+
+  private boolean committing;
+
+  synchronized private void waitForCommit() {
+    // Only allow a single thread to do the commit, at a time:
+    while(committing)
+      doWait();
+    committing = true;
+  }
+
+  synchronized private void doneCommit() {
+    committing = false;
+    notifyAll();
+  }
+
   public final void commit() throws CorruptIndexException, IOException {
 
-    message("commit: start");
+    ensureOpen();
 
-    if (autoCommit || pendingCommit == null) {
-      message("commit: now prepare");
-      prepareCommit(true);
-    } else
-      message("commit: already prepared");
+    // Only let one thread do the prepare/finish at a time
+    waitForCommit();
 
-    finishCommit();
+    try {
+      message("commit: start");
+
+      if (autoCommit || pendingCommit == null) {
+        message("commit: now prepare");
+        prepareCommit(true);
+      } else
+        message("commit: already prepared");
+
+      finishCommit();
+    } finally {
+      doneCommit();
+    }
   }
 
   private synchronized final void finishCommit() throws CorruptIndexException, IOException {
@@ -3234,7 +3525,7 @@
         pendingCommit.finishCommit(directory);
         lastCommitChangeCount = pendingCommitChangeCount;
         segmentInfos.updateGeneration(pendingCommit);
-        setRollbackSegmentInfos();
+        setRollbackSegmentInfos(pendingCommit);
         deleter.checkpoint(pendingCommit, true);
       } finally {
         deleter.decRef(pendingCommit);
@@ -3259,7 +3550,8 @@
    *  be flushed
    */
   protected final void flush(boolean triggerMerge, boolean flushDocStores, boolean flushDeletes) throws CorruptIndexException, IOException {
-    ensureOpen();
+    // We can be called during close, when closing==true, so we must pass false to ensureOpen:
+    ensureOpen(false);
     if (doFlush(flushDocStores, flushDeletes) && triggerMerge)
       maybeMerge();
   }
@@ -3269,12 +3561,14 @@
   // even while a flush is happening
   private synchronized final boolean doFlush(boolean flushDocStores, boolean flushDeletes) throws CorruptIndexException, IOException {
 
-    // Make sure no threads are actively adding a document
+    ensureOpen(false);
 
     assert testPoint("startDoFlush");
 
     flushCount++;
 
+    // Make sure no threads are actively adding a document
+
     flushDeletes |= docWriter.deletesFull();
 
     // When autoCommit=true we must always flush deletes
@@ -3456,7 +3750,7 @@
         if (segmentInfos.indexOf(info) == -1)
           throw new MergePolicy.MergeException("MergePolicy selected a segment (" + info.name + ") that is not in the index", directory);
         else
-          throw new MergePolicy.MergeException("MergePolicy selected non-contiguous segments to merge (" + merge + " vs " + segString() + "), which IndexWriter (currently) cannot handle",
+          throw new MergePolicy.MergeException("MergePolicy selected non-contiguous segments to merge (" + merge.segString(directory) + " vs " + segString() + "), which IndexWriter (currently) cannot handle",
                                                directory);
       }
     }
@@ -3470,7 +3764,7 @@
    *  deletes may have been flushed to the segments since
    *  the merge was started.  This method "carries over"
    *  such new deletes onto the newly merged segment, and
-   *  saves the results deletes file (incrementing the
+   *  saves the resulting deletes file (incrementing the
    *  delete generation for merge.info).  If no deletes were
    *  flushed, no new deletes file is saved. */
   synchronized private void commitMergedDeletes(MergePolicy.OneMerge merge) throws IOException {
@@ -3573,7 +3867,7 @@
       return false;
 
     if (infoStream != null)
-      message("commitMerge: " + merge.segString(directory));
+      message("commitMerge: " + merge.segString(directory) + " index=" + segString());
 
     assert merge.registerDone;
 
@@ -3619,6 +3913,7 @@
     merge.info.setHasProx(merger.hasProx());
 
     segmentInfos.subList(start, start + merge.segments.size()).clear();
+    assert !segmentInfos.contains(merge.info);
     segmentInfos.add(start, merge.info);
 
     // Must checkpoint before decrefing so any newly
@@ -3663,13 +3958,14 @@
           mergeInit(merge);
 
           if (infoStream != null)
-            message("now merge\n  merge=" + merge.segString(directory) + "\n  index=" + segString());
+            message("now merge\n  merge=" + merge.segString(directory) + "\n  merge=" + merge + "\n  index=" + segString());
 
           mergeMiddle(merge);
           success = true;
         } catch (MergePolicy.MergeAbortedException e) {
           merge.setException(e);
           addMergeException(merge);
+
           // We can ignore this exception, unless the merge
           // involves segments from external directories, in
           // which case we must throw it so, for example, the
@@ -3699,10 +3995,6 @@
               updatePendingMerges(merge.maxNumSegmentsOptimize, merge.optimize);
           } finally {
             runningMerges.remove(merge);
-            // Optimize may be waiting on the final optimize
-            // merge to finish; and finishMerges() may be
-            // waiting for all merges to finish:
-            notifyAll();
           }
         }
       }
@@ -3718,11 +4010,16 @@
    *  are now participating in a merge, and true is
    *  returned.  Else (the merge conflicts) false is
    *  returned. */
-  final synchronized boolean registerMerge(MergePolicy.OneMerge merge) {
+  final synchronized boolean registerMerge(MergePolicy.OneMerge merge) throws MergePolicy.MergeAbortedException {
 
     if (merge.registerDone)
       return true;
 
+    if (stopMerges) {
+      merge.abort();
+      throw new MergePolicy.MergeAbortedException("merge is aborted: " + merge.segString(directory));
+    }
+
     final int count = merge.segments.size();
     boolean isExternal = false;
     for(int i=0;i<count;i++) {
@@ -3735,6 +4032,8 @@
         isExternal = true;
     }
 
+    ensureContiguousMerge(merge);
+
     pendingMerges.add(merge);
 
     if (infoStream != null)
@@ -3959,6 +4258,10 @@
   /** Does fininishing for a merge, which is fast but holds
    *  the synchronized lock on IndexWriter instance. */
   final synchronized void mergeFinish(MergePolicy.OneMerge merge) throws IOException {
+    
+    // Optimize, addIndexes or finishMerges may be waiting
+    // on merges to finish.
+    notifyAll();
 
     if (merge.increfDone)
       decrefMergeSegments(merge);
@@ -4052,6 +4355,16 @@
       try {
         merger.createCompoundFile(compoundFileName);
         success = true;
+      } catch (IOException ioe) {
+        synchronized(this) {
+          if (merge.isAborted()) {
+            // This can happen if rollback or close(false)
+            // is called -- fall through to logic below to
+            // remove the partially created CFS:
+            success = true;
+          } else
+            throw ioe;
+        }
       } finally {
         if (!success) {
           if (infoStream != null)
@@ -4170,7 +4483,10 @@
       if (i > 0) {
         buffer.append(' ');
       }
-      buffer.append(infos.info(i).segString(directory));
+      final SegmentInfo info = infos.info(i);
+      buffer.append(info.segString(directory));
+      if (info.dir != directory)
+        buffer.append("**");
     }
     return buffer.toString();
   }
@@ -4258,6 +4574,20 @@
     }
   }
 
+  private synchronized void doWait() {
+    try {
+      // NOTE: the callers of this method should in theory
+      // be able to do simply wait(), but, as a defense
+      // against thread timing hazards where notifyAll()
+      // falls to be called, we wait for at most 1 second
+      // and then return so caller can check if wait
+      // conditions are satisified:
+      wait(1000);
+    } catch (InterruptedException ie) {
+      Thread.currentThread().interrupt();
+    }
+  }
+
   /** Walk through all files referenced by the current
    *  segmentInfos and ask the Directory to sync each file,
    *  if it wasn't already.  If that succeeds, then we
@@ -4283,26 +4613,38 @@
 
       synchronized(this) {
 
-        assert lastCommitChangeCount <= changeCount;
+        // Wait for any running addIndexes to complete
+        // first, then block any from running until we've
+        // copied the segmentInfos we intend to sync:
+        blockAddIndexes(false);
 
-        if (changeCount == lastCommitChangeCount) {
-          if (infoStream != null)
-            message("  skip startCommit(): no changes pending");
-          return;
-        }
+        assert !hasExternalSegments();
 
-        // First, we clone & incref the segmentInfos we intend
-        // to sync, then, without locking, we sync() each file
-        // referenced by toSync, in the background.  Multiple
-        // threads can be doing this at once, if say a large
-        // merge and a small merge finish at the same time:
+        try {
 
-        if (infoStream != null)
-          message("startCommit index=" + segString(segmentInfos) + " changeCount=" + changeCount);
+          assert lastCommitChangeCount <= changeCount;
 
-        toSync = (SegmentInfos) segmentInfos.clone();
-        deleter.incRef(toSync, false);
-        myChangeCount = changeCount;
+          if (changeCount == lastCommitChangeCount) {
+            if (infoStream != null)
+              message("  skip startCommit(): no changes pending");
+            return;
+          }
+
+          // First, we clone & incref the segmentInfos we intend
+          // to sync, then, without locking, we sync() each file
+          // referenced by toSync, in the background.  Multiple
+          // threads can be doing this at once, if say a large
+          // merge and a small merge finish at the same time:
+
+          if (infoStream != null)
+            message("startCommit index=" + segString(segmentInfos) + " changeCount=" + changeCount);
+
+          toSync = (SegmentInfos) segmentInfos.clone();
+          deleter.incRef(toSync, false);
+          myChangeCount = changeCount;
+        } finally {
+          resumeAddIndexes();
+        }
       }
 
       assert testPoint("midStartCommit");
@@ -4326,7 +4668,7 @@
                 try {
                   // Because we incRef'd this commit point, above,
                   // the file had better exist:
-                  assert directory.fileExists(fileName);
+                  assert directory.fileExists(fileName): "file '" + fileName + "' does not exist dir=" + directory;
                   message("now sync " + fileName);
                   directory.sync(fileName);
                   success = true;
@@ -4359,11 +4701,7 @@
             // Wait now for any current pending commit to complete:
             while(pendingCommit != null) {
               message("wait for existing pendingCommit to finish...");
-              try {
-                wait();
-              } catch (InterruptedException ie) {
-                Thread.currentThread().interrupt();
-              }
+              doWait();
             }
 
             if (segmentInfos.getGeneration() > toSync.getGeneration())