Lucene example source code file (ParallelMultiSearcher.java)

This example Lucene source code file (ParallelMultiSearcher.java) is included in the DevDaily.com "Java Source Code Warehouse" project. The intent of this project is to help you "Learn Java by Example" ^TM.

Java - Lucene tags/keywords

documentfrequencycallable, executionhelper, executionhelper, filter, hashmap, integer, io, ioexception, ioexception, override, override, searchable, searchable, term, threading, threads, topdocs, util

The Lucene ParallelMultiSearcher.java source code

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.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.Term;
import org.apache.lucene.util.NamedThreadFactory;
import org.apache.lucene.util.ThreadInterruptedException;

/** Implements parallel search over a set of <code>Searchables.
 *
 * <p>Applications usually need only call the inherited {@link #search(Query,int)}
 * or {@link #search(Query,Filter,int)} methods.
 * 
 * @deprecated Please pass an ExecutorService to {@link
 * IndexSearcher}, instead.
 */
@Deprecated
public class ParallelMultiSearcher extends MultiSearcher {
  private final ExecutorService executor;
  private final Searchable[] searchables;
  private final int[] starts;

  /** Creates a {@link Searchable} which searches <i>searchables with the default 
   * executor service (a cached thread pool). */
  public ParallelMultiSearcher(Searchable... searchables) throws IOException {
    this(Executors.newCachedThreadPool(new NamedThreadFactory(ParallelMultiSearcher.class.getSimpleName())), searchables);
  }

  /**
   * Creates a {@link Searchable} which searches <i>searchables with the specified ExecutorService.
   */
  public ParallelMultiSearcher(ExecutorService executor, Searchable... searchables) throws IOException {
    super(searchables);
    this.searchables = searchables;
    this.starts = getStarts();
    this.executor = executor;
  }
  /**
   * Executes each {@link Searchable}'s docFreq() in its own thread and waits for each search to complete and merge
   * the results back together.
   */
  @Override
  public int docFreq(final Term term) throws IOException {
    final ExecutionHelper<Integer> runner = new ExecutionHelper(executor);
    for(int i = 0; i < searchables.length; i++) {
      final Searchable searchable = searchables[i];
      runner.submit(new Callable<Integer>() {
        public Integer call() throws IOException {
          return Integer.valueOf(searchable.docFreq(term));
        }
      });
    }
    int docFreq = 0;
    for (Integer num : runner) {
      docFreq += num.intValue();
    }
    return docFreq;
  }

  /**
   * A search implementation which executes each 
   * {@link Searchable} in its own thread and waits for each search to complete and merge
   * the results back together.
   */
  @Override
  public TopDocs search(Weight weight, Filter filter, int nDocs) throws IOException {
    final HitQueue hq = new HitQueue(nDocs, false);
    final Lock lock = new ReentrantLock();
    final ExecutionHelper<TopDocs> runner = new ExecutionHelper(executor);
    
    for (int i = 0; i < searchables.length; i++) { // search each searchable
      runner.submit(
          new MultiSearcherCallableNoSort(lock, searchables[i], weight, filter, nDocs, hq, i, starts));
    }

    int totalHits = 0;
    float maxScore = Float.NEGATIVE_INFINITY;
    for (final TopDocs topDocs : runner) {
      totalHits += topDocs.totalHits;
      maxScore = Math.max(maxScore, topDocs.getMaxScore());
    }

    final ScoreDoc[] scoreDocs = new ScoreDoc[hq.size()];
    for (int i = hq.size() - 1; i >= 0; i--) // put docs in array
      scoreDocs[i] = hq.pop();

    return new TopDocs(totalHits, scoreDocs, maxScore);
  }

  /**
   * A search implementation allowing sorting which spans a new thread for each
   * Searchable, waits for each search to complete and merges
   * the results back together.
   */
  @Override
  public TopFieldDocs search(Weight weight, Filter filter, int nDocs, Sort sort) throws IOException {
    if (sort == null) throw new NullPointerException();

    final FieldDocSortedHitQueue hq = new FieldDocSortedHitQueue(nDocs);
    final Lock lock = new ReentrantLock();
    final ExecutionHelper<TopFieldDocs> runner = new ExecutionHelper(executor);
    for (int i = 0; i < searchables.length; i++) { // search each searchable
      runner.submit(
          new MultiSearcherCallableWithSort(lock, searchables[i], weight, filter, nDocs, hq, sort, i, starts));
    }
    int totalHits = 0;
    float maxScore = Float.NEGATIVE_INFINITY;
    for (final TopFieldDocs topFieldDocs : runner) {
      totalHits += topFieldDocs.totalHits;
      maxScore = Math.max(maxScore, topFieldDocs.getMaxScore());
    }
    final ScoreDoc[] scoreDocs = new ScoreDoc[hq.size()];
    for (int i = hq.size() - 1; i >= 0; i--) // put docs in array
      scoreDocs[i] = hq.pop();

    return new TopFieldDocs(totalHits, scoreDocs, hq.getFields(), maxScore);
  }

  /** Lower-level search API.
  *
  * <p>{@link Collector#collect(int)} is called for every matching document.
  *
  * <p>Applications should only use this if they need all of the
  * matching documents.  The high-level search API ({@link
  * Searcher#search(Query,int)}) is usually more efficient, as it skips
  * non-high-scoring hits.
  * 
  * <p>This method cannot be parallelized, because {@link Collector}
  * supports no concurrent access.
  *
  * @param weight to match documents
  * @param filter if non-null, a bitset used to eliminate some documents
  * @param collector to receive hits
  */
  @Override
  public void search(final Weight weight, final Filter filter, final Collector collector)
   throws IOException {
   for (int i = 0; i < searchables.length; i++) {

     final int start = starts[i];

     final Collector hc = new Collector() {
       @Override
       public void setScorer(final Scorer scorer) throws IOException {
         collector.setScorer(scorer);
       }
       
       @Override
       public void collect(final int doc) throws IOException {
         collector.collect(doc);
       }
       
       @Override
       public void setNextReader(final IndexReader reader, final int docBase) throws IOException {
         collector.setNextReader(reader, start + docBase);
       }
       
       @Override
       public boolean acceptsDocsOutOfOrder() {
         return collector.acceptsDocsOutOfOrder();
       }
     };
     
     searchables[i].search(weight, filter, hc);
   }
  }

  @Override
  public void close() throws IOException {
    executor.shutdown();
    super.close();
  }

  @Override
  HashMap<Term, Integer> createDocFrequencyMap(Set terms) throws IOException {
    final Term[] allTermsArray = terms.toArray(new Term[terms.size()]);
    final int[] aggregatedDocFreqs = new int[terms.size()];
    final ExecutionHelper<int[]> runner = new ExecutionHelper(executor);
    for (Searchable searchable : searchables) {
      runner.submit(
          new DocumentFrequencyCallable(searchable, allTermsArray));
    }
    final int docFreqLen = aggregatedDocFreqs.length;
    for (final int[] docFreqs : runner) {
      for(int i=0; i < docFreqLen; i++){
        aggregatedDocFreqs[i] += docFreqs[i];
      }
    }

    final HashMap<Term,Integer> dfMap = new HashMap();
    for(int i=0; i<allTermsArray.length; i++) {
      dfMap.put(allTermsArray[i], Integer.valueOf(aggregatedDocFreqs[i]));
    }
    return dfMap;
  }

  
  /**
   * A {@link Callable} to retrieve the document frequencies for a Term array  
   */
  private static final class DocumentFrequencyCallable implements Callable<int[]> {
    private final Searchable searchable;
    private final Term[] terms;
    
    public DocumentFrequencyCallable(Searchable searchable, Term[] terms) {
      this.searchable = searchable;
      this.terms = terms;
    }
    
    public int[] call() throws Exception {
      return searchable.docFreqs(terms);
    }
  }
  
  /**
   * A helper class that wraps a {@link CompletionService} and provides an
   * iterable interface to the completed {@link Callable} instances.
   * 
   * @param <T>
   *          the type of the {@link Callable} return value
   */
  private static final class ExecutionHelper<T> implements Iterator, Iterable {
    private final CompletionService<T> service;
    private int numTasks;

    ExecutionHelper(final Executor executor) {
      this.service = new ExecutorCompletionService<T>(executor);
    }

    public boolean hasNext() {
      return numTasks > 0;
    }

    public void submit(Callable<T> task) {
      this.service.submit(task);
      ++numTasks;
    }

    public T next() {
     if(!this.hasNext())
       throw new NoSuchElementException();
      try {
        return service.take().get();
      } catch (InterruptedException e) {
        throw new ThreadInterruptedException(e);
      } catch (ExecutionException e) {
        throw new RuntimeException(e);
      } finally {
        --numTasks;
      }
    }

    public void remove() {
      throw new UnsupportedOperationException();
    }

    public Iterator<T> iterator() {
      // use the shortcut here - this is only used in a privat context
      return this;
    }

  }
}