content: package org.apache.lucene.search;
  
  /**
   * 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.Closeable;
  import java.io.IOException;
  import java.util.Collection;
  import java.util.List;
  import java.util.concurrent.CopyOnWriteArrayList;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicLong;
  import java.util.concurrent.locks.Condition;
  import java.util.concurrent.locks.ReentrantLock;
  
  import org.apache.lucene.analysis.Analyzer;
  import org.apache.lucene.document.Document;
  import org.apache.lucene.index.CorruptIndexException;
  import org.apache.lucene.index.IndexReader; // javadocs
  import org.apache.lucene.index.IndexWriter;
  import org.apache.lucene.index.Term;
  import org.apache.lucene.search.IndexSearcher; // javadocs
  import org.apache.lucene.store.Directory;
  import org.apache.lucene.util.IOUtils;
  import org.apache.lucene.util.ThreadInterruptedException;
  
  /**
   * Utility class to manage sharing near-real-time searchers
   * across multiple searching threads.
   *
   * <p>NOTE: to use this class, you must call {@link #maybeReopen(boolean)}
   * periodically.  The {@link NRTManagerReopenThread} is a
   * simple class to do this on a periodic basis.  If you
   * implement your own reopener, be sure to call {@link
   * #addWaitingListener} so your reopener is notified when a
   * caller is waiting for a specific generation searcher. </p>
   *
   * @lucene.experimental
   */
  
  public class NRTManager implements Closeable {
    private static final long MAX_SEARCHER_GEN = Long.MAX_VALUE;
    private final IndexWriter writer;
    private final SearcherManagerRef withoutDeletes;
    private final SearcherManagerRef withDeletes;
    private final AtomicLong indexingGen;
    private final List<WaitingListener> waitingListeners = new CopyOnWriteArrayList<WaitingListener>();
    private final ReentrantLock reopenLock = new ReentrantLock();
    private final Condition newGeneration = reopenLock.newCondition();
  
    /**
     * Create new NRTManager.
     * 
     *  @param writer IndexWriter to open near-real-time
     *         readers
     *  @param warmer optional {@link SearcherWarmer}.  Pass
     *         null if you don't require the searcher to warmed
     *         before going live.  If this is non-null then a
     *         merged segment warmer is installed on the
     *         provided IndexWriter's config.
     *
     *  <p><b>NOTE</b>: the provided {@link SearcherWarmer} is
     *  not invoked for the initial searcher; you should
     *  warm it yourself if necessary.
     */
    public NRTManager(IndexWriter writer, SearcherWarmer warmer) throws IOException {
      this(writer, null, warmer, true);
    }
  
    /**
     * Create new NRTManager.
     * 
     *  @param writer IndexWriter to open near-real-time
     *         readers
     *  @param es optional ExecutorService so different segments can
     *         be searched concurrently (see {@link IndexSearcher#IndexSearcher(IndexReader, ExecutorService)}.
     *         Pass <code>null</code> to search segments sequentially.
     *  @param warmer optional {@link SearcherWarmer}.  Pass
     *         null if you don't require the searcher to warmed
     *         before going live.  If this is non-null then a
     *         merged segment warmer is installed on the
     *         provided IndexWriter's config.
     *
     *  <p><b>NOTE</b>: the provided {@link SearcherWarmer} is
     *  not invoked for the initial searcher; you should
     *  warm it yourself if necessary.
     */
    public NRTManager(IndexWriter writer, ExecutorService es,
        SearcherWarmer warmer) throws IOException {
      this(writer, es, warmer, true);
    }
  
    /**
     * Expert: just like {@link
     * #NRTManager(IndexWriter,ExecutorService,SearcherWarmer)},
     * but you can also specify whether every searcher must
     * apply deletes.  This is useful for cases where certain
     * uses can tolerate seeing some deleted docs, since
     * reopen time is faster if deletes need not be applied. */
    public NRTManager(IndexWriter writer, ExecutorService es,
        SearcherWarmer warmer, boolean alwaysApplyDeletes) throws IOException {
      this.writer = writer;
      if (alwaysApplyDeletes) {
        withoutDeletes = withDeletes = new SearcherManagerRef(true, 0,  new SearcherManager(writer, true, warmer, es));
      } else {
        withDeletes = new SearcherManagerRef(true, 0, new SearcherManager(writer, true, warmer, es));
        withoutDeletes = new SearcherManagerRef(false, 0, new SearcherManager(writer, false, warmer, es));
      }
      indexingGen = new AtomicLong(1);
    }
    
    /** NRTManager invokes this interface to notify it when a
     *  caller is waiting for a specific generation searcher
     *  to be visible. */
    public static interface WaitingListener {
      public void waiting(boolean requiresDeletes, long targetGen);
    }
  
    /** Adds a listener, to be notified when a caller is
     *  waiting for a specific generation searcher to be
     *  visible. */
    public void addWaitingListener(WaitingListener l) {
      waitingListeners.add(l);
    }
  
    /** Remove a listener added with {@link
     *  #addWaitingListener}. */
    public void removeWaitingListener(WaitingListener l) {
      waitingListeners.remove(l);
    }
  
    public long updateDocument(Term t, Document d, Analyzer a) throws IOException {
      writer.updateDocument(t, d, a);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long updateDocument(Term t, Document d) throws IOException {
      writer.updateDocument(t, d);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long updateDocuments(Term t, Collection<Document> docs, Analyzer a) throws IOException {
      writer.updateDocuments(t, docs, a);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long updateDocuments(Term t, Collection<Document> docs) throws IOException {
      writer.updateDocuments(t, docs);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long deleteDocuments(Term t) throws IOException {
      writer.deleteDocuments(t);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long deleteDocuments(Term... terms) throws IOException {
      writer.deleteDocuments(terms);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long deleteDocuments(Query q) throws IOException {
      writer.deleteDocuments(q);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long deleteDocuments(Query... queries) throws IOException {
      writer.deleteDocuments(queries);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long deleteAll() throws IOException {
      writer.deleteAll();
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long addDocument(Document d, Analyzer a) throws IOException {
      writer.addDocument(d, a);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long addDocuments(Collection<Document> docs, Analyzer a) throws IOException {
      writer.addDocuments(docs, a);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long addDocument(Document d) throws IOException {
      writer.addDocument(d);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long addDocuments(Collection<Document> docs) throws IOException {
      writer.addDocuments(docs);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long addIndexes(Directory... dirs) throws CorruptIndexException, IOException {
      writer.addIndexes(dirs);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    public long addIndexes(IndexReader... readers) throws CorruptIndexException, IOException {
      writer.addIndexes(readers);
      // Return gen as of when indexing finished:
      return indexingGen.get();
    }
  
    /**
     * Waits for a given {@link SearcherManager} target generation to be available
     * via {@link #getSearcherManager(boolean)}. If the current generation is less
     * than the given target generation this method will block until the
     * correspondent {@link SearcherManager} is reopened by another thread via
     * {@link #maybeReopen(boolean)} or until the {@link NRTManager} is closed.
     * 
     * @param targetGen the generation to wait for
     * @param requireDeletes <code>true</code> iff the generation requires deletes to be applied otherwise <code>false</code>
     * @return the {@link SearcherManager} with the given target generation
     */
    public SearcherManager waitForGeneration(long targetGen, boolean requireDeletes) {
      return waitForGeneration(targetGen, requireDeletes, -1,  TimeUnit.NANOSECONDS);
    }
  
    /**
     * Waits for a given {@link SearcherManager} target generation to be available
     * via {@link #getSearcherManager(boolean)}. If the current generation is less
     * than the given target generation this method will block until the
     * correspondent {@link SearcherManager} is reopened by another thread via
     * {@link #maybeReopen(boolean)}, the given waiting time has elapsed, or until
     * the {@link NRTManager} is closed.
     * <p>
     * NOTE: if the waiting time elapses before the requested target generation is
     * available the latest {@link SearcherManager} is returned instead.
     * 
     * @param targetGen
     *          the generation to wait for
     * @param requireDeletes
     *          <code>true</code> iff the generation requires deletes to be
     *          applied otherwise <code>false</code>
     * @param time
     *          the time to wait for the target generation
     * @param unit
     *          the waiting time's time unit
     * @return the {@link SearcherManager} with the given target generation or the
     *         latest {@link SearcherManager} if the waiting time elapsed before
     *         the requested generation is available.
     */
    public SearcherManager waitForGeneration(long targetGen, boolean requireDeletes, long time, TimeUnit unit) {
      try {
        final long curGen = indexingGen.get();
        if (targetGen > curGen) {
          throw new IllegalArgumentException("targetGen=" + targetGen + " was never returned by this NRTManager instance (current gen=" + curGen + ")");
        }
        reopenLock.lockInterruptibly();
        try {
          if (targetGen > getCurrentSearchingGen(requireDeletes)) {
            for (WaitingListener listener : waitingListeners) {
              listener.waiting(requireDeletes, targetGen);
            }
            while (targetGen > getCurrentSearchingGen(requireDeletes)) {
              if (!waitOnGenCondition(time, unit)) {
                return getSearcherManager(requireDeletes);
              }
            }
          }
  
        } finally {
          reopenLock.unlock();
        }
      } catch (InterruptedException ie) {
        throw new ThreadInterruptedException(ie);
      }
      return getSearcherManager(requireDeletes);
    }
    
    private boolean waitOnGenCondition(long time, TimeUnit unit)
        throws InterruptedException {
      assert reopenLock.isHeldByCurrentThread();
      if (time < 0) {
        newGeneration.await();
        return true;
      } else {
        return newGeneration.await(time, unit);
      }
    }
  
    /** Returns generation of current searcher. */
    public long getCurrentSearchingGen(boolean applyAllDeletes) {
      if (applyAllDeletes) {
        return withDeletes.generation;
      } else {
        return Math.max(withoutDeletes.generation, withDeletes.generation);
      }
    }
  
    public boolean maybeReopen(boolean applyAllDeletes) throws IOException {
      if (reopenLock.tryLock()) {
        try {
          final SearcherManagerRef reference = applyAllDeletes ? withDeletes : withoutDeletes;
          // Mark gen as of when reopen started:
          final long newSearcherGen = indexingGen.getAndIncrement();
          boolean setSearchGen = false;
          if (reference.generation == MAX_SEARCHER_GEN) {
            newGeneration.signalAll(); // wake up threads if we have a new generation
            return false;
          }
          if (!(setSearchGen = reference.manager.isSearcherCurrent())) {
            setSearchGen = reference.manager.maybeReopen();
          }
          if (setSearchGen) {
            reference.generation = newSearcherGen;// update searcher gen
            newGeneration.signalAll(); // wake up threads if we have a new generation
          }
          return setSearchGen;
        } finally {
          reopenLock.unlock();
        }
      }
      return false;
    }
  
    /**
     * Close this NRTManager to future searching. Any searches still in process in
     * other threads won't be affected, and they should still call
     * {@link SearcherManager#release(IndexSearcher)} after they are done.
     * 
     * <p>
     * <b>NOTE</b>: caller must separately close the writer.
     */
    public void close() throws IOException {
      reopenLock.lock();
      try {
        try {
          IOUtils.close(withDeletes, withoutDeletes);
        } finally { // make sure we signal even if close throws an exception
          newGeneration.signalAll();
        }
      } finally {
        reopenLock.unlock();
        assert withDeletes.generation == MAX_SEARCHER_GEN && withoutDeletes.generation == MAX_SEARCHER_GEN;
      }
    }
  
    /**
     * Returns a {@link SearcherManager}. If <code>applyAllDeletes</code> is
     * <code>true</code> the returned manager is guaranteed to have all deletes
     * applied on the last reopen. Otherwise the latest manager with or without deletes
     * is returned.
     */
    public SearcherManager getSearcherManager(boolean applyAllDeletes) {
      if (applyAllDeletes) {
        return withDeletes.manager;
      } else {
        if (withDeletes.generation > withoutDeletes.generation) {
          return withDeletes.manager;
        } else {
          return withoutDeletes.manager;
        }
      }
    }
    
    static final class SearcherManagerRef implements Closeable {
      final boolean applyDeletes;
      volatile long generation;
      final SearcherManager manager;
  
      SearcherManagerRef(boolean applyDeletes, long generation, SearcherManager manager) {
        super();
        this.applyDeletes = applyDeletes;
        this.generation = generation;
        this.manager = manager;
      }
      
      public void close() throws IOException {
        generation = MAX_SEARCHER_GEN; // max it out to make sure nobody can wait on another gen
        manager.close();
      }
    }
  }
  