--- /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 java.io.IOException;
+import java.io.PrintStream;
+import java.text.NumberFormat;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.lucene.analysis.Analyzer;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.search.Similarity;
+import org.apache.lucene.store.AlreadyClosedException;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.store.RAMFile;
+import org.apache.lucene.util.ArrayUtil;
+import org.apache.lucene.util.BitVector;
+import org.apache.lucene.util.RamUsageEstimator;
+import org.apache.lucene.util.ThreadInterruptedException;
+
+
+/**
+ * This class accepts multiple added documents and directly
+ * writes a single segment file. It does this more
+ * efficiently than creating a single segment per document
+ * (with DocumentWriter) and doing standard merges on those
+ * segments.
+ *
+ * Each added document is passed to the {@link DocConsumer},
+ * which in turn processes the document and interacts with
+ * other consumers in the indexing chain. Certain
+ * consumers, like {@link StoredFieldsWriter} and {@link
+ * TermVectorsTermsWriter}, digest a document and
+ * immediately write bytes to the "doc store" files (ie,
+ * they do not consume RAM per document, except while they
+ * are processing the document).
+ *
+ * Other consumers, eg {@link FreqProxTermsWriter} and
+ * {@link NormsWriter}, buffer bytes in RAM and flush only
+ * when a new segment is produced.
+
+ * Once we have used our allowed RAM buffer, or the number
+ * of added docs is large enough (in the case we are
+ * flushing by doc count instead of RAM usage), we create a
+ * real segment and flush it to the Directory.
+ *
+ * Threads:
+ *
+ * Multiple threads are allowed into addDocument at once.
+ * There is an initial synchronized call to getThreadState
+ * which allocates a ThreadState for this thread. The same
+ * thread will get the same ThreadState over time (thread
+ * affinity) so that if there are consistent patterns (for
+ * example each thread is indexing a different content
+ * source) then we make better use of RAM. Then
+ * processDocument is called on that ThreadState without
+ * synchronization (most of the "heavy lifting" is in this
+ * call). Finally the synchronized "finishDocument" is
+ * called to flush changes to the directory.
+ *
+ * When flush is called by IndexWriter we forcefully idle
+ * all threads and flush only once they are all idle. This
+ * means you can call flush with a given thread even while
+ * other threads are actively adding/deleting documents.
+ *
+ *
+ * Exceptions:
+ *
+ * Because this class directly updates in-memory posting
+ * lists, and flushes stored fields and term vectors
+ * directly to files in the directory, there are certain
+ * limited times when an exception can corrupt this state.
+ * For example, a disk full while flushing stored fields
+ * leaves this file in a corrupt state. Or, an OOM
+ * exception while appending to the in-memory posting lists
+ * can corrupt that posting list. We call such exceptions
+ * "aborting exceptions". In these cases we must call
+ * abort() to discard all docs added since the last flush.
+ *
+ * All other exceptions ("non-aborting exceptions") can
+ * still partially update the index structures. These
+ * updates are consistent, but, they represent only a part
+ * of the document seen up until the exception was hit.
+ * When this happens, we immediately mark the document as
+ * deleted so that the document is always atomically ("all
+ * or none") added to the index.
+ */
+
+final class DocumentsWriter {
+ final AtomicLong bytesUsed = new AtomicLong(0);
+ IndexWriter writer;
+ Directory directory;
+
+ String segment; // Current segment we are working on
+
+ private int nextDocID; // Next docID to be added
+ private int numDocs; // # of docs added, but not yet flushed
+
+ // Max # ThreadState instances; if there are more threads
+ // than this they share ThreadStates
+ private DocumentsWriterThreadState[] threadStates = new DocumentsWriterThreadState[0];
+ private final HashMap<Thread,DocumentsWriterThreadState> threadBindings = new HashMap<Thread,DocumentsWriterThreadState>();
+
+ boolean bufferIsFull; // True when it's time to write segment
+ private boolean aborting; // True if an abort is pending
+
+ PrintStream infoStream;
+ int maxFieldLength = IndexWriter.DEFAULT_MAX_FIELD_LENGTH;
+ Similarity similarity;
+
+ // max # simultaneous threads; if there are more than
+ // this, they wait for others to finish first
+ private final int maxThreadStates;
+
+ // Deletes for our still-in-RAM (to be flushed next) segment
+ private BufferedDeletes pendingDeletes = new BufferedDeletes();
+
+ static class DocState {
+ DocumentsWriter docWriter;
+ Analyzer analyzer;
+ int maxFieldLength;
+ PrintStream infoStream;
+ Similarity similarity;
+ int docID;
+ Document doc;
+ String maxTermPrefix;
+
+ // Only called by asserts
+ public boolean testPoint(String name) {
+ return docWriter.writer.testPoint(name);
+ }
+
+ public void clear() {
+ // don't hold onto doc nor analyzer, in case it is
+ // largish:
+ doc = null;
+ analyzer = null;
+ }
+ }
+
+ /** Consumer returns this on each doc. This holds any
+ * state that must be flushed synchronized "in docID
+ * order". We gather these and flush them in order. */
+ abstract static class DocWriter {
+ DocWriter next;
+ int docID;
+ abstract void finish() throws IOException;
+ abstract void abort();
+ abstract long sizeInBytes();
+
+ void setNext(DocWriter next) {
+ this.next = next;
+ }
+ }
+
+ /**
+ * Create and return a new DocWriterBuffer.
+ */
+ PerDocBuffer newPerDocBuffer() {
+ return new PerDocBuffer();
+ }
+
+ /**
+ * RAMFile buffer for DocWriters.
+ */
+ class PerDocBuffer extends RAMFile {
+
+ /**
+ * Allocate bytes used from shared pool.
+ */
+ @Override
+ protected byte[] newBuffer(int size) {
+ assert size == PER_DOC_BLOCK_SIZE;
+ return perDocAllocator.getByteBlock();
+ }
+
+ /**
+ * Recycle the bytes used.
+ */
+ synchronized void recycle() {
+ if (buffers.size() > 0) {
+ setLength(0);
+
+ // Recycle the blocks
+ perDocAllocator.recycleByteBlocks(buffers);
+ buffers.clear();
+ sizeInBytes = 0;
+
+ assert numBuffers() == 0;
+ }
+ }
+ }
+
+ /**
+ * The IndexingChain must define the {@link #getChain(DocumentsWriter)} method
+ * which returns the DocConsumer that the DocumentsWriter calls to process the
+ * documents.
+ */
+ abstract static class IndexingChain {
+ abstract DocConsumer getChain(DocumentsWriter documentsWriter);
+ }
+
+ static final IndexingChain defaultIndexingChain = new IndexingChain() {
+
+ @Override
+ DocConsumer getChain(DocumentsWriter documentsWriter) {
+ /*
+ This is the current indexing chain:
+
+ DocConsumer / DocConsumerPerThread
+ --> code: DocFieldProcessor / DocFieldProcessorPerThread
+ --> DocFieldConsumer / DocFieldConsumerPerThread / DocFieldConsumerPerField
+ --> code: DocFieldConsumers / DocFieldConsumersPerThread / DocFieldConsumersPerField
+ --> code: DocInverter / DocInverterPerThread / DocInverterPerField
+ --> InvertedDocConsumer / InvertedDocConsumerPerThread / InvertedDocConsumerPerField
+ --> code: TermsHash / TermsHashPerThread / TermsHashPerField
+ --> TermsHashConsumer / TermsHashConsumerPerThread / TermsHashConsumerPerField
+ --> code: FreqProxTermsWriter / FreqProxTermsWriterPerThread / FreqProxTermsWriterPerField
+ --> code: TermVectorsTermsWriter / TermVectorsTermsWriterPerThread / TermVectorsTermsWriterPerField
+ --> InvertedDocEndConsumer / InvertedDocConsumerPerThread / InvertedDocConsumerPerField
+ --> code: NormsWriter / NormsWriterPerThread / NormsWriterPerField
+ --> code: StoredFieldsWriter / StoredFieldsWriterPerThread / StoredFieldsWriterPerField
+ */
+
+ // Build up indexing chain:
+
+ final TermsHashConsumer termVectorsWriter = new TermVectorsTermsWriter(documentsWriter);
+ final TermsHashConsumer freqProxWriter = new FreqProxTermsWriter();
+
+ final InvertedDocConsumer termsHash = new TermsHash(documentsWriter, true, freqProxWriter,
+ new TermsHash(documentsWriter, false, termVectorsWriter, null));
+ final NormsWriter normsWriter = new NormsWriter();
+ final DocInverter docInverter = new DocInverter(termsHash, normsWriter);
+ return new DocFieldProcessor(documentsWriter, docInverter);
+ }
+ };
+
+ final DocConsumer consumer;
+
+ // How much RAM we can use before flushing. This is 0 if
+ // we are flushing by doc count instead.
+
+ private final IndexWriterConfig config;
+
+ private boolean closed;
+ private final FieldInfos fieldInfos;
+
+ private final BufferedDeletesStream bufferedDeletesStream;
+ private final IndexWriter.FlushControl flushControl;
+
+ DocumentsWriter(IndexWriterConfig config, Directory directory, IndexWriter writer, FieldInfos fieldInfos, BufferedDeletesStream bufferedDeletesStream) throws IOException {
+ this.directory = directory;
+ this.writer = writer;
+ this.similarity = config.getSimilarity();
+ this.maxThreadStates = config.getMaxThreadStates();
+ this.fieldInfos = fieldInfos;
+ this.bufferedDeletesStream = bufferedDeletesStream;
+ flushControl = writer.flushControl;
+
+ consumer = config.getIndexingChain().getChain(this);
+ this.config = config;
+ }
+
+ // Buffer a specific docID for deletion. Currently only
+ // used when we hit a exception when adding a document
+ synchronized void deleteDocID(int docIDUpto) {
+ pendingDeletes.addDocID(docIDUpto);
+ // NOTE: we do not trigger flush here. This is
+ // potentially a RAM leak, if you have an app that tries
+ // to add docs but every single doc always hits a
+ // non-aborting exception. Allowing a flush here gets
+ // very messy because we are only invoked when handling
+ // exceptions so to do this properly, while handling an
+ // exception we'd have to go off and flush new deletes
+ // which is risky (likely would hit some other
+ // confounding exception).
+ }
+
+ boolean deleteQueries(Query... queries) {
+ final boolean doFlush = flushControl.waitUpdate(0, queries.length);
+ synchronized(this) {
+ for (Query query : queries) {
+ pendingDeletes.addQuery(query, numDocs);
+ }
+ }
+ return doFlush;
+ }
+
+ boolean deleteQuery(Query query) {
+ final boolean doFlush = flushControl.waitUpdate(0, 1);
+ synchronized(this) {
+ pendingDeletes.addQuery(query, numDocs);
+ }
+ return doFlush;
+ }
+
+ boolean deleteTerms(Term... terms) {
+ final boolean doFlush = flushControl.waitUpdate(0, terms.length);
+ synchronized(this) {
+ for (Term term : terms) {
+ pendingDeletes.addTerm(term, numDocs);
+ }
+ }
+ return doFlush;
+ }
+
+ // TODO: we could check w/ FreqProxTermsWriter: if the
+ // term doesn't exist, don't bother buffering into the
+ // per-DWPT map (but still must go into the global map)
+ boolean deleteTerm(Term term, boolean skipWait) {
+ final boolean doFlush = flushControl.waitUpdate(0, 1, skipWait);
+ synchronized(this) {
+ pendingDeletes.addTerm(term, numDocs);
+ }
+ return doFlush;
+ }
+
+ public FieldInfos getFieldInfos() {
+ return fieldInfos;
+ }
+
+ /** If non-null, various details of indexing are printed
+ * here. */
+ synchronized void setInfoStream(PrintStream infoStream) {
+ this.infoStream = infoStream;
+ for(int i=0;i<threadStates.length;i++) {
+ threadStates[i].docState.infoStream = infoStream;
+ }
+ }
+
+ synchronized void setMaxFieldLength(int maxFieldLength) {
+ this.maxFieldLength = maxFieldLength;
+ for(int i=0;i<threadStates.length;i++) {
+ threadStates[i].docState.maxFieldLength = maxFieldLength;
+ }
+ }
+
+ synchronized void setSimilarity(Similarity similarity) {
+ this.similarity = similarity;
+ for(int i=0;i<threadStates.length;i++) {
+ threadStates[i].docState.similarity = similarity;
+ }
+ }
+
+ /** Get current segment name we are writing. */
+ synchronized String getSegment() {
+ return segment;
+ }
+
+ /** Returns how many docs are currently buffered in RAM. */
+ synchronized int getNumDocs() {
+ return numDocs;
+ }
+
+ void message(String message) {
+ if (infoStream != null) {
+ writer.message("DW: " + message);
+ }
+ }
+
+ synchronized void setAborting() {
+ if (infoStream != null) {
+ message("setAborting");
+ }
+ aborting = true;
+ }
+
+ /** Called if we hit an exception at a bad time (when
+ * updating the index files) and must discard all
+ * currently buffered docs. This resets our state,
+ * discarding any docs added since last flush. */
+ synchronized void abort() throws IOException {
+ if (infoStream != null) {
+ message("docWriter: abort");
+ }
+
+ boolean success = false;
+
+ try {
+
+ // Forcefully remove waiting ThreadStates from line
+ try {
+ waitQueue.abort();
+ } catch (Throwable t) {
+ }
+
+ // Wait for all other threads to finish with
+ // DocumentsWriter:
+ try {
+ waitIdle();
+ } finally {
+ if (infoStream != null) {
+ message("docWriter: abort waitIdle done");
+ }
+
+ assert 0 == waitQueue.numWaiting: "waitQueue.numWaiting=" + waitQueue.numWaiting;
+ waitQueue.waitingBytes = 0;
+
+ pendingDeletes.clear();
+
+ for (DocumentsWriterThreadState threadState : threadStates) {
+ try {
+ threadState.consumer.abort();
+ } catch (Throwable t) {
+ }
+ }
+
+ try {
+ consumer.abort();
+ } catch (Throwable t) {
+ }
+
+ // Reset all postings data
+ doAfterFlush();
+ }
+
+ success = true;
+ } finally {
+ aborting = false;
+ notifyAll();
+ if (infoStream != null) {
+ message("docWriter: done abort; success=" + success);
+ }
+ }
+ }
+
+ /** Reset after a flush */
+ private void doAfterFlush() throws IOException {
+ // All ThreadStates should be idle when we are called
+ assert allThreadsIdle();
+ threadBindings.clear();
+ waitQueue.reset();
+ segment = null;
+ numDocs = 0;
+ nextDocID = 0;
+ bufferIsFull = false;
+ for(int i=0;i<threadStates.length;i++) {
+ threadStates[i].doAfterFlush();
+ }
+ }
+
+ private synchronized boolean allThreadsIdle() {
+ for(int i=0;i<threadStates.length;i++) {
+ if (!threadStates[i].isIdle) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ synchronized boolean anyChanges() {
+ return numDocs != 0 || pendingDeletes.any();
+ }
+
+ // for testing
+ public BufferedDeletes getPendingDeletes() {
+ return pendingDeletes;
+ }
+
+ private void pushDeletes(SegmentInfo newSegment, SegmentInfos segmentInfos) {
+ // Lock order: DW -> BD
+ final long delGen = bufferedDeletesStream.getNextGen();
+ if (pendingDeletes.any()) {
+ if (segmentInfos.size() > 0 || newSegment != null) {
+ final FrozenBufferedDeletes packet = new FrozenBufferedDeletes(pendingDeletes, delGen);
+ if (infoStream != null) {
+ message("flush: push buffered deletes startSize=" + pendingDeletes.bytesUsed.get() + " frozenSize=" + packet.bytesUsed);
+ }
+ bufferedDeletesStream.push(packet);
+ if (infoStream != null) {
+ message("flush: delGen=" + packet.gen);
+ }
+ if (newSegment != null) {
+ newSegment.setBufferedDeletesGen(packet.gen);
+ }
+ } else {
+ if (infoStream != null) {
+ message("flush: drop buffered deletes: no segments");
+ }
+ // We can safely discard these deletes: since
+ // there are no segments, the deletions cannot
+ // affect anything.
+ }
+ pendingDeletes.clear();
+ } else if (newSegment != null) {
+ newSegment.setBufferedDeletesGen(delGen);
+ }
+ }
+
+ public boolean anyDeletions() {
+ return pendingDeletes.any();
+ }
+
+ /** Flush all pending docs to a new segment */
+ // Lock order: IW -> DW
+ synchronized SegmentInfo flush(IndexWriter writer, IndexFileDeleter deleter, MergePolicy mergePolicy, SegmentInfos segmentInfos) throws IOException {
+
+ final long startTime = System.currentTimeMillis();
+
+ // We change writer's segmentInfos:
+ assert Thread.holdsLock(writer);
+
+ waitIdle();
+
+ if (numDocs == 0) {
+ // nothing to do!
+ if (infoStream != null) {
+ message("flush: no docs; skipping");
+ }
+ // Lock order: IW -> DW -> BD
+ pushDeletes(null, segmentInfos);
+ return null;
+ }
+
+ if (aborting) {
+ if (infoStream != null) {
+ message("flush: skip because aborting is set");
+ }
+ return null;
+ }
+
+ boolean success = false;
+
+ SegmentInfo newSegment;
+
+ try {
+ //System.out.println(Thread.currentThread().getName() + ": nw=" + waitQueue.numWaiting);
+ assert nextDocID == numDocs: "nextDocID=" + nextDocID + " numDocs=" + numDocs;
+ assert waitQueue.numWaiting == 0: "numWaiting=" + waitQueue.numWaiting;
+ assert waitQueue.waitingBytes == 0;
+
+ if (infoStream != null) {
+ message("flush postings as segment " + segment + " numDocs=" + numDocs);
+ }
+
+ final SegmentWriteState flushState = new SegmentWriteState(infoStream, directory, segment, fieldInfos,
+ numDocs, writer.getConfig().getTermIndexInterval(),
+ pendingDeletes);
+ // Apply delete-by-docID now (delete-byDocID only
+ // happens when an exception is hit processing that
+ // doc, eg if analyzer has some problem w/ the text):
+ if (pendingDeletes.docIDs.size() > 0) {
+ flushState.deletedDocs = new BitVector(numDocs);
+ for(int delDocID : pendingDeletes.docIDs) {
+ flushState.deletedDocs.set(delDocID);
+ }
+ pendingDeletes.bytesUsed.addAndGet(-pendingDeletes.docIDs.size() * BufferedDeletes.BYTES_PER_DEL_DOCID);
+ pendingDeletes.docIDs.clear();
+ }
+
+ newSegment = new SegmentInfo(segment, numDocs, directory, false, true, fieldInfos.hasProx(), false);
+
+ Collection<DocConsumerPerThread> threads = new HashSet<DocConsumerPerThread>();
+ for (DocumentsWriterThreadState threadState : threadStates) {
+ threads.add(threadState.consumer);
+ }
+
+ double startMBUsed = bytesUsed()/1024./1024.;
+
+ consumer.flush(threads, flushState);
+
+ newSegment.setHasVectors(flushState.hasVectors);
+
+ if (infoStream != null) {
+ message("new segment has " + (flushState.hasVectors ? "vectors" : "no vectors"));
+ if (flushState.deletedDocs != null) {
+ message("new segment has " + flushState.deletedDocs.count() + " deleted docs");
+ }
+ message("flushedFiles=" + newSegment.files());
+ }
+
+ if (mergePolicy.useCompoundFile(segmentInfos, newSegment)) {
+ final String cfsFileName = IndexFileNames.segmentFileName(segment, IndexFileNames.COMPOUND_FILE_EXTENSION);
+
+ if (infoStream != null) {
+ message("flush: create compound file \"" + cfsFileName + "\"");
+ }
+
+ CompoundFileWriter cfsWriter = new CompoundFileWriter(directory, cfsFileName);
+ for(String fileName : newSegment.files()) {
+ cfsWriter.addFile(fileName);
+ }
+ cfsWriter.close();
+ deleter.deleteNewFiles(newSegment.files());
+ newSegment.setUseCompoundFile(true);
+ }
+
+ // Must write deleted docs after the CFS so we don't
+ // slurp the del file into CFS:
+ if (flushState.deletedDocs != null) {
+ final int delCount = flushState.deletedDocs.count();
+ assert delCount > 0;
+ newSegment.setDelCount(delCount);
+ newSegment.advanceDelGen();
+ final String delFileName = newSegment.getDelFileName();
+ if (infoStream != null) {
+ message("flush: write " + delCount + " deletes to " + delFileName);
+ }
+ boolean success2 = false;
+ try {
+ // TODO: in the NRT case it'd be better to hand
+ // this del vector over to the
+ // shortly-to-be-opened SegmentReader and let it
+ // carry the changes; there's no reason to use
+ // filesystem as intermediary here.
+ flushState.deletedDocs.write(directory, delFileName);
+ success2 = true;
+ } finally {
+ if (!success2) {
+ try {
+ directory.deleteFile(delFileName);
+ } catch (Throwable t) {
+ // suppress this so we keep throwing the
+ // original exception
+ }
+ }
+ }
+ }
+
+ if (infoStream != null) {
+ message("flush: segment=" + newSegment);
+ final double newSegmentSizeNoStore = newSegment.sizeInBytes(false)/1024./1024.;
+ final double newSegmentSize = newSegment.sizeInBytes(true)/1024./1024.;
+ message(" ramUsed=" + nf.format(startMBUsed) + " MB" +
+ " newFlushedSize=" + nf.format(newSegmentSize) + " MB" +
+ " (" + nf.format(newSegmentSizeNoStore) + " MB w/o doc stores)" +
+ " docs/MB=" + nf.format(numDocs / newSegmentSize) +
+ " new/old=" + nf.format(100.0 * newSegmentSizeNoStore / startMBUsed) + "%");
+ }
+
+ success = true;
+ } finally {
+ notifyAll();
+ if (!success) {
+ if (segment != null) {
+ deleter.refresh(segment);
+ }
+ abort();
+ }
+ }
+
+ doAfterFlush();
+
+ // Lock order: IW -> DW -> BD
+ pushDeletes(newSegment, segmentInfos);
+ if (infoStream != null) {
+ message("flush time " + (System.currentTimeMillis()-startTime) + " msec");
+ }
+
+ return newSegment;
+ }
+
+ synchronized void close() {
+ closed = true;
+ notifyAll();
+ }
+
+ /** Returns a free (idle) ThreadState that may be used for
+ * indexing this one document. This call also pauses if a
+ * flush is pending. If delTerm is non-null then we
+ * buffer this deleted term after the thread state has
+ * been acquired. */
+ synchronized DocumentsWriterThreadState getThreadState(Term delTerm, int docCount) throws IOException {
+
+ final Thread currentThread = Thread.currentThread();
+ assert !Thread.holdsLock(writer);
+
+ // First, find a thread state. If this thread already
+ // has affinity to a specific ThreadState, use that one
+ // again.
+ DocumentsWriterThreadState state = threadBindings.get(currentThread);
+ if (state == null) {
+
+ // First time this thread has called us since last
+ // flush. Find the least loaded thread state:
+ DocumentsWriterThreadState minThreadState = null;
+ for(int i=0;i<threadStates.length;i++) {
+ DocumentsWriterThreadState ts = threadStates[i];
+ if (minThreadState == null || ts.numThreads < minThreadState.numThreads) {
+ minThreadState = ts;
+ }
+ }
+ if (minThreadState != null && (minThreadState.numThreads == 0 || threadStates.length >= maxThreadStates)) {
+ state = minThreadState;
+ state.numThreads++;
+ } else {
+ // Just create a new "private" thread state
+ DocumentsWriterThreadState[] newArray = new DocumentsWriterThreadState[1+threadStates.length];
+ if (threadStates.length > 0) {
+ System.arraycopy(threadStates, 0, newArray, 0, threadStates.length);
+ }
+ state = newArray[threadStates.length] = new DocumentsWriterThreadState(this);
+ threadStates = newArray;
+ }
+ threadBindings.put(currentThread, state);
+ }
+
+ // Next, wait until my thread state is idle (in case
+ // it's shared with other threads), and no flush/abort
+ // pending
+ waitReady(state);
+
+ // Allocate segment name if this is the first doc since
+ // last flush:
+ if (segment == null) {
+ segment = writer.newSegmentName();
+ assert numDocs == 0;
+ }
+
+ state.docState.docID = nextDocID;
+ nextDocID += docCount;
+
+ if (delTerm != null) {
+ pendingDeletes.addTerm(delTerm, state.docState.docID);
+ }
+
+ numDocs += docCount;
+ state.isIdle = false;
+ return state;
+ }
+
+ boolean addDocument(Document doc, Analyzer analyzer) throws CorruptIndexException, IOException {
+ return updateDocument(doc, analyzer, null);
+ }
+
+ boolean updateDocument(Document doc, Analyzer analyzer, Term delTerm)
+ throws CorruptIndexException, IOException {
+
+ // Possibly trigger a flush, or wait until any running flush completes:
+ boolean doFlush = flushControl.waitUpdate(1, delTerm != null ? 1 : 0);
+
+ // This call is synchronized but fast
+ final DocumentsWriterThreadState state = getThreadState(delTerm, 1);
+
+ final DocState docState = state.docState;
+ docState.doc = doc;
+ docState.analyzer = analyzer;
+
+ boolean success = false;
+ try {
+ // This call is not synchronized and does all the
+ // work
+ final DocWriter perDoc;
+ try {
+ perDoc = state.consumer.processDocument();
+ } finally {
+ docState.clear();
+ }
+
+ // This call is synchronized but fast
+ finishDocument(state, perDoc);
+
+ success = true;
+ } finally {
+ if (!success) {
+
+ // If this thread state had decided to flush, we
+ // must clear it so another thread can flush
+ if (doFlush) {
+ flushControl.clearFlushPending();
+ }
+
+ if (infoStream != null) {
+ message("exception in updateDocument aborting=" + aborting);
+ }
+
+ synchronized(this) {
+
+ state.isIdle = true;
+ notifyAll();
+
+ if (aborting) {
+ abort();
+ } else {
+ skipDocWriter.docID = docState.docID;
+ boolean success2 = false;
+ try {
+ waitQueue.add(skipDocWriter);
+ success2 = true;
+ } finally {
+ if (!success2) {
+ abort();
+ return false;
+ }
+ }
+
+ // Immediately mark this document as deleted
+ // since likely it was partially added. This
+ // keeps indexing as "all or none" (atomic) when
+ // adding a document:
+ deleteDocID(state.docState.docID);
+ }
+ }
+ }
+ }
+
+ doFlush |= flushControl.flushByRAMUsage("new document");
+
+ return doFlush;
+ }
+
+ boolean updateDocuments(Collection<Document> docs, Analyzer analyzer, Term delTerm)
+ throws CorruptIndexException, IOException {
+
+ // Possibly trigger a flush, or wait until any running flush completes:
+ boolean doFlush = flushControl.waitUpdate(docs.size(), delTerm != null ? 1 : 0);
+
+ final int docCount = docs.size();
+
+ // This call is synchronized but fast -- we allocate the
+ // N docIDs up front:
+ final DocumentsWriterThreadState state = getThreadState(null, docCount);
+ final DocState docState = state.docState;
+
+ final int startDocID = docState.docID;
+ int docID = startDocID;
+
+ //System.out.println(Thread.currentThread().getName() + ": A " + docCount);
+ for(Document doc : docs) {
+ docState.doc = doc;
+ docState.analyzer = analyzer;
+ // Assign next docID from our block:
+ docState.docID = docID++;
+
+ boolean success = false;
+ try {
+ // This call is not synchronized and does all the
+ // work
+ final DocWriter perDoc;
+ try {
+ perDoc = state.consumer.processDocument();
+ } finally {
+ docState.clear();
+ }
+
+ // Must call this w/o holding synchronized(this) else
+ // we'll hit deadlock:
+ balanceRAM();
+
+ // Synchronized but fast
+ synchronized(this) {
+ if (aborting) {
+ break;
+ }
+ assert perDoc == null || perDoc.docID == docState.docID;
+ final boolean doPause;
+ if (perDoc != null) {
+ waitQueue.add(perDoc);
+ } else {
+ skipDocWriter.docID = docState.docID;
+ waitQueue.add(skipDocWriter);
+ }
+ }
+
+ success = true;
+ } finally {
+ if (!success) {
+ //System.out.println(Thread.currentThread().getName() + ": E");
+
+ // If this thread state had decided to flush, we
+ // must clear it so another thread can flush
+ if (doFlush) {
+ message("clearFlushPending!");
+ flushControl.clearFlushPending();
+ }
+
+ if (infoStream != null) {
+ message("exception in updateDocuments aborting=" + aborting);
+ }
+
+ synchronized(this) {
+
+ state.isIdle = true;
+ notifyAll();
+
+ if (aborting) {
+ abort();
+ } else {
+
+ // Fill hole in the doc stores for all
+ // docIDs we pre-allocated
+ //System.out.println(Thread.currentThread().getName() + ": F " + docCount);
+ final int endDocID = startDocID + docCount;
+ docID = docState.docID;
+ while(docID < endDocID) {
+ skipDocWriter.docID = docID++;
+ boolean success2 = false;
+ try {
+ waitQueue.add(skipDocWriter);
+ success2 = true;
+ } finally {
+ if (!success2) {
+ abort();
+ return false;
+ }
+ }
+ }
+ //System.out.println(Thread.currentThread().getName() + ": F " + docCount + " done");
+
+ // Mark all pre-allocated docIDs as deleted:
+ docID = startDocID;
+ while(docID < startDocID + docs.size()) {
+ deleteDocID(docID++);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ synchronized(this) {
+ // We must delay pausing until the full doc block is
+ // added, else we can hit deadlock if more than one
+ // thread is adding a block and we need to pause when
+ // both are only part way done:
+ if (waitQueue.doPause()) {
+ waitForWaitQueue();
+ }
+
+ //System.out.println(Thread.currentThread().getName() + ": A " + docCount);
+
+ if (aborting) {
+
+ // We are currently aborting, and another thread is
+ // waiting for me to become idle. We just forcefully
+ // idle this threadState; it will be fully reset by
+ // abort()
+ state.isIdle = true;
+
+ // wakes up any threads waiting on the wait queue
+ notifyAll();
+
+ abort();
+
+ if (doFlush) {
+ message("clearFlushPending!");
+ flushControl.clearFlushPending();
+ }
+
+ return false;
+ }
+
+ // Apply delTerm only after all indexing has
+ // succeeded, but apply it only to docs prior to when
+ // this batch started:
+ if (delTerm != null) {
+ pendingDeletes.addTerm(delTerm, startDocID);
+ }
+
+ state.isIdle = true;
+
+ // wakes up any threads waiting on the wait queue
+ notifyAll();
+ }
+
+ doFlush |= flushControl.flushByRAMUsage("new document");
+
+ //System.out.println(Thread.currentThread().getName() + ": B " + docCount);
+ return doFlush;
+ }
+
+ public synchronized void waitIdle() {
+ while (!allThreadsIdle()) {
+ try {
+ wait();
+ } catch (InterruptedException ie) {
+ throw new ThreadInterruptedException(ie);
+ }
+ }
+ }
+
+ synchronized void waitReady(DocumentsWriterThreadState state) {
+ while (!closed && (!state.isIdle || aborting)) {
+ try {
+ wait();
+ } catch (InterruptedException ie) {
+ throw new ThreadInterruptedException(ie);
+ }
+ }
+
+ if (closed) {
+ throw new AlreadyClosedException("this IndexWriter is closed");
+ }
+ }
+
+ /** Does the synchronized work to finish/flush the
+ * inverted document. */
+ private void finishDocument(DocumentsWriterThreadState perThread, DocWriter docWriter) throws IOException {
+
+ // Must call this w/o holding synchronized(this) else
+ // we'll hit deadlock:
+ balanceRAM();
+
+ synchronized(this) {
+
+ assert docWriter == null || docWriter.docID == perThread.docState.docID;
+
+ if (aborting) {
+
+ // We are currently aborting, and another thread is
+ // waiting for me to become idle. We just forcefully
+ // idle this threadState; it will be fully reset by
+ // abort()
+ if (docWriter != null) {
+ try {
+ docWriter.abort();
+ } catch (Throwable t) {
+ }
+ }
+
+ perThread.isIdle = true;
+
+ // wakes up any threads waiting on the wait queue
+ notifyAll();
+
+ return;
+ }
+
+ final boolean doPause;
+
+ if (docWriter != null) {
+ doPause = waitQueue.add(docWriter);
+ } else {
+ skipDocWriter.docID = perThread.docState.docID;
+ doPause = waitQueue.add(skipDocWriter);
+ }
+
+ if (doPause) {
+ waitForWaitQueue();
+ }
+
+ perThread.isIdle = true;
+
+ // wakes up any threads waiting on the wait queue
+ notifyAll();
+ }
+ }
+
+ synchronized void waitForWaitQueue() {
+ do {
+ try {
+ wait();
+ } catch (InterruptedException ie) {
+ throw new ThreadInterruptedException(ie);
+ }
+ } while (!waitQueue.doResume());
+ }
+
+ private static class SkipDocWriter extends DocWriter {
+ @Override
+ void finish() {
+ }
+ @Override
+ void abort() {
+ }
+ @Override
+ long sizeInBytes() {
+ return 0;
+ }
+ }
+ final SkipDocWriter skipDocWriter = new SkipDocWriter();
+
+ NumberFormat nf = NumberFormat.getInstance();
+
+ /* Initial chunks size of the shared byte[] blocks used to
+ store postings data */
+ final static int BYTE_BLOCK_SHIFT = 15;
+ final static int BYTE_BLOCK_SIZE = 1 << BYTE_BLOCK_SHIFT;
+ final static int BYTE_BLOCK_MASK = BYTE_BLOCK_SIZE - 1;
+ final static int BYTE_BLOCK_NOT_MASK = ~BYTE_BLOCK_MASK;
+
+ private class ByteBlockAllocator extends ByteBlockPool.Allocator {
+ final int blockSize;
+
+ ByteBlockAllocator(int blockSize) {
+ this.blockSize = blockSize;
+ }
+
+ ArrayList<byte[]> freeByteBlocks = new ArrayList<byte[]>();
+
+ /* Allocate another byte[] from the shared pool */
+ @Override
+ byte[] getByteBlock() {
+ synchronized(DocumentsWriter.this) {
+ final int size = freeByteBlocks.size();
+ final byte[] b;
+ if (0 == size) {
+ b = new byte[blockSize];
+ bytesUsed.addAndGet(blockSize);
+ } else
+ b = freeByteBlocks.remove(size-1);
+ return b;
+ }
+ }
+
+ /* Return byte[]'s to the pool */
+
+ @Override
+ void recycleByteBlocks(byte[][] blocks, int start, int end) {
+ synchronized(DocumentsWriter.this) {
+ for(int i=start;i<end;i++) {
+ freeByteBlocks.add(blocks[i]);
+ blocks[i] = null;
+ }
+ }
+ }
+
+ @Override
+ void recycleByteBlocks(List<byte[]> blocks) {
+ synchronized(DocumentsWriter.this) {
+ final int size = blocks.size();
+ for(int i=0;i<size;i++) {
+ freeByteBlocks.add(blocks.get(i));
+ blocks.set(i, null);
+ }
+ }
+ }
+ }
+
+ /* Initial chunks size of the shared int[] blocks used to
+ store postings data */
+ final static int INT_BLOCK_SHIFT = 13;
+ final static int INT_BLOCK_SIZE = 1 << INT_BLOCK_SHIFT;
+ final static int INT_BLOCK_MASK = INT_BLOCK_SIZE - 1;
+
+ private List<int[]> freeIntBlocks = new ArrayList<int[]>();
+
+ /* Allocate another int[] from the shared pool */
+ synchronized int[] getIntBlock() {
+ final int size = freeIntBlocks.size();
+ final int[] b;
+ if (0 == size) {
+ b = new int[INT_BLOCK_SIZE];
+ bytesUsed.addAndGet(INT_BLOCK_SIZE*RamUsageEstimator.NUM_BYTES_INT);
+ } else {
+ b = freeIntBlocks.remove(size-1);
+ }
+ return b;
+ }
+
+ synchronized void bytesUsed(long numBytes) {
+ bytesUsed.addAndGet(numBytes);
+ }
+
+ long bytesUsed() {
+ return bytesUsed.get() + pendingDeletes.bytesUsed.get();
+ }
+
+ /* Return int[]s to the pool */
+ synchronized void recycleIntBlocks(int[][] blocks, int start, int end) {
+ for(int i=start;i<end;i++) {
+ freeIntBlocks.add(blocks[i]);
+ blocks[i] = null;
+ }
+ }
+
+ ByteBlockAllocator byteBlockAllocator = new ByteBlockAllocator(BYTE_BLOCK_SIZE);
+
+ final static int PER_DOC_BLOCK_SIZE = 1024;
+
+ final ByteBlockAllocator perDocAllocator = new ByteBlockAllocator(PER_DOC_BLOCK_SIZE);
+
+
+ /* Initial chunk size of the shared char[] blocks used to
+ store term text */
+ final static int CHAR_BLOCK_SHIFT = 14;
+ final static int CHAR_BLOCK_SIZE = 1 << CHAR_BLOCK_SHIFT;
+ final static int CHAR_BLOCK_MASK = CHAR_BLOCK_SIZE - 1;
+
+ final static int MAX_TERM_LENGTH = CHAR_BLOCK_SIZE-1;
+
+ private ArrayList<char[]> freeCharBlocks = new ArrayList<char[]>();
+
+ /* Allocate another char[] from the shared pool */
+ synchronized char[] getCharBlock() {
+ final int size = freeCharBlocks.size();
+ final char[] c;
+ if (0 == size) {
+ bytesUsed.addAndGet(CHAR_BLOCK_SIZE * RamUsageEstimator.NUM_BYTES_CHAR);
+ c = new char[CHAR_BLOCK_SIZE];
+ } else
+ c = freeCharBlocks.remove(size-1);
+ // We always track allocations of char blocks, for now,
+ // because nothing that skips allocation tracking
+ // (currently only term vectors) uses its own char
+ // blocks.
+ return c;
+ }
+
+ /* Return char[]s to the pool */
+ synchronized void recycleCharBlocks(char[][] blocks, int numBlocks) {
+ for(int i=0;i<numBlocks;i++) {
+ freeCharBlocks.add(blocks[i]);
+ blocks[i] = null;
+ }
+ }
+
+ String toMB(long v) {
+ return nf.format(v/1024./1024.);
+ }
+
+ /* We have four pools of RAM: Postings, byte blocks
+ * (holds freq/prox posting data), char blocks (holds
+ * characters in the term) and per-doc buffers (stored fields/term vectors).
+ * Different docs require varying amount of storage from
+ * these four classes.
+ *
+ * For example, docs with many unique single-occurrence
+ * short terms will use up the Postings RAM and hardly any
+ * of the other two. Whereas docs with very large terms
+ * will use alot of char blocks RAM and relatively less of
+ * the other two. This method just frees allocations from
+ * the pools once we are over-budget, which balances the
+ * pools to match the current docs. */
+ void balanceRAM() {
+
+ final boolean doBalance;
+ final long deletesRAMUsed;
+
+ deletesRAMUsed = bufferedDeletesStream.bytesUsed();
+
+ final long ramBufferSize;
+ final double mb = config.getRAMBufferSizeMB();
+ if (mb == IndexWriterConfig.DISABLE_AUTO_FLUSH) {
+ ramBufferSize = IndexWriterConfig.DISABLE_AUTO_FLUSH;
+ } else {
+ ramBufferSize = (long) (mb*1024*1024);
+ }
+
+ synchronized(this) {
+ if (ramBufferSize == IndexWriterConfig.DISABLE_AUTO_FLUSH || bufferIsFull) {
+ return;
+ }
+
+ doBalance = bytesUsed() + deletesRAMUsed >= ramBufferSize;
+ }
+
+ if (doBalance) {
+
+ if (infoStream != null) {
+ message(" RAM: balance allocations: usedMB=" + toMB(bytesUsed()) +
+ " vs trigger=" + toMB(ramBufferSize) +
+ " deletesMB=" + toMB(deletesRAMUsed) +
+ " byteBlockFree=" + toMB(byteBlockAllocator.freeByteBlocks.size()*BYTE_BLOCK_SIZE) +
+ " perDocFree=" + toMB(perDocAllocator.freeByteBlocks.size()*PER_DOC_BLOCK_SIZE) +
+ " charBlockFree=" + toMB(freeCharBlocks.size()*CHAR_BLOCK_SIZE*RamUsageEstimator.NUM_BYTES_CHAR));
+ }
+
+ final long startBytesUsed = bytesUsed() + deletesRAMUsed;
+
+ int iter = 0;
+
+ // We free equally from each pool in 32 KB
+ // chunks until we are below our threshold
+ // (freeLevel)
+
+ boolean any = true;
+
+ final long freeLevel = (long) (0.95 * ramBufferSize);
+
+ while(bytesUsed()+deletesRAMUsed > freeLevel) {
+
+ synchronized(this) {
+ if (0 == perDocAllocator.freeByteBlocks.size()
+ && 0 == byteBlockAllocator.freeByteBlocks.size()
+ && 0 == freeCharBlocks.size()
+ && 0 == freeIntBlocks.size()
+ && !any) {
+ // Nothing else to free -- must flush now.
+ bufferIsFull = bytesUsed()+deletesRAMUsed > ramBufferSize;
+ if (infoStream != null) {
+ if (bytesUsed()+deletesRAMUsed > ramBufferSize) {
+ message(" nothing to free; set bufferIsFull");
+ } else {
+ message(" nothing to free");
+ }
+ }
+ break;
+ }
+
+ if ((0 == iter % 5) && byteBlockAllocator.freeByteBlocks.size() > 0) {
+ byteBlockAllocator.freeByteBlocks.remove(byteBlockAllocator.freeByteBlocks.size()-1);
+ bytesUsed.addAndGet(-BYTE_BLOCK_SIZE);
+ }
+
+ if ((1 == iter % 5) && freeCharBlocks.size() > 0) {
+ freeCharBlocks.remove(freeCharBlocks.size()-1);
+ bytesUsed.addAndGet(-CHAR_BLOCK_SIZE * RamUsageEstimator.NUM_BYTES_CHAR);
+ }
+
+ if ((2 == iter % 5) && freeIntBlocks.size() > 0) {
+ freeIntBlocks.remove(freeIntBlocks.size()-1);
+ bytesUsed.addAndGet(-INT_BLOCK_SIZE * RamUsageEstimator.NUM_BYTES_INT);
+ }
+
+ if ((3 == iter % 5) && perDocAllocator.freeByteBlocks.size() > 0) {
+ // Remove upwards of 32 blocks (each block is 1K)
+ for (int i = 0; i < 32; ++i) {
+ perDocAllocator.freeByteBlocks.remove(perDocAllocator.freeByteBlocks.size() - 1);
+ bytesUsed.addAndGet(-PER_DOC_BLOCK_SIZE);
+ if (perDocAllocator.freeByteBlocks.size() == 0) {
+ break;
+ }
+ }
+ }
+ }
+
+ if ((4 == iter % 5) && any) {
+ // Ask consumer to free any recycled state
+ any = consumer.freeRAM();
+ }
+
+ iter++;
+ }
+
+ if (infoStream != null) {
+ message(" after free: freedMB=" + nf.format((startBytesUsed-bytesUsed()-deletesRAMUsed)/1024./1024.) + " usedMB=" + nf.format((bytesUsed()+deletesRAMUsed)/1024./1024.));
+ }
+ }
+ }
+
+ final WaitQueue waitQueue = new WaitQueue();
+
+ private class WaitQueue {
+ DocWriter[] waiting;
+ int nextWriteDocID;
+ int nextWriteLoc;
+ int numWaiting;
+ long waitingBytes;
+
+ public WaitQueue() {
+ waiting = new DocWriter[10];
+ }
+
+ synchronized void reset() {
+ // NOTE: nextWriteLoc doesn't need to be reset
+ assert numWaiting == 0;
+ assert waitingBytes == 0;
+ nextWriteDocID = 0;
+ }
+
+ synchronized boolean doResume() {
+ final double mb = config.getRAMBufferSizeMB();
+ final long waitQueueResumeBytes;
+ if (mb == IndexWriterConfig.DISABLE_AUTO_FLUSH) {
+ waitQueueResumeBytes = 2*1024*1024;
+ } else {
+ waitQueueResumeBytes = (long) (mb*1024*1024*0.05);
+ }
+ return waitingBytes <= waitQueueResumeBytes;
+ }
+
+ synchronized boolean doPause() {
+ final double mb = config.getRAMBufferSizeMB();
+ final long waitQueuePauseBytes;
+ if (mb == IndexWriterConfig.DISABLE_AUTO_FLUSH) {
+ waitQueuePauseBytes = 4*1024*1024;
+ } else {
+ waitQueuePauseBytes = (long) (mb*1024*1024*0.1);
+ }
+ return waitingBytes > waitQueuePauseBytes;
+ }
+
+ synchronized void abort() {
+ int count = 0;
+ for(int i=0;i<waiting.length;i++) {
+ final DocWriter doc = waiting[i];
+ if (doc != null) {
+ doc.abort();
+ waiting[i] = null;
+ count++;
+ }
+ }
+ waitingBytes = 0;
+ assert count == numWaiting;
+ numWaiting = 0;
+ }
+
+ private void writeDocument(DocWriter doc) throws IOException {
+ assert doc == skipDocWriter || nextWriteDocID == doc.docID;
+ boolean success = false;
+ try {
+ doc.finish();
+ nextWriteDocID++;
+ nextWriteLoc++;
+ assert nextWriteLoc <= waiting.length;
+ if (nextWriteLoc == waiting.length) {
+ nextWriteLoc = 0;
+ }
+ success = true;
+ } finally {
+ if (!success) {
+ setAborting();
+ }
+ }
+ }
+
+ synchronized public boolean add(DocWriter doc) throws IOException {
+
+ assert doc.docID >= nextWriteDocID;
+
+ if (doc.docID == nextWriteDocID) {
+ writeDocument(doc);
+ while(true) {
+ doc = waiting[nextWriteLoc];
+ if (doc != null) {
+ numWaiting--;
+ waiting[nextWriteLoc] = null;
+ waitingBytes -= doc.sizeInBytes();
+ writeDocument(doc);
+ } else {
+ break;
+ }
+ }
+ } else {
+
+ // I finished before documents that were added
+ // before me. This can easily happen when I am a
+ // small doc and the docs before me were large, or,
+ // just due to luck in the thread scheduling. Just
+ // add myself to the queue and when that large doc
+ // finishes, it will flush me:
+ int gap = doc.docID - nextWriteDocID;
+ if (gap >= waiting.length) {
+ // Grow queue
+ DocWriter[] newArray = new DocWriter[ArrayUtil.oversize(gap, RamUsageEstimator.NUM_BYTES_OBJECT_REF)];
+ assert nextWriteLoc >= 0;
+ System.arraycopy(waiting, nextWriteLoc, newArray, 0, waiting.length-nextWriteLoc);
+ System.arraycopy(waiting, 0, newArray, waiting.length-nextWriteLoc, nextWriteLoc);
+ nextWriteLoc = 0;
+ waiting = newArray;
+ gap = doc.docID - nextWriteDocID;
+ }
+
+ int loc = nextWriteLoc + gap;
+ if (loc >= waiting.length) {
+ loc -= waiting.length;
+ }
+
+ // We should only wrap one time
+ assert loc < waiting.length;
+
+ // Nobody should be in my spot!
+ assert waiting[loc] == null;
+ waiting[loc] = doc;
+ numWaiting++;
+ waitingBytes += doc.sizeInBytes();
+ }
+
+ return doPause();
+ }
+ }
+}
fnp / pylucene.git / blobdiff