You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by mi...@apache.org on 2016/06/05 09:51:06 UTC
[05/19] lucene-solr:master: sequence numbers: resolve last nocommits
sequence numbers: resolve last nocommits
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/39de6896
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/39de6896
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/39de6896
Branch: refs/heads/master
Commit: 39de68963b13450a8ca0f044a9deac857c1c823c
Parents: 38e9822
Author: Mike McCandless <mi...@apache.org>
Authored: Wed May 25 18:41:21 2016 -0400
Committer: Mike McCandless <mi...@apache.org>
Committed: Wed May 25 18:41:21 2016 -0400
----------------------------------------------------------------------
.../index/DocumentsWriterDeleteQueue.java | 37 ++------------------
.../index/DocumentsWriterFlushControl.java | 4 +--
.../org/apache/lucene/index/IndexWriter.java | 20 ++++++++---
.../search/ControlledRealTimeReopenThread.java | 3 --
.../index/TestIndexingSequenceNumbers.java | 25 ++++++-------
5 files changed, 31 insertions(+), 58 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/39de6896/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java
----------------------------------------------------------------------
diff --git a/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java b/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java
index 5d0e83d..abb735d 100644
--- a/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java
+++ b/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java
@@ -150,41 +150,10 @@ final class DocumentsWriterDeleteQueue implements Accountable {
return seqNo;
}
- // nocommit can we remove the sync'd
synchronized long add(Node<?> newNode) {
- /*
- * this non-blocking / 'wait-free' linked list add was inspired by Apache
- * Harmony's ConcurrentLinkedQueue Implementation.
- */
- while (true) {
- final Node<?> currentTail = this.tail;
- final Node<?> tailNext = currentTail.next;
- if (tail == currentTail) {
- if (tailNext != null) {
- /*
- * we are in intermediate state here. the tails next pointer has been
- * advanced but the tail itself might not be updated yet. help to
- * advance the tail and try again updating it.
- */
- tailUpdater.compareAndSet(this, currentTail, tailNext); // can fail
- } else {
- /*
- * we are in quiescent state and can try to insert the new node to the
- * current tail if we fail to insert we just retry the operation since
- * somebody else has already added its item
- */
- if (currentTail.casNext(null, newNode)) {
- /*
- * 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, newNode);
- return seqNo.getAndIncrement();
- }
- }
- }
- }
+ tail.next = newNode;
+ tail = newNode;
+ return seqNo.getAndIncrement();
}
boolean anyChanges() {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/39de6896/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java
----------------------------------------------------------------------
diff --git a/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java b/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java
index f388f46..bd8015d 100644
--- a/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java
+++ b/lucene/core/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java
@@ -484,8 +484,8 @@ final class DocumentsWriterFlushControl implements Accountable {
// jump over any possible in flight ops:
seqNo = documentsWriter.deleteQueue.seqNo.get() + perThreadPool.getActiveThreadStateCount();
- // nocommit is this (active thread state count) always enough of a gap? what if new indexing thread sneaks in just now? it would
- // have to get this next delete queue?
+ // Insert a gap in seqNo of current active thread count, in the worst case those threads now have one operation in flight. It's fine
+ // if we have some sequence numbers that were never assigned:
DocumentsWriterDeleteQueue newQueue = new DocumentsWriterDeleteQueue(flushingQueue.generation+1, seqNo+1);
documentsWriter.deleteQueue = newQueue;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/39de6896/lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
----------------------------------------------------------------------
diff --git a/lucene/core/src/java/org/apache/lucene/index/IndexWriter.java b/lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
index 9a520b1..945399c 100644
--- a/lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
+++ b/lucene/core/src/java/org/apache/lucene/index/IndexWriter.java
@@ -426,7 +426,7 @@ public class IndexWriter implements Closeable, TwoPhaseCommit, Accountable {
boolean success = false;
synchronized (fullFlushLock) {
try {
- // nocommit should we make this available in the returned NRT reader?
+ // TODO: should we somehow make this available in the returned NRT reader?
long seqNo = docWriter.flushAllThreads();
if (seqNo < 0) {
anyChanges = true;
@@ -2984,7 +2984,11 @@ public class IndexWriter implements Closeable, TwoPhaseCommit, Accountable {
}
startCommit(toCommit);
success = true;
- return seqNo;
+ if (pendingCommit == null) {
+ return -1;
+ } else {
+ return seqNo;
+ }
} finally {
if (!success) {
synchronized (this) {
@@ -3058,6 +3062,12 @@ public class IndexWriter implements Closeable, TwoPhaseCommit, Accountable {
* loss it may still lose data. Lucene cannot guarantee
* consistency on such devices. </p>
*
+ * <p> If nothing was committed, because there were no
+ * pending changes, this returns -1. Otherwise, it returns
+ * the sequence number such that all indexing operations
+ * prior to this sequence will be included in the commit
+ * point, and all other operations will not. </p>
+ *
* @see #prepareCommit
*/
@Override
@@ -4978,9 +4988,11 @@ public class IndexWriter implements Closeable, TwoPhaseCommit, Accountable {
};
}
- // nocommit javadocs
+ /** Returns the last sequence number.
+ *
+ * @lucene.experimental */
public long getLastSequenceNumber() {
ensureOpen();
- return docWriter.deleteQueue.seqNo.get();
+ return docWriter.deleteQueue.seqNo.get()-1;
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/39de6896/lucene/core/src/java/org/apache/lucene/search/ControlledRealTimeReopenThread.java
----------------------------------------------------------------------
diff --git a/lucene/core/src/java/org/apache/lucene/search/ControlledRealTimeReopenThread.java b/lucene/core/src/java/org/apache/lucene/search/ControlledRealTimeReopenThread.java
index d015ae9..466d793 100644
--- a/lucene/core/src/java/org/apache/lucene/search/ControlledRealTimeReopenThread.java
+++ b/lucene/core/src/java/org/apache/lucene/search/ControlledRealTimeReopenThread.java
@@ -151,9 +151,6 @@ public class ControlledRealTimeReopenThread<T> extends Thread implements Closeab
*/
public synchronized boolean waitForGeneration(long targetGen, int maxMS) throws InterruptedException {
final long curGen = writer.getLastSequenceNumber();
- if (targetGen > curGen) {
- throw new IllegalArgumentException("targetGen=" + targetGen + " was never returned by the ReferenceManager instance (current gen=" + curGen + ")");
- }
if (targetGen > searchingGen) {
// Notify the reopen thread that the waitingGen has
// changed, so it may wake up and realize it should
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/39de6896/lucene/core/src/test/org/apache/lucene/index/TestIndexingSequenceNumbers.java
----------------------------------------------------------------------
diff --git a/lucene/core/src/test/org/apache/lucene/index/TestIndexingSequenceNumbers.java b/lucene/core/src/test/org/apache/lucene/index/TestIndexingSequenceNumbers.java
index f2e9636..fb9b9ab 100644
--- a/lucene/core/src/test/org/apache/lucene/index/TestIndexingSequenceNumbers.java
+++ b/lucene/core/src/test/org/apache/lucene/index/TestIndexingSequenceNumbers.java
@@ -74,8 +74,7 @@ public class TestIndexingSequenceNumbers extends LuceneTestCase {
int iters = atLeast(100);
for(int iter=0;iter<iters;iter++) {
Directory dir = newDirectory();
- // nocommit use RandomIndexWriter
- final IndexWriter w = new IndexWriter(dir, newIndexWriterConfig());
+ final RandomIndexWriter w = new RandomIndexWriter(random(), dir);
Thread[] threads = new Thread[TestUtil.nextInt(random(), 2, 5)];
final CountDownLatch startingGun = new CountDownLatch(1);
final long[] seqNos = new long[threads.length];
@@ -117,7 +116,7 @@ public class TestIndexingSequenceNumbers extends LuceneTestCase {
}
// make sure all sequence numbers were different
assertEquals(threads.length, allSeqNos.size());
- DirectoryReader r = DirectoryReader.open(w);
+ DirectoryReader r = w.getReader();
IndexSearcher s = newSearcher(r);
TopDocs hits = s.search(new TermQuery(id), 1);
assertEquals(1, hits.totalHits);
@@ -142,10 +141,12 @@ public class TestIndexingSequenceNumbers extends LuceneTestCase {
final int idCount = TestUtil.nextInt(random(), 10, 1000);
Directory dir = newDirectory();
- // nocommit use RandomIndexWriter
IndexWriterConfig iwc = newIndexWriterConfig();
iwc.setIndexDeletionPolicy(NoDeletionPolicy.INSTANCE);
+
+ // Cannot use RIW since it randomly commits:
final IndexWriter w = new IndexWriter(dir, iwc);
+
final int numThreads = TestUtil.nextInt(random(), 2, 5);
Thread[] threads = new Thread[numThreads];
//System.out.println("TEST: iter=" + iter + " opCount=" + opCount + " idCount=" + idCount + " threadCount=" + threads.length);
@@ -171,11 +172,10 @@ public class TestIndexingSequenceNumbers extends LuceneTestCase {
if (random().nextInt(500) == 17) {
op.what = 2;
synchronized(commitLock) {
- if (w.hasUncommittedChanges()) {
- op.seqNo = w.commit();
+ op.seqNo = w.commit();
+ if (op.seqNo != -1) {
commits.add(op);
}
- //System.out.println("done commit seqNo=" + op.seqNo);
}
} else {
op.id = random().nextInt(idCount);
@@ -215,14 +215,11 @@ public class TestIndexingSequenceNumbers extends LuceneTestCase {
thread.join();
}
- /*
- // nocommit: why does this make the assertEquals angry...?
- if (w.hasUncommittedChanges()) {
- Operation commitOp = new Operation();
- commitOp.seqNo = w.commit();
+ Operation commitOp = new Operation();
+ commitOp.seqNo = w.commit();
+ if (commitOp.seqNo != -1) {
commits.add(commitOp);
}
- */
List<IndexCommit> indexCommits = DirectoryReader.listCommits(dir);
assertEquals(commits.size(), indexCommits.size());
@@ -296,6 +293,4 @@ public class TestIndexingSequenceNumbers extends LuceneTestCase {
dir.close();
}
-
- // nocommit test that does n ops across threads, then does it again with a single index / single thread, and assert indices are the same
}