lucene/src/test-framework/org/apache/lucene/index/ThreadedIndexingAndSearchingTestCase.java - manifoldcf-integration-solr-3.x - Git at Google

 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.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Collection;
 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.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.Directory;
 import org.apache.lucene.store.MockDirectoryWrapper;
 import org.apache.lucene.util.LineFileDocs;
 import org.apache.lucene.util.LuceneTestCase;
 import org.apache.lucene.util.NamedThreadFactory;
 import org.apache.lucene.util._TestUtil;

 // TODO
 //   - mix in optimize, addIndexes
 //   - randomoly mix in non-congruent docs

 /** Utility class that spawns multiple indexing and
  *  searching threads. */
 public abstract class ThreadedIndexingAndSearchingTestCase extends LuceneTestCase {

   protected final AtomicBoolean failed = new AtomicBoolean();
   protected final AtomicInteger addCount = new AtomicInteger();
   protected final AtomicInteger delCount = new AtomicInteger();
   protected final AtomicInteger packCount = new AtomicInteger();

   protected Directory dir;
   protected IndexWriter writer;

   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;
     }
   }

   // Called per-search
   protected abstract IndexSearcher getCurrentSearcher() throws Exception;

   protected abstract IndexSearcher getFinalSearcher() throws Exception;

   protected void releaseSearcher(IndexSearcher s) throws Exception {
   }

   // Called once to run searching
   protected abstract void doSearching(ExecutorService es, long stopTime) throws Exception;

   protected Directory getDirectory(Directory in) {
     return in;
   }

   protected void updateDocuments(Term id, Collection<Document> docs) throws Exception {
     writer.updateDocuments(id, docs);
   }

   protected void addDocuments(Term id, Collection<Document> docs) throws Exception {
     writer.addDocuments(docs);
   }

   protected void addDocument(Term id, Document doc) throws Exception {
     writer.addDocument(doc);
   }

   protected void updateDocument(Term term, Document doc) throws Exception {
     writer.updateDocument(term, doc);
   }

   protected void deleteDocuments(Term term) throws Exception {
     writer.deleteDocuments(term);
   }

   protected void doAfterIndexingThreadDone() {
   }

   private Thread[] launchIndexingThreads(final LineFileDocs docs,
                                          int numThreads,
                                          final long stopTime,
                                          final Set<String> delIDs,
                                          final Set<String> delPackIDs,
                                          final List<SubDocs> allSubDocs)
     throws Exception {
     final Thread[] threads = new Thread[numThreads];
     for(int thread=0;thread<numThreads;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 {

                 // Occasional longish pause if running
                 // nightly
                 if (LuceneTestCase.TEST_NIGHTLY && random.nextInt(6) == 3) {
                   if (VERBOSE) {
                     System.out.println(Thread.currentThread().getName() + ": now long sleep");
                   }
                   Thread.sleep(_TestUtil.nextInt(random, 50, 500));
                 }

                 // Rate limit ingest rate:
                 if (random.nextInt(7) == 5) {
                   Thread.sleep(_TestUtil.nextInt(random, 1, 10));
                   if (VERBOSE) {
                     System.out.println(Thread.currentThread().getName() + ": done sleep");
                   }
                 }

                 Document doc = docs.nextDoc();
                 if (doc == null) {
                   break;
                 }

                 // Maybe add randomly named field
                 final String addedField;
                 if (random.nextBoolean()) {
                   addedField = "extra" + random.nextInt(40);
                   doc.add(newField(addedField, "a random field", Field.Store.YES, Field.Index.ANALYZED));
                 } else {
                   addedField = null;
                 }

                 if (random.nextBoolean()) {

                   if (random.nextBoolean()) {
                     // Add/update doc block:
                     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);
                       // Update doc block, replacing prior packID
                       packID = delSubDocs.packID;
                     } else {
                       delSubDocs = null;
                       // Add doc block, using new packID
                       packID = packCount.getAndIncrement() + "";
                     }

                     final Field packIDField = newField("packID", packID, Field.Store.YES, Field.Index.NOT_ANALYZED);
                     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());

                     final Term packIDTerm = new Term("packID", packID);

                     if (delSubDocs != null) {
                       delSubDocs.deleted = true;
                       delIDs.addAll(delSubDocs.subIDs);
                       delCount.addAndGet(delSubDocs.subIDs.size());
                       if (VERBOSE) {
                         System.out.println(Thread.currentThread().getName() + ": update pack packID=" + delSubDocs.packID + " count=" + docsList.size() + " docs=" + docIDs);
                       }
                       updateDocuments(packIDTerm, docsList);
                     } else {
                       if (VERBOSE) {
                         System.out.println(Thread.currentThread().getName() + ": add pack packID=" + packID + " count=" + docsList.size() + " docs=" + docIDs);
                       }
                       addDocuments(packIDTerm, docsList);
                     }
                     doc.removeField("packID");

                     if (random.nextInt(5) == 2) {
                       if (VERBOSE) {
                         System.out.println(Thread.currentThread().getName() + ": buffer del id:" + packID);
                       }
                       toDeleteSubDocs.add(subDocs);
                     }

                   } else {
                     // Add single doc
                     final String docid = doc.get("docid");
                     if (VERBOSE) {
                       System.out.println(Thread.currentThread().getName() + ": add doc docid:" + docid);
                     }
                     addDocument(new Term("docid", 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(docid);
                     }
                   }
                 } else {

                   // Update single doc, but we never re-use
                   // and ID so the delete will never
                   // actually happen:
                   if (VERBOSE) {
                     System.out.println(Thread.currentThread().getName() + ": update doc id:" + doc.get("docid"));
                   }
                   final String docid = doc.get("docid");
                   updateDocument(new Term("docid", 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(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);
                     }
                     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;
                     delPackIDs.add(subDocs.packID);
                     deleteDocuments(new Term("packID", subDocs.packID));
                     subDocs.deleted = true;
                     if (VERBOSE) {
                       System.out.println(Thread.currentThread().getName() + ": 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");
             }

             doAfterIndexingThreadDone();
           }
         };
       threads[thread].setDaemon(true);
       threads[thread].start();
     }

     return threads;
   }

   protected void runSearchThreads(final long stopTimeMS) throws Exception {
     final int numThreads = _TestUtil.nextInt(random, 1, 5);
     final Thread[] searchThreads = new Thread[numThreads];
     final AtomicInteger totHits = new AtomicInteger();

     // silly starting guess:
     final AtomicInteger totTermCount = new AtomicInteger(100);

     // TODO: we should enrich this to do more interesting searches
     for(int thread=0;thread<searchThreads.length;thread++) {
       searchThreads[thread] = new Thread() {
           @Override
           public void run() {
             while (System.currentTimeMillis() < stopTimeMS) {
               try {
                 final IndexSearcher s = getCurrentSearcher();
                 try {
                   if (s.getIndexReader().numDocs() > 0) {
                     smokeTestSearcher(s);
                     TermEnum termEnum = s.getIndexReader().terms(new Term("body", ""));
                     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() < stopTimeMS) {
                       Term term = termEnum.term();
                       if (term == null) {
                         if (seenTermCount == 0) {
                           break;
                         }
                         totTermCount.set(seenTermCount);
                         seenTermCount = 0;
                         if (totTermCount.get() < 10) {
                           shift = 0;
                           trigger = 1;
                         } else {
                           trigger = totTermCount.get()/10;
                           //System.out.println("trigger " + trigger);
                           shift = random.nextInt(trigger);
                         }
                         termEnum = s.getIndexReader().terms(new Term("body", ""));
                         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(term)));
                       }
                     }
                     //if (VERBOSE) {
                     //System.out.println(Thread.currentThread().getName() + ": search done");
                     //}
                   }
                 } finally {
                   releaseSearcher(s);
                 }
               } 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<searchThreads.length;thread++) {
       searchThreads[thread].join();
     }

     if (VERBOSE) {
       System.out.println("TEST: DONE search: totHits=" + totHits);
     }
   }

   protected void doAfterWriter(ExecutorService es) throws Exception {
   }

   protected void doClose() throws Exception {
   }

   public void runTest(String testName) throws Exception {

     failed.set(false);
     addCount.set(0);
     delCount.set(0);
     packCount.set(0);

     final long t0 = System.currentTimeMillis();

     final LineFileDocs docs = new LineFileDocs(random);
     final File tempDir = _TestUtil.getTempDir(testName);
     dir = newFSDirectory(tempDir);
     ((MockDirectoryWrapper) 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();
         int sum = 0;
         final int inc = Math.max(1, maxDoc/50);
         for(int docID=0;docID<maxDoc;docID += inc) {
           if (!reader.isDeleted(docID)) {
             final Document doc = reader.document(docID);
             sum += doc.getFields().size();
           }
         }

         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");
         }
       }
       });

     writer = new IndexWriter(dir, conf);
     if (VERBOSE) {
       writer.setInfoStream(System.out);
     }
     _TestUtil.reduceOpenFiles(writer);

     final ExecutorService es = random.nextBoolean() ? null : Executors.newCachedThreadPool(new NamedThreadFactory(testName));

     doAfterWriter(es);

     final int NUM_INDEX_THREADS = _TestUtil.nextInt(random, 2, 4);

     final int RUN_TIME_SEC = LuceneTestCase.TEST_NIGHTLY ? 300 : RANDOM_MULTIPLIER;

     final Set<String> delIDs = Collections.synchronizedSet(new HashSet<String>());
     final Set<String> delPackIDs = Collections.synchronizedSet(new HashSet<String>());
     final List<SubDocs> allSubDocs = Collections.synchronizedList(new ArrayList<SubDocs>());

     final long stopTime = System.currentTimeMillis() + RUN_TIME_SEC*1000;

     final Thread[] indexThreads = launchIndexingThreads(docs, NUM_INDEX_THREADS, stopTime, delIDs, delPackIDs, allSubDocs);

     if (VERBOSE) {
       System.out.println("TEST: DONE start indexing threads [" + (System.currentTimeMillis()-t0) + " ms]");
     }

     // Let index build up a bit
     Thread.sleep(100);

     doSearching(es, stopTime);

     if (VERBOSE) {
       System.out.println("TEST: all searching done [" + (System.currentTimeMillis()-t0) + " ms]");
     }

     for(int thread=0;thread<indexThreads.length;thread++) {
       indexThreads[thread].join();
     }

     if (VERBOSE) {
       System.out.println("TEST: done join indexing threads [" + (System.currentTimeMillis()-t0) + " ms]; addCount=" + addCount + " delCount=" + delCount);
     }

     final IndexSearcher s = getFinalSearcher();
     if (VERBOSE) {
       System.out.println("TEST: finalSearcher=" + s);
     }
     boolean doFail = false;

     // Verify: make sure delIDs are in fact deleted:
     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 " + hits.totalHits + " hits; first docID=" + hits.scoreDocs[0].doc);
         doFail = true;
       }
     }

     // Verify: make sure delPackIDs are in fact deleted:
     for(String id : delPackIDs) {
       final TopDocs hits = s.search(new TermQuery(new Term("packID", id)), 1);
       if (hits.totalHits != 0) {
         System.out.println("packID=" + id + " is supposed to be deleted, but got " + hits.totalHits + " matches");
         doFail = true;
       }
     }

     // Verify: make sure each group of sub-docs are still in docID order:
     for(SubDocs subDocs : allSubDocs) {
       TopDocs hits = s.search(new TermQuery(new Term("packID", subDocs.packID)), 20);
       if (!subDocs.deleted) {
         // We sort by relevance but the scores should be identical so sort falls back to by docID:
         if (hits.totalHits != subDocs.subIDs.size()) {
           System.out.println("packID=" + subDocs.packID + ": expected " + subDocs.subIDs.size() + " hits but got " + hits.totalHits);
           doFail = true;
         } else {
           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"));
           }

           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 {
         // Pack was deleted -- make sure its docs are
         // deleted.  We can't verify packID is deleted
         // because we can re-use packID for update:
         for(String subID : subDocs.subIDs) {
           assertEquals(0, s.search(new TermQuery(new Term("docid", subID)), 1).totalHits);
         }
       }
     }

     // Verify: make sure all not-deleted docs are in fact
     // not deleted:
     final int endID = Integer.parseInt(docs.nextDoc().get("docid"));
     docs.close();

     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(), s.getIndexReader().numDocs());
     releaseSearcher(s);

     if (es != null) {
       es.shutdown();
       es.awaitTermination(1, TimeUnit.SECONDS);
     }

     writer.commit();
     assertEquals("index=" + writer.segString() + " addCount=" + addCount + " delCount=" + delCount, addCount.get() - delCount.get(), writer.numDocs());

     assertFalse(writer.anyNonBulkMerges);
     doClose();
     writer.close(false);
     _TestUtil.checkIndex(dir);
     dir.close();
     _TestUtil.rmDir(tempDir);

     if (VERBOSE) {
       System.out.println("TEST: done [" + (System.currentTimeMillis()-t0) + " ms]");
     }
   }

   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.STRING))).totalHits;
   }

   protected void smokeTestSearcher(IndexSearcher s) throws Exception {
     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);
   }
 }
	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.io.File;
	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.Collection;
	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.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.Directory;
	import org.apache.lucene.store.MockDirectoryWrapper;
	import org.apache.lucene.util.LineFileDocs;
	import org.apache.lucene.util.LuceneTestCase;
	import org.apache.lucene.util.NamedThreadFactory;
	import org.apache.lucene.util._TestUtil;

	// TODO
	// - mix in optimize, addIndexes
	// - randomoly mix in non-congruent docs

	/** Utility class that spawns multiple indexing and
	* searching threads. */
	public abstract class ThreadedIndexingAndSearchingTestCase extends LuceneTestCase {

	protected final AtomicBoolean failed = new AtomicBoolean();
	protected final AtomicInteger addCount = new AtomicInteger();
	protected final AtomicInteger delCount = new AtomicInteger();
	protected final AtomicInteger packCount = new AtomicInteger();

	protected Directory dir;
	protected IndexWriter writer;

	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;
	}
	}

	// Called per-search
	protected abstract IndexSearcher getCurrentSearcher() throws Exception;

	protected abstract IndexSearcher getFinalSearcher() throws Exception;

	protected void releaseSearcher(IndexSearcher s) throws Exception {
	}

	// Called once to run searching
	protected abstract void doSearching(ExecutorService es, long stopTime) throws Exception;

	protected Directory getDirectory(Directory in) {
	return in;
	}

	protected void updateDocuments(Term id, Collection<Document> docs) throws Exception {
	writer.updateDocuments(id, docs);
	}

	protected void addDocuments(Term id, Collection<Document> docs) throws Exception {
	writer.addDocuments(docs);
	}

	protected void addDocument(Term id, Document doc) throws Exception {
	writer.addDocument(doc);
	}

	protected void updateDocument(Term term, Document doc) throws Exception {
	writer.updateDocument(term, doc);
	}

	protected void deleteDocuments(Term term) throws Exception {
	writer.deleteDocuments(term);
	}

	protected void doAfterIndexingThreadDone() {
	}

	private Thread[] launchIndexingThreads(final LineFileDocs docs,
	int numThreads,
	final long stopTime,
	final Set<String> delIDs,
	final Set<String> delPackIDs,
	final List<SubDocs> allSubDocs)
	throws Exception {
	final Thread[] threads = new Thread[numThreads];
	for(int thread=0;thread<numThreads;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 {

	// Occasional longish pause if running
	// nightly
	if (LuceneTestCase.TEST_NIGHTLY && random.nextInt(6) == 3) {
	if (VERBOSE) {
	System.out.println(Thread.currentThread().getName() + ": now long sleep");
	}
	Thread.sleep(_TestUtil.nextInt(random, 50, 500));
	}

	// Rate limit ingest rate:
	if (random.nextInt(7) == 5) {
	Thread.sleep(_TestUtil.nextInt(random, 1, 10));
	if (VERBOSE) {
	System.out.println(Thread.currentThread().getName() + ": done sleep");
	}
	}

	Document doc = docs.nextDoc();
	if (doc == null) {
	break;
	}

	// Maybe add randomly named field
	final String addedField;
	if (random.nextBoolean()) {
	addedField = "extra" + random.nextInt(40);
	doc.add(newField(addedField, "a random field", Field.Store.YES, Field.Index.ANALYZED));
	} else {
	addedField = null;
	}

	if (random.nextBoolean()) {

	if (random.nextBoolean()) {
	// Add/update doc block:
	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);
	// Update doc block, replacing prior packID
	packID = delSubDocs.packID;
	} else {
	delSubDocs = null;
	// Add doc block, using new packID
	packID = packCount.getAndIncrement() + "";
	}

	final Field packIDField = newField("packID", packID, Field.Store.YES, Field.Index.NOT_ANALYZED);
	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());

	final Term packIDTerm = new Term("packID", packID);

	if (delSubDocs != null) {
	delSubDocs.deleted = true;
	delIDs.addAll(delSubDocs.subIDs);
	delCount.addAndGet(delSubDocs.subIDs.size());
	if (VERBOSE) {
	System.out.println(Thread.currentThread().getName() + ": update pack packID=" + delSubDocs.packID + " count=" + docsList.size() + " docs=" + docIDs);
	}
	updateDocuments(packIDTerm, docsList);
	} else {
	if (VERBOSE) {
	System.out.println(Thread.currentThread().getName() + ": add pack packID=" + packID + " count=" + docsList.size() + " docs=" + docIDs);
	}
	addDocuments(packIDTerm, docsList);
	}
	doc.removeField("packID");

	if (random.nextInt(5) == 2) {
	if (VERBOSE) {
	System.out.println(Thread.currentThread().getName() + ": buffer del id:" + packID);
	}
	toDeleteSubDocs.add(subDocs);
	}

	} else {
	// Add single doc
	final String docid = doc.get("docid");
	if (VERBOSE) {
	System.out.println(Thread.currentThread().getName() + ": add doc docid:" + docid);
	}
	addDocument(new Term("docid", 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(docid);
	}
	}
	} else {

	// Update single doc, but we never re-use
	// and ID so the delete will never
	// actually happen:
	if (VERBOSE) {
	System.out.println(Thread.currentThread().getName() + ": update doc id:" + doc.get("docid"));
	}
	final String docid = doc.get("docid");
	updateDocument(new Term("docid", 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(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);
	}
	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;
	delPackIDs.add(subDocs.packID);
	deleteDocuments(new Term("packID", subDocs.packID));
	subDocs.deleted = true;
	if (VERBOSE) {
	System.out.println(Thread.currentThread().getName() + ": 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");
	}

	doAfterIndexingThreadDone();
	}
	};
	threads[thread].setDaemon(true);
	threads[thread].start();
	}

	return threads;
	}

	protected void runSearchThreads(final long stopTimeMS) throws Exception {
	final int numThreads = _TestUtil.nextInt(random, 1, 5);
	final Thread[] searchThreads = new Thread[numThreads];
	final AtomicInteger totHits = new AtomicInteger();

	// silly starting guess:
	final AtomicInteger totTermCount = new AtomicInteger(100);

	// TODO: we should enrich this to do more interesting searches
	for(int thread=0;thread<searchThreads.length;thread++) {
	searchThreads[thread] = new Thread() {
	@Override
	public void run() {
	while (System.currentTimeMillis() < stopTimeMS) {
	try {
	final IndexSearcher s = getCurrentSearcher();
	try {
	if (s.getIndexReader().numDocs() > 0) {
	smokeTestSearcher(s);
	TermEnum termEnum = s.getIndexReader().terms(new Term("body", ""));
	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() < stopTimeMS) {
	Term term = termEnum.term();
	if (term == null) {
	if (seenTermCount == 0) {
	break;
	}
	totTermCount.set(seenTermCount);
	seenTermCount = 0;
	if (totTermCount.get() < 10) {
	shift = 0;
	trigger = 1;
	} else {
	trigger = totTermCount.get()/10;
	//System.out.println("trigger " + trigger);
	shift = random.nextInt(trigger);
	}
	termEnum = s.getIndexReader().terms(new Term("body", ""));
	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(term)));
	}
	}
	//if (VERBOSE) {
	//System.out.println(Thread.currentThread().getName() + ": search done");
	//}
	}
	} finally {
	releaseSearcher(s);
	}
	} 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<searchThreads.length;thread++) {
	searchThreads[thread].join();
	}

	if (VERBOSE) {
	System.out.println("TEST: DONE search: totHits=" + totHits);
	}
	}

	protected void doAfterWriter(ExecutorService es) throws Exception {
	}

	protected void doClose() throws Exception {
	}

	public void runTest(String testName) throws Exception {

	failed.set(false);
	addCount.set(0);
	delCount.set(0);
	packCount.set(0);

	final long t0 = System.currentTimeMillis();

	final LineFileDocs docs = new LineFileDocs(random);
	final File tempDir = _TestUtil.getTempDir(testName);
	dir = newFSDirectory(tempDir);
	((MockDirectoryWrapper) 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();
	int sum = 0;
	final int inc = Math.max(1, maxDoc/50);
	for(int docID=0;docID<maxDoc;docID += inc) {
	if (!reader.isDeleted(docID)) {
	final Document doc = reader.document(docID);
	sum += doc.getFields().size();
	}
	}

	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");
	}
	}
	});

	writer = new IndexWriter(dir, conf);
	if (VERBOSE) {
	writer.setInfoStream(System.out);
	}
	_TestUtil.reduceOpenFiles(writer);

	final ExecutorService es = random.nextBoolean() ? null : Executors.newCachedThreadPool(new NamedThreadFactory(testName));

	doAfterWriter(es);

	final int NUM_INDEX_THREADS = _TestUtil.nextInt(random, 2, 4);

	final int RUN_TIME_SEC = LuceneTestCase.TEST_NIGHTLY ? 300 : RANDOM_MULTIPLIER;

	final Set<String> delIDs = Collections.synchronizedSet(new HashSet<String>());
	final Set<String> delPackIDs = Collections.synchronizedSet(new HashSet<String>());
	final List<SubDocs> allSubDocs = Collections.synchronizedList(new ArrayList<SubDocs>());

	final long stopTime = System.currentTimeMillis() + RUN_TIME_SEC*1000;

	final Thread[] indexThreads = launchIndexingThreads(docs, NUM_INDEX_THREADS, stopTime, delIDs, delPackIDs, allSubDocs);

	if (VERBOSE) {
	System.out.println("TEST: DONE start indexing threads [" + (System.currentTimeMillis()-t0) + " ms]");
	}

	// Let index build up a bit
	Thread.sleep(100);

	doSearching(es, stopTime);

	if (VERBOSE) {
	System.out.println("TEST: all searching done [" + (System.currentTimeMillis()-t0) + " ms]");
	}

	for(int thread=0;thread<indexThreads.length;thread++) {
	indexThreads[thread].join();
	}

	if (VERBOSE) {
	System.out.println("TEST: done join indexing threads [" + (System.currentTimeMillis()-t0) + " ms]; addCount=" + addCount + " delCount=" + delCount);
	}

	final IndexSearcher s = getFinalSearcher();
	if (VERBOSE) {
	System.out.println("TEST: finalSearcher=" + s);
	}
	boolean doFail = false;

	// Verify: make sure delIDs are in fact deleted:
	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 " + hits.totalHits + " hits; first docID=" + hits.scoreDocs[0].doc);
	doFail = true;
	}
	}

	// Verify: make sure delPackIDs are in fact deleted:
	for(String id : delPackIDs) {
	final TopDocs hits = s.search(new TermQuery(new Term("packID", id)), 1);
	if (hits.totalHits != 0) {
	System.out.println("packID=" + id + " is supposed to be deleted, but got " + hits.totalHits + " matches");
	doFail = true;
	}
	}

	// Verify: make sure each group of sub-docs are still in docID order:
	for(SubDocs subDocs : allSubDocs) {
	TopDocs hits = s.search(new TermQuery(new Term("packID", subDocs.packID)), 20);
	if (!subDocs.deleted) {
	// We sort by relevance but the scores should be identical so sort falls back to by docID:
	if (hits.totalHits != subDocs.subIDs.size()) {
	System.out.println("packID=" + subDocs.packID + ": expected " + subDocs.subIDs.size() + " hits but got " + hits.totalHits);
	doFail = true;
	} else {
	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"));
	}

	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 {
	// Pack was deleted -- make sure its docs are
	// deleted. We can't verify packID is deleted
	// because we can re-use packID for update:
	for(String subID : subDocs.subIDs) {
	assertEquals(0, s.search(new TermQuery(new Term("docid", subID)), 1).totalHits);
	}
	}
	}

	// Verify: make sure all not-deleted docs are in fact
	// not deleted:
	final int endID = Integer.parseInt(docs.nextDoc().get("docid"));
	docs.close();

	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(), s.getIndexReader().numDocs());
	releaseSearcher(s);

	if (es != null) {
	es.shutdown();
	es.awaitTermination(1, TimeUnit.SECONDS);
	}

	writer.commit();
	assertEquals("index=" + writer.segString() + " addCount=" + addCount + " delCount=" + delCount, addCount.get() - delCount.get(), writer.numDocs());

	assertFalse(writer.anyNonBulkMerges);
	doClose();
	writer.close(false);
	_TestUtil.checkIndex(dir);
	dir.close();
	_TestUtil.rmDir(tempDir);

	if (VERBOSE) {
	System.out.println("TEST: done [" + (System.currentTimeMillis()-t0) + " ms]");
	}
	}

	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.STRING))).totalHits;
	}

	protected void smokeTestSearcher(IndexSearcher s) throws Exception {
	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);
	}
	}