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 2011/09/25 17:18:13 UTC
svn commit: r1175413 [2/2] - in /lucene/dev/trunk/lucene: contrib/
contrib/misc/src/java/org/apache/lucene/index/
contrib/misc/src/java/org/apache/lucene/search/
contrib/misc/src/test/org/apache/lucene/index/
contrib/misc/src/test/org/apache/lucene/sea...
Modified: lucene/dev/trunk/lucene/src/test/org/apache/lucene/index/TestNRTThreads.java
URL: http://svn.apache.org/viewvc/lucene/dev/trunk/lucene/src/test/org/apache/lucene/index/TestNRTThreads.java?rev=1175413&r1=1175412&r2=1175413&view=diff
==============================================================================
--- lucene/dev/trunk/lucene/src/test/org/apache/lucene/index/TestNRTThreads.java (original)
+++ lucene/dev/trunk/lucene/src/test/org/apache/lucene/index/TestNRTThreads.java Sun Sep 25 15:18:13 2011
@@ -17,329 +17,26 @@ package org.apache.lucene.index;
* limitations under the License.
*/
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import org.apache.lucene.analysis.MockAnalyzer;
-import org.apache.lucene.document.Document;
-import org.apache.lucene.document.Field;
-import org.apache.lucene.document.StringField;
-import org.apache.lucene.document.TextField;
-import org.apache.lucene.index.codecs.CodecProvider;
+
import org.apache.lucene.search.IndexSearcher;
-import org.apache.lucene.search.PhraseQuery;
-import org.apache.lucene.search.Query;
-import org.apache.lucene.search.ScoreDoc;
-import org.apache.lucene.search.Sort;
-import org.apache.lucene.search.SortField;
-import org.apache.lucene.search.TermQuery;
-import org.apache.lucene.search.TopDocs;
import org.apache.lucene.store.MockDirectoryWrapper;
-import org.apache.lucene.util.Bits;
-import org.apache.lucene.util.BytesRef;
-import org.apache.lucene.util.LineFileDocs;
-import org.apache.lucene.util.LuceneTestCase;
-import org.apache.lucene.util.NamedThreadFactory;
-import org.apache.lucene.util._TestUtil;
-import org.junit.Test;
// TODO
// - mix in optimize, addIndexes
// - randomoly mix in non-congruent docs
-public class TestNRTThreads extends LuceneTestCase {
-
- private static class SubDocs {
- public final String packID;
- public final List<String> subIDs;
- public boolean deleted;
-
- public SubDocs(String packID, List<String> subIDs) {
- this.packID = packID;
- this.subIDs = subIDs;
- }
- }
-
- @Test
- public void testNRTThreads() throws Exception {
-
- final long t0 = System.currentTimeMillis();
-
- final String defaultCodec = CodecProvider.getDefault().getDefaultFieldCodec();
- if (defaultCodec.equals("SimpleText") || defaultCodec.equals("Memory")) {
- // no
- CodecProvider.getDefault().setDefaultFieldCodec("Standard");
- }
-
- final LineFileDocs docs = new LineFileDocs(random);
- final File tempDir = _TestUtil.getTempDir("nrtopenfiles");
- final MockDirectoryWrapper dir = newFSDirectory(tempDir);
- dir.setCheckIndexOnClose(false); // don't double-checkIndex, we do it ourselves.
- final IndexWriterConfig conf = newIndexWriterConfig(TEST_VERSION_CURRENT, new MockAnalyzer(random));
-
- if (LuceneTestCase.TEST_NIGHTLY) {
- // newIWConfig makes smallish max seg size, which
- // results in tons and tons of segments for this test
- // when run nightly:
- MergePolicy mp = conf.getMergePolicy();
- if (mp instanceof TieredMergePolicy) {
- ((TieredMergePolicy) mp).setMaxMergedSegmentMB(5000.);
- } else if (mp instanceof LogByteSizeMergePolicy) {
- ((LogByteSizeMergePolicy) mp).setMaxMergeMB(1000.);
- } else if (mp instanceof LogMergePolicy) {
- ((LogMergePolicy) mp).setMaxMergeDocs(100000);
- }
- }
-
- conf.setMergedSegmentWarmer(new IndexWriter.IndexReaderWarmer() {
- @Override
- public void warm(IndexReader reader) throws IOException {
- if (VERBOSE) {
- System.out.println("TEST: now warm merged reader=" + reader);
- }
- final int maxDoc = reader.maxDoc();
- final Bits liveDocs = reader.getLiveDocs();
- int sum = 0;
- final int inc = Math.max(1, maxDoc/50);
- for(int docID=0;docID<maxDoc;docID += inc) {
- if (liveDocs == null || liveDocs.get(docID)) {
- final Document doc = reader.document(docID);
- sum += doc.getFields().size();
- }
- }
+public class TestNRTThreads extends ThreadedIndexingAndSearchingTestCase {
+
+ @Override
+ protected void doSearching(ExecutorService es, long stopTime) throws Exception {
- IndexSearcher searcher = newSearcher(reader);
- sum += searcher.search(new TermQuery(new Term("body", "united")), 10).totalHits;
- searcher.close();
-
- if (VERBOSE) {
- System.out.println("TEST: warm visited " + sum + " fields");
- }
- }
- });
-
- final IndexWriter writer = new IndexWriter(dir, conf);
- if (VERBOSE) {
- writer.setInfoStream(System.out);
- }
- _TestUtil.reduceOpenFiles(writer);
-
- final int NUM_INDEX_THREADS = 2;
- final int NUM_SEARCH_THREADS = 3;
-
- final int RUN_TIME_SEC = LuceneTestCase.TEST_NIGHTLY ? 300 : RANDOM_MULTIPLIER;
-
- final AtomicBoolean failed = new AtomicBoolean();
- final AtomicInteger addCount = new AtomicInteger();
- final AtomicInteger delCount = new AtomicInteger();
- final AtomicInteger packCount = new AtomicInteger();
-
- final Set<String> delIDs = Collections.synchronizedSet(new HashSet<String>());
- final List<SubDocs> allSubDocs = Collections.synchronizedList(new ArrayList<SubDocs>());
-
- final long stopTime = System.currentTimeMillis() + RUN_TIME_SEC*1000;
- Thread[] threads = new Thread[NUM_INDEX_THREADS];
- for(int thread=0;thread<NUM_INDEX_THREADS;thread++) {
- threads[thread] = new Thread() {
- @Override
- public void run() {
- // TODO: would be better if this were cross thread, so that we make sure one thread deleting anothers added docs works:
- final List<String> toDeleteIDs = new ArrayList<String>();
- final List<SubDocs> toDeleteSubDocs = new ArrayList<SubDocs>();
- while(System.currentTimeMillis() < stopTime && !failed.get()) {
- try {
- Document doc = docs.nextDoc();
- if (doc == null) {
- break;
- }
- final String addedField;
- if (random.nextBoolean()) {
- addedField = "extra" + random.nextInt(10);
- doc.add(new TextField(addedField, "a random field"));
- } else {
- addedField = null;
- }
- if (random.nextBoolean()) {
- if (VERBOSE) {
- //System.out.println(Thread.currentThread().getName() + ": add doc id:" + doc.get("docid"));
- }
-
- if (random.nextBoolean()) {
- // Add a pack of adjacent sub-docs
- final String packID;
- final SubDocs delSubDocs;
- if (toDeleteSubDocs.size() > 0 && random.nextBoolean()) {
- delSubDocs = toDeleteSubDocs.get(random.nextInt(toDeleteSubDocs.size()));
- assert !delSubDocs.deleted;
- toDeleteSubDocs.remove(delSubDocs);
- // reuse prior packID
- packID = delSubDocs.packID;
- } else {
- delSubDocs = null;
- // make new packID
- packID = packCount.getAndIncrement() + "";
- }
-
- final Field packIDField = newField("packID", packID, StringField.TYPE_STORED);
- final List<String> docIDs = new ArrayList<String>();
- final SubDocs subDocs = new SubDocs(packID, docIDs);
- final List<Document> docsList = new ArrayList<Document>();
-
- allSubDocs.add(subDocs);
- doc.add(packIDField);
- docsList.add(_TestUtil.cloneDocument(doc));
- docIDs.add(doc.get("docid"));
-
- final int maxDocCount = _TestUtil.nextInt(random, 1, 10);
- while(docsList.size() < maxDocCount) {
- doc = docs.nextDoc();
- if (doc == null) {
- break;
- }
- docsList.add(_TestUtil.cloneDocument(doc));
- docIDs.add(doc.get("docid"));
- }
- addCount.addAndGet(docsList.size());
-
- if (delSubDocs != null) {
- delSubDocs.deleted = true;
- delIDs.addAll(delSubDocs.subIDs);
- delCount.addAndGet(delSubDocs.subIDs.size());
- if (VERBOSE) {
- System.out.println("TEST: update pack packID=" + delSubDocs.packID + " count=" + docsList.size() + " docs=" + docIDs);
- }
- writer.updateDocuments(new Term("packID", delSubDocs.packID), docsList);
- /*
- // non-atomic:
- writer.deleteDocuments(new Term("packID", delSubDocs.packID));
- for(Document subDoc : docsList) {
- writer.addDocument(subDoc);
- }
- */
- } else {
- if (VERBOSE) {
- System.out.println("TEST: add pack packID=" + packID + " count=" + docsList.size() + " docs=" + docIDs);
- }
- writer.addDocuments(docsList);
-
- /*
- // non-atomic:
- for(Document subDoc : docsList) {
- writer.addDocument(subDoc);
- }
- */
- }
- doc.removeField("packID");
-
- if (random.nextInt(5) == 2) {
- if (VERBOSE) {
- //System.out.println(Thread.currentThread().getName() + ": buffer del id:" + packID);
- }
- toDeleteSubDocs.add(subDocs);
- }
-
- } else {
- writer.addDocument(doc);
- addCount.getAndIncrement();
-
- if (random.nextInt(5) == 3) {
- if (VERBOSE) {
- //System.out.println(Thread.currentThread().getName() + ": buffer del id:" + doc.get("docid"));
- }
- toDeleteIDs.add(doc.get("docid"));
- }
- }
- } else {
- // we use update but it never replaces a
- // prior doc
- if (VERBOSE) {
- //System.out.println(Thread.currentThread().getName() + ": update doc id:" + doc.get("docid"));
- }
- writer.updateDocument(new Term("docid", doc.get("docid")), doc);
- addCount.getAndIncrement();
-
- if (random.nextInt(5) == 3) {
- if (VERBOSE) {
- //System.out.println(Thread.currentThread().getName() + ": buffer del id:" + doc.get("docid"));
- }
- toDeleteIDs.add(doc.get("docid"));
- }
- }
-
- if (random.nextInt(30) == 17) {
- if (VERBOSE) {
- //System.out.println(Thread.currentThread().getName() + ": apply " + toDeleteIDs.size() + " deletes");
- }
- for(String id : toDeleteIDs) {
- if (VERBOSE) {
- //System.out.println(Thread.currentThread().getName() + ": del term=id:" + id);
- }
- writer.deleteDocuments(new Term("docid", id));
- }
- final int count = delCount.addAndGet(toDeleteIDs.size());
- if (VERBOSE) {
- //System.out.println(Thread.currentThread().getName() + ": tot " + count + " deletes");
- }
- delIDs.addAll(toDeleteIDs);
- toDeleteIDs.clear();
-
- for(SubDocs subDocs : toDeleteSubDocs) {
- assert !subDocs.deleted;
- writer.deleteDocuments(new Term("packID", subDocs.packID));
- subDocs.deleted = true;
- if (VERBOSE) {
- System.out.println(" del subs: " + subDocs.subIDs + " packID=" + subDocs.packID);
- }
- delIDs.addAll(subDocs.subIDs);
- delCount.addAndGet(subDocs.subIDs.size());
- }
- toDeleteSubDocs.clear();
- }
- if (addedField != null) {
- doc.removeField(addedField);
- }
- } catch (Throwable t) {
- System.out.println(Thread.currentThread().getName() + ": hit exc");
- t.printStackTrace();
- failed.set(true);
- throw new RuntimeException(t);
- }
- }
- if (VERBOSE) {
- System.out.println(Thread.currentThread().getName() + ": indexing done");
- }
- }
- };
- threads[thread].setDaemon(true);
- threads[thread].start();
- }
-
- if (VERBOSE) {
- System.out.println("TEST: DONE start indexing threads [" + (System.currentTimeMillis()-t0) + " ms]");
- }
-
- // let index build up a bit
- Thread.sleep(100);
+ boolean anyOpenDelFiles = false;
IndexReader r = IndexReader.open(writer, true);
- boolean any = false;
- // silly starting guess:
- final AtomicInteger totTermCount = new AtomicInteger(100);
-
- final ExecutorService es = Executors.newCachedThreadPool(new NamedThreadFactory("NRT search threads"));
-
- while(System.currentTimeMillis() < stopTime && !failed.get()) {
+ while (System.currentTimeMillis() < stopTime && !failed.get()) {
if (random.nextBoolean()) {
if (VERBOSE) {
System.out.println("TEST: now reopen r=" + r);
@@ -355,11 +52,11 @@ public class TestNRTThreads extends Luce
}
r.close();
writer.commit();
- final Set<String> openDeletedFiles = dir.getOpenDeletedFiles();
+ final Set<String> openDeletedFiles = ((MockDirectoryWrapper) dir).getOpenDeletedFiles();
if (openDeletedFiles.size() > 0) {
System.out.println("OBD files: " + openDeletedFiles);
}
- any |= openDeletedFiles.size() > 0;
+ anyOpenDelFiles |= openDeletedFiles.size() > 0;
//assertEquals("open but deleted: " + openDeletedFiles, 0, openDeletedFiles.size());
if (VERBOSE) {
System.out.println("TEST: now open");
@@ -372,203 +69,52 @@ public class TestNRTThreads extends Luce
//System.out.println("numDocs=" + r.numDocs() + "
//openDelFileCount=" + dir.openDeleteFileCount());
- smokeTestReader(r);
-
if (r.numDocs() > 0) {
-
- final IndexSearcher s = new IndexSearcher(r, es);
-
- // run search threads
- final long searchStopTime = System.currentTimeMillis() + 500;
- final Thread[] searchThreads = new Thread[NUM_SEARCH_THREADS];
- final AtomicInteger totHits = new AtomicInteger();
- for(int thread=0;thread<NUM_SEARCH_THREADS;thread++) {
- searchThreads[thread] = new Thread() {
- @Override
- public void run() {
- try {
- TermsEnum termsEnum = MultiFields.getTerms(s.getIndexReader(), "body").iterator();
- int seenTermCount = 0;
- int shift;
- int trigger;
- if (totTermCount.get() < 10) {
- shift = 0;
- trigger = 1;
- } else {
- trigger = totTermCount.get()/10;
- shift = random.nextInt(trigger);
- }
- while(System.currentTimeMillis() < searchStopTime) {
- BytesRef term = termsEnum.next();
- if (term == null) {
- if (seenTermCount < 10) {
- break;
- }
- totTermCount.set(seenTermCount);
- seenTermCount = 0;
- trigger = totTermCount.get()/10;
- //System.out.println("trigger " + trigger);
- shift = random.nextInt(trigger);
- termsEnum.seekCeil(new BytesRef(""));
- continue;
- }
- seenTermCount++;
- // search 10 terms
- if (trigger == 0) {
- trigger = 1;
- }
- if ((seenTermCount + shift) % trigger == 0) {
- //if (VERBOSE) {
- //System.out.println(Thread.currentThread().getName() + " now search body:" + term.utf8ToString());
- //}
- totHits.addAndGet(runQuery(s, new TermQuery(new Term("body", term))));
- }
- }
- if (VERBOSE) {
- System.out.println(Thread.currentThread().getName() + ": search done");
- }
- } catch (Throwable t) {
- System.out.println(Thread.currentThread().getName() + ": hit exc");
- failed.set(true);
- t.printStackTrace(System.out);
- throw new RuntimeException(t);
- }
- }
- };
- searchThreads[thread].setDaemon(true);
- searchThreads[thread].start();
- }
-
- for(int thread=0;thread<NUM_SEARCH_THREADS;thread++) {
- searchThreads[thread].join();
- }
-
- if (VERBOSE) {
- System.out.println("TEST: DONE search: totHits=" + totHits);
- }
- } else {
- Thread.sleep(100);
+ fixedSearcher = new IndexSearcher(r, es);
+ smokeTestSearcher(fixedSearcher);
+ runSearchThreads(System.currentTimeMillis() + 500);
}
}
-
- es.shutdown();
- es.awaitTermination(1, TimeUnit.SECONDS);
-
- if (VERBOSE) {
- System.out.println("TEST: all searching done [" + (System.currentTimeMillis()-t0) + " ms]");
- }
+ r.close();
//System.out.println("numDocs=" + r.numDocs() + " openDelFileCount=" + dir.openDeleteFileCount());
- r.close();
- final Set<String> openDeletedFiles = dir.getOpenDeletedFiles();
+ final Set<String> openDeletedFiles = ((MockDirectoryWrapper) dir).getOpenDeletedFiles();
if (openDeletedFiles.size() > 0) {
System.out.println("OBD files: " + openDeletedFiles);
}
- any |= openDeletedFiles.size() > 0;
+ anyOpenDelFiles |= openDeletedFiles.size() > 0;
- assertFalse("saw non-zero open-but-deleted count", any);
- if (VERBOSE) {
- System.out.println("TEST: now join");
- }
- for(int thread=0;thread<NUM_INDEX_THREADS;thread++) {
- threads[thread].join();
- }
- if (VERBOSE) {
- System.out.println("TEST: done join [" + (System.currentTimeMillis()-t0) + " ms]; addCount=" + addCount + " delCount=" + delCount);
- }
+ assertFalse("saw non-zero open-but-deleted count", anyOpenDelFiles);
+ }
- final IndexReader r2 = writer.getReader();
- final IndexSearcher s = newSearcher(r2);
- boolean doFail = false;
- for(String id : delIDs) {
- final TopDocs hits = s.search(new TermQuery(new Term("docid", id)), 1);
- if (hits.totalHits != 0) {
- System.out.println("doc id=" + id + " is supposed to be deleted, but got docID=" + hits.scoreDocs[0].doc);
- doFail = true;
- }
- }
+ private IndexSearcher fixedSearcher;
- // Make sure each group of sub-docs are still in docID order:
- for(SubDocs subDocs : allSubDocs) {
- if (!subDocs.deleted) {
- // We sort by relevance but the scores should be identical so sort falls back to by docID:
- TopDocs hits = s.search(new TermQuery(new Term("packID", subDocs.packID)), 20);
- assertEquals(subDocs.subIDs.size(), hits.totalHits);
- int lastDocID = -1;
- int startDocID = -1;
- for(ScoreDoc scoreDoc : hits.scoreDocs) {
- final int docID = scoreDoc.doc;
- if (lastDocID != -1) {
- assertEquals(1+lastDocID, docID);
- } else {
- startDocID = docID;
- }
- lastDocID = docID;
- final Document doc = s.doc(docID);
- assertEquals(subDocs.packID, doc.get("packID"));
- }
+ protected IndexSearcher getCurrentSearcher() throws Exception {
+ return fixedSearcher;
+ }
- lastDocID = startDocID - 1;
- for(String subID : subDocs.subIDs) {
- hits = s.search(new TermQuery(new Term("docid", subID)), 1);
- assertEquals(1, hits.totalHits);
- final int docID = hits.scoreDocs[0].doc;
- if (lastDocID != -1) {
- assertEquals(1+lastDocID, docID);
- }
- lastDocID = docID;
- }
- } else {
- for(String subID : subDocs.subIDs) {
- assertEquals(0, s.search(new TermQuery(new Term("docid", subID)), 1).totalHits);
- }
- }
- }
-
- final int endID = Integer.parseInt(docs.nextDoc().get("docid"));
- for(int id=0;id<endID;id++) {
- String stringID = ""+id;
- if (!delIDs.contains(stringID)) {
- final TopDocs hits = s.search(new TermQuery(new Term("docid", stringID)), 1);
- if (hits.totalHits != 1) {
- System.out.println("doc id=" + stringID + " is not supposed to be deleted, but got hitCount=" + hits.totalHits);
- doFail = true;
- }
- }
- }
- assertFalse(doFail);
-
- assertEquals("index=" + writer.segString() + " addCount=" + addCount + " delCount=" + delCount, addCount.get() - delCount.get(), r2.numDocs());
- r2.close();
-
- writer.commit();
- assertEquals("index=" + writer.segString() + " addCount=" + addCount + " delCount=" + delCount, addCount.get() - delCount.get(), writer.numDocs());
-
- assertFalse(writer.anyNonBulkMerges);
- writer.close(false);
- _TestUtil.checkIndex(dir);
- s.close();
- dir.close();
- _TestUtil.rmDir(tempDir);
- docs.close();
- if (VERBOSE) {
- System.out.println("TEST: done [" + (System.currentTimeMillis()-t0) + " ms]");
+ @Override
+ protected void releaseSearcher(IndexSearcher s) throws Exception {
+ if (s != fixedSearcher) {
+ // Final searcher:
+ s.getIndexReader().close();
+ s.close();
}
}
- private int runQuery(IndexSearcher s, Query q) throws Exception {
- s.search(q, 10);
- return s.search(q, null, 10, new Sort(new SortField("title", SortField.Type.STRING))).totalHits;
+ @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);
}
- private void smokeTestReader(IndexReader r) throws Exception {
- IndexSearcher s = newSearcher(r);
- runQuery(s, new TermQuery(new Term("body", "united")));
- runQuery(s, new TermQuery(new Term("titleTokenized", "states")));
- PhraseQuery pq = new PhraseQuery();
- pq.add(new Term("body", "united"));
- pq.add(new Term("body", "states"));
- runQuery(s, pq);
- s.close();
+ public void testNRTThreads() throws Exception {
+ runTest("TestNRTThreads");
}
}