multiindex.java

来自「jsr170接口的java实现。是个apache的开源项目。」· Java 代码 · 共 1,855 行 · 第 1/5 页

/*
 * 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.
 */
package org.apache.jackrabbit.core.query.lucene;

import org.apache.jackrabbit.core.NodeId;
import org.apache.jackrabbit.core.fs.FileSystemException;
import org.apache.jackrabbit.core.fs.local.LocalFileSystem;
import org.apache.jackrabbit.core.state.ItemStateException;
import org.apache.jackrabbit.core.state.ItemStateManager;
import org.apache.jackrabbit.core.state.NoSuchItemStateException;
import org.apache.jackrabbit.core.state.NodeState;
import org.apache.jackrabbit.uuid.Constants;
import org.apache.jackrabbit.uuid.UUID;
import org.apache.jackrabbit.util.Timer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.lucene.document.Document;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.Term;
import org.apache.commons.collections.iterators.EmptyIterator;

import javax.jcr.RepositoryException;
import java.io.IOException;
import java.io.File;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Arrays;
import java.util.Set;
import java.util.HashSet;
import java.util.HashMap;
import java.util.Map;
import java.util.Collection;

/**
 * A <code>MultiIndex</code> consists of a {@link VolatileIndex} and multiple
 * {@link PersistentIndex}es. The goal is to keep most parts of the index open
 * with index readers and write new index data to the volatile index. When
 * the volatile index reaches a certain size (see {@link SearchIndex#setMinMergeDocs(int)})
 * a new persistent index is created with the index data from the volatile index,
 * the same happens when the volatile index has been idle for some time (see
 * {@link SearchIndex#setVolatileIdleTime(int)}).
 * The new persistent index is then added to the list of already existing
 * persistent indexes. Further operations on the new persistent index will
 * however only require an <code>IndexReader</code> which serves for queries
 * but also for delete operations on the index.
 * <p/>
 * The persistent indexes are merged from time to time. The merge behaviour
 * is configurable using the methods: {@link SearchIndex#setMaxMergeDocs(int)},
 * {@link SearchIndex#setMergeFactor(int)} and {@link SearchIndex#setMinMergeDocs(int)}.
 * For detailed description of the configuration parameters see also the lucene
 * <code>IndexWriter</code> class.
 * <p/>
 * This class is thread-safe.
 * <p/>
 * Note on implementation: Multiple modifying threads are synchronized on a
 * <code>MultiIndex</code> instance itself. Sychronization between a modifying
 * thread and reader threads is done using {@link #updateMonitor} and
 * {@link #updateInProgress}.
 */
public class MultiIndex {

    /**
     * The logger instance for this class
     */
    private static final Logger log = LoggerFactory.getLogger(MultiIndex.class);

    /**
     * Default name of the redo log file
     */
    private static final String REDO_LOG = "redo.log";

    /**
     * Name of the file that contains the indexing queue log.
     */
    private static final String INDEXING_QUEUE_FILE = "indexing_queue.log";

    /**
     * Names of active persistent index directories.
     */
    private final IndexInfos indexNames = new IndexInfos("indexes");

    /**
     * Names of index directories that can be deleted.
     */
    private final IndexInfos deletable = new IndexInfos("deletable");

    /**
     * List of open persistent indexes. This list may also contain an open
     * PersistentIndex owned by the IndexMerger daemon. Such an index is not
     * registered with indexNames and <b>must not</b> be used in regular index
     * operations (delete node, etc.)!
     */
    private final List indexes = new ArrayList();

    /**
     * The internal namespace mappings of the query manager.
     */
    private final NamespaceMappings nsMappings;

    /**
     * The base filesystem to store the index.
     */
    private final File indexDir;

    /**
     * The query handler
     */
    private final SearchIndex handler;

    /**
     * The volatile index.
     */
    private VolatileIndex volatileIndex;

    /**
     * Flag indicating whether an update operation is in progress.
     */
    private boolean updateInProgress = false;

    /**
     * If not <code>null</code> points to a valid <code>IndexReader</code> that
     * reads from all indexes, including volatile and persistent indexes.
     */
    private CachingMultiReader multiReader;

    /**
     * Shared document number cache across all persistent indexes.
     */
    private final DocNumberCache cache;

    /**
     * Monitor to use to synchronize access to {@link #multiReader} and
     * {@link #updateInProgress}.
     */
    private final Object updateMonitor = new Object();

    /**
     * <code>true</code> if the redo log contained entries on startup.
     */
    private boolean redoLogApplied = false;

    /**
     * The time this index was last flushed or a transaction was committed.
     */
    private long lastFlushTime;

    /**
     * The <code>IndexMerger</code> for this <code>MultiIndex</code>.
     */
    private final IndexMerger merger;

    /**
     * Timer to schedule flushes of this index after some idle time.
     */
    private static final Timer FLUSH_TIMER = new Timer(true);

    /**
     * Task that is periodically called by {@link #FLUSH_TIMER} and checks
     * if index should be flushed.
     */
    private final Timer.Task flushTask;

    /**
     * The RedoLog of this <code>MultiIndex</code>.
     */
    private final RedoLog redoLog;

    /**
     * The indexing queue with pending text extraction jobs.
     */
    private IndexingQueue indexingQueue;

    /**
     * Set&lt;NodeId> of uuids that should not be indexed.
     */
    private final Set excludedIDs;

    /**
     * The next transaction id.
     */
    private long nextTransactionId = 0;

    /**
     * The current transaction id.
     */
    private long currentTransactionId = -1;

    /**
     * Flag indicating whether re-indexing is running.
     */
    private boolean reindexing = false;

    /**
     * Creates a new MultiIndex.
     *
     * @param indexDir the base file system
     * @param handler the search handler
     * @param stateMgr shared item state manager
     * @param rootId id of the root node
     * @param excludedIDs   Set&lt;NodeId> that contains uuids that should not
     *                      be indexed nor further traversed.
     * @param mapping the namespace mapping to use
     * @throws IOException if an error occurs
     */
    MultiIndex(File indexDir,
               SearchIndex handler,
               ItemStateManager stateMgr,
               NodeId rootId,
               Set excludedIDs,
               NamespaceMappings mapping) throws IOException {

        this.indexDir = indexDir;
        this.handler = handler;
        this.cache = new DocNumberCache(handler.getCacheSize());
        this.redoLog = new RedoLog(new File(indexDir, REDO_LOG));
        this.excludedIDs = new HashSet(excludedIDs);
        this.nsMappings = mapping;

        if (indexNames.exists(indexDir)) {
            indexNames.read(indexDir);
        }
        if (deletable.exists(indexDir)) {
            deletable.read(indexDir);
        }

        // try to remove deletable files if there are any
        attemptDelete();

        // initialize IndexMerger
        merger = new IndexMerger(this);
        merger.setMaxMergeDocs(handler.getMaxMergeDocs());
        merger.setMergeFactor(handler.getMergeFactor());
        merger.setMinMergeDocs(handler.getMinMergeDocs());

        IndexingQueueStore store;
        try {
            LocalFileSystem fs = new LocalFileSystem();
            fs.setRoot(indexDir);
            fs.init();
            store = new IndexingQueueStore(fs, INDEXING_QUEUE_FILE);
        } catch (FileSystemException e) {
            IOException ex = new IOException();
            ex.initCause(e);
            throw ex;
        }

        // initialize indexing queue
        this.indexingQueue = new IndexingQueue(store, this);


        try {
            // open persistent indexes
            for (int i = 0; i < indexNames.size(); i++) {
                File sub = new File(indexDir, indexNames.getName(i));
                // only open if it still exists
                // it is possible that indexNames still contains a name for
                // an index that has been deleted, but indexNames has not been
                // written to disk.
                if (!sub.exists()) {
                    log.debug("index does not exist anymore: " + sub.getAbsolutePath());
                    // move on to next index
                    continue;
                }
                PersistentIndex index = new PersistentIndex(
                        indexNames.getName(i), sub, false,
                        handler.getTextAnalyzer(), cache, indexingQueue);
                index.setMaxMergeDocs(handler.getMaxMergeDocs());
                index.setMergeFactor(handler.getMergeFactor());
                index.setMinMergeDocs(handler.getMinMergeDocs());
                index.setMaxFieldLength(handler.getMaxFieldLength());
                index.setUseCompoundFile(handler.getUseCompoundFile());
                indexes.add(index);
                merger.indexAdded(index.getName(), index.getNumDocuments());
            }

            // init volatile index
            resetVolatileIndex();

            redoLogApplied = redoLog.hasEntries();

            // run recovery
            Recovery.run(this, redoLog);

            // now that we are ready, start index merger
            merger.start();

            if (redoLogApplied) {
                // wait for the index merge to finish pending jobs
                try {
                    merger.waitUntilIdle();
                } catch (InterruptedException e) {
                    // move on
                }
                flush();
            }

            // do an initial index if there are no indexes at all
            if (indexNames.size() == 0) {
                reindexing = true;
                // traverse and index workspace
                executeAndLog(new Start(Action.INTERNAL_TRANSACTION));
                NodeState rootState = (NodeState) stateMgr.getItemState(rootId);
                createIndex(rootState, stateMgr);
                executeAndLog(new Commit(getTransactionId()));
                reindexing = false;
            }
        } catch (ItemStateException e) {
            throw new IOException("Error indexing root node: " + e.getMessage());
        } catch (RepositoryException e) {
            throw new IOException("Error indexing root node: " + e.getMessage());
        }

        lastFlushTime = System.currentTimeMillis();
        flushTask = new Timer.Task() {
            public void run() {
                // check if there are any indexing jobs finished
                checkIndexingQueue();
                // check if volatile index should be flushed
                checkFlush();
            }
        };
        FLUSH_TIMER.schedule(flushTask, 0, 1000);
    }

    /**
     * Atomically updates the index by removing some documents and adding
     * others.
     *
     * @param remove Iterator of <code>UUID</code>s that identify documents to
     *               remove
     * @param add    Iterator of <code>Document</code>s to add. Calls to
     *               <code>next()</code> on this iterator may return
     *               <code>null</code>, to indicate that a node could not be
     *               indexed successfully.
     */
    synchronized void update(Iterator remove, Iterator add) throws IOException {
        synchronized (updateMonitor) {
            updateInProgress = true;
        }
        try {
            long transactionId = nextTransactionId++;
            executeAndLog(new Start(transactionId));

            boolean flush = false;
            while (remove.hasNext()) {
                executeAndLog(new DeleteNode(transactionId, (UUID) remove.next()));
            }
            while (add.hasNext()) {
                Document doc = (Document) add.next();
                if (doc != null) {
                    executeAndLog(new AddNode(transactionId, doc));
                    // commit volatile index if needed
                    flush |= checkVolatileCommit();
                }
            }
            executeAndLog(new Commit(transactionId));