--- /dev/null
+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 org.apache.lucene.store.Directory;
+import org.apache.lucene.util.ThreadInterruptedException;
+import org.apache.lucene.util.CollectionUtil;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Comparator;
+
+/** A {@link MergeScheduler} that runs each merge using a
+ * separate thread.
+ *
+ * <p>Specify the max number of threads that may run at
+ * once with {@link #setMaxThreadCount}.</p>
+ *
+ * <p>Separately specify the maximum number of simultaneous
+ * merges with {@link #setMaxMergeCount}. If the number of
+ * merges exceeds the max number of threads then the
+ * largest merges are paused until one of the smaller
+ * merges completes.</p>
+ *
+ * <p>If more than {@link #getMaxMergeCount} merges are
+ * requested then this class will forcefully throttle the
+ * incoming threads by pausing until one more more merges
+ * complete.</p>
+ */
+public class ConcurrentMergeScheduler extends MergeScheduler {
+
+ private int mergeThreadPriority = -1;
+
+ protected List<MergeThread> mergeThreads = new ArrayList<MergeThread>();
+
+ // Max number of merge threads allowed to be running at
+ // once. When there are more merges then this, we
+ // forcefully pause the larger ones, letting the smaller
+ // ones run, up until maxMergeCount merges at which point
+ // we forcefully pause incoming threads (that presumably
+ // are the ones causing so much merging). We dynamically
+ // default this from 1 to 3, depending on how many cores
+ // you have:
+ private int maxThreadCount = Math.max(1, Math.min(3, Runtime.getRuntime().availableProcessors()/2));
+
+ // Max number of merges we accept before forcefully
+ // throttling the incoming threads
+ private int maxMergeCount = maxThreadCount+2;
+
+ protected Directory dir;
+
+ private volatile boolean closed;
+ protected IndexWriter writer;
+ protected int mergeThreadCount;
+
+ public ConcurrentMergeScheduler() {
+ if (allInstances != null) {
+ // Only for testing
+ addMyself();
+ }
+ }
+
+ /** Sets the max # simultaneous merge threads that should
+ * be running at once. This must be <= {@link
+ * #setMaxMergeCount}. */
+ public void setMaxThreadCount(int count) {
+ if (count < 1) {
+ throw new IllegalArgumentException("count should be at least 1");
+ }
+ if (count > maxMergeCount) {
+ throw new IllegalArgumentException("count should be <= maxMergeCount (= " + maxMergeCount + ")");
+ }
+ maxThreadCount = count;
+ }
+
+ /** @see #setMaxThreadCount(int) */
+ public int getMaxThreadCount() {
+ return maxThreadCount;
+ }
+
+ /** Sets the max # simultaneous merges that are allowed.
+ * If a merge is necessary yet we already have this many
+ * threads running, the incoming thread (that is calling
+ * add/updateDocument) will block until a merge thread
+ * has completed. Note that we will only run the
+ * smallest {@link #setMaxThreadCount} merges at a time. */
+ public void setMaxMergeCount(int count) {
+ if (count < 1) {
+ throw new IllegalArgumentException("count should be at least 1");
+ }
+ if (count < maxThreadCount) {
+ throw new IllegalArgumentException("count should be >= maxThreadCount (= " + maxThreadCount + ")");
+ }
+ maxMergeCount = count;
+ }
+
+ /** See {@link #setMaxMergeCount}. */
+ public int getMaxMergeCount() {
+ return maxMergeCount;
+ }
+
+ /** Return the priority that merge threads run at. By
+ * default the priority is 1 plus the priority of (ie,
+ * slightly higher priority than) the first thread that
+ * calls merge. */
+ public synchronized int getMergeThreadPriority() {
+ initMergeThreadPriority();
+ return mergeThreadPriority;
+ }
+
+ /** Set the base priority that merge threads run at.
+ * Note that CMS may increase priority of some merge
+ * threads beyond this base priority. It's best not to
+ * set this any higher than
+ * Thread.MAX_PRIORITY-maxThreadCount, so that CMS has
+ * room to set relative priority among threads. */
+ public synchronized void setMergeThreadPriority(int pri) {
+ if (pri > Thread.MAX_PRIORITY || pri < Thread.MIN_PRIORITY)
+ throw new IllegalArgumentException("priority must be in range " + Thread.MIN_PRIORITY + " .. " + Thread.MAX_PRIORITY + " inclusive");
+ mergeThreadPriority = pri;
+ updateMergeThreads();
+ }
+
+ // Larger merges come first
+ protected static final Comparator<MergeThread> compareByMergeDocCount = new Comparator<MergeThread>() {
+ public int compare(MergeThread t1, MergeThread t2) {
+ final MergePolicy.OneMerge m1 = t1.getCurrentMerge();
+ final MergePolicy.OneMerge m2 = t2.getCurrentMerge();
+
+ final int c1 = m1 == null ? Integer.MAX_VALUE : m1.totalDocCount;
+ final int c2 = m2 == null ? Integer.MAX_VALUE : m2.totalDocCount;
+
+ return c2 - c1;
+ }
+ };
+
+ /**
+ * Called whenever the running merges have changed, to pause & unpause
+ * threads. This method sorts the merge threads by their merge size in
+ * descending order and then pauses/unpauses threads from first to last --
+ * that way, smaller merges are guaranteed to run before larger ones.
+ */
+ protected synchronized void updateMergeThreads() {
+
+ // Only look at threads that are alive & not in the
+ // process of stopping (ie have an active merge):
+ final List<MergeThread> activeMerges = new ArrayList<MergeThread>();
+
+ int threadIdx = 0;
+ while (threadIdx < mergeThreads.size()) {
+ final MergeThread mergeThread = mergeThreads.get(threadIdx);
+ if (!mergeThread.isAlive()) {
+ // Prune any dead threads
+ mergeThreads.remove(threadIdx);
+ continue;
+ }
+ if (mergeThread.getCurrentMerge() != null) {
+ activeMerges.add(mergeThread);
+ }
+ threadIdx++;
+ }
+
+ // Sort the merge threads in descending order.
+ CollectionUtil.mergeSort(activeMerges, compareByMergeDocCount);
+
+ int pri = mergeThreadPriority;
+ final int activeMergeCount = activeMerges.size();
+ for (threadIdx=0;threadIdx<activeMergeCount;threadIdx++) {
+ final MergeThread mergeThread = activeMerges.get(threadIdx);
+ final MergePolicy.OneMerge merge = mergeThread.getCurrentMerge();
+ if (merge == null) {
+ continue;
+ }
+
+ // pause the thread if maxThreadCount is smaller than the number of merge threads.
+ final boolean doPause = threadIdx < activeMergeCount - maxThreadCount;
+
+ if (verbose()) {
+ if (doPause != merge.getPause()) {
+ if (doPause) {
+ message("pause thread " + mergeThread.getName());
+ } else {
+ message("unpause thread " + mergeThread.getName());
+ }
+ }
+ }
+ if (doPause != merge.getPause()) {
+ merge.setPause(doPause);
+ }
+
+ if (!doPause) {
+ if (verbose()) {
+ message("set priority of merge thread " + mergeThread.getName() + " to " + pri);
+ }
+ mergeThread.setThreadPriority(pri);
+ pri = Math.min(Thread.MAX_PRIORITY, 1+pri);
+ }
+ }
+ }
+
+ /**
+ * Returns true if verbosing is enabled. This method is usually used in
+ * conjunction with {@link #message(String)}, like that:
+ *
+ * <pre>
+ * if (verbose()) {
+ * message("your message");
+ * }
+ * </pre>
+ */
+ protected boolean verbose() {
+ return writer != null && writer.verbose();
+ }
+
+ /**
+ * Outputs the given message - this method assumes {@link #verbose()} was
+ * called and returned true.
+ */
+ protected void message(String message) {
+ writer.message("CMS: " + message);
+ }
+
+ private synchronized void initMergeThreadPriority() {
+ if (mergeThreadPriority == -1) {
+ // Default to slightly higher priority than our
+ // calling thread
+ mergeThreadPriority = 1+Thread.currentThread().getPriority();
+ if (mergeThreadPriority > Thread.MAX_PRIORITY)
+ mergeThreadPriority = Thread.MAX_PRIORITY;
+ }
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ sync();
+ }
+
+ /** Wait for any running merge threads to finish */
+ public void sync() {
+ while (true) {
+ MergeThread toSync = null;
+ synchronized (this) {
+ for (MergeThread t : mergeThreads) {
+ if (t.isAlive()) {
+ toSync = t;
+ break;
+ }
+ }
+ }
+ if (toSync != null) {
+ try {
+ toSync.join();
+ } catch (InterruptedException ie) {
+ throw new ThreadInterruptedException(ie);
+ }
+ } else {
+ break;
+ }
+ }
+ }
+
+ /**
+ * Returns the number of merge threads that are alive. Note that this number
+ * is ≤ {@link #mergeThreads} size.
+ */
+ protected synchronized int mergeThreadCount() {
+ int count = 0;
+ for (MergeThread mt : mergeThreads) {
+ if (mt.isAlive() && mt.getCurrentMerge() != null) {
+ count++;
+ }
+ }
+ return count;
+ }
+
+ @Override
+ public void merge(IndexWriter writer) throws IOException {
+
+ assert !Thread.holdsLock(writer);
+
+ this.writer = writer;
+
+ initMergeThreadPriority();
+
+ dir = writer.getDirectory();
+
+ // First, quickly run through the newly proposed merges
+ // and add any orthogonal merges (ie a merge not
+ // involving segments already pending to be merged) to
+ // the queue. If we are way behind on merging, many of
+ // these newly proposed merges will likely already be
+ // registered.
+
+ if (verbose()) {
+ message("now merge");
+ message(" index: " + writer.segString());
+ }
+
+ // Iterate, pulling from the IndexWriter's queue of
+ // pending merges, until it's empty:
+ while (true) {
+
+ synchronized(this) {
+ long startStallTime = 0;
+ while (mergeThreadCount() >= 1+maxMergeCount) {
+ startStallTime = System.currentTimeMillis();
+ if (verbose()) {
+ message(" too many merges; stalling...");
+ }
+ try {
+ wait();
+ } catch (InterruptedException ie) {
+ throw new ThreadInterruptedException(ie);
+ }
+ }
+
+ if (verbose()) {
+ if (startStallTime != 0) {
+ message(" stalled for " + (System.currentTimeMillis()-startStallTime) + " msec");
+ }
+ }
+ }
+
+
+ // TODO: we could be careful about which merges to do in
+ // the BG (eg maybe the "biggest" ones) vs FG, which
+ // merges to do first (the easiest ones?), etc.
+ MergePolicy.OneMerge merge = writer.getNextMerge();
+ if (merge == null) {
+ if (verbose())
+ message(" no more merges pending; now return");
+ return;
+ }
+
+ // We do this w/ the primary thread to keep
+ // deterministic assignment of segment names
+ writer.mergeInit(merge);
+
+ boolean success = false;
+ try {
+ synchronized(this) {
+ message(" consider merge " + merge.segString(dir));
+
+ // OK to spawn a new merge thread to handle this
+ // merge:
+ final MergeThread merger = getMergeThread(writer, merge);
+ mergeThreads.add(merger);
+ if (verbose()) {
+ message(" launch new thread [" + merger.getName() + "]");
+ }
+
+ merger.start();
+
+ // Must call this after starting the thread else
+ // the new thread is removed from mergeThreads
+ // (since it's not alive yet):
+ updateMergeThreads();
+
+ success = true;
+ }
+ } finally {
+ if (!success) {
+ writer.mergeFinish(merge);
+ }
+ }
+ }
+ }
+
+ /** Does the actual merge, by calling {@link IndexWriter#merge} */
+ protected void doMerge(MergePolicy.OneMerge merge) throws IOException {
+ writer.merge(merge);
+ }
+
+ /** Create and return a new MergeThread */
+ protected synchronized MergeThread getMergeThread(IndexWriter writer, MergePolicy.OneMerge merge) throws IOException {
+ final MergeThread thread = new MergeThread(writer, merge);
+ thread.setThreadPriority(mergeThreadPriority);
+ thread.setDaemon(true);
+ thread.setName("Lucene Merge Thread #" + mergeThreadCount++);
+ return thread;
+ }
+
+ protected class MergeThread extends Thread {
+
+ IndexWriter tWriter;
+ MergePolicy.OneMerge startMerge;
+ MergePolicy.OneMerge runningMerge;
+ private volatile boolean done;
+
+ public MergeThread(IndexWriter writer, MergePolicy.OneMerge startMerge) throws IOException {
+ this.tWriter = writer;
+ this.startMerge = startMerge;
+ }
+
+ public synchronized void setRunningMerge(MergePolicy.OneMerge merge) {
+ runningMerge = merge;
+ }
+
+ public synchronized MergePolicy.OneMerge getRunningMerge() {
+ return runningMerge;
+ }
+
+ public synchronized MergePolicy.OneMerge getCurrentMerge() {
+ if (done) {
+ return null;
+ } else if (runningMerge != null) {
+ return runningMerge;
+ } else {
+ return startMerge;
+ }
+ }
+
+ public void setThreadPriority(int pri) {
+ try {
+ setPriority(pri);
+ } catch (NullPointerException npe) {
+ // Strangely, Sun's JDK 1.5 on Linux sometimes
+ // throws NPE out of here...
+ } catch (SecurityException se) {
+ // Ignore this because we will still run fine with
+ // normal thread priority
+ }
+ }
+
+ @Override
+ public void run() {
+
+ // First time through the while loop we do the merge
+ // that we were started with:
+ MergePolicy.OneMerge merge = this.startMerge;
+
+ try {
+
+ if (verbose())
+ message(" merge thread: start");
+
+ while(true) {
+ setRunningMerge(merge);
+ doMerge(merge);
+
+ // Subsequent times through the loop we do any new
+ // merge that writer says is necessary:
+ merge = tWriter.getNextMerge();
+ if (merge != null) {
+ tWriter.mergeInit(merge);
+ updateMergeThreads();
+ if (verbose())
+ message(" merge thread: do another merge " + merge.segString(dir));
+ } else {
+ break;
+ }
+ }
+
+ if (verbose())
+ message(" merge thread: done");
+
+ } catch (Throwable exc) {
+
+ // Ignore the exception if it was due to abort:
+ if (!(exc instanceof MergePolicy.MergeAbortedException)) {
+ if (!suppressExceptions) {
+ // suppressExceptions is normally only set during
+ // testing.
+ anyExceptions = true;
+ handleMergeException(exc);
+ }
+ }
+ } finally {
+ done = true;
+ synchronized(ConcurrentMergeScheduler.this) {
+ updateMergeThreads();
+ ConcurrentMergeScheduler.this.notifyAll();
+ }
+ }
+ }
+
+ @Override
+ public String toString() {
+ MergePolicy.OneMerge merge = getRunningMerge();
+ if (merge == null)
+ merge = startMerge;
+ return "merge thread: " + merge.segString(dir);
+ }
+ }
+
+ /** Called when an exception is hit in a background merge
+ * thread */
+ protected void handleMergeException(Throwable exc) {
+ try {
+ // When an exception is hit during merge, IndexWriter
+ // removes any partial files and then allows another
+ // merge to run. If whatever caused the error is not
+ // transient then the exception will keep happening,
+ // so, we sleep here to avoid saturating CPU in such
+ // cases:
+ Thread.sleep(250);
+ } catch (InterruptedException ie) {
+ throw new ThreadInterruptedException(ie);
+ }
+ throw new MergePolicy.MergeException(exc, dir);
+ }
+
+ static boolean anyExceptions = false;
+
+ /** Used for testing */
+ public static boolean anyUnhandledExceptions() {
+ if (allInstances == null) {
+ throw new RuntimeException("setTestMode() was not called; often this is because your test case's setUp method fails to call super.setUp in LuceneTestCase");
+ }
+ synchronized(allInstances) {
+ final int count = allInstances.size();
+ // Make sure all outstanding threads are done so we see
+ // any exceptions they may produce:
+ for(int i=0;i<count;i++)
+ allInstances.get(i).sync();
+ boolean v = anyExceptions;
+ anyExceptions = false;
+ return v;
+ }
+ }
+
+ public static void clearUnhandledExceptions() {
+ synchronized(allInstances) {
+ anyExceptions = false;
+ }
+ }
+
+ /** Used for testing */
+ private void addMyself() {
+ synchronized(allInstances) {
+ final int size = allInstances.size();
+ int upto = 0;
+ for(int i=0;i<size;i++) {
+ final ConcurrentMergeScheduler other = allInstances.get(i);
+ if (!(other.closed && 0 == other.mergeThreadCount()))
+ // Keep this one for now: it still has threads or
+ // may spawn new threads
+ allInstances.set(upto++, other);
+ }
+ allInstances.subList(upto, allInstances.size()).clear();
+ allInstances.add(this);
+ }
+ }
+
+ private boolean suppressExceptions;
+
+ /** Used for testing */
+ void setSuppressExceptions() {
+ suppressExceptions = true;
+ }
+
+ /** Used for testing */
+ void clearSuppressExceptions() {
+ suppressExceptions = false;
+ }
+
+ /** Used for testing */
+ private static List<ConcurrentMergeScheduler> allInstances;
+
+ /** @deprecated this test mode code will be removed in a future release */
+ @Deprecated
+ public static void setTestMode() {
+ allInstances = new ArrayList<ConcurrentMergeScheduler>();
+ }
+}
fnp / pylucene.git / blobdiff