lucene-java-3.4.0/lucene/contrib/misc/src/java/org/apache/lucene/index/NRTManagerReopenThread.java

   1 package org.apache.lucene.index;
   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
  82 public class NRTManagerReopenThread extends Thread implements NRTManager.WaitingListener, Closeable {
  83   private final NRTManager manager;
  84   private final long targetMaxStaleNS;
  85   private final long targetMinStaleNS;
  86   private boolean finish;
  87   private boolean waitingNeedsDeletes;
  88   private long waitingGen;
  89
  90   /**
  91    * Create NRTManagerReopenThread, to periodically reopen the NRT searcher.
  92    *
  93    * @param targetMaxStaleSec Maximum time until a new
  94    *        reader must be opened; this sets the upper bound
  95    *        on how slowly reopens may occur
  96    *
  97    * @param targetMinStaleSec Mininum time until a new
  98    *        reader can be opened; this sets the lower bound
  99    *        on how quickly reopens may occur, when a caller
 100    *        is waiting for a specific indexing change to
 101    *        become visible.
 102    */
 103
 104   public NRTManagerReopenThread(NRTManager manager, double targetMaxStaleSec, double targetMinStaleSec) {
 105     if (targetMaxStaleSec < targetMinStaleSec) {
 106       throw new IllegalArgumentException("targetMaxScaleSec (= " + targetMaxStaleSec + ") < targetMinStaleSec (=" + targetMinStaleSec + ")");
 107     }
 108     this.manager = manager;
 109     this.targetMaxStaleNS = (long) (1000000000*targetMaxStaleSec);
 110     this.targetMinStaleNS = (long) (1000000000*targetMinStaleSec);
 111     manager.addWaitingListener(this);
 112   }
 113
 114   public synchronized void close() {
 115     //System.out.println("NRT: set finish");
 116     manager.removeWaitingListener(this);
 117     this.finish = true;
 118     notify();
 119     try {
 120       join();
 121     } catch (InterruptedException ie) {
 122       throw new ThreadInterruptedException(ie);
 123     }
 124   }
 125
 126   public synchronized void waiting(boolean needsDeletes, long targetGen) {
 127     waitingNeedsDeletes |= needsDeletes;
 128     waitingGen = Math.max(waitingGen, targetGen);
 129     notify();
 130     //System.out.println(Thread.currentThread().getName() + ": force wakeup waitingGen=" + waitingGen + " applyDeletes=" + applyDeletes + " waitingNeedsDeletes=" + waitingNeedsDeletes);
 131   }
 132
 133   @Override
 134     public void run() {
 135     // TODO: maybe use private thread ticktock timer, in
 136     // case clock shift messes up nanoTime?
 137     long lastReopenStartNS = System.nanoTime();
 138
 139     //System.out.println("reopen: start");
 140     try {
 141       while (true) {
 142
 143         final boolean doApplyDeletes;
 144
 145         boolean hasWaiting = false;
 146
 147         synchronized(this) {
 148           // TODO: try to guestimate how long reopen might
 149           // take based on past data?
 150
 151           while (!finish) {
 152             //System.out.println("reopen: cycle");
 153
 154             // True if we have someone waiting for reopen'd searcher:
 155             hasWaiting = waitingGen > manager.getCurrentSearchingGen(waitingNeedsDeletes);
 156             final long nextReopenStartNS = lastReopenStartNS + (hasWaiting ? targetMinStaleNS : targetMaxStaleNS);
 157
 158             final long sleepNS = nextReopenStartNS - System.nanoTime();
 159
 160             if (sleepNS > 0) {
 161               //System.out.println("reopen: sleep " + (sleepNS/1000000.0) + " ms (hasWaiting=" + hasWaiting + ")");
 162               try {
 163                 wait(sleepNS/1000000, (int) (sleepNS%1000000));
 164               } catch (InterruptedException ie) {
 165                 Thread.currentThread().interrupt();
 166                 //System.out.println("NRT: set finish on interrupt");
 167                 finish = true;
 168                 break;
 169               }
 170             } else {
 171               break;
 172             }
 173           }
 174
 175           if (finish) {
 176             //System.out.println("reopen: finish");
 177             return;
 178           }
 179
 180           doApplyDeletes = hasWaiting ? waitingNeedsDeletes : true;
 181           waitingNeedsDeletes = false;
 182           //System.out.println("reopen: start hasWaiting=" + hasWaiting);
 183         }
 184
 185         lastReopenStartNS = System.nanoTime();
 186         try {
 187           //final long t0 = System.nanoTime();
 188           manager.reopen(doApplyDeletes);
 189           //System.out.println("reopen took " + ((System.nanoTime()-t0)/1000000.0) + " msec");
 190         } catch (IOException ioe) {
 191           //System.out.println(Thread.currentThread().getName() + ": IOE");
 192           //ioe.printStackTrace();
 193           throw new RuntimeException(ioe);
 194         }
 195       }
 196     } catch (Throwable t) {
 197       //System.out.println("REOPEN EXC");
 198       //t.printStackTrace(System.out);
 199       throw new RuntimeException(t);
 200     }
 201   }
 202 }