Index: src/java/org/apache/lucene/index/DocumentsWriterThreadPool.java =================================================================== --- src/java/org/apache/lucene/index/DocumentsWriterThreadPool.java (revision 1055743) +++ src/java/org/apache/lucene/index/DocumentsWriterThreadPool.java (working copy) @@ -60,7 +60,7 @@ protected final int maxNumThreadStates; protected ThreadState[] allThreadStates = new ThreadState[0]; - private final Lock lock = new ReentrantLock(); + final Lock lock = new ReentrantLock(); private final Condition threadStateAvailable = lock.newCondition(); private boolean globalLock; private boolean aborting; @@ -225,11 +225,16 @@ } } } - + + // nocommit: should we give an option to only return non-flushing DWPTs here? final Iterator getPerThreadIterator() { return getPerThreadIterator(allThreadStates); } + int getNumThreads() { + return allThreadStates.length; + } + private static final Iterator getPerThreadIterator(final ThreadState[] localAllThreads) { return new Iterator() { int i = 0; Index: src/java/org/apache/lucene/index/DocumentsWriterPerThread.java =================================================================== --- src/java/org/apache/lucene/index/DocumentsWriterPerThread.java (revision 1055743) +++ src/java/org/apache/lucene/index/DocumentsWriterPerThread.java (working copy) @@ -82,7 +82,7 @@ }; // Deletes for our still-in-RAM (to be flushed next) segment - private SegmentDeletes pendingDeletes = new SegmentDeletes(); + SegmentDeletes pendingDeletes = new SegmentDeletes(); static class DocState { final DocumentsWriterPerThread docWriter; @@ -251,22 +251,22 @@ void deleteQueries(Query... queries) { for (Query query : queries) { - pendingDeletes.addQuery(query, numDocsInRAM); + pendingDeletes.addQuery(query, numDocsInRAM, bytesUsed, parent.bytesUsed); } } void deleteQuery(Query query) { - pendingDeletes.addQuery(query, numDocsInRAM); + pendingDeletes.addQuery(query, numDocsInRAM, bytesUsed, parent.bytesUsed); } void deleteTerms(Term... terms) { for (Term term : terms) { - pendingDeletes.addTerm(term, numDocsInRAM); + pendingDeletes.addTerm(term, numDocsInRAM, bytesUsed, parent.bytesUsed); } } void deleteTerm(Term term) { - pendingDeletes.addTerm(term, numDocsInRAM); + pendingDeletes.addTerm(term, numDocsInRAM, bytesUsed, parent.bytesUsed); } public void commitDocument() { Index: src/java/org/apache/lucene/index/SegmentDeletes.java =================================================================== --- src/java/org/apache/lucene/index/SegmentDeletes.java (revision 1055743) +++ src/java/org/apache/lucene/index/SegmentDeletes.java (working copy) @@ -139,9 +139,12 @@ // should already be cleared } - public void addQuery(Query query, int docIDUpto) { + public void addQuery(Query query, int docIDUpto, AtomicLong... used) { queries.put(query, docIDUpto); bytesUsed.addAndGet(BYTES_PER_DEL_QUERY); + for (AtomicLong u : used) { + u.addAndGet(BYTES_PER_DEL_QUERY); + } } public void addDocID(int docID) { @@ -149,7 +152,7 @@ bytesUsed.addAndGet(BYTES_PER_DEL_DOCID); } - public void addTerm(Term term, int docIDUpto) { + public void addTerm(Term term, int docIDUpto, AtomicLong... used) { Integer current = terms.get(term); if (current != null && docIDUpto < current) { // Only record the new number if it's greater than the @@ -166,6 +169,9 @@ numTermDeletes.incrementAndGet(); if (current == null) { bytesUsed.addAndGet(BYTES_PER_DEL_TERM + term.bytes.length); + for (AtomicLong u : used) { + u.addAndGet(BYTES_PER_DEL_TERM + term.bytes.length); + } } } Index: src/java/org/apache/lucene/index/DocumentsWriter.java =================================================================== --- src/java/org/apache/lucene/index/DocumentsWriter.java (revision 1055743) +++ src/java/org/apache/lucene/index/DocumentsWriter.java (working copy) @@ -162,6 +162,8 @@ final BufferedDeletes bufferedDeletes; private final IndexWriter.FlushControl flushControl; private final IndexingChain chain; + SegmentDeletes defaultDeletes = new SegmentDeletes(); + FlushPolicy flushPolicy; DocumentsWriter(Directory directory, IndexWriter writer, IndexingChain chain, DocumentsWriterThreadPool indexerThreadPool, FieldInfos fieldInfos, BufferedDeletes bufferedDeletes) throws IOException { this.directory = directory; @@ -172,42 +174,96 @@ this.threadPool = indexerThreadPool; this.chain = chain; flushControl = writer.flushControl; + flushPolicy = new DefaultFlushPolicy(writer); } - + + boolean hasThreads() { + return threadPool.getNumThreads() > 0; + } + + boolean flush() { + return flushPolicy.flush(this); + } + + // nocommit: we need to add the delete's bytes used boolean deleteQueries(Query... queries) { - final boolean doFlush = flushControl.waitUpdate(0, queries.length); - Iterator it = threadPool.getPerThreadIterator(); - while (it.hasNext()) { - it.next().deleteQueries(queries); + threadPool.lock.lock(); + try { + if (hasThreads()) { + Iterator it = threadPool.getPerThreadIterator(); + while (it.hasNext()) { + it.next().deleteQueries(queries); + } + } else { + synchronized (defaultDeletes) { + for (Query query : queries) { + defaultDeletes.addQuery(query, SegmentDeletes.MAX_INT); + } + } + } + } finally { + threadPool.lock.unlock(); } - return doFlush; + return flush(); } boolean deleteQuery(Query query) { - final boolean doFlush = flushControl.waitUpdate(0, 1); - Iterator it = threadPool.getPerThreadIterator(); - while (it.hasNext()) { - it.next().deleteQuery(query); + threadPool.lock.lock(); + try { + if (hasThreads()) { + Iterator it = threadPool.getPerThreadIterator(); + while (it.hasNext()) { + it.next().deleteQuery(query); + } + } else { + synchronized (defaultDeletes) { + defaultDeletes.addQuery(query, SegmentDeletes.MAX_INT); + } + } + } finally { + threadPool.lock.unlock(); } - return doFlush; + return flush(); } boolean deleteTerms(Term... terms) { - final boolean doFlush = flushControl.waitUpdate(0, terms.length); - Iterator it = threadPool.getPerThreadIterator(); - while (it.hasNext()) { - it.next().deleteTerms(terms); + threadPool.lock.lock(); + try { + if (hasThreads()) { + Iterator it = threadPool.getPerThreadIterator(); + while (it.hasNext()) { + it.next().deleteTerms(terms); + } + } else { + synchronized (defaultDeletes) { + for (Term term : terms) { + defaultDeletes.addTerm(term, SegmentDeletes.MAX_INT, ramUsed); + } + } + } + } finally { + threadPool.lock.unlock(); } - return doFlush; + return flush(); } - boolean deleteTerm(Term term, boolean skipWait) { - final boolean doFlush = flushControl.waitUpdate(0, 1, skipWait); - Iterator it = threadPool.getPerThreadIterator(); - while (it.hasNext()) { - it.next().deleteTerm(term); + boolean deleteTerm(Term term) { + threadPool.lock.lock(); + try { + if (hasThreads()) { + Iterator it = threadPool.getPerThreadIterator(); + while (it.hasNext()) { + it.next().deleteTerm(term); + } + } else { + synchronized (defaultDeletes) { + defaultDeletes.addTerm(term, SegmentDeletes.MAX_INT, ramUsed); + } + } + } finally { + threadPool.lock.unlock(); } - return doFlush; + return flush(); } public FieldInfos getFieldInfos() { @@ -322,14 +378,29 @@ return numDocsInRAM.get() != 0; //return numDocsInRAM.get() != 0 || pendingDeletes.any(); } - + // for testing - public SegmentDeletes getPendingDeletes() { - return null; - // nocommit - //return pendingDeletes; + public int getBufferedDeleteTermsSize() { + int total = 0; + Iterator it = threadPool.getPerThreadIterator(); + while (it.hasNext()) { + DocumentsWriterPerThread dwpt = it.next(); + total += dwpt.pendingDeletes.terms.size(); + } + return total; } + //for testing + public int getNumBufferedDeleteTerms() { + int total = 0; + Iterator it = threadPool.getPerThreadIterator(); + while (it.hasNext()) { + DocumentsWriterPerThread dwpt = it.next(); + total += dwpt.pendingDeletes.numTermDeletes.get(); + } + return total; + } + public boolean anyDeletions() { // nocommit return true; @@ -360,7 +431,11 @@ synchronized(DocumentsWriter.this) { ensureOpen(); if (delTerm != null) { - deleteTerm(delTerm, true); + // nocommit: getPerThreadIterator isn't threadsafe? + Iterator it = threadPool.getPerThreadIterator(); + while (it.hasNext()) { + it.next().deleteTerm(delTerm); + } } perThread.commitDocument(); numDocsInRAM.incrementAndGet(); @@ -421,13 +496,23 @@ return true; } - + if (flush()) { + flushSegment(perThread); + return true; + } return false; } final boolean flushAllThreads(final boolean flushDeletes) throws IOException { - + + if (flushDeletes) { + if (indexWriter.segmentInfos.size() > 0 && defaultDeletes.any()) { + bufferedDeletes.pushDeletes(defaultDeletes, indexWriter.segmentInfos.lastElement(), true); + defaultDeletes = new SegmentDeletes(); + } + } + return threadPool.executeAllThreads(this, new DocumentsWriterThreadPool.AllThreadsTask() { @Override public Boolean process(Iterator threadsIterator) throws IOException { Index: src/java/org/apache/lucene/index/IndexWriter.java =================================================================== --- src/java/org/apache/lucene/index/IndexWriter.java (revision 1055743) +++ src/java/org/apache/lucene/index/IndexWriter.java (working copy) @@ -1275,7 +1275,7 @@ public void deleteDocuments(Term term) throws CorruptIndexException, IOException { ensureOpen(); try { - if (docWriter.deleteTerm(term, false)) { + if (docWriter.deleteTerm(term)) { flush(true, false); } } catch (OutOfMemoryError oom) { @@ -3209,12 +3209,12 @@ // For test purposes. final int getBufferedDeleteTermsSize() { - return docWriter.getPendingDeletes().terms.size(); + return docWriter.getBufferedDeleteTermsSize(); } // For test purposes. final int getNumBufferedDeleteTerms() { - return docWriter.getPendingDeletes().numTermDeletes.get(); + return docWriter.getNumBufferedDeleteTerms(); } // utility routines for tests @@ -3564,7 +3564,7 @@ public synchronized void clearDeletes() { delCount = 0; } - + /** public synchronized boolean waitUpdate(int docInc, int delInc) { return waitUpdate(docInc, delInc, false); } @@ -3604,8 +3604,9 @@ return flushByRAMUsage("add delete/doc"); } - + **/ public synchronized boolean flushByRAMUsage(String reason) { + return docWriter.flush(); // nocommit // final double ramBufferSizeMB = config.getRAMBufferSizeMB(); // if (ramBufferSizeMB != IndexWriterConfig.DISABLE_AUTO_FLUSH) { @@ -3624,7 +3625,7 @@ // } // } // } - return false; + //return false; } }