You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by si...@apache.org on 2011/04/26 17:18:02 UTC
svn commit: r1096780 - in /lucene/dev/branches/realtime_search/lucene/src:
java/org/apache/lucene/index/ test/org/apache/lucene/index/
Author: simonw
Date: Tue Apr 26 15:18:02 2011
New Revision: 1096780
URL: http://svn.apache.org/viewvc?rev=1096780&view=rev
Log:
LUCENE-3023: more cleanup, asserts and javadoc
Modified:
lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriter.java
lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java
lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java
lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java
lucene/dev/branches/realtime_search/lucene/src/test/org/apache/lucene/index/TestFlushByRamOrCountsPolicy.java
Modified: lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriter.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriter.java?rev=1096780&r1=1096779&r2=1096780&view=diff
==============================================================================
--- lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriter.java (original)
+++ lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriter.java Tue Apr 26 15:18:02 2011
@@ -338,6 +338,11 @@ final class DocumentsWriter {
if (flushingDWPT != null) {
maybeMerge |= doFlush(flushingDWPT);
+ } else {
+ final DocumentsWriterPerThread nextPendingFlush = flushControl.nextPendingFlush();
+ if (nextPendingFlush != null) {
+ maybeMerge |= doFlush(nextPendingFlush);
+ }
}
return maybeMerge;
}
Modified: lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java?rev=1096780&r1=1096779&r2=1096780&view=diff
==============================================================================
--- lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java (original)
+++ lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterDeleteQueue.java Tue Apr 26 15:18:02 2011
@@ -64,7 +64,7 @@ import org.apache.lucene.search.Query;
final class DocumentsWriterDeleteQueue {
private volatile Node tail;
-
+
private static final AtomicReferenceFieldUpdater<DocumentsWriterDeleteQueue, Node> tailUpdater = AtomicReferenceFieldUpdater
.newUpdater(DocumentsWriterDeleteQueue.class, Node.class, "tail");
Modified: lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java?rev=1096780&r1=1096779&r2=1096780&view=diff
==============================================================================
--- lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java (original)
+++ lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java Tue Apr 26 15:18:02 2011
@@ -128,7 +128,7 @@ public final class DocumentsWriterFlushC
}
}
}
- final DocumentsWriterPerThread flushingDWPT = getFlushIfPending(perThread);
+ final DocumentsWriterPerThread flushingDWPT = tryCheckoutForFlush(perThread, false);
healthiness.updateStalled(this);
return flushingDWPT;
}
@@ -226,18 +226,6 @@ public final class DocumentsWriterFlushC
return null;
}
- private DocumentsWriterPerThread getFlushIfPending(ThreadState perThread) {
- if (numPending > 0) {
- final DocumentsWriterPerThread dwpt = perThread == null ? null
- : tryCheckoutForFlush(perThread, false);
- if (dwpt == null) {
- return nextPendingFlush();
- }
- return dwpt;
- }
- return null;
- }
-
@Override
public String toString() {
return "DocumentsWriterFlushControl [activeBytes=" + activeBytes
@@ -257,7 +245,7 @@ public final class DocumentsWriterFlushC
while (allActiveThreads.hasNext() && numPending > 0) {
ThreadState next = allActiveThreads.next();
if (next.flushPending) {
- DocumentsWriterPerThread dwpt = tryCheckoutForFlush(next, false);
+ final DocumentsWriterPerThread dwpt = tryCheckoutForFlush(next, false);
if (dwpt != null) {
return dwpt;
}
@@ -327,6 +315,7 @@ public final class DocumentsWriterFlushC
if (!next.isActive()) {
continue;
}
+ assert next.perThread.deleteQueue == flushingQueue || next.perThread.deleteQueue == documentsWriter.deleteQueue;
if (next.perThread.deleteQueue != flushingQueue) {
// this one is already a new DWPT
continue;
@@ -346,6 +335,7 @@ public final class DocumentsWriterFlushC
}
}
synchronized (this) {
+ assert assertBlockedFlushes(flushingQueue);
flushQueue.addAll(blockedFlushes);
blockedFlushes.clear();
flushQueue.addAll(toFlush);
@@ -357,6 +347,7 @@ public final class DocumentsWriterFlushC
assert flushQueue.isEmpty();
try {
if (!blockedFlushes.isEmpty()) {
+ assert assertBlockedFlushes(documentsWriter.deleteQueue);
flushQueue.addAll(blockedFlushes);
blockedFlushes.clear();
}
@@ -364,6 +355,14 @@ public final class DocumentsWriterFlushC
fullFlush = false;
}
}
+
+ boolean assertBlockedFlushes(DocumentsWriterDeleteQueue flushingQueue) {
+ Queue<DocumentsWriterPerThread> flushes = this.blockedFlushes;
+ for (DocumentsWriterPerThread documentsWriterPerThread : flushes) {
+ assert documentsWriterPerThread.deleteQueue == flushingQueue;
+ }
+ return true;
+ }
synchronized void abortFullFlushes() {
try {
Modified: lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java?rev=1096780&r1=1096779&r2=1096780&view=diff
==============================================================================
--- lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java (original)
+++ lucene/dev/branches/realtime_search/lucene/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java Tue Apr 26 15:18:02 2011
@@ -322,6 +322,11 @@ public class DocumentsWriterPerThread {
numDocsInRAM = 0;
}
+ /**
+ * Prepares this DWPT for flushing. This method will freeze and return the
+ * {@link DocumentsWriterDeleteQueue}s global buffer and apply all pending
+ * deletes to this DWPT.
+ */
FrozenBufferedDeletes prepareFlush() {
assert numDocsInRAM > 0;
final FrozenBufferedDeletes globalDeletes = deleteQueue.freezeGlobalBuffer(deleteSlice);
@@ -330,6 +335,7 @@ public class DocumentsWriterPerThread {
if (deleteSlice != null) {
// apply all deletes before we flush and release the delete slice
deleteSlice.apply(pendingDeletes, numDocsInRAM);
+ assert deleteSlice.isEmpty();
deleteSlice = null;
}
return globalDeletes;
@@ -338,6 +344,7 @@ public class DocumentsWriterPerThread {
/** Flush all pending docs to a new segment */
FlushedSegment flush() throws IOException {
assert numDocsInRAM > 0;
+ assert deleteSlice == null : "all deletes must be applied in prepareFlush";
flushState = new SegmentWriteState(infoStream, directory, segment, fieldInfos,
numDocsInRAM, writer.getConfig().getTermIndexInterval(),
fieldInfos.buildSegmentCodecs(true), pendingDeletes);
Modified: lucene/dev/branches/realtime_search/lucene/src/test/org/apache/lucene/index/TestFlushByRamOrCountsPolicy.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/realtime_search/lucene/src/test/org/apache/lucene/index/TestFlushByRamOrCountsPolicy.java?rev=1096780&r1=1096779&r2=1096780&view=diff
==============================================================================
--- lucene/dev/branches/realtime_search/lucene/src/test/org/apache/lucene/index/TestFlushByRamOrCountsPolicy.java (original)
+++ lucene/dev/branches/realtime_search/lucene/src/test/org/apache/lucene/index/TestFlushByRamOrCountsPolicy.java Tue Apr 26 15:18:02 2011
@@ -199,7 +199,6 @@ public class TestFlushByRamOrCountsPolic
for (int x = 0; x < threads.length; x++) {
threads[x].join();
}
-
assertEquals(" all flushes must be due", 0, flushControl.flushBytes());
assertEquals(numDocumentsToIndex, writer.numDocs());
assertEquals(numDocumentsToIndex, writer.maxDoc());
@@ -334,6 +333,7 @@ public class TestFlushByRamOrCountsPolic
}
}
}
+ writer.commit();
} catch (Throwable ex) {
throw new RuntimeException(ex);
}