Apache
/
lucene
mirror of https://github.com/apache/lucene.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

LUCENE-1022: share writer's infoStream with LogMergePolicy & ConcurrentMergeScheduler 

git-svn-id: https://svn.apache.org/repos/asf/lucene/java/trunk@587101 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Michael McCandless 2007-10-22 14:04:31 +00:00
parent 1df7007219
commit 38bbbb51de
3 changed files with 40 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java
View File

@ -33,8 +33,6 @@ import java.util.ArrayList;
public class ConcurrentMergeScheduler implements MergeScheduler {
public static boolean VERBOSE = false;
private int mergeThreadPriority = -1;
private List mergeThreads = new ArrayList();
@ -44,6 +42,7 @@ public class ConcurrentMergeScheduler implements MergeScheduler {
private Directory dir;
private boolean closed;
private IndexWriter writer;
public ConcurrentMergeScheduler() {
if (allInstances != null) {
@ -94,7 +93,8 @@ public class ConcurrentMergeScheduler implements MergeScheduler {
}
private void message(String message) {
System.out.println("CMS [" + Thread.currentThread().getName() + "]: " + message);
if (writer != null)
writer.message("CMS: " + message);
}
private synchronized void initMergeThreadPriority() {
@ -110,11 +110,10 @@ public class ConcurrentMergeScheduler implements MergeScheduler {
public synchronized void sync() {
while(mergeThreadCount() > 0) {
if (VERBOSE) {
message("now wait for threads; currently " + mergeThreads.size() + " still running");
for(int i=0;i<mergeThreads.size();i++)
final int count = mergeThreads.size();
for(int i=0;i<count;i++)
message(" " + i + ": " + ((MergeThread) mergeThreads.get(i)));
}
try {
wait();
@ -129,6 +128,8 @@ public class ConcurrentMergeScheduler implements MergeScheduler {
public void merge(IndexWriter writer)
throws CorruptIndexException, IOException {
this.writer = writer;
initMergeThreadPriority();
dir = writer.getDirectory();
@ -140,10 +141,8 @@ public class ConcurrentMergeScheduler implements MergeScheduler {
// 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 its empty:
@ -155,7 +154,6 @@ public class ConcurrentMergeScheduler implements MergeScheduler {
MergePolicy.OneMerge merge = writer.getNextMerge();
if (merge == null) {
if (VERBOSE)
message(" no more merges pending; now return");
return;
}
@ -164,11 +162,9 @@ public class ConcurrentMergeScheduler implements MergeScheduler {
// deterministic assignment of segment names
writer.mergeInit(merge);
if (VERBOSE)
message(" consider merge " + merge.segString(dir));
if (merge.isExternal) {
if (VERBOSE)
message(" merge involves segments from an external directory; now run in foreground");
} else {
synchronized(this) {
@ -177,7 +173,6 @@ public class ConcurrentMergeScheduler implements MergeScheduler {
// merge:
MergeThread merger = new MergeThread(writer, merge);
mergeThreads.add(merger);
if (VERBOSE)
message(" launch new thread [" + merger.getName() + "]");
try {
merger.setPriority(mergeThreadPriority);
@ -187,7 +182,7 @@ public class ConcurrentMergeScheduler implements MergeScheduler {
}
merger.start();
continue;
} else if (VERBOSE)
} else
message(" too many merge threads running; run merge in foreground");
}
}
@ -225,7 +220,6 @@ public class ConcurrentMergeScheduler implements MergeScheduler {
try {
if (VERBOSE)
message(" merge thread: start");
while(true) {
@ -237,13 +231,11 @@ public class ConcurrentMergeScheduler implements MergeScheduler {
merge = writer.getNextMerge();
if (merge != null) {
writer.mergeInit(merge);
if (VERBOSE)
message(" merge thread: do another merge " + merge.segString(dir));
} else
break;
}
if (VERBOSE)
message(" merge thread: done");
} catch (Throwable exc) {

8
src/java/org/apache/lucene/index/IndexWriter.java
View File

@ -304,7 +304,13 @@ public class IndexWriter {
}
}
private void message(String message) {
/**
* Prints a message to the infoStream (if non-null),
* prefixed with the identifying information for this
* writer and the thread that's calling it.
*/
public void message(String message) {
if (infoStream != null)
infoStream.println("IW " + messageID + " [" + Thread.currentThread().getName() + "]: " + message);
}

14
src/java/org/apache/lucene/index/LogMergePolicy.java
View File

@ -61,6 +61,12 @@ public abstract class LogMergePolicy implements MergePolicy {
private boolean useCompoundFile = true;
private boolean useCompoundDocStore = true;
private IndexWriter writer;
private void message(String message) {
if (writer != null)
writer.message("LMP: " + message);
}
/** <p>Returns the number of segments that are merged at
* once and also controls the total number of segments
@ -211,6 +217,8 @@ public abstract class LogMergePolicy implements MergePolicy {
public MergeSpecification findMerges(SegmentInfos infos, IndexWriter writer) throws IOException {
final int numSegments = infos.size();
this.writer = writer;
message("findMerges: " + numSegments + " segments");
// Compute levels, which is just log (base mergeFactor)
// of the size of each segment
@ -284,6 +292,7 @@ public abstract class LogMergePolicy implements MergePolicy {
}
upto--;
}
message(" level " + levelBottom + " to " + maxLevel + ": " + (1+upto-start) + " segments");
// Finally, record all merges that are viable at this level:
int end = start + mergeFactor;
@ -297,8 +306,11 @@ public abstract class LogMergePolicy implements MergePolicy {
if (!anyTooLarge) {
if (spec == null)
spec = new MergeSpecification();
message(" " + start + " to " + end + ": add this merge");
spec.add(new OneMerge(infos.range(start, end), useCompoundFile));
}
} else
message(" " + start + " to " + end + ": contains segment over maxMergeSize or maxMergeDocs; skipping");
start = end;
end = start + mergeFactor;
}