--- /dev/null
+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();
+ }
+ }
+}
fnp / pylucene.git / blobdiff