Index: src/test/org/apache/lucene/index/TestIndexWriter.java =================================================================== --- src/test/org/apache/lucene/index/TestIndexWriter.java (revision 1055743) +++ src/test/org/apache/lucene/index/TestIndexWriter.java (working copy) @@ -2598,6 +2598,7 @@ if (VERBOSE) { System.out.println("TEST: iter doIndexing=" + doIndexing); } + try { if (doIndexing) { // Add docs until a flush is triggered final int startFlushCount = w.flushCount; @@ -2613,6 +2614,10 @@ count++; } } + } catch (Throwable th) { + th.printStackTrace(); + throw new Exception("doindexing:"+doIndexing, th); + } assertTrue("flush happened too quickly during " + (doIndexing ? "indexing" : "deleting") + " count=" + count, count > 2500); } w.close(); Index: src/java/org/apache/lucene/index/FlushPolicy.java =================================================================== --- src/java/org/apache/lucene/index/FlushPolicy.java (revision 0) +++ src/java/org/apache/lucene/index/FlushPolicy.java (revision 0) @@ -0,0 +1,22 @@ +package org.apache.lucene.index; + +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +public abstract class FlushPolicy { + public abstract boolean flush(DocumentsWriter docWriter); +} Index: src/java/org/apache/lucene/index/DefaultFlushPolicy.java =================================================================== --- src/java/org/apache/lucene/index/DefaultFlushPolicy.java (revision 0) +++ src/java/org/apache/lucene/index/DefaultFlushPolicy.java (revision 0) @@ -0,0 +1,34 @@ +package org.apache.lucene.index; + +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +public class DefaultFlushPolicy extends FlushPolicy { + IndexWriter writer; + + public DefaultFlushPolicy(IndexWriter writer) { + this.writer = writer; + } + + public boolean flush(DocumentsWriter docWriter) { + if (writer.getConfig().getRAMBufferSizeMB() != IndexWriterConfig.DISABLE_AUTO_FLUSH) { + long maxRam = (long)((writer.getConfig().getRAMBufferSizeMB() * 1024 * 1024) * 0.9); + return docWriter.bytesUsed.get() > maxRam; + } + return false; + } +} 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; @@ -209,9 +209,14 @@ ThreadState state = acquireThreadState(documentsWriter, doc); boolean success = false; try { - T result = task.process(state.perThread); - success = true; - return result; + try { + T result = task.process(state.perThread); + success = true; + return result; + } catch (OutOfMemoryError oom) { + System.out.println("OOM mem used:"+documentsWriter.bytesUsed.get()); + throw oom; + } } finally { boolean abort = false; if (!success && state.perThread.aborting) { @@ -225,11 +230,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; @@ -250,7 +260,7 @@ } }; } - + protected final T addNewThreadState(DocumentsWriter documentsWriter, T threadState) { // Just create a new "private" thread state ThreadState[] newArray = new ThreadState[1+allThreadStates.length]; 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; @@ -150,6 +150,7 @@ final AtomicLong bytesUsed = new AtomicLong(0); FieldInfos fieldInfos = new FieldInfos(); + boolean flushing = false; public DocumentsWriterPerThread(Directory directory, DocumentsWriter parent, IndexingChain indexingChain) { parent.indexWriter.testPoint("DocumentsWriterPerThread.init start"); @@ -251,22 +252,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() { @@ -297,7 +298,9 @@ /** Flush all pending docs to a new segment */ SegmentInfo flush() throws IOException { assert numDocsInRAM > 0; - + + flushing = true; + flushState = new SegmentWriteState(infoStream, directory, segment, fieldInfos, numDocsInRAM, writer.getConfig().getTermIndexInterval(), SegmentCodecs.build(fieldInfos, writer.codecs)); @@ -352,6 +355,7 @@ if (!success) { setAborting(); } + flushing = false; } } 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) @@ -106,7 +106,6 @@ int numDocsInStore; // # docs written to doc stores - boolean bufferIsFull; // True when it's time to write segment private boolean closed; PrintStream infoStream; @@ -119,7 +118,6 @@ final IndexWriter indexWriter; private AtomicInteger numDocsInRAM = new AtomicInteger(0); - private AtomicLong ramUsed = new AtomicLong(0); private int numDocumentsWriterPerThreads; static class DocState { @@ -162,6 +160,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 +172,70 @@ 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(); + DocumentsWriterPerThread dwpt = it.next(); + if (!dwpt.flushing) { + dwpt.deleteQueries(queries); + } + } else { + synchronized (defaultDeletes) { + for (Query query : queries) { + defaultDeletes.addQuery(query, SegmentDeletes.MAX_INT, bytesUsed); + } + } + } + } 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); - } - return doFlush; + return deleteQueries(query); } 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()) { + DocumentsWriterPerThread dwpt = it.next(); + if (!dwpt.flushing) { + dwpt.deleteTerms(terms); + } + } + } else { + synchronized (defaultDeletes) { + for (Term term : terms) { + defaultDeletes.addTerm(term, SegmentDeletes.MAX_INT, bytesUsed); + } + } + } + } 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); - } - return doFlush; + boolean deleteTerm(Term term) { + return deleteTerms(term); } public FieldInfos getFieldInfos() { @@ -322,14 +350,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 +403,14 @@ synchronized(DocumentsWriter.this) { ensureOpen(); if (delTerm != null) { - deleteTerm(delTerm, true); + // nocommit: is getPerThreadIterator threadsafe? + Iterator it = threadPool.getPerThreadIterator(); + while (it.hasNext()) { + DocumentsWriterPerThread dwpt = it.next(); + if (!dwpt.flushing) { + dwpt.deleteTerm(delTerm); + } + } } perThread.commitDocument(); numDocsInRAM.incrementAndGet(); @@ -373,7 +423,9 @@ return false; } }); - + + assert bytesUsed.get() > 0; + if (flushed) { indexWriter.maybeMerge(); return true; @@ -394,11 +446,10 @@ } long deltaRAM = perThread.bytesUsed() - perThreadRAMUsedBeforeAdd; - long oldValue = ramUsed.get(); - while (!ramUsed.compareAndSet(oldValue, oldValue + deltaRAM)) { - oldValue = ramUsed.get(); + long oldValue = bytesUsed.get(); + while (!bytesUsed.compareAndSet(oldValue, oldValue + deltaRAM)) { + oldValue = bytesUsed.get(); } - return flushed; } @@ -421,13 +472,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 @@ -3423,6 +3423,7 @@ message("hit OutOfMemoryError inside " + location); } hitOOM = true; + System.out.println("docwriter RAM used:"+docWriter.bytesUsed.get()); throw oom; } @@ -3564,7 +3565,7 @@ public synchronized void clearDeletes() { delCount = 0; } - + /** public synchronized boolean waitUpdate(int docInc, int delInc) { return waitUpdate(docInc, delInc, false); } @@ -3604,8 +3605,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 +3626,7 @@ // } // } // } - return false; + //return false; } }