Index: lucene/core/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java =================================================================== --- lucene/core/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java (revision 1329354) +++ lucene/core/src/java/org/apache/lucene/index/DocumentsWriterPerThread.java (working copy) @@ -37,7 +37,7 @@ import org.apache.lucene.util.RamUsageEstimator; import org.apache.lucene.util.MutableBits; -public class DocumentsWriterPerThread { +class DocumentsWriterPerThread { /** * The IndexingChain must define the {@link #getChain(DocumentsWriterPerThread)} method Index: lucene/core/src/java/org/apache/lucene/index/ThreadAffinityDocumentsWriterThreadPool.java =================================================================== --- lucene/core/src/java/org/apache/lucene/index/ThreadAffinityDocumentsWriterThreadPool.java (revision 1329354) +++ lucene/core/src/java/org/apache/lucene/index/ThreadAffinityDocumentsWriterThreadPool.java (working copy) @@ -30,7 +30,7 @@ * {@link ThreadAffinityDocumentsWriterThreadPool} tries to find the currently * minimal contended {@link ThreadState}. */ -public class ThreadAffinityDocumentsWriterThreadPool extends DocumentsWriterPerThreadPool { +class ThreadAffinityDocumentsWriterThreadPool extends DocumentsWriterPerThreadPool { private Map threadBindings = new ConcurrentHashMap(); /** Index: lucene/core/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java =================================================================== --- lucene/core/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java (revision 1329354) +++ lucene/core/src/java/org/apache/lucene/index/DocumentsWriterFlushControl.java (working copy) @@ -40,7 +40,7 @@ * {@link IndexWriterConfig#getRAMPerThreadHardLimitMB()} to prevent address * space exhaustion. */ -public final class DocumentsWriterFlushControl { +final class DocumentsWriterFlushControl { private final long hardMaxBytesPerDWPT; private long activeBytes = 0; Index: lucene/core/src/java/org/apache/lucene/index/FlushPolicy.java =================================================================== --- lucene/core/src/java/org/apache/lucene/index/FlushPolicy.java (revision 1329354) +++ lucene/core/src/java/org/apache/lucene/index/FlushPolicy.java (working copy) @@ -50,7 +50,7 @@ * @see DocumentsWriterPerThread * @see IndexWriterConfig#setFlushPolicy(FlushPolicy) */ -public abstract class FlushPolicy { +abstract class FlushPolicy { protected final SetOnce writer = new SetOnce(); protected IndexWriterConfig indexWriterConfig; Index: lucene/core/src/java/org/apache/lucene/index/DocumentsWriterPerThreadPool.java =================================================================== --- lucene/core/src/java/org/apache/lucene/index/DocumentsWriterPerThreadPool.java (revision 1329354) +++ lucene/core/src/java/org/apache/lucene/index/DocumentsWriterPerThreadPool.java (working copy) @@ -36,7 +36,7 @@ * new {@link DocumentsWriterPerThread} instance. *

*/ -public abstract class DocumentsWriterPerThreadPool { +abstract class DocumentsWriterPerThreadPool { /** * {@link ThreadState} references and guards a @@ -50,7 +50,7 @@ * before accessing the state. */ @SuppressWarnings("serial") - public final static class ThreadState extends ReentrantLock { + final static class ThreadState extends ReentrantLock { DocumentsWriterPerThread dwpt; // TODO this should really be part of DocumentsWriterFlushControl // write access guarded by DocumentsWriterFlushControl @@ -127,7 +127,7 @@ /** * Creates a new {@link DocumentsWriterPerThreadPool} with a given maximum of {@link ThreadState}s. */ - public DocumentsWriterPerThreadPool(int maxNumThreadStates) { + DocumentsWriterPerThreadPool(int maxNumThreadStates) { if (maxNumThreadStates < 1) { throw new IllegalArgumentException("maxNumThreadStates must be >= 1 but was: " + maxNumThreadStates); } @@ -135,7 +135,7 @@ numThreadStatesActive = 0; } - public void initialize(DocumentsWriter documentsWriter, FieldNumberBiMap globalFieldMap, IndexWriterConfig config) { + void initialize(DocumentsWriter documentsWriter, FieldNumberBiMap globalFieldMap, IndexWriterConfig config) { this.documentsWriter.set(documentsWriter); // thread pool is bound to DW this.globalFieldMap.set(globalFieldMap); for (int i = 0; i < threadStates.length; i++) { @@ -148,14 +148,14 @@ * Returns the max number of {@link ThreadState} instances available in this * {@link DocumentsWriterPerThreadPool} */ - public int getMaxThreadStates() { + int getMaxThreadStates() { return threadStates.length; } /** * Returns the active number of {@link ThreadState} instances. */ - public int getActiveThreadState() { + int getActiveThreadState() { return numThreadStatesActive; } @@ -169,7 +169,7 @@ * @return a new {@link ThreadState} iff any new state is available otherwise * null */ - public synchronized ThreadState newThreadState() { + synchronized ThreadState newThreadState() { if (numThreadStatesActive < threadStates.length) { final ThreadState threadState = threadStates[numThreadStatesActive]; threadState.lock(); // lock so nobody else will get this ThreadState @@ -211,7 +211,7 @@ /** * Deactivate all unreleased threadstates */ - protected synchronized void deactivateUnreleasedStates() { + synchronized void deactivateUnreleasedStates() { for (int i = numThreadStatesActive; i < threadStates.length; i++) { final ThreadState threadState = threadStates[i]; threadState.lock(); @@ -223,7 +223,7 @@ } } - protected DocumentsWriterPerThread replaceForFlush(ThreadState threadState, boolean closed) { + DocumentsWriterPerThread replaceForFlush(ThreadState threadState, boolean closed) { assert threadState.isHeldByCurrentThread(); assert globalFieldMap.get() != null; final DocumentsWriterPerThread dwpt = threadState.dwpt; @@ -238,7 +238,7 @@ return dwpt; } - public void recycle(DocumentsWriterPerThread dwpt) { + void recycle(DocumentsWriterPerThread dwpt) { // don't recycle DWPT by default } @@ -266,7 +266,7 @@ * waiting to acquire its lock or null if no {@link ThreadState} * is yet visible to the calling thread. */ - protected ThreadState minContendedThreadState() { + ThreadState minContendedThreadState() { ThreadState minThreadState = null; final int limit = numThreadStatesActive; for (int i = 0; i < limit; i++) { Index: lucene/core/src/java/org/apache/lucene/index/DocumentsWriterFlushQueue.java =================================================================== --- lucene/core/src/java/org/apache/lucene/index/DocumentsWriterFlushQueue.java (revision 1329354) +++ lucene/core/src/java/org/apache/lucene/index/DocumentsWriterFlushQueue.java (working copy) @@ -27,7 +27,7 @@ /** * @lucene.internal */ -public class DocumentsWriterFlushQueue { +class DocumentsWriterFlushQueue { private final Queue queue = new LinkedList(); // we track tickets separately since count must be present even before the ticket is // constructed ie. queue.size would not reflect it. Index: lucene/core/src/java/org/apache/lucene/index/FlushByRamOrCountsPolicy.java =================================================================== --- lucene/core/src/java/org/apache/lucene/index/FlushByRamOrCountsPolicy.java (revision 1329354) +++ lucene/core/src/java/org/apache/lucene/index/FlushByRamOrCountsPolicy.java (working copy) @@ -47,7 +47,7 @@ * pending iff the global active RAM consumption is >= the configured max RAM * buffer. */ -public class FlushByRamOrCountsPolicy extends FlushPolicy { +class FlushByRamOrCountsPolicy extends FlushPolicy { @Override public void onDelete(DocumentsWriterFlushControl control, ThreadState state) { Index: lucene/core/src/java/org/apache/lucene/index/IndexWriterConfig.java =================================================================== --- lucene/core/src/java/org/apache/lucene/index/IndexWriterConfig.java (revision 1329354) +++ lucene/core/src/java/org/apache/lucene/index/IndexWriterConfig.java (working copy) @@ -571,8 +571,8 @@ *

*

* NOTE: This only takes effect when IndexWriter is first created.

*/ - public IndexWriterConfig setIndexerThreadPool(DocumentsWriterPerThreadPool threadPool) { - if(threadPool == null) { + IndexWriterConfig setIndexerThreadPool(DocumentsWriterPerThreadPool threadPool) { + if (threadPool == null) { throw new IllegalArgumentException("DocumentsWriterPerThreadPool must not be nul"); } this.indexerThreadPool = threadPool; @@ -582,10 +582,32 @@ /** Returns the configured {@link DocumentsWriterPerThreadPool} instance. * @see #setIndexerThreadPool(DocumentsWriterPerThreadPool) * @return the configured {@link DocumentsWriterPerThreadPool} instance.*/ - public DocumentsWriterPerThreadPool getIndexerThreadPool() { + DocumentsWriterPerThreadPool getIndexerThreadPool() { return this.indexerThreadPool; } + /** + * Sets the max number of simultaneous threads that may be indexing documents + * at once in IndexWriter. Values < 1 are invalid and if passed + * maxThreadStates will be set to + * {@link #DEFAULT_MAX_THREAD_STATES}. + * + *

Only takes effect when IndexWriter is first created. */ + public IndexWriterConfig setMaxThreadStates(int maxThreadStates) { + this.indexerThreadPool = new ThreadAffinityDocumentsWriterThreadPool(maxThreadStates); + return this; + } + + /** Returns the max number of simultaneous threads that + * may be indexing documents at once in IndexWriter. */ + public int getMaxThreadStates() { + try { + return ((ThreadAffinityDocumentsWriterThreadPool) indexerThreadPool).getMaxThreadStates(); + } catch (ClassCastException cce) { + throw new IllegalStateException(cce); + } + } + /** By default, IndexWriter does not pool the * SegmentReaders it must open for deletions and * merging, unless a near-real-time reader has been Index: lucene/test-framework/src/java/org/apache/lucene/index/RandomDocumentsWriterPerThreadPool.java =================================================================== --- lucene/test-framework/src/java/org/apache/lucene/index/RandomDocumentsWriterPerThreadPool.java (revision 1329354) +++ lucene/test-framework/src/java/org/apache/lucene/index/RandomDocumentsWriterPerThreadPool.java (working copy) @@ -20,7 +20,7 @@ /** * - * A {@link DocumentsWriterPerThreadPool} that selects thread states at random. + * A DocumentsWriterPerThreadPool that selects thread states at random. * * @lucene.internal * @lucene.experimental Index: lucene/test-framework/src/java/org/apache/lucene/util/LuceneTestCase.java =================================================================== --- lucene/test-framework/src/java/org/apache/lucene/util/LuceneTestCase.java (revision 1329354) +++ lucene/test-framework/src/java/org/apache/lucene/util/LuceneTestCase.java (working copy) @@ -26,6 +26,7 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; +import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -35,8 +36,8 @@ import java.util.Iterator; import java.util.List; import java.util.Locale; +import java.util.Map.Entry; import java.util.Map; -import java.util.Map.Entry; import java.util.Random; import java.util.Set; import java.util.TimeZone; @@ -53,11 +54,11 @@ import org.apache.lucene.document.FieldType; import org.apache.lucene.index.AtomicReader; import org.apache.lucene.index.CompositeReader; +import org.apache.lucene.index.DirectoryReader; import org.apache.lucene.index.FieldFilterAtomicReader; import org.apache.lucene.index.FieldInfo; -import org.apache.lucene.index.DirectoryReader; +import org.apache.lucene.index.IndexReader.ReaderClosedListener; import org.apache.lucene.index.IndexReader; -import org.apache.lucene.index.IndexReader.ReaderClosedListener; import org.apache.lucene.index.IndexWriterConfig; import org.apache.lucene.index.LogByteSizeMergePolicy; import org.apache.lucene.index.LogDocMergePolicy; @@ -70,25 +71,24 @@ import org.apache.lucene.index.SegmentReader; import org.apache.lucene.index.SerialMergeScheduler; import org.apache.lucene.index.SlowCompositeReaderWrapper; -import org.apache.lucene.index.ThreadAffinityDocumentsWriterThreadPool; import org.apache.lucene.index.TieredMergePolicy; import org.apache.lucene.search.AssertingIndexSearcher; import org.apache.lucene.search.BooleanQuery; +import org.apache.lucene.search.FieldCache.CacheEntry; import org.apache.lucene.search.FieldCache; -import org.apache.lucene.search.FieldCache.CacheEntry; import org.apache.lucene.search.IndexSearcher; +import org.apache.lucene.search.QueryUtils.FCInvisibleMultiReader; import org.apache.lucene.search.RandomSimilarityProvider; import org.apache.lucene.search.similarities.DefaultSimilarity; import org.apache.lucene.search.similarities.Similarity; -import org.apache.lucene.search.QueryUtils.FCInvisibleMultiReader; import org.apache.lucene.store.Directory; import org.apache.lucene.store.FSDirectory; import org.apache.lucene.store.FlushInfo; import org.apache.lucene.store.IOContext; import org.apache.lucene.store.LockFactory; import org.apache.lucene.store.MergeInfo; +import org.apache.lucene.store.MockDirectoryWrapper.Throttling; import org.apache.lucene.store.MockDirectoryWrapper; -import org.apache.lucene.store.MockDirectoryWrapper.Throttling; import org.apache.lucene.store.NRTCachingDirectory; import org.apache.lucene.util.FieldCacheSanityChecker.Insanity; import org.junit.After; @@ -108,7 +108,6 @@ import org.junit.runner.notification.RunListener; import org.junit.runners.model.MultipleFailureException; import org.junit.runners.model.Statement; - import com.carrotsearch.randomizedtesting.JUnit4MethodProvider; import com.carrotsearch.randomizedtesting.MixWithSuiteName; import com.carrotsearch.randomizedtesting.RandomizedContext; @@ -1068,13 +1067,38 @@ if (r.nextBoolean()) { int maxNumThreadStates = rarely(r) ? _TestUtil.nextInt(r, 5, 20) // crazy value : _TestUtil.nextInt(r, 1, 4); // reasonable value - if (rarely(r)) { - // random thread pool - c.setIndexerThreadPool(new RandomDocumentsWriterPerThreadPool(maxNumThreadStates, r)); - } else { - // random thread pool - c.setIndexerThreadPool(new ThreadAffinityDocumentsWriterThreadPool(maxNumThreadStates)); + + Method setIndexerThreadPoolMethod = null; + try { + // Retrieve the package-private setIndexerThreadPool + // method: + for(Method m : IndexWriterConfig.class.getDeclaredMethods()) { + if (m.getName().equals("setIndexerThreadPool")) { + m.setAccessible(true); + setIndexerThreadPoolMethod = m; + break; + } + } + } catch (Exception e) { + // Should not happen? + throw new RuntimeException(e); } + + if (setIndexerThreadPoolMethod == null) { + throw new RuntimeException("failed to lookup IndexWriterConfig.setIndexerThreadPool method"); + } + + try { + if (rarely(r)) { + // random thread pool + setIndexerThreadPoolMethod.invoke(c, new RandomDocumentsWriterPerThreadPool(maxNumThreadStates, r)); + } else { + // random thread pool + c.setMaxThreadStates(maxNumThreadStates); + } + } catch (Exception e) { + throw new RuntimeException(e); + } } if (rarely(r)) {