content: 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);
          final String segFileNamepos = IndexFileNames.segmentFileName(segment, IndexFileNames.COMPOUND_FILE_EXTENSION_POS);
  
          if (infoStream != null) {
            message("flush: create compound file \"" + cfsFileName + "\"");
          }
  
          CompoundFileWriter cfsWriter = new CompoundFileWriter(directory, cfsFileName,segFileNamepos);
          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");
      }
    }
  
  //  boolean doPauseWaitQueue = false;
  
    /** 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();
      }
    }
  }
  