diff --git a/lucene/contrib/misc/src/test/org/apache/lucene/index/TestNRTManager.java b/lucene/contrib/misc/src/test/org/apache/lucene/index/TestNRTManager.java index cb79651..7b0ef13 100644 --- a/lucene/contrib/misc/src/test/org/apache/lucene/index/TestNRTManager.java +++ b/lucene/contrib/misc/src/test/org/apache/lucene/index/TestNRTManager.java @@ -61,7 +61,7 @@ public class TestNRTManager extends ThreadedIndexingAndSearchingTestCase { } @Override - protected void updateDocuments(Term id, List> docs) throws Exception { + protected long updateDocuments(Term id, List> docs) throws Exception { final long gen = nrt.updateDocuments(id, docs); // Randomly verify the update "took": @@ -82,10 +82,11 @@ public class TestNRTManager extends ThreadedIndexingAndSearchingTestCase { } lastGens.set(gen); + return gen; } @Override - protected void addDocuments(Term id, List> docs) throws Exception { + protected long addDocuments(Term id, List> docs) throws Exception { final long gen = nrt.addDocuments(docs); // Randomly verify the add "took": if (random.nextInt(20) == 2) { @@ -104,10 +105,11 @@ public class TestNRTManager extends ThreadedIndexingAndSearchingTestCase { } } lastGens.set(gen); + return gen; } @Override - protected void addDocument(Term id, Iterable doc) throws Exception { + protected long addDocument(Term id, Iterable doc) throws Exception { final long gen = nrt.addDocument(doc); // Randomly verify the add "took": @@ -127,10 +129,11 @@ public class TestNRTManager extends ThreadedIndexingAndSearchingTestCase { } } lastGens.set(gen); + return gen; } @Override - protected void updateDocument(Term id, Iterable doc) throws Exception { + protected long updateDocument(Term id, Iterable doc) throws Exception { final long gen = nrt.updateDocument(id, doc); // Randomly verify the udpate "took": if (random.nextInt(20) == 2) { @@ -149,10 +152,11 @@ public class TestNRTManager extends ThreadedIndexingAndSearchingTestCase { } } lastGens.set(gen); + return gen; } @Override - protected void deleteDocuments(Term id) throws Exception { + protected long deleteDocuments(Term id) throws Exception { final long gen = nrt.deleteDocuments(id); // randomly verify the delete "took": if (random.nextInt(20) == 7) { @@ -171,6 +175,7 @@ public class TestNRTManager extends ThreadedIndexingAndSearchingTestCase { } } lastGens.set(gen); + return gen; } private NRTManager nrt; diff --git a/lucene/src/java/org/apache/lucene/index/DocumentsWriter.java b/lucene/src/java/org/apache/lucene/index/DocumentsWriter.java index e04e659..cb17952 100644 --- a/lucene/src/java/org/apache/lucene/index/DocumentsWriter.java +++ b/lucene/src/java/org/apache/lucene/index/DocumentsWriter.java @@ -139,24 +139,26 @@ final class DocumentsWriter { flushControl = new DocumentsWriterFlushControl(this, config); } - synchronized void deleteQueries(final Query... queries) throws IOException { - deleteQueue.addDelete(queries); + synchronized long deleteQueries(final Query... queries) throws IOException { + final long seqId = deleteQueue.addDelete(queries); flushControl.doOnDelete(); if (flushControl.doApplyAllDeletes()) { applyAllDeletes(deleteQueue); } + return seqId; } // TODO: we could check w/ FreqProxTermsWriter: if the // term doesn't exist, don't bother buffering into the // per-DWPT map (but still must go into the global map) - synchronized void deleteTerms(final Term... terms) throws IOException { + synchronized long deleteTerms(final Term... terms) throws IOException { final DocumentsWriterDeleteQueue deleteQueue = this.deleteQueue; - deleteQueue.addDelete(terms); + final long seqId = deleteQueue.addDelete(terms); flushControl.doOnDelete(); if (flushControl.doApplyAllDeletes()) { applyAllDeletes(deleteQueue); } + return seqId; } DocumentsWriterDeleteQueue currentDeleteSession() { @@ -319,13 +321,13 @@ final class DocumentsWriter { return maybeMerge; } - boolean updateDocuments(final Iterable> docs, final Analyzer analyzer, + long updateDocuments(final Iterable> docs, final Analyzer analyzer, final Term delTerm) throws CorruptIndexException, IOException { boolean maybeMerge = preUpdate(); final ThreadState perThread = flushControl.obtainAndLock(); final DocumentsWriterPerThread flushingDWPT; - + final long seqId; try { if (!perThread.isActive()) { ensureOpen(); @@ -334,8 +336,9 @@ final class DocumentsWriter { final DocumentsWriterPerThread dwpt = perThread.perThread; try { - final int docCount = dwpt.updateDocuments(docs, analyzer, delTerm); - numDocsInRAM.addAndGet(docCount); + int docCount = dwpt.getNumDocsInRAM(); + seqId = dwpt.updateDocuments(docs, analyzer, delTerm) << 1; + numDocsInRAM.addAndGet(dwpt.getNumDocsInRAM() - docCount); } finally { if (dwpt.checkAndResetHasAborted()) { flushControl.doOnAbort(perThread); @@ -346,11 +349,11 @@ final class DocumentsWriter { } finally { perThread.unlock(); } - - return postUpdate(flushingDWPT, maybeMerge); + maybeMerge = postUpdate(flushingDWPT, maybeMerge); + return maybeMerge ? seqId | 1 : seqId; } - boolean updateDocument(final Iterable doc, final Analyzer analyzer, + long updateDocument(final Iterable doc, final Analyzer analyzer, final Term delTerm) throws CorruptIndexException, IOException { boolean maybeMerge = preUpdate(); @@ -358,7 +361,7 @@ final class DocumentsWriter { final ThreadState perThread = flushControl.obtainAndLock(); final DocumentsWriterPerThread flushingDWPT; - + final long seqId; try { if (!perThread.isActive()) { @@ -368,7 +371,7 @@ final class DocumentsWriter { final DocumentsWriterPerThread dwpt = perThread.perThread; try { - dwpt.updateDocument(doc, analyzer, delTerm); + seqId = dwpt.updateDocument(doc, analyzer, delTerm) << 1; numDocsInRAM.incrementAndGet(); } finally { if (dwpt.checkAndResetHasAborted()) { @@ -381,7 +384,8 @@ final class DocumentsWriter { perThread.unlock(); } - return postUpdate(flushingDWPT, maybeMerge); + maybeMerge = postUpdate(flushingDWPT, maybeMerge); + return maybeMerge ? seqId | 1 : seqId; } private boolean doFlush(DocumentsWriterPerThread flushingDWPT) throws IOException { @@ -542,7 +546,7 @@ final class DocumentsWriter { * two stage operation; the caller must ensure (in try/finally) that finishFlush * is called after this method, to release the flush lock in DWFlushControl */ - final boolean flushAllThreads() + final long flushAllThreads() throws IOException { final DocumentsWriterDeleteQueue flushingDeleteQueue; if (infoStream != null) { @@ -581,7 +585,8 @@ final class DocumentsWriter { } finally { assert flushingDeleteQueue == currentFullFlushDelQueue; } - return anythingFlushed; + long seqId = flushingDeleteQueue.getMaxSeqId() << 1 ; + return anythingFlushed ? seqId | 1 : seqId; } final void finishFullFlush(boolean success) { diff --git a/lucene/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java b/lucene/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java index 3489fd2..d635829 100644 --- a/lucene/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java +++ b/lucene/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java @@ -79,14 +79,14 @@ final class DocumentsWriterDeleteQueue { final long generation; DocumentsWriterDeleteQueue() { - this(0); + this(0, 0); } - DocumentsWriterDeleteQueue(long generation) { - this(new BufferedDeletes(), generation); + DocumentsWriterDeleteQueue(long generation, long seqId) { + this(new BufferedDeletes(), generation, seqId); } - DocumentsWriterDeleteQueue(BufferedDeletes globalBufferedDeletes, long generation) { + DocumentsWriterDeleteQueue(BufferedDeletes globalBufferedDeletes, long generation, long seqId) { this.globalBufferedDeletes = globalBufferedDeletes; this.generation = generation; /* @@ -94,26 +94,29 @@ final class DocumentsWriterDeleteQueue { * apply this tail since the head is always omitted. */ tail = new Node(null); // sentinel + tail.seqId = seqId; globalSlice = new DeleteSlice(tail); } - void addDelete(Query... queries) { - add(new QueryArrayNode(queries)); + long addDelete(Query... queries) { + long seqId = add(new QueryArrayNode(queries)); tryApplyGlobalSlice(); + return seqId; } - void addDelete(Term... terms) { - add(new TermArrayNode(terms)); + long addDelete(Term... terms) { + final long seqId = add(new TermArrayNode(terms)); tryApplyGlobalSlice(); + return seqId; } /** * invariant for document update */ - void add(Term term, DeleteSlice slice) { + long add(Term term, DeleteSlice slice) { final TermNode termNode = new TermNode(term); // System.out.println(Thread.currentThread().getName() + ": push " + termNode + " this=" + this); - add(termNode); + final long seqId = add(termNode); /* * this is an update request where the term is the updated documents * delTerm. in that case we need to guarantee that this insert is atomic @@ -128,9 +131,10 @@ final class DocumentsWriterDeleteQueue { assert slice.sliceHead != slice.sliceTail : "slice head and tail must differ after add"; tryApplyGlobalSlice(); // TODO doing this each time is not necessary maybe // we can do it just every n times or so? + return seqId; } - void add(Node item) { + long add(Node item) { /* * this non-blocking / 'wait-free' linked list add was inspired by Apache * Harmony's ConcurrentLinkedQueue Implementation. @@ -152,19 +156,26 @@ final class DocumentsWriterDeleteQueue { * current tail if we fail to insert we just retry the operation since * somebody else has already added its item */ - if (currentTail.casNext(null, item)) { + long seqId; + if (currentTail.casNext(null, item.updateSeqId(seqId = currentTail.seqId+1))) { /* * now that we are done we need to advance the tail while another * thread could have advanced it already so we can ignore the return * type of this CAS call */ tailUpdater.compareAndSet(this, currentTail, item); - return; + return seqId; } } } } } + + + long getMaxSeqId() { + return tail.seqId; + } + boolean anyChanges() { globalBufferLock.lock(); @@ -276,7 +287,7 @@ final class DocumentsWriterDeleteQueue { // Reset to a 0 length slice sliceHead = sliceTail; } - + /** * Returns true iff the given item is identical to the item * hold by the slices tail, otherwise false. @@ -306,11 +317,19 @@ final class DocumentsWriterDeleteQueue { } private static class Node { + long seqId = Long.MIN_VALUE; volatile Node next; final T item; Node(T item) { this.item = item; + + } + + Node updateSeqId(long seqId) { + assert next == null; + this.seqId = seqId; + return this; } @SuppressWarnings("rawtypes") @@ -334,7 +353,9 @@ final class DocumentsWriterDeleteQueue { @Override void apply(BufferedDeletes bufferedDeletes, int docIDUpto) { - bufferedDeletes.addTerm(item, docIDUpto); + if (item != null) { + bufferedDeletes.addTerm(item, docIDUpto); + } } @Override @@ -408,5 +429,4 @@ final class DocumentsWriterDeleteQueue { return "DWDQ: [ generation: " + generation + " ]"; } - } diff --git a/lucene/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java b/lucene/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java index cdef910..bad469f 100644 --- a/lucene/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java +++ b/lucene/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java @@ -351,11 +351,11 @@ public final class DocumentsWriterFlushControl { return flushingWriters.size(); } - public boolean doApplyAllDeletes() { + public boolean doApplyAllDeletes() { return flushDeletes.getAndSet(false); } - public void setApplyAllDeletes() { + public void setApplyAllDeletes() { flushDeletes.set(true); } @@ -385,55 +385,61 @@ public final class DocumentsWriterFlushControl { } } - void markForFullFlush() { - final DocumentsWriterDeleteQueue flushingQueue; - synchronized (this) { - assert !fullFlush : "called DWFC#markForFullFlush() while full flush is still running"; - assert fullFlushBuffer.isEmpty() : "full flush buffer should be empty: "+ fullFlushBuffer; - fullFlush = true; - flushingQueue = documentsWriter.deleteQueue; - // Set a new delete queue - all subsequent DWPT will use this queue until - // we do another full flush - DocumentsWriterDeleteQueue newQueue = new DocumentsWriterDeleteQueue(flushingQueue.generation+1); - documentsWriter.deleteQueue = newQueue; - } - final Iterator allActiveThreads = perThreadPool.getActivePerThreadsIterator(); - while (allActiveThreads.hasNext()) { - final ThreadState next = allActiveThreads.next(); - next.lock(); - try { - if (!next.isActive()) { - continue; - } - assert next.perThread.deleteQueue == flushingQueue - || next.perThread.deleteQueue == documentsWriter.deleteQueue : " flushingQueue: " - + flushingQueue - + " currentqueue: " - + documentsWriter.deleteQueue - + " perThread queue: " - + next.perThread.deleteQueue - + " numDocsInRam: " + next.perThread.getNumDocsInRAM(); - if (next.perThread.deleteQueue != flushingQueue) { - // this one is already a new DWPT - continue; + long markForFullFlush() { + try { + final DocumentsWriterDeleteQueue flushingQueue; + synchronized (this) { + assert !fullFlush : "called DWFC#markForFullFlush() while full flush is still running"; + assert fullFlushBuffer.isEmpty() : "full flush buffer should be empty: "+ fullFlushBuffer; + fullFlush = true; + flushingQueue = documentsWriter.deleteQueue; + final int numActiveDWPT = perThreadPool.freezeUnreleasedStates(true); + // Set a new delete queue - all subsequent DWPT will use this queue until + // we do another full flush + DocumentsWriterDeleteQueue newQueue = new DocumentsWriterDeleteQueue(flushingQueue.generation+1, numActiveDWPT + flushingQueue.getMaxSeqId()); + documentsWriter.deleteQueue = newQueue; + } + final Iterator allActiveThreads = perThreadPool.getActivePerThreadsIterator(); + while (allActiveThreads.hasNext()) { + final ThreadState next = allActiveThreads.next(); + next.lock(); + try { + if (!next.isActive()) { + continue; + } + assert next.perThread.deleteQueue == flushingQueue + || next.perThread.deleteQueue == documentsWriter.deleteQueue : " flushingQueue: " + + flushingQueue + + " currentqueue: " + + documentsWriter.deleteQueue + + " perThread queue: " + + next.perThread.deleteQueue + + " numDocsInRam: " + next.perThread.getNumDocsInRAM(); + if (next.perThread.deleteQueue != flushingQueue) { + // this one is already a new DWPT + continue; + } + addFlushableState(next); + } finally { + next.unlock(); } - addFlushableState(next); - } finally { - next.unlock(); } + synchronized (this) { + /* make sure we move all DWPT that are where concurrently marked as + * pending and moved to blocked are moved over to the flushQueue. There is + * a chance that this happens since we marking DWPT for full flush without + * blocking indexing.*/ + pruneBlockedQueue(flushingQueue); + assert assertBlockedFlushes(documentsWriter.deleteQueue); + flushQueue.addAll(fullFlushBuffer); + fullFlushBuffer.clear(); + stallControl.updateStalled(this); + } + assert assertActiveDeleteQueue(documentsWriter.deleteQueue); + return flushingQueue.getMaxSeqId(); + } finally { + perThreadPool.freezeUnreleasedStates(false); } - synchronized (this) { - /* make sure we move all DWPT that are where concurrently marked as - * pending and moved to blocked are moved over to the flushQueue. There is - * a chance that this happens since we marking DWPT for full flush without - * blocking indexing.*/ - pruneBlockedQueue(flushingQueue); - assert assertBlockedFlushes(documentsWriter.deleteQueue); - flushQueue.addAll(fullFlushBuffer); - fullFlushBuffer.clear(); - stallControl.updateStalled(this); - } - assert assertActiveDeleteQueue(documentsWriter.deleteQueue); } private boolean assertActiveDeleteQueue(DocumentsWriterDeleteQueue queue) { diff --git a/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java b/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java index fae432f..fb6c484 100644 --- a/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java +++ b/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java @@ -215,7 +215,7 @@ public class DocumentsWriterPerThread { return retval; } - public void updateDocument(Iterable doc, Analyzer analyzer, Term delTerm) throws IOException { + public long updateDocument(Iterable doc, Analyzer analyzer, Term delTerm) throws IOException { assert writer.testPoint("DocumentsWriterPerThread addDocument start"); assert deleteQueue != null; docState.doc = doc; @@ -262,10 +262,10 @@ public class DocumentsWriterPerThread { abort(); } } - finishDocument(delTerm); + return finishDocument(delTerm); } - public int updateDocuments(Iterable> docs, Analyzer analyzer, Term delTerm) throws IOException { + public long updateDocuments(Iterable> docs, Analyzer analyzer, Term delTerm) throws IOException { assert writer.testPoint("DocumentsWriterPerThread addDocuments start"); assert deleteQueue != null; docState.analyzer = analyzer; @@ -326,27 +326,22 @@ public class DocumentsWriterPerThread { abort(); } } - - finishDocument(null); + ++numDocsInRAM; } - // Apply delTerm only after all indexing has // succeeded, but apply it only to docs prior to when // this batch started: - if (delTerm != null) { - deleteQueue.add(delTerm, deleteSlice); - assert deleteSlice.isTailItem(delTerm) : "expected the delete term as the tail item"; - deleteSlice.apply(pendingDeletes, numDocsInRAM-docCount); - } + return finishDocument(delTerm, numDocsInRAM-docCount); } finally { docState.clear(); } - - return docCount; + } + private long finishDocument(Term delTerm) throws IOException { + return finishDocument(delTerm, numDocsInRAM++); } - private void finishDocument(Term delTerm) throws IOException { + private long finishDocument(Term delTerm, int upTo) throws IOException { /* * here we actually finish the document in two steps 1. push the delete into * the queue and update our slice. 2. increment the DWPT private document @@ -355,23 +350,19 @@ public class DocumentsWriterPerThread { * the updated slice we get from 1. holds all the deletes that have occurred * since we updated the slice the last time. */ + long seqId; if (deleteSlice == null) { deleteSlice = deleteQueue.newSlice(); + seqId = deleteQueue.add(delTerm, deleteSlice); if (delTerm != null) { - deleteQueue.add(delTerm, deleteSlice); deleteSlice.reset(); } - } else { - if (delTerm != null) { - deleteQueue.add(delTerm, deleteSlice); - assert deleteSlice.isTailItem(delTerm) : "expected the delete term as the tail item"; - deleteSlice.apply(pendingDeletes, numDocsInRAM); - } else if (deleteQueue.updateSlice(deleteSlice)) { - deleteSlice.apply(pendingDeletes, numDocsInRAM); - } + seqId = deleteQueue.add(delTerm, deleteSlice); + assert deleteSlice.isTailItem(delTerm) : "expected the delete term as the tail item"; + deleteSlice.apply(pendingDeletes, upTo); } - ++numDocsInRAM; + return seqId; } // Buffer a specific docID for deletion. Currently only diff --git a/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThreadPool.java b/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThreadPool.java index eb6f209..2ced15d 100644 --- a/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThreadPool.java +++ b/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThreadPool.java @@ -131,6 +131,7 @@ public abstract class DocumentsWriterPerThreadPool { private CodecProvider codecProvider; private FieldNumberBiMap globalFieldMap; private final SetOnce documentsWriter = new SetOnce(); + private boolean unreleasedThreadStatesFrozen; /** * Creates a new {@link DocumentsWriterPerThreadPool} with max. @@ -183,7 +184,7 @@ public abstract class DocumentsWriterPerThreadPool { * null */ public synchronized ThreadState newThreadState() { - if (numThreadStatesActive < perThreads.length) { + if (!unreleasedThreadStatesFrozen && numThreadStatesActive < perThreads.length) { final ThreadState threadState = perThreads[numThreadStatesActive]; threadState.lock(); // lock so nobody else will get this ThreadState boolean unlock = true; @@ -236,6 +237,11 @@ public abstract class DocumentsWriterPerThreadPool { } } + synchronized int freezeUnreleasedStates(boolean freeze) { + unreleasedThreadStatesFrozen = freeze; + return numThreadStatesActive; + } + protected DocumentsWriterPerThread replaceForFlush(ThreadState threadState, boolean closed) { assert threadState.isHeldByCurrentThread(); final DocumentsWriterPerThread dwpt = threadState.perThread; diff --git a/lucene/src/java/org/apache/lucene/index/IndexWriter.java b/lucene/src/java/org/apache/lucene/index/IndexWriter.java index 0f5b284..6d3ca66 100644 --- a/lucene/src/java/org/apache/lucene/index/IndexWriter.java +++ b/lucene/src/java/org/apache/lucene/index/IndexWriter.java @@ -192,6 +192,7 @@ import org.apache.lucene.util.TwoPhaseCommit; * referenced by the "front" of the index). For this, IndexFileDeleter * keeps track of the last non commit checkpoint. */ +// nocommit -add javadoc for seq IDs public class IndexWriter implements Closeable, TwoPhaseCommit { /** * Name of the write lock in the index. @@ -220,6 +221,7 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { private List rollbackSegments; // list of segmentInfo we will fallback to if the commit fails volatile SegmentInfos pendingCommit; // set when a commit is pending (after prepareCommit() & before commit()) + volatile long pendingCommitSeqId; volatile long pendingCommitChangeCount; final SegmentInfos segmentInfos; // the segments @@ -363,7 +365,8 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { synchronized (fullFlushLock) { boolean success = false; try { - anySegmentFlushed = docWriter.flushAllThreads(); + long seqId = docWriter.flushAllThreads(); + anySegmentFlushed = (seqId & 1) == 1; if (!anySegmentFlushed) { // prevent double increment since docWriter#doFlush increments the flushcount // if we flushed anything. @@ -1295,8 +1298,8 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * @throws CorruptIndexException if the index is corrupt * @throws IOException if there is a low-level IO error */ - public void addDocument(Iterable doc) throws CorruptIndexException, IOException { - addDocument(doc, analyzer); + public long addDocument(Iterable doc) throws CorruptIndexException, IOException { + return addDocument(doc, analyzer); } /** @@ -1314,8 +1317,8 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * @throws CorruptIndexException if the index is corrupt * @throws IOException if there is a low-level IO error */ - public void addDocument(Iterable doc, Analyzer analyzer) throws CorruptIndexException, IOException { - updateDocument(null, doc, analyzer); + public long addDocument(Iterable doc, Analyzer analyzer) throws CorruptIndexException, IOException { + return updateDocument(null, doc, analyzer); } /** @@ -1352,8 +1355,8 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * * @lucene.experimental */ - public void addDocuments(Iterable> docs) throws CorruptIndexException, IOException { - addDocuments(docs, analyzer); + public long addDocuments(Iterable> docs) throws CorruptIndexException, IOException { + return addDocuments(docs, analyzer); } /** @@ -1367,8 +1370,8 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * * @lucene.experimental */ - public void addDocuments(Iterable> docs, Analyzer analyzer) throws CorruptIndexException, IOException { - updateDocuments(null, docs, analyzer); + public long addDocuments(Iterable> docs, Analyzer analyzer) throws CorruptIndexException, IOException { + return updateDocuments(null, docs, analyzer); } /** @@ -1384,8 +1387,8 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * * @lucene.experimental */ - public void updateDocuments(Term delTerm, Iterable> docs) throws CorruptIndexException, IOException { - updateDocuments(delTerm, docs, analyzer); + public long updateDocuments(Term delTerm, Iterable> docs) throws CorruptIndexException, IOException { + return updateDocuments(delTerm, docs, analyzer); } /** @@ -1402,13 +1405,15 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * * @lucene.experimental */ - public void updateDocuments(Term delTerm, Iterable> docs, Analyzer analyzer) throws CorruptIndexException, IOException { + public long updateDocuments(Term delTerm, Iterable> docs, Analyzer analyzer) throws CorruptIndexException, IOException { ensureOpen(); try { boolean success = false; boolean anySegmentFlushed = false; + final long seqId; try { - anySegmentFlushed = docWriter.updateDocuments(docs, analyzer, delTerm); + seqId = docWriter.updateDocuments(docs, analyzer, delTerm); + anySegmentFlushed = (seqId & 1) == 1; success = true; } finally { if (!success && infoStream != null) { @@ -1418,8 +1423,10 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { if (anySegmentFlushed) { maybeMerge(); } + return seqId >>> 1; } catch (OutOfMemoryError oom) { handleOOM(oom, "updateDocuments"); + return -1; } } @@ -1434,12 +1441,13 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * @throws CorruptIndexException if the index is corrupt * @throws IOException if there is a low-level IO error */ - public void deleteDocuments(Term term) throws CorruptIndexException, IOException { + public long deleteDocuments(Term term) throws CorruptIndexException, IOException { ensureOpen(); try { - docWriter.deleteTerms(term); + return docWriter.deleteTerms(term); } catch (OutOfMemoryError oom) { handleOOM(oom, "deleteDocuments(Term)"); + return -1; // won't happen but compiler wants it } } @@ -1457,12 +1465,13 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * @throws CorruptIndexException if the index is corrupt * @throws IOException if there is a low-level IO error */ - public void deleteDocuments(Term... terms) throws CorruptIndexException, IOException { + public long deleteDocuments(Term... terms) throws CorruptIndexException, IOException { ensureOpen(); try { - docWriter.deleteTerms(terms); + return docWriter.deleteTerms(terms); } catch (OutOfMemoryError oom) { handleOOM(oom, "deleteDocuments(Term..)"); + return -1; // won't happen but compiler wants it } } @@ -1477,12 +1486,13 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * @throws CorruptIndexException if the index is corrupt * @throws IOException if there is a low-level IO error */ - public void deleteDocuments(Query query) throws CorruptIndexException, IOException { + public long deleteDocuments(Query query) throws CorruptIndexException, IOException { ensureOpen(); try { - docWriter.deleteQueries(query); + return docWriter.deleteQueries(query); } catch (OutOfMemoryError oom) { handleOOM(oom, "deleteDocuments(Query)"); + return -1; // won't happen but compiler wants it } } @@ -1499,12 +1509,13 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * @throws CorruptIndexException if the index is corrupt * @throws IOException if there is a low-level IO error */ - public void deleteDocuments(Query... queries) throws CorruptIndexException, IOException { + public long deleteDocuments(Query... queries) throws CorruptIndexException, IOException { ensureOpen(); try { - docWriter.deleteQueries(queries); + return docWriter.deleteQueries(queries); } catch (OutOfMemoryError oom) { handleOOM(oom, "deleteDocuments(Query..)"); + return -1; // won't happen but compiler wants it } } @@ -1525,9 +1536,9 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * @throws CorruptIndexException if the index is corrupt * @throws IOException if there is a low-level IO error */ - public void updateDocument(Term term, Iterable doc) throws CorruptIndexException, IOException { + public long updateDocument(Term term, Iterable doc) throws CorruptIndexException, IOException { ensureOpen(); - updateDocument(term, doc, getAnalyzer()); + return updateDocument(term, doc, getAnalyzer()); } /** @@ -1548,14 +1559,16 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * @throws CorruptIndexException if the index is corrupt * @throws IOException if there is a low-level IO error */ - public void updateDocument(Term term, Iterable doc, Analyzer analyzer) + public long updateDocument(Term term, Iterable doc, Analyzer analyzer) throws CorruptIndexException, IOException { ensureOpen(); try { boolean success = false; boolean anySegmentFlushed = false; + final long seqId; try { - anySegmentFlushed = docWriter.updateDocument(doc, analyzer, term); + seqId = docWriter.updateDocument(doc, analyzer, term); + anySegmentFlushed = (seqId & 1) == 1; success = true; } finally { if (!success && infoStream != null) @@ -1565,8 +1578,10 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { if (anySegmentFlushed) { maybeMerge(); } + return seqId >>> 1; } catch (OutOfMemoryError oom) { handleOOM(oom, "updateDocument"); + return -1; } } @@ -2707,9 +2722,9 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * href="#OOME">above for details.

* * @see #prepareCommit(Map) */ - public final void prepareCommit() throws CorruptIndexException, IOException { + public final long prepareCommit() throws CorruptIndexException, IOException { ensureOpen(); - prepareCommit(null); + return prepareCommit(null); } /**

Expert: prepare for commit, specifying @@ -2742,7 +2757,7 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * only "stick" if there are actually changes in the * index to commit. */ - public final void prepareCommit(Map commitUserData) throws CorruptIndexException, IOException { + public final long prepareCommit(Map commitUserData) throws CorruptIndexException, IOException { ensureOpen(false); if (infoStream != null) { @@ -2762,6 +2777,7 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { assert testPoint("startDoFlush"); SegmentInfos toCommit = null; boolean anySegmentsFlushed = false; + long seqId; // This is copied from doFlush, except it's modified to // clone & incRef the flushed SegmentInfos inside the @@ -2773,7 +2789,10 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { boolean flushSuccess = false; boolean success = false; try { - anySegmentsFlushed = docWriter.flushAllThreads(); + seqId = docWriter.flushAllThreads(); + + anySegmentsFlushed = (seqId & 1) == 1; + seqId = seqId >>> 1; if (!anySegmentsFlushed) { // prevent double increment since docWriter#doFlush increments the flushcount // if we flushed anything. @@ -2794,6 +2813,7 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { toCommit = (SegmentInfos) segmentInfos.clone(); pendingCommitChangeCount = changeCount; + pendingCommitSeqId = seqId; // This protects the segmentInfos we are now going // to commit. This is important in case, eg, while @@ -2814,6 +2834,7 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { } } catch (OutOfMemoryError oom) { handleOOM(oom, "prepareCommit"); + return -1; } boolean success = false; @@ -2831,6 +2852,7 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { } startCommit(toCommit, commitUserData); + return seqId; } // Used only by commit, below; lock order is commitLock -> IW @@ -2866,8 +2888,8 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * @see #prepareCommit * @see #commit(Map) */ - public final void commit() throws CorruptIndexException, IOException { - commit(null); + public final long commit() throws CorruptIndexException, IOException { + return commit(null); } /** Commits all changes to the index, specifying a @@ -2879,18 +2901,19 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { * you should immediately close the writer. See above for details.

*/ - public final void commit(Map commitUserData) throws CorruptIndexException, IOException { + public final long commit(Map commitUserData) throws CorruptIndexException, IOException { ensureOpen(); - commitInternal(commitUserData); + return commitInternal(commitUserData); } - private final void commitInternal(Map commitUserData) throws CorruptIndexException, IOException { + private final long commitInternal(Map commitUserData) throws CorruptIndexException, IOException { if (infoStream != null) { message("commit: start"); } + long seqId = pendingCommitSeqId; synchronized(commitLock) { if (infoStream != null) { @@ -2901,12 +2924,13 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { if (infoStream != null) { message("commit: now prepare"); } - prepareCommit(commitUserData); + seqId = prepareCommit(commitUserData); } else if (infoStream != null) { message("commit: already prepared"); } finishCommit(); + return seqId; } } @@ -2985,7 +3009,8 @@ public class IndexWriter implements Closeable, TwoPhaseCommit { synchronized (fullFlushLock) { try { - anySegmentFlushed = docWriter.flushAllThreads(); + long seqId = docWriter.flushAllThreads(); + anySegmentFlushed = (seqId & 1) == 1; success = true; } finally { docWriter.finishFullFlush(success); diff --git a/lucene/src/java/org/apache/lucene/util/TwoPhaseCommit.java b/lucene/src/java/org/apache/lucene/util/TwoPhaseCommit.java index efd2d69..4c74527 100644 --- a/lucene/src/java/org/apache/lucene/util/TwoPhaseCommit.java +++ b/lucene/src/java/org/apache/lucene/util/TwoPhaseCommit.java @@ -35,7 +35,7 @@ public interface TwoPhaseCommit { * 2-phase commit fails, {@link #rollback()} is called to discard all changes * since last successful commit. */ - public void prepareCommit() throws IOException; + public long prepareCommit() throws IOException; /** * Like {@link #commit()}, but takes an additional commit data to be included @@ -47,7 +47,7 @@ public interface TwoPhaseCommit { * * @see #prepareCommit() */ - public void prepareCommit(Map commitData) throws IOException; + public long prepareCommit(Map commitData) throws IOException; /** * The second phase of a 2-phase commit. Implementations should ideally do @@ -55,7 +55,7 @@ public interface TwoPhaseCommit { * after it returns, the caller can assume that the changes were successfully * committed to the underlying storage. */ - public void commit() throws IOException; + public long commit() throws IOException; /** * Like {@link #commit()}, but takes an additional commit data to be included @@ -64,7 +64,7 @@ public interface TwoPhaseCommit { * @see #commit() * @see #prepareCommit(Map) */ - public void commit(Map commitData) throws IOException; + public long commit(Map commitData) throws IOException; /** * Discards any changes that have occurred since the last commit. In a 2-phase diff --git a/lucene/src/java/org/apache/lucene/util/TwoPhaseCommitTool.java b/lucene/src/java/org/apache/lucene/util/TwoPhaseCommitTool.java index 66f3526..269162a 100644 --- a/lucene/src/java/org/apache/lucene/util/TwoPhaseCommitTool.java +++ b/lucene/src/java/org/apache/lucene/util/TwoPhaseCommitTool.java @@ -44,20 +44,20 @@ public final class TwoPhaseCommitTool { this.commitData = commitData; } - public void prepareCommit() throws IOException { - prepareCommit(commitData); + public long prepareCommit() throws IOException { + return prepareCommit(commitData); } - public void prepareCommit(Map commitData) throws IOException { - tpc.prepareCommit(this.commitData); + public long prepareCommit(Map commitData) throws IOException { + return tpc.prepareCommit(this.commitData); } - public void commit() throws IOException { - commit(commitData); + public long commit() throws IOException { + return commit(commitData); } - public void commit(Map commitData) throws IOException { - tpc.commit(this.commitData); + public long commit(Map commitData) throws IOException { + return tpc.commit(this.commitData); } public void rollback() throws IOException { diff --git a/lucene/src/test-framework/org/apache/lucene/index/ThreadedIndexingAndSearchingTestCase.java b/lucene/src/test-framework/org/apache/lucene/index/ThreadedIndexingAndSearchingTestCase.java index 425fe7b..bf8e1d3 100644 --- a/lucene/src/test-framework/org/apache/lucene/index/ThreadedIndexingAndSearchingTestCase.java +++ b/lucene/src/test-framework/org/apache/lucene/index/ThreadedIndexingAndSearchingTestCase.java @@ -95,24 +95,24 @@ public abstract class ThreadedIndexingAndSearchingTestCase extends LuceneTestCas return in; } - protected void updateDocuments(Term id, List> docs) throws Exception { - writer.updateDocuments(id, docs); + protected long updateDocuments(Term id, List> docs) throws Exception { + return writer.updateDocuments(id, docs); } - protected void addDocuments(Term id, List> docs) throws Exception { - writer.addDocuments(docs); + protected long addDocuments(Term id, List> docs) throws Exception { + return writer.addDocuments(docs); } - protected void addDocument(Term id, Iterable doc) throws Exception { - writer.addDocument(doc); + protected long addDocument(Term id, Iterable doc) throws Exception { + return writer.addDocument(doc); } - protected void updateDocument(Term term, Iterable doc) throws Exception { - writer.updateDocument(term, doc); + protected long updateDocument(Term term, Iterable doc) throws Exception { + return writer.updateDocument(term, doc); } - protected void deleteDocuments(Term term) throws Exception { - writer.deleteDocuments(term); + protected long deleteDocuments(Term term) throws Exception { + return writer.deleteDocuments(term); } protected void doAfterIndexingThreadDone() { @@ -523,7 +523,7 @@ public abstract class ThreadedIndexingAndSearchingTestCase extends LuceneTestCas for(int thread=0;thread operations = new ConcurrentSkipListMap(); + + public void testReBuildIndex() throws Exception { + runTest("TestSequenceIdsIndexing"); + } + + @Override + protected void doAfterIndexing() throws Exception { + IndexWriterConfig conf = new IndexWriterConfig(TEST_VERSION_CURRENT, new MockAnalyzer(random)); + Directory newDirectory = newDirectory(); + this.writer.commit(); + + IndexWriter writer = new IndexWriter(newDirectory, conf); + // rebuild the index from the sequence ids + for (Operation op : operations.values()) { + op.apply(writer); + } + writer.commit(); + + writer.close(); + IndexReader sourceReader = IndexReader.open(dir); + IndexReader targetReader = IndexReader.open(newDirectory); + IndexSearcher searcher = new IndexSearcher(targetReader); + + + assertEquals(sourceReader.numDocs(), targetReader.numDocs()); + int numDocs = sourceReader.maxDoc(); + Bits liveDocs = MultiFields.getLiveDocs(sourceReader); + for (int i = 0; i < numDocs; i++) { + Document sourceDocument = sourceReader.document(i); + String idValue = sourceDocument.get("docid"); + if (liveDocs.get(i)) { + TopDocs search = searcher.search(new TermQuery(new Term("docid", idValue)), 10); + assertEquals(1, search.totalHits); + Document targetDoc = searcher.doc(search.scoreDocs[0].doc); + for (IndexableField indexableField : sourceDocument) { + String string = targetDoc.get(indexableField.name()); + assertEquals(string, indexableField.stringValue()); + } + } else { + TopDocs search = searcher.search(new TermQuery(new Term("docid", idValue)), 10); + assertEquals(0, search.totalHits); + } + } + searcher.close(); + sourceReader.close(); + targetReader.close(); + newDirectory.close(); + } + + + @Override + protected long updateDocuments(Term id, + List> docs) throws Exception { + long seqId = super.updateDocuments(id, docs); + this.operations.put(seqId, new UpdateOperation(cloneDocs(docs), id)); + return seqId; + } + + @Override + protected long addDocuments(Term id, + List> docs) throws Exception { + long seqId = super.addDocuments(id, docs); + assertNull(this.operations.put(seqId, new AddOperation(cloneDocs(docs)))); + return seqId; + } + + private static List> cloneDocs( List> docs) { + List resultDocs = new ArrayList(); + for (Iterable doc : docs) { + resultDocs.add(_TestUtil.cloneDocument((Document)doc)); + } + return resultDocs; + } + + @Override + protected long addDocument(Term id, Iterable doc) + throws Exception { + long seqId = super.addDocument(id, doc); + assertNull(this.operations.put(seqId, new AddOperation(_TestUtil.cloneDocument((Document)doc)))); + return seqId; + } + + @Override + protected long updateDocument(Term term, + Iterable doc) throws Exception { + long seqId = super.updateDocument(term, doc); + assertNull(this.operations.put(seqId, new UpdateOperation(_TestUtil.cloneDocument((Document)doc), term))); + return seqId; + } + + @Override + protected long deleteDocuments(Term term) throws Exception { + long seqId = super.deleteDocuments(term); + assertNull(this.operations.put(seqId, new DeleteOperation(term))); + return seqId; + } + + private IndexSearcher fixedSearcher; + + protected IndexSearcher getCurrentSearcher() throws Exception { + return fixedSearcher; + } + + @Override + protected void releaseSearcher(IndexSearcher s) throws Exception { + if (s != fixedSearcher) { + // Final searcher: + s.getIndexReader().close(); + s.close(); + } + } + + @Override + protected void doSearching(ExecutorService es, long stopTime) + throws Exception { + } + + @Override + protected IndexSearcher getFinalSearcher() throws Exception { + final IndexReader r2; + if (random.nextBoolean()) { + r2 = writer.getReader(); + } else { + writer.commit(); + r2 = IndexReader.open(dir); + } + return newSearcher(r2); + } + + public abstract static class Operation { + public abstract void apply(IndexWriter writer) throws IOException; + } + + public static class AddOperation extends Operation { + public List> docs; + public Iterable doc; + + public AddOperation(List> docs) { + this.docs = docs; + } + + public AddOperation(Iterable doc) { + this.doc = doc; + } + + @Override + public void apply(IndexWriter writer) throws IOException { + if (doc != null) { + writer.addDocument(doc); + } else { + writer.addDocuments(docs); + } + } + } + + public static class UpdateOperation extends Operation { + public List> docs; + public Term delTerm; + public Iterable doc; + + public UpdateOperation(List> docs, + Term delTerm) { + this.docs = docs; + this.delTerm = delTerm; + } + + public UpdateOperation(Iterable doc, Term delTerm) { + this.doc = doc; + this.delTerm = delTerm; + } + + @Override + public void apply(IndexWriter writer) throws IOException { + if (doc != null) { + writer.updateDocument(delTerm, doc); + } else { + writer.updateDocuments(delTerm, docs); + } + } + } + + public static class DeleteOperation extends Operation { + public Term delTerm; + + public DeleteOperation(Term term) { + this.delTerm = term; + } + + @Override + public void apply(IndexWriter writer) throws IOException { + writer.deleteDocuments(delTerm); + } + } + +} diff --git a/lucene/src/test/org/apache/lucene/index/TestStressSeqIds.java b/lucene/src/test/org/apache/lucene/index/TestStressSeqIds.java new file mode 100644 index 0000000..f609300 --- /dev/null +++ b/lucene/src/test/org/apache/lucene/index/TestStressSeqIds.java @@ -0,0 +1,175 @@ +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. + */ + + +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import java.util.TreeSet; + +import org.apache.lucene.analysis.MockAnalyzer; +import org.apache.lucene.document.Document; +import org.apache.lucene.document.StringField; +import org.apache.lucene.document.TextField; +import org.apache.lucene.index.IndexWriterConfig.OpenMode; +import org.apache.lucene.store.Directory; +import org.apache.lucene.util.English; +import org.apache.lucene.util.LuceneTestCase; + +public class TestStressSeqIds extends LuceneTestCase { + private static abstract class TimedThread extends Thread { + volatile boolean failed; + int count; + private TimedThread[] allThreads; + private final long maxSeqId; + + abstract public long doWork() throws Throwable; + + TimedThread(TimedThread[] threads, long maxSeqId) { + this.allThreads = threads; + this.maxSeqId = maxSeqId; + } + + @Override + public void run() { + count = 0; + long seqId; + try { + do { + if (anyErrors()) break; + seqId = doWork(); + count++; + } while(seqId < maxSeqId); + } catch (Throwable e) { + System.out.println(Thread.currentThread() + ": exc"); + e.printStackTrace(System.out); + failed = true; + } + } + + private boolean anyErrors() { + for(int i=0;i seqIds = new HashSet(); + + public IndexerThread(IndexWriter writer, TimedThread[] threads, long maxSeqId) { + super(threads, maxSeqId); + this.writer = writer; + } + + @Override + public long doWork() throws Exception { + Long lastSeqId = Long.MIN_VALUE; + int numOps = atLeast(30); + // Add 10 docs: + for(int j=0; j< numOps; j++) { + + int n = random.nextInt(); + + int op = random.nextInt(10); + Document d = new Document(); + switch (op) { + case 5: + d.add(newField("id", Integer.toString(nextID++), StringField.TYPE_STORED)); + d.add(newField("contents", English.intToEnglish(n), TextField.TYPE_UNSTORED)); + lastSeqId = writer.addDocument(d); + break; + case 4: + case 3: + case 2: + lastSeqId = writer.deleteDocuments(new Term("id", ""+ random.nextInt(nextID))); + break; + case 1: + { + String id = Integer.toString(nextID++); + Document d1 = new Document(); + d.add(newField("id", id, StringField.TYPE_STORED)); + d.add(newField("contents", English.intToEnglish(n), + TextField.TYPE_UNSTORED)); + d1.add(newField("id", id, StringField.TYPE_STORED)); + d1.add(newField("contents", English.intToEnglish(n), + TextField.TYPE_UNSTORED)); + lastSeqId = writer.updateDocuments(new Term("id", id), + Arrays.asList(d, d1)); + } + break; + + default: + String id = Integer.toString(nextID++); + d.add(newField("id", id, StringField.TYPE_STORED)); + d.add(newField("contents", English.intToEnglish(n), TextField.TYPE_UNSTORED)); + lastSeqId = writer.updateDocument(new Term("id", id), d); + break; + } + assertFalse(seqIds.contains(lastSeqId)); + seqIds.add(lastSeqId); + } + return lastSeqId.longValue(); + } + } + + + public void runStressTest(Directory directory, MergeScheduler mergeScheduler) throws Exception { + IndexWriter modifier = new IndexWriter(directory, newIndexWriterConfig( + TEST_VERSION_CURRENT, new MockAnalyzer(random)) + .setOpenMode(OpenMode.CREATE).setMaxBufferedDocs(10).setMergeScheduler( + mergeScheduler)); + modifier.commit(); + + IndexerThread[] threads = new IndexerThread[10 + random.nextInt(TEST_NIGHTLY ? 5 : 3)]; + int numOps = atLeast(2000); + + for (int i = 0; i < threads.length; i++) { + IndexerThread indexerThread = new IndexerThread(modifier, threads, numOps); + threads[i] = indexerThread; + indexerThread.start(); + } + + for(int i=0;i allSeqIds = new TreeSet(); + int size = 0; + for (IndexerThread timedThread : threads) { + Set seqIds = timedThread.seqIds; + size += seqIds.size(); + assertTrue(allSeqIds.addAll(seqIds)); + } + assertEquals(size, allSeqIds.size()); + assertNull(allSeqIds.higher(maxSeqId)); + for(int i=0;i commitData) throws IOException { + public long prepareCommit(Map commitData) throws IOException { this.prepareCommitData = commitData; assertFalse("commit should not have been called before all prepareCommit were", commitCalled); if (failOnPrepare) { throw new IOException("failOnPrepare"); } + return -1; } - public void commit() throws IOException { - commit(null); + public long commit() throws IOException { + return commit(null); } - public void commit(Map commitData) throws IOException { + public long commit(Map commitData) throws IOException { this.commitData = commitData; commitCalled = true; if (failOnCommit) { throw new RuntimeException("failOnCommit"); } + return -1; } public void rollback() throws IOException {