Revision: 14574
http://gate.svn.sourceforge.net/gate/?rev=14574&view=rev
Author: valyt
Date: 2011-11-18 15:42:20 +0000 (Fri, 18 Nov 2011)
Log Message:
-----------
[My head hurts. Not being selfish, I'm checking this in so others can enjoy the
sensation ;) ]
More work done on the ranking query runner:
- background/foreground thread logic now designed
- document IDs are being collected
- most of the major new public API implemented.
Still TODO:
- actual hits collection
- the rest of the public API (direct copy from old implementation?)
Modified Paths:
--------------
mimir/trunk/mimir-core/src/gate/mimir/search/RankingQueryRunnerImpl.java
Modified:
mimir/trunk/mimir-core/src/gate/mimir/search/RankingQueryRunnerImpl.java
===================================================================
--- mimir/trunk/mimir-core/src/gate/mimir/search/RankingQueryRunnerImpl.java
2011-11-18 15:39:17 UTC (rev 14573)
+++ mimir/trunk/mimir-core/src/gate/mimir/search/RankingQueryRunnerImpl.java
2011-11-18 15:42:20 UTC (rev 14574)
@@ -15,71 +15,161 @@
*/
package gate.mimir.search;
-import gate.mimir.index.IndexException;
import gate.mimir.search.query.Binding;
import gate.mimir.search.query.QueryExecutor;
import gate.mimir.search.query.QueryNode;
import gate.mimir.search.score.MimirScorer;
-
import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
import it.unimi.dsi.fastutil.doubles.DoubleList;
-import it.unimi.dsi.fastutil.ints.Int2IntFunction;
-import it.unimi.dsi.fastutil.ints.IntAVLTreeSet;
import it.unimi.dsi.fastutil.ints.IntArrayList;
-import it.unimi.dsi.fastutil.ints.IntIterator;
import it.unimi.dsi.fastutil.ints.IntList;
-import it.unimi.dsi.fastutil.ints.IntSortedSet;
+import it.unimi.dsi.fastutil.objects.Object2ObjectAVLTreeMap;
import it.unimi.dsi.fastutil.objects.ObjectArrayList;
import it.unimi.dsi.fastutil.objects.ObjectList;
import java.io.IOException;
-import java.io.Serializable;
+import java.util.Comparator;
import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.SortedMap;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.LinkedBlockingQueue;
import org.apache.log4j.Logger;
/**
- * A QueryRunner implementation that does ranking.
+ * A QueryRunner implementation that can perform ranking.
+ * This query runner has two modes of functioning: ranking and non-ranking,
+ * depending on whether a {@link MimirScorer} is provided during construction
+ * or not.
*/
-public class RankingQueryRunnerImpl implements Runnable {
+public class RankingQueryRunnerImpl {
+ private static final Runnable NO_MORE_JOBS = new Runnable(){
+ public void run() {}
+ };
+
+ protected class BackgroundRunner implements Runnable {
+ @Override
+ public void run() {
+ try {
+ while(true) {
+ Runnable job = backgroundTasks.take();
+ if(job == NO_MORE_JOBS) break;
+ else job.run();
+ }
+ } catch(InterruptedException e) {
+ Thread.currentThread().interrupt();
+ e.printStackTrace();
+ }
+ }
+ }
+
+
/**
- * When doing ranking, this class is used to delegate the iteration of
- * document IDs to an IntIterator that is limited to a finite set of
documents
- * that have just been ranked. When that iterator is emptied, the
- * {@link RankingQueryRunnerImpl#documentsById} field is nullified before
- * hasNext() returns false, to indicate that more documents may be available.
- */
- protected class RankedDocIdIterator implements IntIterator {
+ * Collects the document hits (i.e. {@link Binding}s) for the documents
+ * between the two provided ranks (indexes in the {@link #documentsOrder}
+ * list. If ranking is not being performed ( {@link #documentsOrder} is
+ * <code>null</null>, then the indexes are used against the
+ * {@link #documentIds} list.
+ *
+ * This is the only actor that writes to the {@link #documentHits} list.
+ */
+ protected class HitsCollector implements Runnable {
+ int start;
+ int end;
- public RankedDocIdIterator(IntIterator underlyingIterator) {
- this.underlyingIterator = underlyingIterator;
+ public HitsCollector(int rangeStart, int rangeEnd) {
+ this.start = rangeStart;
+ this.end = rangeEnd;
}
-
- protected IntIterator underlyingIterator;
-
- public boolean hasNext() {
- return underlyingIterator.hasNext();
+
+ @Override
+ public void run() {
+ // TODO Auto-generated method stub
+ if(documentScores != null) {
+ // we're ranking -> first calculate the range of documents in ID order
+
+ }
+
}
-
- public Integer next() {
- return underlyingIterator.next();
+ }
+
+
+ /**
+ * The first action started when a new {@link RankingQueryRunnerImpl} is
+ * created. It performs the following actions:
+ * <ul>
+ * <li>collects all document IDs in
+ * {@link RankingQueryRunnerImpl#documentIds}</li>
+ * <li>if ranking enabled
+ * <ul>
+ * <li>it collects all document scores
+ * </ul>
+ * </li>
+ * <li>if ranking not enabled
+ * <ul>
+ * <li>it collects the document hits for the first
+ * block of documents</li>
+ * </ul>
+ * </li>
+ * <li>If ranking enabled, after all document IDs are obtained, it starts
+ * the work for ranking the first block of documents (which, upon
+ * completion, will also start a background job to collect all the hits
for
+ * that block).</li>
+ * </ul>
+ */
+ protected class DocIdsCollector implements Runnable {
+ @Override
+ public void run() {
+ try{
+ // collect all documents and their scores
+ final boolean scoring = scorer != null;
+ if(scoring) scorer.wrap(queryExecutor);
+ int docId = scoring ? scorer.nextDocument(-1) :
queryExecutor.nextDocument(-1);
+ if(!scoring) { // then also collect some hits
+ synchronized(hitCollectors) {
+ hitCollectors.put(new int[]{0, docBlockSize},
+ new FutureTask<Object>(this, null));
+ }
+ }
+ while(docId >= 0) {
+ documentIds.add(docId);
+ if(scoring){
+ documentScores.add(scorer.score());
+ documentHits.add(null);
+ } else {
+ // not scoring: also collect the hits for the first block of
documents
+ if(docId < docBlockSize) {
+ ObjectList<Binding> hits = new ObjectArrayList<Binding>();
+ Binding hit = queryExecutor.nextHit();
+ while(hit != null) {
+ hits.add(hit);
+ hit = queryExecutor.nextHit();
+ }
+ documentHits.add(hits);
+ } else {
+ documentHits.add(null);
+ }
+ }
+ docId = scoring ? scorer.nextDocument(-1) :
queryExecutor.nextDocument(-1);
+ }
+ allDocIdsCollected = true;
+ if(scoring) {
+ // now rank the first batch of documents
+ // this will also start a second background job to collect the hits
+ rankDocuments(queryExecutor.getQueryEngine().getRankingDocCount()
-1);
+ }
+ }catch (IOException e) {
+ logger.error("Exception while collecting document IDs", e);
+ try {
+ close();
+ } catch(IOException e1) {
+ logger.error("Exception while closing, after exception.", e1);
+ }
+ }
}
-
- public void remove() {
- underlyingIterator.remove();
- }
-
- public int nextInt() {
- return underlyingIterator.nextInt();
- }
-
- public int skip(int n) {
- return underlyingIterator.skip(n);
- }
-
}
protected Logger logger = Logger.getLogger(RankingQueryRunnerImpl.class);
@@ -95,6 +185,12 @@
protected MimirScorer scorer;
/**
+ * The number of documents to be ranked (of have their hits collected) as a
+ * block.
+ */
+ protected int docBlockSize;
+
+ /**
* The document IDs for the documents found to contain hits. This list is
* sorted in ascending documentID order.
*/
@@ -111,8 +207,7 @@
* The sets of hits for each returned document. This data structure is
lazily
* built, so some elements may be null.
*/
- protected ObjectList<Binding[]> documentHits;
-
+ protected ObjectList<List<Binding>> documentHits;
/**
* The order the documents should be returned in (elements in this list are
@@ -121,18 +216,25 @@
protected IntList documentsOrder;
/**
- * An iterator supplying documentIDs in ascending order. These are used when
- * collecting the hits.
+ * Data structure holding references to {@link Future}s that are currently
+ * working (or have worked) on collecting hits for a range of document
+ * indexes.
*/
- protected IntIterator documentsById;
+ protected SortedMap<int[], Future<?>> hitCollectors;
/**
- * The thread used for executing the query. This is a separate thread from
one
- * that created the query runner.
+ * The background thread used for collecting hits.
*/
protected Thread runningThread;
/**
+ * A queue with tasks to be executed by the background thread.
+ */
+ protected BlockingQueue<Runnable> backgroundTasks;
+
+ protected volatile boolean allDocIdsCollected = false;
+
+ /**
* Creates a query runner in ranking mode.
* @param qNode the {@link QueryNode} for the query being executed.
* @param scorer the {@link MimirScorer} to use for ranking.
@@ -142,36 +244,45 @@
public RankingQueryRunnerImpl(QueryExecutor executor, MimirScorer scorer)
throws IOException {
this.queryExecutor = executor;
this.scorer = scorer;
- documentsById = null;
- // start the search
- getMoreHits();
- }
-
-
- /* (non-Javadoc)
- * @see gate.mimir.search.QueryRunner#getMoreHits()
- */
- protected synchronized void getMoreHits() throws IOException {
- if(runningThread != null){
- //we're already running -> ignore
- return;
+ docBlockSize = queryExecutor.getQueryEngine().getRankingDocCount();
+ documentIds = new IntArrayList();
+ documentHits = new ObjectArrayList<List<Binding>>();
+ if(scorer != null) {
+ documentScores = new DoubleArrayList();
+ documentsOrder = new IntArrayList(
+ queryExecutor.getQueryEngine().getRankingDocCount());
}
- // get a thread from the executor, if one exists
+ hitCollectors = new Object2ObjectAVLTreeMap<int[], Future<?>>(
+ new Comparator<int[]>(){
+ @Override
+ public int compare(int[] o1, int[] o2) { return o1[0] - o2[0]; }
+ });
+ // start the background thread
+ backgroundTasks = new LinkedBlockingQueue<Runnable>();
+ Runnable backgroundRunner = new BackgroundRunner();
+ //get a thread from the executor, if one exists
if(queryExecutor.getQueryEngine().getExecutor() != null){
- queryExecutor.getQueryEngine().getExecutor().execute(this);
+ queryExecutor.getQueryEngine().getExecutor().execute(backgroundRunner);
}else{
- new Thread(this, getClass().getName()).start();
+ new Thread(backgroundRunner, getClass().getName()).start();
}
+
+ // queue a job for collecting all document ids
+ try {
+ backgroundTasks.put(new DocIdsCollector());
+ } catch(InterruptedException e) {
+ Thread.currentThread().interrupt();
+ logger.error("Could not queue a background task.", e);
+ }
}
-
/**
- * Gets the number of documents found to contain hits. If the search has not
- * yet completed, then -1 is returned.
+ * Gets the number of result documents. If the search has not yet completed,
+ * then -1 is returned.
* @return
*/
public int getDocumentsCount() {
- if(queryExecutor == null) return documentIds.size();
+ if(allDocIdsCollected) return documentIds.size();
else return -1;
}
@@ -186,57 +297,99 @@
}
/**
- * Gets the ID of a document found to contain hits.
- * @param index the index of the desired document in the list of documents.
+ * Gets the ID of a result document.
+ * @param rank the index of the desired document in the list of documents.
* This should be a value between 0 and {@link #getDocumentsCount()} -1.
*
+ * If the requested document position has not yet been ranked (i.e. we know
+ * there is a document at that position, but we don't yet know which one)
then
+ * the necessary ranking is performed before this method returns.
+ *
* @return an int value, representing the ID of the requested document.
* @throws IndexOutOfBoundsException is the index provided is less than
zero,
* or greater than {@link #getDocumentsCount()} -1.
+ * @throws IOException
*/
- public int getDocumentID(int index) throws IndexOutOfBoundsException {
- // TODO: check position index has been ranked yet
+ public int getDocumentID(int rank) throws IndexOutOfBoundsException,
IOException {
+ return documentIds.getInt(getDocumentIndex(rank));
+ }
+
+ /**
+ * Retrieves the hits withing a given result document.
+ * @param rank the index of the desired document in the list of documents.
+ * This should be a value between 0 and {@link #getDocumentsCount()} -1.
+ *
+ * This method call waits until the requested data is available before
+ * returning (document hits are being collected by a background thread).
+ *
+ * @return
+ * @throws IOException
+ * @throws IndexOutOfBoundsException
+ */
+ public List<Binding> getDocumentHits(int rank) throws
IndexOutOfBoundsException, IOException {
+ int documentIndex = getDocumentIndex(rank);
+ List<Binding> hits = documentHits.get(documentIndex);
+ if(hits == null) {
+ // hits not collected yet
+ try {
+ // find the Future working on it, or start a new one,
+ // then wait for it to complete
+ collectHits(new int[]{documentIndex, documentIndex}).get();
+ hits = documentHits.get(documentIndex);
+ } catch(Exception e) {
+ logger.error("Exception while waiting for hits collection", e);
+ throw new RuntimeException(
+ "Exception while waiting for hits collection", e);
+ }
+ }
+ return hits;
+ }
+
+ /**
+ * Given a document rank, return its index in the {@link #documentIds} list.
+ * @param rank
+ * @return
+ * @throws IOException, IndexOutOfBoundsException
+ */
+ protected int getDocumentIndex(int rank) throws IOException,
IndexOutOfBoundsException {
+ int maxIndex = documentIds.size();
+ if(rank >= maxIndex) throw new IndexOutOfBoundsException(
+ "Document index too large (" + rank + " > " + maxIndex + ".");
if(documentsOrder != null) {
- return documentIds.getInt(documentsOrder.getInt(index));
+ // we're in ranking mode
+ if(rank >= documentsOrder.size()) {
+ // document exists, but has not been ranked yet
+ rankDocuments(rank);
+ }
+ return documentsOrder.getInt(rank);
} else {
- return documentIds.getInt(index);
+ return rank;
}
}
/**
- * Creates an {@link IntIterator} used for enumerating the documents in the
- * correct order for returning to the user.
- * If no more documents are available, this should return null
- * @return
+ * Ranks some more documents (i.e. adds more entries to the
+ * {@link #documentsOrder} list, making sure that the document at provided
+ * index is included.
+ * This is the only method that writes to the {@link #documentsOrder} list.
+ * This method is executed synchronously in the client thread.
+ *
+ * @param index
* @throws IOException
*/
- protected IntIterator getDocumentIterator() throws IOException {
- if(scorer != null) {
- // we're doing ranking
- if(documentIds == null) {
- // first stage: collect all documents and their scores
- documentIds = new IntArrayList();
- documentScores = new DoubleArrayList();
- documentHits = new ObjectArrayList<Binding[]>();
- documentsOrder = new IntArrayList(
- queryExecutor.getQueryEngine().getRankingDocCount());
-
- scorer.wrap(queryExecutor);
- int docId = scorer.nextDocument(-1);
- while(docId >= 0) {
- documentIds.add(docId);
- documentScores.add(scorer.score());
- documentHits.add(null);
- docId = scorer.nextDocument(-1);
- }
+ protected void rankDocuments(int index) throws IOException {
+ synchronized(documentsOrder) {
+ // rank some documents
+ int rankRangeStart = documentsOrder.size();
+ int rankRangeEnd = index;
+ if(rankRangeEnd - rankRangeStart <
+ queryExecutor.getQueryEngine().getRankingDocCount()) {
+ // extend the size of the chunk of documents to be ranked
+ rankRangeEnd = rankRangeStart +
+ queryExecutor.getQueryEngine().getRankingDocCount();
}
- // collect some more ranked documents
+ int documentsOrderWriteIndex = rankRangeStart;
- int rankRangeStart = documentsOrder.size();
- int rankRangeEnd = documentsOrder.size() +
- queryExecutor.getQueryEngine().getRankingDocCount();
- int docsByRankWriteIndex = rankRangeStart;
-
// the document with the minimum score already ranked.
int smallestOldScoreDocId = rankRangeStart > 0 ?
documentIds.getInt(documentsOrder.getInt(rankRangeStart -1))
@@ -245,8 +398,6 @@
double smallestOldScore = rankRangeStart > 0 ?
documentScores.getDouble(documentsOrder.getInt(rankRangeStart -1))
: -1;
- // the documentIds for newly ranked documents
- IntSortedSet newDocuments = new IntAVLTreeSet();
// now collect some more documents
for(int i = 0; i < documentIds.size(); i++) {
int documentId = documentIds.getInt(i);
@@ -264,18 +415,17 @@
// smaller score than the maximum permitted score (i.e. it has not
// already been ranked)., or
// - it's a new document with the same score as the largest permitted
score
- if(docsByRankWriteIndex < rankRangeEnd
+ if(documentsOrderWriteIndex < rankRangeEnd
||
(documentScore > smallestNewScore &&
(smallestOldScore < 0 || documentScore < smallestOldScore))
||
documentScore == smallestOldScore && documentId !=
smallestOldScoreDocId) {
- if(docsByRankWriteIndex == rankRangeEnd) {
+ if(documentsOrderWriteIndex == rankRangeEnd) {
// we need to remove the newly ranked document
// with the smallest score
- docsByRankWriteIndex--;
- int oldDocIndex = documentsOrder.removeInt(docsByRankWriteIndex);
- newDocuments.remove(documentIds.getInt(oldDocIndex));
+ documentsOrderWriteIndex--;
+ documentsOrder.removeInt(documentsOrderWriteIndex);
}
// find the rank for the new doc
int rank = rankRangeStart;
@@ -284,46 +434,67 @@
rank++;
}
documentsOrder.add(rank, i);
- newDocuments.add(documentId);
- docsByRankWriteIndex++;
+ documentsOrderWriteIndex++;
}
}
- if(newDocuments.isEmpty()){
- return null;
- } else {
- return new RankedDocIdIterator(newDocuments.iterator());
+ // start collecting the hits for the newly ranked documents (in a new
thread)
+ if(documentsOrderWriteIndex > rankRangeStart){
+ collectHits(new int[] {rankRangeStart, documentsOrderWriteIndex});
}
- } else {
- // we're not doing scoring, simply use the queryExecutor as an
intIterator
- return queryExecutor;
}
}
- public void run() {
- //store the running thread
- synchronized(this) {
- if(runningThread != null){
- //some task is already running
- return;
+ /**
+ * Makes sure all the documents in the specified range are queued for hit
+ * collection.
+ * @param interval the interval specified by 2 document ranks
+ * @return the future that has been queued for collecting the hits.
+ */
+ protected Future collectHits(int[] interval) {
+ // expand the interval to block size
+ if(interval[1] - interval[0] < docBlockSize) {
+ interval[0] -= docBlockSize / 2;
+ interval[1] += docBlockSize / 2;
+ }
+ HitsCollector hitsCollector = null;
+ Future<?> future;
+ synchronized(hitCollectors) {
+ SortedMap<int[], Future<?>> headMap = hitCollectors.headMap(interval);
+ int[] previousInterval = headMap.isEmpty() ? new int[]{0,0} :
+ headMap.lastKey();
+ if(previousInterval[1] >= interval[1]) {
+ // we're part of previous interval
+ future = hitCollectors.get(previousInterval);
+ } else {
+ // calculate an appropriate interval to collect hits for
+ SortedMap<int[], Future<?>> tailMap = hitCollectors.tailMap(interval);
+ int[] followingInterval = tailMap.isEmpty() ?
+ new int[]{Integer.MAX_VALUE, Integer.MAX_VALUE} : tailMap.firstKey();
+ int start = Math.max(previousInterval[1], interval[0]);
+ int end = Math.min(followingInterval[0], interval[1]);
+ hitsCollector = new HitsCollector(start, end);
+ future = new FutureTask(hitsCollector, null);
+ hitCollectors.put(new int[]{start, end}, future);
}
- runningThread = Thread.currentThread();
}
- try {
- if(documentsById == null) {
- documentsById = getDocumentIterator();
+ if(hitsCollector != null) {
+ try {
+ backgroundTasks.put(hitsCollector);
+ } catch(InterruptedException e) {
+ logger.error("Error while queuing background work", e);
+ throw new RuntimeException("Error while queuing background work", e);
}
- // collect the hits
- while(documentsById != null) {
-
- }
- } catch(IOException e) {
- //something went bad!
- logger.error("IOException during search!", e);
- }finally{
- //this search stage has completed -> clear the running thread
- synchronized(this) {
- runningThread = null;
- }
- }
+ }
+ return future;
}
-}
+
+ public void close() throws IOException {
+ queryExecutor.close();
+ scorer = null;
+ try {
+ backgroundTasks.put(NO_MORE_JOBS);
+ } catch(InterruptedException e) {
+ // ignore
+ }
+ }
+}
\ No newline at end of file
This was sent by the SourceForge.net collaborative development platform, the
world's largest Open Source development site.
------------------------------------------------------------------------------
All the data continuously generated in your IT infrastructure
contains a definitive record of customers, application performance,
security threats, fraudulent activity, and more. Splunk takes this
data and makes sense of it. IT sense. And common sense.
http://p.sf.net/sfu/splunk-novd2d
_______________________________________________
GATE-cvs mailing list
[email protected]
https://lists.sourceforge.net/lists/listinfo/gate-cvs
