lucene-java-3.5.0/lucene/src/java/org/apache/lucene/search/NRTManagerReopenThread.java

   1 package org.apache.lucene.search;
   2
   3 /**
   4  * Licensed to the Apache Software Foundation (ASF) under one or more
   5  * contributor license agreements.  See the NOTICE file distributed with
   6  * this work for additional information regarding copyright ownership.
   7  * The ASF licenses this file to You under the Apache License, Version 2.0
   8  * (the "License"); you may not use this file except in compliance with
   9  * the License.  You may obtain a copy of the License at
  10  *
  11  *     http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  */
  19
  20 import java.io.Closeable;
  21 import java.io.IOException;
  22
  23 import org.apache.lucene.util.ThreadInterruptedException;
  24
  25 /**
  26  * Utility class that runs a reopen thread to periodically
  27  * reopen the NRT searchers in the provided {@link
  28  * NRTManager}.
  29  *
  30  * <p> Typical usage looks like this:
  31  *
  32  * <pre>
  33  *   ... open your own writer ...
  34  *
  35  *   NRTManager manager = new NRTManager(writer);
  36  *
  37  *   // Refreshes searcher every 5 seconds when nobody is waiting, and up to 100 msec delay
  38  *   // when somebody is waiting:
  39  *   NRTManagerReopenThread reopenThread = new NRTManagerReopenThread(manager, 5.0, 0.1);
  40  *   reopenThread.setName("NRT Reopen Thread");
  41  *   reopenThread.setPriority(Math.min(Thread.currentThread().getPriority()+2, Thread.MAX_PRIORITY));
  42  *   reopenThread.setDaemon(true);
  43  *   reopenThread.start();
  44  * </pre>
  45  *
  46  * Then, for each incoming query, do this:
  47  *
  48  * <pre>
  49  *   // For each incoming query:
  50  *   IndexSearcher searcher = manager.get();
  51  *   try {
  52  *     // Use searcher to search...
  53  *   } finally {
  54  *     manager.release(searcher);
  55  *   }
  56  * </pre>
  57  *
  58  * You should make changes using the <code>NRTManager</code>; if you later need to obtain
  59  * a searcher reflecting those changes:
  60  *
  61  * <pre>
  62  *   // ... or updateDocument, deleteDocuments, etc:
  63  *   long gen = manager.addDocument(...);
  64  *
  65  *   // Returned searcher is guaranteed to reflect the just added document
  66  *   IndexSearcher searcher = manager.get(gen);
  67  *   try {
  68  *     // Use searcher to search...
  69  *   } finally {
  70  *     manager.release(searcher);
  71  *   }
  72  * </pre>
  73  *
  74  *
  75  * When you are done be sure to close both the manager and the reopen thrad:
  76  * <pre>
  77  *   reopenThread.close();
  78  *   manager.close();
  79  * </pre>
  80  *
  81  * @lucene.experimental
  82  */
  83
  84 public class NRTManagerReopenThread extends Thread implements NRTManager.WaitingListener, Closeable {
  85
  86   private final NRTManager manager;
  87   private final long targetMaxStaleNS;
  88   private final long targetMinStaleNS;
  89   private boolean finish;
  90   private long waitingGen;
  91   private boolean waitingNeedsDeletes;
  92
  93   /**
  94    * Create NRTManagerReopenThread, to periodically reopen the NRT searcher.
  95    *
  96    * @param targetMaxStaleSec Maximum time until a new
  97    *        reader must be opened; this sets the upper bound
  98    *        on how slowly reopens may occur
  99    *
 100    * @param targetMinStaleSec Mininum time until a new
 101    *        reader can be opened; this sets the lower bound
 102    *        on how quickly reopens may occur, when a caller
 103    *        is waiting for a specific indexing change to
 104    *        become visible.
 105    */
 106
 107   public NRTManagerReopenThread(NRTManager manager, double targetMaxStaleSec, double targetMinStaleSec) {
 108     if (targetMaxStaleSec < targetMinStaleSec) {
 109       throw new IllegalArgumentException("targetMaxScaleSec (= " + targetMaxStaleSec + ") < targetMinStaleSec (=" + targetMinStaleSec + ")");
 110     }
 111     this.manager = manager;
 112     this.targetMaxStaleNS = (long) (1000000000*targetMaxStaleSec);
 113     this.targetMinStaleNS = (long) (1000000000*targetMinStaleSec);
 114     manager.addWaitingListener(this);
 115   }
 116
 117   public synchronized void close() {
 118     //System.out.println("NRT: set finish");
 119     manager.removeWaitingListener(this);
 120     this.finish = true;
 121     notify();
 122     try {
 123       join();
 124     } catch (InterruptedException ie) {
 125       throw new ThreadInterruptedException(ie);
 126     }
 127   }
 128
 129   public synchronized void waiting(boolean needsDeletes, long targetGen) {
 130     waitingNeedsDeletes |= needsDeletes;
 131     waitingGen = Math.max(waitingGen, targetGen);
 132     notify();
 133     //System.out.println(Thread.currentThread().getName() + ": force wakeup waitingGen=" + waitingGen + " applyDeletes=" + applyDeletes + " waitingNeedsDeletes=" + waitingNeedsDeletes);
 134   }
 135
 136   @Override
 137   public void run() {
 138     // TODO: maybe use private thread ticktock timer, in
 139     // case clock shift messes up nanoTime?
 140     long lastReopenStartNS = System.nanoTime();
 141
 142     //System.out.println("reopen: start");
 143     try {
 144       while (true) {
 145
 146         boolean hasWaiting = false;
 147
 148         synchronized(this) {
 149           // TODO: try to guestimate how long reopen might
 150           // take based on past data?
 151
 152           while (!finish) {
 153             //System.out.println("reopen: cycle");
 154
 155             // True if we have someone waiting for reopen'd searcher:
 156             hasWaiting = waitingGen > manager.getCurrentSearchingGen(waitingNeedsDeletes);
 157             final long nextReopenStartNS = lastReopenStartNS + (hasWaiting ? targetMinStaleNS : targetMaxStaleNS);
 158
 159             final long sleepNS = nextReopenStartNS - System.nanoTime();
 160
 161             if (sleepNS > 0) {
 162               //System.out.println("reopen: sleep " + (sleepNS/1000000.0) + " ms (hasWaiting=" + hasWaiting + ")");
 163               try {
 164                 wait(sleepNS/1000000, (int) (sleepNS%1000000));
 165               } catch (InterruptedException ie) {
 166                 Thread.currentThread().interrupt();
 167                 //System.out.println("NRT: set finish on interrupt");
 168                 finish = true;
 169                 break;
 170               }
 171             } else {
 172               break;
 173             }
 174           }
 175
 176           if (finish) {
 177             //System.out.println("reopen: finish");
 178             return;
 179           }
 180           //System.out.println("reopen: start hasWaiting=" + hasWaiting);
 181         }
 182
 183         lastReopenStartNS = System.nanoTime();
 184         try {
 185           //final long t0 = System.nanoTime();
 186           manager.maybeReopen(waitingNeedsDeletes);
 187           //System.out.println("reopen took " + ((System.nanoTime()-t0)/1000000.0) + " msec");
 188         } catch (IOException ioe) {
 189           //System.out.println(Thread.currentThread().getName() + ": IOE");
 190           //ioe.printStackTrace();
 191           throw new RuntimeException(ioe);
 192         }
 193       }
 194     } catch (Throwable t) {
 195       //System.out.println("REOPEN EXC");
 196       //t.printStackTrace(System.out);
 197       throw new RuntimeException(t);
 198     }
 199   }
 200 }