indexmerger.java

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

/*
 * 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.lucene.index.Term;
import org.apache.lucene.index.IndexReader;
import org.apache.commons.collections.Buffer;
import org.apache.commons.collections.BufferUtils;
import org.apache.commons.collections.buffer.UnboundedFifoBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Collections;
import java.util.ArrayList;
import java.util.Iterator;
import java.io.IOException;

import EDU.oswego.cs.dl.util.concurrent.Sync;
import EDU.oswego.cs.dl.util.concurrent.Mutex;

/**
 * Merges indexes in a separate deamon thread.
 */
class IndexMerger extends Thread implements IndexListener {

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

    /**
     * Marker task to signal the background thread to quit.
     */
    private static final Merge QUIT = new Merge(new Index[0]);

    /**
     * minMergeDocs config parameter.
     */
    private int minMergeDocs = SearchIndex.DEFAULT_MIN_MERGE_DOCS;

    /**
     * maxMergeDocs config parameter
     */
    private int maxMergeDocs = SearchIndex.DEFAULT_MAX_MERGE_DOCS;

    /**
     * mergeFactor config parameter
     */
    private int mergeFactor = SearchIndex.DEFAULT_MERGE_FACTOR;

    /**
     * Queue of merge Tasks
     */
    private final Buffer mergeTasks = BufferUtils.blockingBuffer(new UnboundedFifoBuffer());

    /**
     * List of id <code>Term</code> that identify documents that were deleted
     * while a merge was running.
     */
    private final List deletedDocuments = Collections.synchronizedList(new ArrayList());

    /**
     * List of <code>IndexBucket</code>s in ascending document limit.
     */
    private final List indexBuckets = new ArrayList();

    /**
     * The <code>MultiIndex</code> this index merger is working on.
     */
    private final MultiIndex multiIndex;

    /**
     * Monitor object to synchronize merge calculation.
     */
    private final Object lock = new Object();

    /**
     * Mutex that is acquired when replacing indexes on MultiIndex.
     */
    private final Sync indexReplacement = new Mutex();

    /**
     * When released, indicates that this index merger is idle.
     */
    private final Sync mergerIdle = new Mutex();

    /**
     * Creates an <code>IndexMerger</code>.
     *
     * @param multiIndex the <code>MultiIndex</code>.
     */
    IndexMerger(MultiIndex multiIndex) {
        this.multiIndex = multiIndex;
        setName("IndexMerger");
        setDaemon(true);
        try {
            mergerIdle.acquire();
        } catch (InterruptedException e) {
            // will never happen, lock is free upon construction
            throw new InternalError("Unable to acquire mutex after construction");
        }
    }

    /**
     * Informs the index merger that an index was added / created.
     *
     * @param name the name of the index.
     * @param numDocs the number of documents it contains.
     */
    void indexAdded(String name, int numDocs) {
        if (numDocs < 0) {
            throw new IllegalArgumentException("numDocs must be positive");
        }
        // multiple threads may enter this method:
        // - the background thread of this IndexMerger, when it replaces indexes
        //   after a successful merge
        // - a regular thread that updates the workspace
        //
        // therefore we have to synchronize this block
        synchronized (lock) {
            // initially create buckets
            if (indexBuckets.size() == 0) {
                int lower = 0;
                int upper = minMergeDocs;
                while (upper < maxMergeDocs) {
                    indexBuckets.add(new IndexBucket(lower, upper, true));
                    lower = upper + 1;
                    upper *= mergeFactor;
                }
                // one with upper = maxMergeDocs
                indexBuckets.add(new IndexBucket(lower, maxMergeDocs, false));
                // and another one as overflow, just in case...
                indexBuckets.add(new IndexBucket(maxMergeDocs + 1, Integer.MAX_VALUE, false));
            }

            // put index in bucket
            IndexBucket bucket = (IndexBucket) indexBuckets.get(indexBuckets.size() - 1);
            for (int i = 0; i < indexBuckets.size(); i++) {
                bucket = (IndexBucket) indexBuckets.get(i);
                if (bucket.fits(numDocs)) {
                    break;
                }
            }
            bucket.add(new Index(name, numDocs));

            if (log.isDebugEnabled()) {
                log.debug("index added: name=" + name + ", numDocs=" + numDocs);
            }

            // if bucket does not allow merge, we don't have to continue
            if (!bucket.allowsMerge()) {
                return;
            }

            // check if we need a merge
            if (bucket.size() >= mergeFactor) {
                long targetMergeDocs = bucket.upper;
                targetMergeDocs = Math.min(targetMergeDocs * mergeFactor, maxMergeDocs);
                // sum up docs in bucket
                List indexesToMerge = new ArrayList();
                int mergeDocs = 0;
                for (Iterator it = bucket.iterator(); it.hasNext() && mergeDocs <= targetMergeDocs;) {
                    indexesToMerge.add(it.next());
                }
                if (indexesToMerge.size() > 2) {
                    // found merge
                    Index[] idxs = (Index[]) indexesToMerge.toArray(new Index[indexesToMerge.size()]);
                    bucket.removeAll(indexesToMerge);
                    if (log.isDebugEnabled()) {
                        log.debug("requesting merge for " + indexesToMerge);
                    }
                    mergeTasks.add(new Merge(idxs));
                    log.debug("merge queue now contains " + mergeTasks.size() + " tasks.");
                }
            }
        }
    }

    /**
     * @inheritDoc
     */
    public void documentDeleted(Term id) {
        log.debug("document deleted: " + id.text());
        deletedDocuments.add(id);
    }

    /**
     * When the calling thread returns this index merger will be idle, that is
     * there will be no merge tasks pending anymore. The method returns immediately
     * if there are currently no tasks pending at all.
     */
    void waitUntilIdle() throws InterruptedException {
        mergerIdle.acquire();
        // and immediately release again
        mergerIdle.release();
    }

    /**
     * Signals this <code>IndexMerger</code> to stop and waits until it
     * has terminated.
     */
    void dispose() {
        log.debug("dispose IndexMerger");
        // get mutex for index replacements
        try {
            indexReplacement.acquire();
        } catch (InterruptedException e) {
            log.warn("Interrupted while acquiring index replacement sync: " + e);
            // try to stop IndexMerger without the sync
        }

        // clear task queue
        mergeTasks.clear();

        // send quit
        mergeTasks.add(QUIT);
        log.debug("quit sent");

        try {
            // give the merger thread some time to quit,
            // it is possible that the merger is busy working on a large index.
            // if that is the case we will just ignore it and the deamon will
            // die without being able to finish the merge.
            // at this point it is not possible anymore to replace indexes
            // on the MultiIndex because we hold the indexReplacement Sync.
            this.join(500);
            if (isAlive()) {
                log.info("Unable to stop IndexMerger. Deamon is busy.");
            } else {
                log.debug("IndexMerger thread stopped");
            }
            log.debug("merge queue size: " + mergeTasks.size());
        } catch (InterruptedException e) {
            log.warn("Interrupted while waiting for IndexMerger thread to terminate.");
        }