Apache
/
lucene
mirror of https://github.com/apache/lucene.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Concurrently flush next buffer during commit in RandomIndexWriter (#607) 

This is a spinn-off from `LUCENE-8700` that is satisfied by IndexWriter#flushNextBuffer.
The idea here is to additionally call flushNextBuffer in RandomIndexWriter for better
test coverage. This is a test-only change.
This commit is contained in:
Simon Willnauer 2019-03-14 15:43:35 +01:00 committed by GitHub
parent c2a6772f1e
commit ffb1fc83de
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 41 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
lucene/test-framework/src/java/org/apache/lucene/index/RandomIndexWriter.java
View File

@ -21,6 +21,7 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Random;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.MockAnalyzer;
@ -198,16 +199,7 @@ public class RandomIndexWriter implements Closeable {
LuceneTestCase.maybeChangeLiveIndexWriterConfig(r, w.getConfig());
if (docCount++ == flushAt) {
if (r.nextBoolean()) {
if (LuceneTestCase.VERBOSE) {
System.out.println("RIW.add/updateDocument: now flushing the largest writer at docCount=" + docCount);
}
int activeThreadStateCount = w.docWriter.perThreadPool.getActiveThreadStateCount();
int numFlushes = Math.min(1, r.nextInt(activeThreadStateCount+1));
for (int i = 0; i < numFlushes; i++) {
if (w.flushNextBuffer() == false) {
break; // stop once we didn't flush anything
}
}
flushAllBuffersSequentially();
} else if (r.nextBoolean()) {
if (LuceneTestCase.VERBOSE) {
System.out.println("RIW.add/updateDocument: now doing a flush at docCount=" + docCount);
@ -226,6 +218,19 @@ public class RandomIndexWriter implements Closeable {
}
}
}
private void flushAllBuffersSequentially() throws IOException {
if (LuceneTestCase.VERBOSE) {
System.out.println("RIW.add/updateDocument: now flushing the largest writer at docCount=" + docCount);
}
int activeThreadStateCount = w.docWriter.perThreadPool.getActiveThreadStateCount();
int numFlushes = Math.min(1, r.nextInt(activeThreadStateCount+1));
for (int i = 0; i < numFlushes; i++) {
if (w.flushNextBuffer() == false) {
break; // stop once we didn't flush anything
}
}
}
public long addDocuments(Iterable<? extends Iterable<? extends IndexableField>> docs) throws IOException {
LuceneTestCase.maybeChangeLiveIndexWriterConfig(r, w.getConfig());
@ -312,9 +317,30 @@ public class RandomIndexWriter implements Closeable {
public long commit() throws IOException {
LuceneTestCase.maybeChangeLiveIndexWriterConfig(r, w.getConfig());
if (r.nextInt(10) == 0) {
AtomicReference<Throwable> exception = new AtomicReference<>();
Thread t = new Thread(() -> {
try {
flushAllBuffersSequentially();
} catch (IOException e) {
exception.set(e);
}
});
t.start();
long seqId = w.commit();
try {
t.join();
} catch (InterruptedException e) {
throw new AssertionError(e);
}
if (exception.get() != null) {
throw new AssertionError(exception.get());
}
return seqId;
}
return w.commit();
}
public IndexWriter.DocStats getDocStats() {
return w.getDocStats();
}
@ -328,8 +354,8 @@ public class RandomIndexWriter implements Closeable {
return getReader(true, false);
}
private boolean doRandomForceMerge = true;
private boolean doRandomForceMergeAssert = true;
private boolean doRandomForceMerge;
private boolean doRandomForceMergeAssert;
public void forceMergeDeletes(boolean doWait) throws IOException {
LuceneTestCase.maybeChangeLiveIndexWriterConfig(r, w.getConfig());
@ -491,7 +517,7 @@ public class RandomIndexWriter implements Closeable {
* Simple interface that is executed for each <tt>TP</tt> {@link InfoStream} component
* message. See also {@link RandomIndexWriter#mockIndexWriter(Random, Directory, IndexWriterConfig, TestPoint)}
*/
public static interface TestPoint {
public abstract void apply(String message);
public interface TestPoint {
void apply(String message);
}
}