Index: src/test/org/apache/lucene/index/TestRAMNRT.java =================================================================== --- src/test/org/apache/lucene/index/TestRAMNRT.java (revision 0) +++ src/test/org/apache/lucene/index/TestRAMNRT.java (revision 0) @@ -0,0 +1,224 @@ +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.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.lucene.analysis.WhitespaceAnalyzer; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.MockRAMDirectory; +import org.apache.lucene.store.RAMDirectory; +import org.apache.lucene.util.LuceneTestCase; + +public class TestRAMNRT extends LuceneTestCase { + Random random = new Random(); + + public void testSimple() throws Exception { + for (int x=0; x < 10; x++) { + doTestSimple(); + } + } + + public void doTestSimple() throws Exception { + Directory primaryDir = new MockRAMDirectory(); + RAMDirectory ramDir = new MockRAMDirectory(); + + RAMNRT ramnrt = new RAMNRT(primaryDir, new WhitespaceAnalyzer(), true, + IndexWriter.MaxFieldLength.LIMITED, ramDir, 1024 * 150); // 150k + ramnrt.setInfoStream(System.out); + + IndexWriter primaryWriter = ramnrt.getPrimaryWriter(); + for (int i = 0; i < 100; i++) { + primaryWriter.addDocument(TestIndexWriterReader.createDocument(i, + "primary", 4)); + } + for (int i = 0; i < 100; i++) { + ramnrt.addDocument(TestIndexWriterReader.createDocument(i, "ram", 4)); + } + + IndexReader reader = ramnrt.getReader(); + assertEquals(200, reader.maxDoc()); + IndexReader ramReader = ramnrt.getRAMWriter().getReader(); + assertEquals(100, ramReader.maxDoc()); + IndexReader primaryReader = ramnrt.getPrimaryWriter().getReader(); + assertEquals(100, primaryReader.maxDoc()); + ramReader.close(); + primaryReader.close(); + reader.close(); + + ramnrt.flush(); + ((ConcurrentMergeScheduler)ramnrt.getPrimaryWriter().getMergeScheduler()).sync(); + primaryReader = ramnrt.getPrimaryWriter().getReader(); + assertEquals(200, primaryReader.maxDoc()); + ramReader = ramnrt.getRAMWriter().getReader(); + assertEquals(0, ramReader.maxDoc()); + + ramReader.close(); + primaryReader.close(); + + ramnrt.close(); + + primaryDir.close(); + ramDir.close(); + } + + /** + * This tests synchronization of the primary and ram writers. + * + * @throws Exception + */ + public void testRandomThreads() throws Exception { + long duration = 1000 * 35; + AtomicInteger totalAdded = new AtomicInteger(0); + + Directory primaryDir = new MockRAMDirectory(); + RAMDirectory ramDir = new MockRAMDirectory(); + + RAMNRT ramnrt = new RAMNRT(primaryDir, new WhitespaceAnalyzer(), true, + IndexWriter.MaxFieldLength.LIMITED, ramDir, 1024 * 500); // 500k + ramnrt.setInfoStream(System.out); + + GetReadersThread[] grts = new GetReadersThread[3]; + for (int x = 0; x < grts.length; x++) { + grts[x] = new GetReadersThread("getreadersthread" + x, ramnrt); + } + AddDocsThread[] adts = new AddDocsThread[3]; + for (int x = 0; x < adts.length; x++) { + adts[x] = new AddDocsThread("adddocsthread " + x, ramnrt, totalAdded); + } + for (int x = 0; x < adts.length; x++) { + adts[x].start(); + } + for (int x = 0; x < grts.length; x++) { + grts[x].start(); + } + long startTime = System.currentTimeMillis(); + while (true) { + if ((System.currentTimeMillis() - startTime) > duration) { + break; + } + } + System.out.println(duration+" elapsed"); + for (int x = 0; x < adts.length; x++) { + adts[x].dorun = false; + } + for (int x = 0; x < grts.length; x++) { + grts[x].dorun = false; + } + for (int x = 0; x < adts.length; x++) { + adts[x].waitForThread(); + System.out.println("add docs errors "+x); + //for (Throwable th : adts[x].errors) { + // th.printStackTrace(); + //} + } + for (int x = 0; x < grts.length; x++) { + grts[x].waitForThread(); + } + System.out.println("Completed..."); + IndexReader reader = ramnrt.getReader(); + assertEquals(totalAdded.get(), reader.maxDoc()); + ramnrt.close(); + primaryDir.close(); + ramDir.close(); + } + + public class GenericThread extends Thread { + boolean dorun = true; + + List errors = new ArrayList(); + + RAMNRT ramnrt; + + public GenericThread(String threadName, RAMNRT ramnrt) { + this.ramnrt = ramnrt; + setName(threadName); + } + + public void waitForThread() throws InterruptedException { + System.out.println("waiting for "+getName()); + join(); + } + } + + /** + * Check to insure the readers are not identical + */ + public class GetReadersThread extends GenericThread { + List failures = new ArrayList(); + + public GetReadersThread(String threadName, RAMNRT ramnrt) { + super(threadName, ramnrt); + } + + public void run() { + try { + while (dorun) { + int num = random.nextInt(100); + for (int x=0; x < num; x++) { + IndexReader reader = ramnrt.getReader(); + reader.close(); + } + Thread.sleep(1000); + } + } catch (Throwable ex) { + ex.printStackTrace(); + errors.add(ex); + if (errors.size() > 5) { + throw new RuntimeException(ex); + } + } + } + } + + public class AddDocsThread extends GenericThread { + AtomicInteger totalAdded; + + public AddDocsThread(String threadName, RAMNRT ramnrt, + AtomicInteger totalAdded) { + super(threadName, ramnrt); + this.totalAdded = totalAdded; + } + + public void run() { + try { + while (dorun) { + int numdocs = random.nextInt(2000); + for (int x = 0; x < numdocs; x++) { + if (!dorun) + return; + ramnrt.addDocument(TestIndexWriterReader.createDocument(x, "", 3)); + totalAdded.incrementAndGet(); + } + System.out.println("added " + numdocs + " docs"); + ramnrt.flush(); + System.out.println("added " + numdocs + " docs after flush"); + } + } catch (Throwable th) { + th.printStackTrace(); + errors.add(th); + } + } + } +} Index: src/java/org/apache/lucene/index/RAMNRT.java =================================================================== --- src/java/org/apache/lucene/index/RAMNRT.java (revision 0) +++ src/java/org/apache/lucene/index/RAMNRT.java (revision 0) @@ -0,0 +1,402 @@ +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.IOException; +import java.io.PrintStream; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.ReentrantLock; + +import org.apache.lucene.analysis.Analyzer; +import org.apache.lucene.document.Document; +import org.apache.lucene.index.DocumentsWriter.IndexingChain; +import org.apache.lucene.index.IndexWriter.MaxFieldLength; +import org.apache.lucene.search.Query; +import org.apache.lucene.store.Directory; +import org.apache.lucene.store.FileSwitchDirectory; +import org.apache.lucene.store.LockObtainFailedException; +import org.apache.lucene.store.RAMDirectory; + +/** + * Writers that use the same directory. The RAMDir segment names are generated + * from the primary writer. Updates are made to the ramWriter, which is flushed + * when the maximum allocated RAM is exceeded. + */ +// TODO: mergingSegments in ram writer needs to be cleared on abort and exit +// TODO: flush can be performed in the background? +public class RAMNRT { + private final static Set SWITCH_FILE_EXTS = new HashSet(); + static { + SWITCH_FILE_EXTS.add("fdx"); + SWITCH_FILE_EXTS.add("fdt"); + SWITCH_FILE_EXTS.add("tvx"); + SWITCH_FILE_EXTS.add("tvf"); + SWITCH_FILE_EXTS.add("tvd"); + } + + private RAMIndexWriter ramWriter; + + private PrimaryWriter primaryWriter; + + private RAMDirectory ramDirectory; + + AtomicBoolean flushing = new AtomicBoolean(false); + + private long maxRam; + + private PrintStream infoStream; + + ReentrantLock lock = new ReentrantLock(); + + public RAMNRT(Directory d, Analyzer a, boolean create, MaxFieldLength mfl, + RAMDirectory ramDir, long maxRam) throws IOException { + this.primaryWriter = new PrimaryWriter(d, a, create, mfl); + ConcurrentMergeScheduler primaryScheduler = new ConcurrentMergeScheduler( + "primary"); + primaryWriter.setMergeScheduler(primaryScheduler); + this.maxRam = maxRam; + Directory primaryDir = primaryWriter.getDirectory(); + ramDirectory = ramDir; + FileSwitchDirectory fsd = new FileSwitchDirectory(SWITCH_FILE_EXTS, + primaryDir, ramDirectory, true); + ramWriter = new RAMIndexWriter(fsd, primaryWriter.getAnalyzer(), true, + new MaxFieldLength(primaryWriter.getMaxFieldLength())); + ConcurrentMergeScheduler ramScheduler = new ConcurrentMergeScheduler("ram"); + ramWriter.setMergeScheduler(ramScheduler); + // because these segments are in RAM, + // we don't need to use the compound format + ramWriter.setUseCompoundFile(false); + } + + public void setInfoStream(PrintStream infoStream) { + this.infoStream = infoStream; + } + + public IndexWriter getPrimaryWriter() { + return primaryWriter; + } + + public IndexWriter getRAMWriter() { + return ramWriter; + } + + /** + * Closes the underlying primary and ram writers + * + * @throws IOException + */ + public void close() throws IOException { + assert !ramWriter.hasExternalSegments(); + ramWriter.close(); + primaryWriter.close(); + } + + /** + * Returns a MultiReader of the readers from the primary and ram writers. + * Duplicates may exist as ram segments are merged to the primary writer, this + * is OK because we remove them here. + * + * @return + * @throws IOException + */ + public IndexReader getReader() throws IOException { + IndexReader[] ramReaders = ramWriter.getReader().getSequentialSubReaders(); + IndexReader[] primaryReaders = primaryWriter.getReader() + .getSequentialSubReaders(); + List readers = new ArrayList(ramReaders.length + + primaryReaders.length); + int i = 0; + Set infos = new HashSet(); + for (int x = 0; x < primaryReaders.length; x++) { + SegmentInfo info = ((SegmentReader) primaryReaders[x]).getSegmentInfo(); + infos.add(info); + readers.add(primaryReaders[x]); + } + for (int x = 0; x < ramReaders.length; x++) { + SegmentInfo info = ((SegmentReader) ramReaders[x]).getSegmentInfo(); + if (!infos.contains(info)) { + infos.add(info); + readers.add(ramReaders[x]); + } + } + MultiReader reader = new MultiReader((IndexReader[]) readers + .toArray(new IndexReader[0])); + return reader; + } + + /** + * Returns the total ram usage, the ram writer's buffer plus the ram + * directory. + */ + public long getRAMSize() { + return ramWriter.ramSizeInBytes() + ramDirectory.sizeInBytes(); + } + + public void addDocument(Document doc) throws IOException, + InterruptedException { + addDocument(doc, primaryWriter.getAnalyzer()); + } + + public void addDocument(Document doc, Analyzer analyzer) throws IOException, + InterruptedException { + ramWriter.addDocument(doc, analyzer); + ifFlush(); + } + + public void deleteDocuments(Query query) throws CorruptIndexException, + IOException, InterruptedException { + ramWriter.deleteDocuments(query); + primaryWriter.deleteDocuments(query); + ifFlush(); + } + + public void deleteDocuments(Query... queries) throws CorruptIndexException, + IOException, InterruptedException { + ramWriter.deleteDocuments(queries); + primaryWriter.deleteDocuments(queries); + ifFlush(); + } + + public void updateDocument(Term term, Document doc, Analyzer analyzer) + throws CorruptIndexException, IOException, InterruptedException { + ramWriter.updateDocument(term, doc, analyzer); + primaryWriter.deleteDocuments(term); + ifFlush(); + } + + public void updateDocument(Term term, Document doc) + throws CorruptIndexException, IOException, InterruptedException { + ramWriter.updateDocument(term, doc); + primaryWriter.deleteDocuments(term); + ifFlush(); + } + + private void ifFlush() throws IOException, InterruptedException { + if (getRAMSize() > maxRam) { + long ramSizeBytes = ramWriter.ramSizeInBytes(); + long ramDirBytes = ramDirectory.sizeInBytes(); + // if (infoStream != null) + // infoStream.println("ramSize:"+ramSizeBytes+" ramDir:"+ramDirBytes); + flush(); + } + } + + /** + * Flushes the RAM writer, and merge all RAM segments to one primary directory + * segment. + * + * @throws IOException + */ + public void flush() throws IOException, InterruptedException { + flushing.set(true); + try { + // lock.tryLock(10*1000, TimeUnit.MILLISECONDS); + // try { + ramWriter.flush(false, true, true); + // merge all RAM segments to one primary directory segment + // get a copy of the infos because we don't want them changing + // from underneath us + SegmentInfos ramInfos = ramWriter.getSegmentInfosCopy(); + MergePolicy.OneMerge merge = new MergePolicy.OneMerge(ramInfos, + primaryWriter.getUseCompoundFile()); + merge.ramnrt = true; + boolean mergeRegistered = false; + synchronized (primaryWriter) { + // tell the ram writer we're merging these segments + // to the primary writer + if (ramWriter.addRAMMerge(merge)) { + System.out.println("adding ram merge files size:"+merge.segments.sizeOfFiles()+" "+merge.segString(ramWriter.getDirectory())); + for (SegmentInfo ramInfo : ramInfos) { + if (primaryWriter.segmentInfos.contains(ramInfo)) { + System.out.println("primary info contains ramInfo: " + ramInfo); + return; + } + } + for (SegmentInfo ramInfo : ramInfos) { + primaryWriter.segmentInfos.add(ramInfo); + } + mergeRegistered = primaryWriter.registerMerge(merge); + assert mergeRegistered; + } + } + if (mergeRegistered) + primaryWriter.maybeMerge(); + } finally { + flushing.set(false); + } + // perform the ram nrt merges synchronously + // primaryWriter.resolveExternalSegments(true); + + // there should not be any duplicates after resolve segments + // assert isUniques(primaryWriter.getSegmentInfosCopy()); + // assert isUniques(ramWriter.getSegmentInfosCopy()); + // assert Collections.disjoint(primaryWriter.getSegmentInfos(), ramWriter + // .getSegmentInfos()); + // } finally { + // lock.unlock(); + // } + } + + /** + * private static boolean isUniques(Collection col) { Set set = new + * HashSet(col); return col.size() == set.size(); } + **/ + public class PrimaryWriter extends IndexWriter { + public PrimaryWriter(Directory d, Analyzer a, boolean create, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl, + IndexingChain indexingChain, IndexCommit commit) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, create, deletionPolicy, mfl, indexingChain, commit); + } + + public PrimaryWriter(Directory d, Analyzer a, boolean create, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, create, deletionPolicy, mfl); + } + + public PrimaryWriter(Directory d, Analyzer a, boolean create, + MaxFieldLength mfl) throws CorruptIndexException, + LockObtainFailedException, IOException { + super(d, a, create, mfl); + } + + public PrimaryWriter(Directory d, Analyzer a, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl, + IndexCommit commit) throws CorruptIndexException, + LockObtainFailedException, IOException { + super(d, a, deletionPolicy, mfl, commit); + } + + public PrimaryWriter(Directory d, Analyzer a, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, deletionPolicy, mfl); + } + + public PrimaryWriter(Directory d, Analyzer a, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, mfl); + } + + void mergeSuccess(MergePolicy.OneMerge merge) { + try { + // here we remove segments from the ram writer that have been merged + // away + if (merge.ramnrt) { + if (infoStream != null) + infoStream.println("primaryWriter.mergeSuccess " + + merge.segString(primaryWriter.getDirectory())); + // it's one of our ram merges + // we need to remove these segments from the ram writer + synchronized (ramWriter) { + System.out.println("ramwriter remove merged segments: " + + merge.segments); + // TODO: assert there are no duplicate segments + ramWriter.removeSegments(merge.segments); + // this removes the segments from the ram writer + ramWriter.mergeFinish(merge); + ramWriter.removeMerging(merge.segments); + } + } + } catch (IOException ioe) { + // TODO: do something better with this exception + ioe.printStackTrace(); + } + } + } + + public class RAMIndexWriter extends IndexWriter { + Set ramMergingSegments = new HashSet(); + + public RAMIndexWriter(Directory d, Analyzer a, boolean create, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl, + IndexingChain indexingChain, IndexCommit commit) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, create, deletionPolicy, mfl, indexingChain, commit); + } + + public RAMIndexWriter(Directory d, Analyzer a, boolean create, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, create, deletionPolicy, mfl); + } + + public RAMIndexWriter(Directory d, Analyzer a, boolean create, + MaxFieldLength mfl) throws CorruptIndexException, + LockObtainFailedException, IOException { + super(d, a, create, mfl); + } + + public RAMIndexWriter(Directory d, Analyzer a, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl, + IndexCommit commit) throws CorruptIndexException, + LockObtainFailedException, IOException { + super(d, a, deletionPolicy, mfl, commit); + } + + public RAMIndexWriter(Directory d, Analyzer a, + IndexDeletionPolicy deletionPolicy, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, deletionPolicy, mfl); + } + + public RAMIndexWriter(Directory d, Analyzer a, MaxFieldLength mfl) + throws CorruptIndexException, LockObtainFailedException, IOException { + super(d, a, mfl); + } + + public synchronized boolean addRAMMerge(MergePolicy.OneMerge merge) + throws IOException { + int count = merge.segments.size(); + for (int i = 0; i < count; i++) { + final SegmentInfo info = merge.segments.info(i); + if (mergingSegments.contains(info)) + return false; + } + for (SegmentInfo info : merge.segments) { + mergingSegments.add(info); + } + return true; + } + + public synchronized void removeMerging(SegmentInfos infos) + throws IOException { + mergingSegments.removeAll(infos); + } + + protected synchronized void finishMerges(boolean waitForMerges) + throws IOException { + // clear out the somewhat fake ram segment merges + // they're fake because ram writer isn't actually merging them + // however they're marked as being merged so ram writer + // doesn't waste time merging segments that are being + // moved to the primary writer + mergingSegments.removeAll(ramMergingSegments); + super.finishMerges(waitForMerges); + } + + protected String newSegmentName() { + return primaryWriter.newSegmentName(); + } + } +} Index: src/java/org/apache/lucene/index/MergePolicy.java =================================================================== --- src/java/org/apache/lucene/index/MergePolicy.java (revision 831174) +++ src/java/org/apache/lucene/index/MergePolicy.java (working copy) @@ -85,6 +85,7 @@ final boolean useCompoundFile; boolean aborted; Throwable error; + boolean ramnrt = false; public OneMerge(SegmentInfos segments, boolean useCompoundFile) { if (0 == segments.size()) @@ -136,6 +137,9 @@ if (mergeDocStores) { b.append(" [mergeDocStores]"); } + if (ramnrt) { + b.append(" [ramnrt]"); + } return b.toString(); } } Index: src/java/org/apache/lucene/index/SegmentInfos.java =================================================================== --- src/java/org/apache/lucene/index/SegmentInfos.java (revision 831174) +++ src/java/org/apache/lucene/index/SegmentInfos.java (working copy) @@ -915,7 +915,15 @@ public Map getUserData() { return userData; } - + + public long sizeOfFiles() throws IOException { + long total = 0; + for (SegmentInfo info : this) { + total += info.sizeInBytes(); + } + return total; + } + void setUserData(Map data) { if (data == null) { userData = Collections.emptyMap(); Index: src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java =================================================================== --- src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java (revision 831174) +++ src/java/org/apache/lucene/index/ConcurrentMergeScheduler.java (working copy) @@ -46,7 +46,16 @@ private boolean closed; protected IndexWriter writer; protected int mergeThreadCount; - + String name; + + public ConcurrentMergeScheduler(String name) { + this.name = name; + if (allInstances != null) { + // Only for testing + addMyself(); + } + } + public ConcurrentMergeScheduler() { if (allInstances != null) { // Only for testing @@ -241,7 +250,7 @@ final MergeThread thread = new MergeThread(writer, merge); thread.setThreadPriority(mergeThreadPriority); thread.setDaemon(true); - thread.setName("Lucene Merge Thread #" + mergeThreadCount++); + thread.setName(name+" Lucene Merge Thread #" + mergeThreadCount++); return thread; } @@ -314,6 +323,9 @@ // suppressExceptions is normally only set during // testing. anyExceptions = true; + if (merge.ramnrt) { + System.out.println("ramnrt merge exception"); + } handleMergeException(exc); } } Index: src/java/org/apache/lucene/index/IndexWriter.java =================================================================== --- src/java/org/apache/lucene/index/IndexWriter.java (revision 831174) +++ src/java/org/apache/lucene/index/IndexWriter.java (working copy) @@ -253,7 +253,7 @@ private SegmentInfos localRollbackSegmentInfos; // segmentInfos we will fallback to if the commit fails private int localFlushedDocCount; // saved docWriter.getFlushedDocCount during local transaction - private SegmentInfos segmentInfos = new SegmentInfos(); // the segments + protected SegmentInfos segmentInfos = new SegmentInfos(); // the segments private DocumentsWriter docWriter; private IndexFileDeleter deleter; @@ -269,15 +269,15 @@ // Holds all SegmentInfo instances currently involved in // merges - private HashSet mergingSegments = new HashSet(); + protected HashSet mergingSegments = new HashSet(); private MergePolicy mergePolicy = new LogByteSizeMergePolicy(this); private MergeScheduler mergeScheduler = new ConcurrentMergeScheduler(); - private LinkedList pendingMerges = new LinkedList(); - private Set runningMerges = new HashSet(); + protected LinkedList pendingMerges = new LinkedList(); + protected Set runningMerges = new HashSet(); private List mergeExceptions = new ArrayList(); private long mergeGen; - private boolean stopMerges; + protected boolean stopMerges; private int flushCount; private int flushDeletesCount; @@ -2127,7 +2127,7 @@ return flushDeletesCount; } - final String newSegmentName() { + protected String newSegmentName() { // Cannot synchronize on IndexWriter because that causes // deadlock synchronized(segmentInfos) { @@ -2236,7 +2236,39 @@ public void optimize(boolean doWait) throws CorruptIndexException, IOException { optimize(1, doWait); } - + + /** + * Remove readers and segments . + * @param infos + * @throws CorruptIndexException + * @throws IOException + */ + void removeSegments(SegmentInfos infos) throws CorruptIndexException, IOException { + for (int i=0;i 0; if (stopMerges) @@ -2473,28 +2510,34 @@ } final MergePolicy.MergeSpecification spec; - if (optimize) { - spec = mergePolicy.findMergesForOptimize(segmentInfos, maxNumSegmentsOptimize, segmentsToOptimize); + if (mergeSpec != null) { + spec = mergeSpec; + } else { + if (optimize) { + spec = mergePolicy.findMergesForOptimize(segmentInfos, maxNumSegmentsOptimize, segmentsToOptimize); - if (spec != null) { - final int numMerges = spec.merges.size(); - for(int i=0;i it = pendingMerges.iterator(); while(it.hasNext()) { MergePolicy.OneMerge merge = it.next(); - if (merge.isExternal) { + if (merge.isExternal && (!ramnrt || (ramnrt && merge.ramnrt))) { // Advance the merge from pending to running it.remove(); runningMerges.add(merge); @@ -2806,7 +2853,7 @@ } } - private synchronized void finishMerges(boolean waitForMerges) throws IOException { + protected synchronized void finishMerges(boolean waitForMerges) throws IOException { if (!waitForMerges) { stopMerges = true; @@ -2825,7 +2872,7 @@ message("now abort running merge " + merge.segString(directory)); merge.abort(); } - + // Ensure any running addIndexes finishes. It's fine // if a new one attempts to start because its merges // will quickly see the stopMerges == true and abort. @@ -2875,7 +2922,9 @@ while(pendingMerges.size() > 0 || runningMerges.size() > 0) { doWait(); } - + //if (mergingSegments.size() > 0) { + // System.out.println("mergingSegments > 0:"+mergingSegments); + //} // sanity check assert 0 == mergingSegments.size(); } @@ -3057,10 +3106,10 @@ } } - private boolean hasExternalSegments() { + boolean hasExternalSegments() { return segmentInfos.hasExternalSegments(directory); } - + /* If any of our segments are using a directory != ours * then we have to either copy them over one by one, merge * them (if merge policy has chosen to) or wait until @@ -3068,8 +3117,11 @@ * We don't return until the SegmentInfos has no more * external segments. Currently this is only used by * addIndexesNoOptimize(). */ - private void resolveExternalSegments() throws CorruptIndexException, IOException { - + void resolveExternalSegments() throws CorruptIndexException, IOException { + resolveExternalSegments(false); + } + + void resolveExternalSegments(boolean ramnrt) throws CorruptIndexException, IOException { boolean any = false; boolean done = false; @@ -3135,7 +3187,7 @@ // more merges may become necessary: mergeScheduler.merge(this); } - + /** Merges the provided indexes into this index. *

After this completes, the index is optimized.

*

The provided IndexReaders are not closed.

@@ -3664,10 +3716,9 @@ } private int ensureContiguousMerge(MergePolicy.OneMerge merge) { - int first = segmentInfos.indexOf(merge.segments.info(0)); if (first == -1) - throw new MergePolicy.MergeException("could not find segment " + merge.segments.info(0).name + " in current index " + segString(), directory); + throw new MergePolicy.MergeException("ramnrt:"+merge.ramnrt+" could not find segment " + merge.segments.info(0).name + " in current index " + segString(), directory); final int numSegments = segmentInfos.size(); @@ -3709,7 +3760,6 @@ // started merging: int docUpto = 0; int delCount = 0; - for(int i=0; i < sourceSegments.size(); i++) { SegmentInfo info = sourceSegments.info(i); int docCount = info.docCount; @@ -3816,9 +3866,14 @@ merge.info.setHasProx(merger.hasProx()); - segmentInfos.subList(start, start + merge.segments.size()).clear(); + //segmentInfos.subList(start, start + merge.segments.size()).clear(); + // remove the segmentinfo's one by one + for (int x=0; x < merge.segments.size(); x++) { + segmentInfos.remove(merge.segments.info(x)); + } assert !segmentInfos.contains(merge.info); - segmentInfos.add(start, merge.info); + if (start >=0) segmentInfos.add(start, merge.info); + else segmentInfos.add(merge.info); // Must note the change to segmentInfos so any commits // in-flight don't lose it: @@ -3888,6 +3943,7 @@ if (infoStream != null) message("now merge\n merge=" + merge.segString(directory) + "\n merge=" + merge + "\n index=" + segString()); + System.out.println(((ConcurrentMergeScheduler)mergeScheduler).name+" now merge\n merge=" + merge.segString(directory) + "\n merge=" + merge + "\n index=" + segString()); mergeMiddle(merge); mergeSuccess(merge); @@ -3917,7 +3973,19 @@ handleOOM(oom, "merge"); } } - + + SegmentInfos getSegmentInfos() { + return segmentInfos; + } + + synchronized SegmentInfos getSegmentInfosCopy() { + SegmentInfos copy = new SegmentInfos(); + for (SegmentInfo info : segmentInfos) { + copy.add(info); + } + return copy; + } + /** Hook that's called when the specified merge is complete. */ void mergeSuccess(MergePolicy.OneMerge merge) { } @@ -3944,13 +4012,13 @@ final SegmentInfo info = merge.segments.info(i); if (mergingSegments.contains(info)) return false; - if (segmentInfos.indexOf(info) == -1) + if (segmentInfos.indexOf(info) == -1 && !merge.ramnrt) return false; if (info.dir != directory) isExternal = true; } - ensureContiguousMerge(merge); + if (!merge.ramnrt) ensureContiguousMerge(merge); pendingMerges.add(merge); @@ -4191,7 +4259,9 @@ message("merging " + merge.segString(directory)); merger = new SegmentMerger(this, mergedName, merge); - + + assert merge.readers == null; + assert merge.readersClone == null; merge.readers = new SegmentReader[numSegments]; merge.readersClone = new SegmentReader[numSegments];