[HER-1827] Have Frontier report include top X longest queues

* Histotable.java
    expose static method useful for frequency-reporting
* TopNSet.java, TopNSet.java
    improve handling of case where updates may decrease frequency counts
    add convenience methods for common needs
* WorkQueueFrontier.java
    replace single longestActiveQueue with largestQueues TopNSet, of configurable size
    make max-queues-to-report-per-category configurable
This commit is contained in:
gojomo committed 2010-09-28 22:12:28 +00:00
1 parent 74d3fb54ce
commit 8f28d493a2
4 files changed
+213 -70

No files matched your search

@@ -56,6 +56,7 @@ import org.archive.crawler.event.CrawlURIDispositionEvent;
import org.archive.crawler.framework.ToeThread;
import org.archive.crawler.frontier.precedence.BaseQueuePrecedencePolicy;
import org.archive.crawler.frontier.precedence.QueuePrecedencePolicy;
import org.archive.crawler.util.TopNSet;
import org.archive.modules.CrawlURI;
import org.archive.spring.KeyedProperties;
import org.archive.util.ArchiveUtils;
@@ -85,9 +86,6 @@ implements Closeable,
ApplicationContextAware {
private static final long serialVersionUID = 570384305871965843L;
/** truncate reporting of queues at some large but not unbounded number */
private static final int REPORT_MAX_QUEUES = 2000;
/**
* If we know that only a small amount of queues is held in memory,
* we can avoid using a disk-based BigMap.
@@ -189,7 +187,14 @@ implements Closeable,
this.precedenceFloor = floor;
}
/** truncate reporting of queues at this large but not unbounded number */
protected int maxQueuesPerReportCategory = 2000;
public int getMaxQueuesPerReportCategory() {
return this.maxQueuesPerReportCategory;
}
public void setMaxQueuesPerReportCategory(int max) {
this.maxQueuesPerReportCategory = max;
}
/** All known queues.
*/
@@ -217,8 +222,19 @@ implements Closeable,
/** URIs scheduled to be re-enqueued at future date */
protected StoredSortedMap<Long, CrawlURI> futureUris;
transient protected WorkQueue longestActiveQueue = null;
/** remember keys of small number of largest queues for reporting */
transient protected TopNSet largestQueues = new TopNSet(20);
/** remember this many largest queues for reporting's sake; actual tracking
* can be somewhat approximate when some queues shrink before others'
* sizes are again noted, or if the size is adjusted mid-crawl. */
public int getLargestQueuesCount() {
return largestQueues.getMaxSize();
}
public void setLargestQueuesCount(int count) {
largestQueues.setMaxSize(count);
}
protected int highestPrecedenceWaiting = Integer.MAX_VALUE;
/** The UriUniqFilter to use, tracking those UURIs which are
@@ -436,11 +452,7 @@ implements Closeable,
readyQueue(wq);
}
}
WorkQueue laq = longestActiveQueue;
if(((laq==null) || wq.getCount() > laq.getCount())) {
longestActiveQueue = wq;
}
largestQueues.update(wq.getClassKey(), wq.getCount());
}
/**
@@ -1441,39 +1453,36 @@ implements Closeable,
w.print("\n -----===== MANAGER THREAD =====-----\n");
ToeThread.reportThread(managerThread, w);
WorkQueue longest = longestActiveQueue;
if (longest != null) {
w.print("\n -----===== LONGEST QUEUE =====-----\n");
longest.reportTo(w);
}
w.print("\n -----===== "+largestQueues.size()+" LONGEST QUEUES =====-----\n");
appendQueueReports(w, "LONGEST", largestQueues.getEntriesDescending().iterator(), largestQueues.size(), largestQueues.size());
w.print("\n -----===== IN-PROCESS QUEUES =====-----\n");
@SuppressWarnings("unchecked")
Collection<WorkQueue> inProcess = inProcessQueues;
ArrayList<WorkQueue> copy = extractSome(inProcess, REPORT_MAX_QUEUES);
appendQueueReports(w, "IN-PROCESS", copy.iterator(), copy.size(), REPORT_MAX_QUEUES);
ArrayList<WorkQueue> copy = extractSome(inProcess, maxQueuesPerReportCategory);
appendQueueReports(w, "IN-PROCESS", copy.iterator(), copy.size(), maxQueuesPerReportCategory);
w.print("\n -----===== READY QUEUES =====-----\n");
appendQueueReports(w, "READY", this.readyClassQueues.iterator(),
this.readyClassQueues.size(), REPORT_MAX_QUEUES);
this.readyClassQueues.size(), maxQueuesPerReportCategory);
w.print("\n -----===== SNOOZED QUEUES =====-----\n");
Object[] objs = snoozedClassQueues.toArray();
DelayedWorkQueue[] qs = Arrays.copyOf(objs,objs.length,DelayedWorkQueue[].class);
Arrays.sort(qs);
appendQueueReports(w, "SNOOZED", new ObjectArrayIterator(qs), getSnoozedCount(), REPORT_MAX_QUEUES);
appendQueueReports(w, "SNOOZED", new ObjectArrayIterator(qs), getSnoozedCount(), maxQueuesPerReportCategory);
w.print("\n -----===== INACTIVE QUEUES =====-----\n");
SortedMap<Integer,Queue<String>> sortedInactives = getInactiveQueuesByPrecedence();
for(Integer prec : sortedInactives.keySet()) {
Queue<String> inactiveQueues = sortedInactives.get(prec);
appendQueueReports(w, "INACTIVE-p"+prec, inactiveQueues.iterator(),
inactiveQueues.size(), REPORT_MAX_QUEUES);
inactiveQueues.size(), maxQueuesPerReportCategory);
}
w.print("\n -----===== RETIRED QUEUES =====-----\n");
appendQueueReports(w, "RETIRED", getRetiredQueues().iterator(),
getRetiredQueues().size(), REPORT_MAX_QUEUES);
getRetiredQueues().size(), maxQueuesPerReportCategory);
w.flush();
}
@@ -1513,6 +1522,7 @@ implements Closeable,
* @param total
* @param max
*/
@SuppressWarnings("unchecked")
protected void appendQueueReports(PrintWriter w, String label, Iterator<?> iterator,
int total, int max) {
Object obj;
@@ -1527,6 +1537,8 @@ implements Closeable,
q = (WorkQueue)obj;
} else if (obj instanceof DelayedWorkQueue) {
q = (WorkQueue)((DelayedWorkQueue)obj).getWorkQueue(this);
} else if (obj instanceof Map.Entry) {
q = this.allQueues.get((String)((Map.Entry)obj).getKey());
} else {
q = this.allQueues.get((String)obj);
}
@@ -1616,7 +1628,7 @@ implements Closeable,
return (float)(activeCount + eligibleInactiveCount) / (inProcessCount + snoozedCount);
}
public long deepestUri() {
return longestActiveQueue==null ? -1 : longestActiveQueue.getCount();
return largestQueues.getTopSet().size()==0 ? -1 : largestQueues.getTopSet().get(largestQueues.getLargest());
}
/**
@@ -1640,5 +1652,4 @@ implements Closeable,
return inProcessQueues.size();
}
} // TODO: slim class! Suspect it should be < 800 lines, shedding budgeting/reporting
} // TODO: slim class! Suspect it should be < 800 lines, shedding budgeting/reporting
@@ -21,12 +21,24 @@ package org.archive.crawler.util;
import java.io.Serializable;
import java.util.HashMap;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.Map.Entry;
import org.archive.util.Histotable;
/**
* Counting Set which only remembers the 'top N' of all String values
* reported (with counts) to it. Assumes counts reported for a
* certain String key only ever increase.
*
* reported (with counts) to it. Precise if counts reported for a
* certain String key only ever increase. Otherwise, the current top N
* might decrement to be smaller than other tallies, which won't be noticed
* until those other, previously ignored tallies, are re-reported.
*
* TODO: Histotable and TopNSet that they could possibly
* have a closer relationship and share some code (even though Histotable
* tracks 'all' keys, and handles increments, while TopNSet only remembers
* a small subset of keys, and requires a fresh full value on each update.)
*
* @contributor gojomo
*/
public class TopNSet implements Serializable {
@@ -37,53 +49,62 @@ public class TopNSet implements Serializable {
HashMap<String, Long> set;
long smallestKnownValue;
String smallestKnownKey;
long largestKnownValue;
String largestKnownKey;
public TopNSet(int size){
maxsize = size;
set = new HashMap<String, Long>(size);
}
/**
* Update the given String key with a new total value, perhaps displacing
* an existing top-valued entry, and updating the fields recording max/min
* keys in any case.
*
* @param key String key to update
* @param value long new total value (*not* increment/decrement)
*/
public synchronized void update(String key, long value){
if(set.containsKey(key)) {
// Update the value of an existing key
if(set.containsKey(key) || set.size() < maxsize || value > smallestKnownValue) {
set.put(key,value);
// This may promote the key if it was the smallest
if(smallestKnownKey == null || smallestKnownKey.equals(key)){
updateSmallest();
}
} else if(set.size()<maxsize) {
// Can add a new key/value pair as we still have space
set.put(key, value);
// Check if this is new smallest known value
if(value<smallestKnownValue){
smallestKnownValue = value;
smallestKnownKey = key;
}
} else {
// Determine if value is large enough for inclusion
if(value>smallestKnownValue){
// Replace current smallest
if(set.size() > maxsize) {
set.remove(smallestKnownKey);
updateSmallest();
set.put(key, value);
} // Else do nothing.
updateBounds();
} else if(value < smallestKnownValue
|| value > largestKnownValue
|| key.equals(smallestKnownKey)
|| key.equals(largestKnownKey)) {
updateBounds();
}
}
}
private void updateSmallest(){
// Need to scan through for new smallest value.
long oldSmallest = smallestKnownValue;
public String getLargest() {
return largestKnownKey;
}
public String getSmallest() {
return smallestKnownKey;
}
/**
* After an operation invalidating the previous largest/smallest entry,
* find the new largest/smallest.
*/
protected void updateBounds(){
// freshly determine
smallestKnownValue = Long.MAX_VALUE;
largestKnownValue = Long.MIN_VALUE;
for(String k : set.keySet()){
long v = set.get(k);
if(v<smallestKnownValue){
smallestKnownValue = v;
smallestKnownKey = k;
if(v==oldSmallest){
// Found another key matching old smallest known value
// Can not be anything smaller.
return;
}
}
if(v>largestKnownValue){
largestKnownValue = v;
largestKnownKey = k;
}
}
}
@@ -95,4 +116,26 @@ public class TopNSet implements Serializable {
public HashMap<String,Long> getTopSet() {
return set;
}
public int size() {
return set.size();
}
/**
* Get descending ordered list of key,count Entries.
*
* @return SortedSet of Entry<key, count> descending-frequency
*/
public SortedSet<Entry<?, Long>> getEntriesDescending() {
TreeSet<Entry<?, Long>> sorted = Histotable.getEntryByFrequencySortedSet();
sorted.addAll(getTopSet().entrySet());
return sorted;
}
public int getMaxSize() {
return maxsize;
}
public void setMaxSize(int max) {
maxsize = max;
}
}