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 2010/12/20 11:41:05 UTC
svn commit: r1051044 - in /lucene/dev/branches/branch_3x: ./ lucene/
lucene/src/java/org/apache/lucene/index/
lucene/src/test/org/apache/lucene/index/
lucene/src/test/org/apache/lucene/store/ solr/
Author: mikemccand
Date: Mon Dec 20 10:41:05 2010
New Revision: 1051044
URL: http://svn.apache.org/viewvc?rev=1051044&view=rev
Log:
LUCENE-2820: fix CMS to stop its threads without deadlocking!
Modified:
lucene/dev/branches/branch_3x/ (props changed)
lucene/dev/branches/branch_3x/lucene/ (props changed)
lucene/dev/branches/branch_3x/lucene/src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java
lucene/dev/branches/branch_3x/lucene/src/test/org/apache/lucene/index/TestIndexWriterExceptions.java
lucene/dev/branches/branch_3x/lucene/src/test/org/apache/lucene/store/MockDirectoryWrapper.java
lucene/dev/branches/branch_3x/solr/ (props changed)
Modified: lucene/dev/branches/branch_3x/lucene/src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_3x/lucene/src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java?rev=1051044&r1=1051043&r2=1051044&view=diff
==============================================================================
--- lucene/dev/branches/branch_3x/lucene/src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java (original)
+++ lucene/dev/branches/branch_3x/lucene/src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java Mon Dec 20 10:41:05 2010
@@ -65,7 +65,7 @@ public class ConcurrentMergeScheduler ex
protected Directory dir;
- private boolean closed;
+ private volatile boolean closed;
protected IndexWriter writer;
protected int mergeThreadCount;
@@ -154,18 +154,37 @@ public class ConcurrentMergeScheduler ex
* pause & unpause threads. */
protected synchronized void updateMergeThreads() {
- CollectionUtil.mergeSort(mergeThreads, compareByMergeDocCount);
+ // 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++;
+ }
+
+ CollectionUtil.mergeSort(activeMerges, compareByMergeDocCount);
- final int count = mergeThreads.size();
int pri = mergeThreadPriority;
- for(int i=0;i<count;i++) {
- final MergeThread mergeThread = mergeThreads.get(i);
+ 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) {
+ if (merge == null) {
continue;
}
+
final boolean doPause;
- if (i < count-maxThreadCount) {
+ if (threadIdx < activeMergeCount-maxThreadCount) {
doPause = true;
} else {
doPause = false;
@@ -216,22 +235,29 @@ public class ConcurrentMergeScheduler ex
@Override
public void close() {
closed = true;
+ sync();
}
- public synchronized void sync() {
- while(mergeThreadCount() > 0) {
- if (verbose())
- message("now wait for threads; currently " + mergeThreads.size() + " still running");
- final int count = mergeThreads.size();
- if (verbose()) {
- for(int i=0;i<count;i++)
- message(" " + i + ": " + mergeThreads.get(i));
+ /** 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;
+ }
+ }
}
-
- try {
- wait();
- } catch (InterruptedException ie) {
- throw new ThreadInterruptedException(ie);
+ if (toSync != null) {
+ try {
+ toSync.join();
+ } catch (InterruptedException ie) {
+ throw new ThreadInterruptedException(ie);
+ }
+ } else {
+ break;
}
}
}
@@ -239,9 +265,12 @@ public class ConcurrentMergeScheduler ex
private synchronized int mergeThreadCount() {
int count = 0;
final int numThreads = mergeThreads.size();
- for(int i=0;i<numThreads;i++)
- if (mergeThreads.get(i).isAlive())
+ for(int i=0;i<numThreads;i++) {
+ final MergeThread t = mergeThreads.get(i);
+ if (t.isAlive() && t.getCurrentMerge() != null) {
count++;
+ }
+ }
return count;
}
@@ -318,11 +347,17 @@ public class ConcurrentMergeScheduler ex
// merge:
merger = getMergeThread(writer, merge);
mergeThreads.add(merger);
- updateMergeThreads();
- if (verbose())
+ 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 {
@@ -415,8 +450,6 @@ public class ConcurrentMergeScheduler ex
if (verbose())
message(" merge thread: do another merge " + merge.segString(dir));
} else {
- done = true;
- updateMergeThreads();
break;
}
}
@@ -436,11 +469,10 @@ public class ConcurrentMergeScheduler ex
}
}
} finally {
+ done = true;
synchronized(ConcurrentMergeScheduler.this) {
- ConcurrentMergeScheduler.this.notifyAll();
- boolean removed = mergeThreads.remove(this);
- assert removed;
updateMergeThreads();
+ ConcurrentMergeScheduler.this.notifyAll();
}
}
}
Modified: lucene/dev/branches/branch_3x/lucene/src/test/org/apache/lucene/index/TestIndexWriterExceptions.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_3x/lucene/src/test/org/apache/lucene/index/TestIndexWriterExceptions.java?rev=1051044&r1=1051043&r2=1051044&view=diff
==============================================================================
--- lucene/dev/branches/branch_3x/lucene/src/test/org/apache/lucene/index/TestIndexWriterExceptions.java (original)
+++ lucene/dev/branches/branch_3x/lucene/src/test/org/apache/lucene/index/TestIndexWriterExceptions.java Mon Dec 20 10:41:05 2010
@@ -841,10 +841,14 @@ public class TestIndexWriterExceptions e
w.close();
for(int i=0;i<200;i++) {
+ if (VERBOSE) {
+ System.out.println("TEST: iter " + i);
+ }
MockDirectoryWrapper dir = new MockDirectoryWrapper(random, new RAMDirectory(startDir));
conf = newIndexWriterConfig( TEST_VERSION_CURRENT, new MockAnalyzer()).setMergeScheduler(new ConcurrentMergeScheduler());
((ConcurrentMergeScheduler) conf.getMergeScheduler()).setSuppressExceptions();
w = new IndexWriter(dir, conf);
+ w.setInfoStream(VERBOSE ? System.out : null);
dir.setRandomIOExceptionRate(0.5);
try {
w.optimize();
Modified: lucene/dev/branches/branch_3x/lucene/src/test/org/apache/lucene/store/MockDirectoryWrapper.java
URL: http://svn.apache.org/viewvc/lucene/dev/branches/branch_3x/lucene/src/test/org/apache/lucene/store/MockDirectoryWrapper.java?rev=1051044&r1=1051043&r2=1051044&view=diff
==============================================================================
--- lucene/dev/branches/branch_3x/lucene/src/test/org/apache/lucene/store/MockDirectoryWrapper.java (original)
+++ lucene/dev/branches/branch_3x/lucene/src/test/org/apache/lucene/store/MockDirectoryWrapper.java Mon Dec 20 10:41:05 2010
@@ -18,18 +18,20 @@ package org.apache.lucene.store;
*/
import java.io.Closeable;
-import java.io.IOException;
import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.Iterator;
-import java.util.Random;
import java.util.Map;
-import java.util.HashMap;
-import java.util.HashSet;
+import java.util.Random;
import java.util.Set;
-import java.util.ArrayList;
+
+import org.apache.lucene.util.LuceneTestCase;
/**
* This is a Directory Wrapper that adds methods
@@ -231,6 +233,10 @@ public class MockDirectoryWrapper extend
if (randomIOExceptionRate > 0.0) {
int number = Math.abs(randomState.nextInt() % 1000);
if (number < randomIOExceptionRate*1000) {
+ if (LuceneTestCase.VERBOSE) {
+ System.out.println(Thread.currentThread().getName() + ": MockDirectoryWrapper: now throw random exception");
+ new Throwable().printStackTrace(System.out);
+ }
throw new IOException("a random IOException");
}
}