lee/heritrix3
1
0 Fork 0
mirror of https://github.com/internetarchive/heritrix3.git synced 2026-11-09 15:36:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove synchronization on frontier methods; rely on safe queues, synchronization-per-WorkQueue instead

Browse Source 
* AbstractFrontier
    remove inbound/outbound queues; simplify managerThread
    perform findEligible/schedule/receive/finish immediately
* WorkQueueFrontier
    eliminate holdQueues setting 
    synchronize sendToQueue on target queue
    make findEligibleUri reentrant; rely on [Blocking|Stored]Queue thread-safety
    make arriving at front of ready trigger for 'active'/budget-session
    simplify inactiveQueues management
    simplify waking from overflow
* WorkQueue
    redefine 'active' as in-budget-session
    rename 'held' as 'managed'
    synchronize major methods, relying on allQueues cache that operations on same intended queue use identical instance
    split over-budget test to isOverSessionBudget and isOverTotalBudget
    have toString show classKey
* BdbFrontier
    checkpoint fixes: remember nextOrdinal, rotate frontier-recover log, reset queues on recovery
    consistencyCheck method for probing state during stress tests
This commit is contained in:
gojomo committed 2010-11-15 22:56:03 +00:00
1 parent 881c829fe5
commit cb7f96682b
4 files changed
+325 -647
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java
+34 -280
View File
@@ -21,7 +21,6 @@ package org.archive.crawler.frontier;
import static org.archive.modules.CoreAttributeConstants.A_NONFATAL_ERRORS;
import static org.archive.modules.fetcher.FetchStatusCodes.S_BLOCKED_BY_CUSTOM_PROCESSOR;
import static org.archive.modules.fetcher.FetchStatusCodes.S_UNATTEMPTED;
import static org.archive.modules.fetcher.FetchStatusCodes.S_BLOCKED_BY_USER;
import static org.archive.modules.fetcher.FetchStatusCodes.S_CONNECT_FAILED;
import static org.archive.modules.fetcher.FetchStatusCodes.S_CONNECT_LOST;
@@ -32,6 +31,7 @@ import static org.archive.modules.fetcher.FetchStatusCodes.S_OUT_OF_SCOPE;
import static org.archive.modules.fetcher.FetchStatusCodes.S_ROBOTS_PRECLUDED;
import static org.archive.modules.fetcher.FetchStatusCodes.S_TOO_MANY_EMBED_HOPS;
import static org.archive.modules.fetcher.FetchStatusCodes.S_TOO_MANY_LINK_HOPS;
import static org.archive.modules.fetcher.FetchStatusCodes.S_UNATTEMPTED;
import java.io.BufferedReader;
import java.io.EOFException;
@@ -42,9 +42,6 @@ import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.util.Collection;
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
@@ -155,24 +152,6 @@ public abstract class AbstractFrontier
kp.put("independentExtractors", enabled);
}
/** inbound updates: URIs to be scheduled, finished; requested state changes */
protected BlockingQueue<InEvent> inbound;
public BlockingQueue<InEvent> getInbound() {
return inbound;
}
public void setInbound(BlockingQueue<InEvent> inbound) {
this.inbound = inbound;
}
/** outbound URIs */
protected BlockingQueue<CrawlURI> outbound;
public BlockingQueue<CrawlURI> getOutbound() {
return outbound;
}
public void setOutbound(BlockingQueue<CrawlURI> outbound) {
this.outbound = outbound;
}
public boolean isRunning() {
return managerThread!=null && managerThread.isAlive();
}
@@ -282,7 +261,7 @@ public abstract class AbstractFrontier
* Currently captures Frontier/URI transitions.
* Can be null if user chose not to run a recovery.log.
*/
private FrontierJournal recover = null;
protected FrontierJournal recover = null;
/**
* @param name Name of this frontier.
@@ -292,9 +271,6 @@ public abstract class AbstractFrontier
}
/** reusable no-op inbound event, to force reeval of state/eligible URIs */
protected InEvent NOOP = new InEvent() { public void process() {} };
/**
* lock to allow holding all worker ToeThreads from taking URIs already
* on the outbound queue; they acquire read permission before take()ing;
@@ -313,12 +289,11 @@ public abstract class AbstractFrontier
/** last Frontier.State reached; used to suppress duplicate notifications */
State lastReachedState = null;
/** Frontier.state that manager thread should seek to reach */
State targetState = State.PAUSE;
volatile State targetState = State.PAUSE;
/**
* Start the manager thread. Subclasses should invoke this in a custom
* readObject() method so that the manager thread begins after checkpoint
* recovery.
* Start the dedicated thread with an independent view of the frontier's
* state.
*/
protected void startManagerThread() {
managerThread = new Thread(this+".managerThread") {
@@ -340,13 +315,6 @@ public abstract class AbstractFrontier
} catch (IOException e) {
throw new IllegalStateException(e);
}
if(outbound==null) {
outbound = new ArrayBlockingQueue<CrawlURI>(200, true);
}
if(inbound==null) {
inbound = new ArrayBlockingQueue<InEvent>(4000, true);
}
pause();
startManagerThread();
}
@@ -372,10 +340,9 @@ public abstract class AbstractFrontier
outboundLock.writeLock().unlock();
}
reachedState(State.RUN);
// fill to-do 'on-deck' queue
fillOutbound();
// process discovered and finished URIs
drainInbound();
Thread.sleep(1000);
if(isEmpty()) {
// pause when frontier exhausted; controller will
// determine if this means to finish or not
@@ -390,29 +357,21 @@ public abstract class AbstractFrontier
outboundLock.writeLock().lock();
// process all inbound
while (targetState == State.PAUSE) {
if (outbound.size() == getInProcessCount()) {
// if all 'in-process' URIs are actually
// waiting in outbound, we are at PAUSE
if (getInProcessCount()==0) {
reachedState(State.PAUSE);
}
// continue to process discovered and finished URIs
InEvent ev = inbound.take();
synchronized(this) {
ev.process();
}
Thread.sleep(1000);
}
break;
case FINISH:
// prevent all outbound takes
outboundLock.writeLock().lock();
// process all inbound
while (outbound.size() != getInProcessCount()) {
// continue to process discovered and finished URIs
InEvent ev = inbound.take();
synchronized(this) {
ev.process();
}
while (getInProcessCount()>0) {
Thread.sleep(1000);
}
finalTasks();
// TODO: more cleanup?
reachedState(State.FINISH);
@@ -448,52 +407,6 @@ public abstract class AbstractFrontier
protected void finalTasks() {
// by default; nothing
}
/**
* Fill the outbound queue with eligible CrawlURIs, to capacity
* or as much as possible.
*
* @throws InterruptedException
*/
protected void fillOutbound() throws InterruptedException {
while (outbound.remainingCapacity() > 0) {
CrawlURI crawlable = findEligibleURI();
if (crawlable != null) {
outbound.put(crawlable);
} else {
break;
}
}
}
/**
* Drain the inbound queue of update events, or at the very least
* wait until some additional delayed-queue URI becomes available.
*
* @throws InterruptedException
*/
protected void drainInbound() throws InterruptedException {
int batch = inbound.size();
for(int i = 0; i < batch; i++) {
InEvent ev = inbound.poll();
if(ev==null) {
break;
}
synchronized(this) {
ev.process();
}
}
if(batch==0) {
// always do at least one timed try
InEvent ev = inbound.poll(getMaxInWait(),
TimeUnit.MILLISECONDS);
if (ev != null) {
synchronized(this) {
ev.process();
}
}
}
}
/**
* The given state has been reached; if it is a new state, generate
@@ -512,30 +425,14 @@ public abstract class AbstractFrontier
* @see org.archive.crawler.framework.Frontier#next()
*/
public CrawlURI next() throws InterruptedException {
// perhaps hold without taking ready outbound items
outboundLock.readLock().lockInterruptibly();
outboundLock.readLock().unlock();
CrawlURI retval = outbound.poll();
while(retval==null) {
CrawlURI crawlable = null;
while(crawlable==null) {
outboundLock.readLock().lockInterruptibly();
// try filling outbound until we get something to work on
CrawlURI crawlable = findEligibleURI();
if (crawlable != null) {
outbound.put(crawlable);
retval = outbound.poll();
} else {
// or if nothing ready, wait for other threads to fill for us
// (no busy spin)
retval = outbound.take();
}
crawlable = findEligibleURI();
outboundLock.readLock().unlock();
}
// // TODO: consider if following necessary for maintaining throughput
// if(outbound.size()<=1) {
// doOrEnqueue(NOOP);
// }
return retval;
return crawlable;
}
/**
@@ -610,11 +507,11 @@ public abstract class AbstractFrontier
try {
KeyedProperties.loadOverridesFrom(curi);
preparer.prepare(curi);
processScheduleIfUnique(curi);
} finally {
KeyedProperties.clearOverridesFrom(curi);
}
}
enqueueOrDo(new ScheduleIfUnique(curi));
}
/**
@@ -629,7 +526,12 @@ public abstract class AbstractFrontier
public void receive(CrawlURI curi) {
sheetOverlaysManager.applyOverlaysTo(curi);
// prefer doing asap if already in manager thread
doOrEnqueue(new ScheduleAlways(curi));
try {
KeyedProperties.loadOverridesFrom(curi);
processScheduleAlways(curi);
} finally {
KeyedProperties.clearOverridesFrom(curi);
}
}
/**
@@ -644,7 +546,12 @@ public abstract class AbstractFrontier
* @see org.archive.crawler.framework.Frontier#finished(org.archive.modules.CrawlURI)
*/
public void finished(CrawlURI curi) {
enqueueOrDo(new Finish(curi));
try {
KeyedProperties.loadOverridesFrom(curi);
processFinish(curi);
} finally {
KeyedProperties.clearOverridesFrom(curi);
}
}
private void initJournal(String logsDisk) throws IOException {
@@ -663,15 +570,6 @@ public abstract class AbstractFrontier
* @see org.archive.crawler.framework.Frontier#requestState(org.archive.crawler.framework.Frontier.State)
*/
public void requestState(State target) {
enqueueOrDo(new SetTargetState(target));
}
/**
* Actually set a new target Frontier.State. Should only be called in
* managerThread, as by an InEvent.
*/
protected void processSetTargetState(State target) {
// assert Thread.currentThread() == managerThread;
targetState = target;
}
@@ -683,7 +581,7 @@ public abstract class AbstractFrontier
requestState(State.RUN);
}
synchronized public void terminate() {
public void terminate() {
requestState(State.FINISH);
}
@@ -1214,150 +1112,6 @@ public abstract class AbstractFrontier
public void reportTo(PrintWriter writer) {
reportTo(null, writer);
}
/**
* Arrange for the given InEvent to be done by the managerThread, via
* enqueueing with other events if possible, but directly if not possible
* and this is the managerThread.
* @param ev InEvent to be done
*/
protected void enqueue(InEvent ev) {
if(!inbound.offer(ev)) {
try {
drainInbound();
inbound.put(ev);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
/**
* Arrange for the given InEvent to be done by the managerThread, via
* enqueueing with other events if possible, but directly if not possible
* and this is the managerThread.
* @param ev InEvent to be done
*/
protected void enqueueOrDo(InEvent ev) {
// for now, treat same as enqueue
// TODO: reevaluate
enqueue(ev);
// if(!inbound.offer(ev)) {
// // if can't defer,
// if(Thread.currentThread()==managerThread) {
// // if can't enqueue, ok to just do
// synchronized(this) {
// ev.process();
// }
// return;
// } else {
// try {
// drainInbound();
// inbound.put(ev);
// } catch (InterruptedException e) {
// throw new RuntimeException(e);
// }
// }
// }
}
/**
* Arrange for the given InEvent to be done by the managerThread,
* immediately if this is the managerThread, of via enqueueing with
* other inbound events otherwise.
* @param ev InEvent to be done
*/
protected void doOrEnqueue(InEvent ev) {
if (Thread.currentThread() == managerThread) {
// if can't enqueue, ok to just do
synchronized(this) {
ev.process();
}
return;
} else {
enqueue(ev);
}
}
/**
* An event/update for the managerThread to process from the inbound queue.
*/
public abstract class InEvent {
abstract public void process();
}
/**
* A CrawlURI to be scheduled by the managerThread without regard to
* whether the CrawlURI was already-seen.
*/
public class ScheduleAlways extends InEvent {
CrawlURI curi;
public ScheduleAlways(CrawlURI c) {
this.curi = c;
}
public void process() {
try {
KeyedProperties.loadOverridesFrom(curi);
processScheduleAlways(curi);
} finally {
KeyedProperties.clearOverridesFrom(curi);
}
}
}
/**
* A CrawlURI to be scheduled by the managerThread if it has not been
* already-seen. (That is, if it passes the UriUniqFilter.)
*/
public class ScheduleIfUnique extends InEvent {
CrawlURI curi;
public ScheduleIfUnique(CrawlURI c) {
this.curi = c;
}
public void process() {
try {
KeyedProperties.loadOverridesFrom(curi);
processScheduleIfUnique(curi);
} finally {
KeyedProperties.clearOverridesFrom(curi);
}
}
}
/**
* A CrawlURI, previously issued via the outbound queue, that has finished
* its processing chain with update implications for the frontier state.
*/
public class Finish extends InEvent {
CrawlURI caUri;
public Finish(CrawlURI c) {
this.caUri = c;
}
public void process() {
try {
KeyedProperties.loadOverridesFrom(caUri);
processFinish(caUri);
} finally {
KeyedProperties.clearOverridesFrom(caUri);
}
}
}
/**
* An request that the frontier enter a new Frontier.State.
*/
public class SetTargetState extends InEvent {
State target;
public SetTargetState(State target) {
this.target = target;
}
@Override
public void process() {
processSetTargetState(target);
// TODO: perhaps null reachedState, because until new state is
// reached it's misleading?
}
}
public void onApplicationEvent(ApplicationEvent event) {
if(event instanceof CrawlStateEvent) {
engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java
+100 -23
View File
@@ -20,7 +20,8 @@ package org.archive.crawler.frontier;
import java.util.Queue;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
@@ -31,6 +32,7 @@ import javax.management.openmbean.CompositeData;
import org.apache.commons.collections.Closure;
import org.archive.bdb.BdbModule;
import org.archive.bdb.DisposableStoredSortedMap;
import org.archive.bdb.StoredQueue;
import org.archive.checkpointing.Checkpoint;
import org.archive.checkpointing.Checkpointable;
@@ -231,6 +233,7 @@ implements Checkpointable, BeanNameAware {
JSONObject json = new JSONObject();
try {
json.put("nextOrdinal", nextOrdinal.get());
json.put("queuedUriCount", queuedUriCount.get());
json.put("futureUriCount", futureUriCount.get());
json.put("succeededFetchCount", succeededFetchCount.get());
@@ -242,6 +245,9 @@ implements Checkpointable, BeanNameAware {
// impossible
throw new RuntimeException(e);
}
if(this.recover!=null) {
recover.rotateForCheckpoint(checkpointInProgress);
}
}
public void finishCheckpoint(Checkpoint checkpointInProgress) {
@@ -261,6 +267,7 @@ implements Checkpointable, BeanNameAware {
if(isRecovery) {
JSONObject json = recoveryCheckpoint.loadJson(beanName);
try {
nextOrdinal.set(json.getLong("nextOrdinal"));
queuedUriCount.set(json.getLong("queuedUriCount"));
futureUriCount.set(json.getLong("futureUriCount"));
succeededFetchCount.set(json.getLong("succeededFetchCount"));
@@ -271,41 +278,31 @@ implements Checkpointable, BeanNameAware {
throw new RuntimeException(e);
}
// restore WorkQueues to internal management queues
enqueueOrDo(new Recover());
}
}
/**
* Frontier managerThread action to restore the placement of
* all queues to either the 'retired' collection or one of the
* inactive tiers (from which they will become ready/active as
* necessary).
*/
public class Recover extends InEvent {
@Override
public void process() {
// restore WorkQueues to internal management queues
// either retired or inactive tiers
for (String key : allQueues.keySet()) {
WorkQueue q = allQueues.get(key);
q.getOnInactiveQueues().clear();
if(q.isRetired()) {
getRetiredQueues().add(key);
} else {
} else if (q.getCount()>0){
deactivateQueue(q);
} else {
// ensure quques that were empty-but-not-yet-recognized
// (ready/snoozed) at time of checkpoint are reset
q.noteDeactivated();
}
}
}
}
@Override
protected void initOtherQueues() throws DatabaseException {
// small risk of OutOfMemoryError: if 'hold-queues' is false,
// readyClassQueues may grow in size without bound
// tiny risk of OutOfMemoryError: if giant number of snoozed
// queues all wake-to-ready at once
readyClassQueues = new LinkedBlockingQueue<String>();
inactiveQueuesByPrecedence = new TreeMap<Integer,Queue<String>>();
inactiveQueuesByPrecedence = new ConcurrentSkipListMap<Integer,Queue<String>>();
retiredQueues = bdb.getStoredQueue("retiredQueues", String.class, false);
retiredQueues = bdb.getStoredQueue("retiredQueues", String.class);
// primary snoozed queues
snoozedClassQueues = new DelayQueue<DelayedWorkQueue>();
@@ -326,7 +323,7 @@ implements Checkpointable, BeanNameAware {
*/
@Override
Queue<String> createInactiveQueueForPrecedence(int precedence) {
return bdb.getStoredQueue("inactiveQueues-"+precedence, String.class, false);
return bdb.getStoredQueue("inactiveQueues-"+precedence, String.class);
}
/**
@@ -344,4 +341,84 @@ implements Checkpointable, BeanNameAware {
};
pendingUris.forAllPendingDo(tolog);
}
/**
* Run a self-consistency check over queue collections, queues-of-queues,
* etc. for testing purposes. Requires one of the same locks as for PAUSE,
* so should only be run while crawl is running.
*/
public void consistencyCheck() {
// outboundLock.writeLock().lock();
dispositionInProgressLock.writeLock().lock();
System.err.println("<<<CHECKING FRONTIER CONSISTENCY");
DisposableStoredSortedMap<String,String> queueSummaries =
bdb.getStoredMap(
null,
String.class,
String.class,
false,
false);
// mark every queue with the 'managed' collections it's in
consistencyMarkup(queueSummaries, inProcessQueues, "i");
consistencyMarkup(queueSummaries,readyClassQueues, "r");
consistencyMarkup(queueSummaries,snoozedClassQueues, "s");
consistencyMarkup(queueSummaries,snoozedOverflow.values(), "S");
for( Entry<Integer, Queue<String>> entry : getInactiveQueuesByPrecedence().entrySet()) {
consistencyMarkup(queueSummaries,entry.getValue(),Integer.toString(entry.getKey()));
}
consistencyMarkup(queueSummaries,retiredQueues, "R");
// report problems where a queue isn't as expected or ideal
int anomalies = 0;
for(String q : allQueues.keySet()) {
WorkQueue wq = allQueues.get(q);
String summary = queueSummaries.get(q);
if(wq.getCount()>0 && summary == null) {
// every non-empty queue should have been in at least one collection
System.err.println("FRONTIER ANOMALY: "+q+" "+wq.getCount()+" "+wq.isManaged()+" but not in managed collections");
// System.err.println(wq.shortReportLegend()+"\n"+inactiveByClass.get(q)+"\n"+wq.shortReportLine());
anomalies++;
}
if(wq.getCount()==0 && summary == null && wq.isManaged()) {
// any empty queue should only report isManaged if in a collection
System.err.println("FRONTIER ANOMALY: "+q+" "+wq.getCount()+" "+wq.isManaged()+" but not in managed collections");
// System.err.println(wq.shortReportLegend()+"\n"+inactiveByClass.get(q)+"\n"+wq.shortReportLine());
anomalies++;
}
}
System.err.println(anomalies+" ANOMALIES");
int concerns = 0;
for(String q : queueSummaries.keySet()) {
String summary = queueSummaries.get(q);
if(summary != null && summary.split(",").length>1) {
// ideally queues won't be more than one place (though frontier
// should operate if they are, and changing precedence values
// will cause multiple entries by design)
WorkQueue wq = allQueues.get(q);
System.err.println("FRONTIER CONCERN: "+q+" "+wq.getCount()+" multiple places: "+summary);
System.err.println("\n"+wq.shortReportLegend()+"\n"+wq.shortReportLine());
concerns++;
}
}
System.err.println(concerns+" CONCERNS");
System.err.println("END CHECKING FRONTIER>>>");
queueSummaries.dispose();
dispositionInProgressLock.writeLock().unlock();
// outboundLock.writeLock().unlock();
}
protected void consistencyMarkup(
DisposableStoredSortedMap<String, String> queueSummaries,
Iterable<?> queues, String mark) {
for(Object qq : queues) {
String key = (qq instanceof String)
? (String)qq
: (qq instanceof WorkQueue)
? ((WorkQueue)qq).getClassKey()
: ((DelayedWorkQueue)qq).getClassKey();
String val = queueSummaries.get(key);
val = (val==null) ? mark : val+","+mark;
queueSummaries.put(key, val);
}
}
}
engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java
+57 -104
View File
@@ -23,10 +23,8 @@ import java.io.IOException;
import java.io.PrintWriter;
import java.io.Serializable;
import java.util.Date;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
@@ -58,7 +56,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
protected final String classKey;
/** whether queue is active (ready/in-process/snoozed) or on a waiting queue */
protected boolean active = true;
protected boolean active = false;
/** Total number of stored items */
protected long count = 0;
@@ -67,17 +65,14 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
protected long enqueueCount = 0;
/** Whether queue is already in lifecycle stage */
protected boolean isHeld = false;
protected boolean isManaged = false;
/** Time to wake, if snoozed */
protected long wakeTime = 0;
/** assigned precedence */
protected PrecedenceProvider precedenceProvider = new SimplePrecedenceProvider(1);
/** set of by-precedence inactive-queues on which WorkQueue is waiting */
protected Set<Integer> onInactiveQueues = new HashSet<Integer>();
/** Per-session 'budget' controlling activity duration */
protected int sessionBudget = 0;
@@ -127,7 +122,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
* @param match
* @return count of deleted URIs
*/
public long deleteMatching(final WorkQueueFrontier frontier, String match) {
public synchronized long deleteMatching(final WorkQueueFrontier frontier, String match) {
try {
final long deleteCount = deleteMatchingFromQueue(frontier, match);
this.count -= deleteCount;
@@ -146,7 +141,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
* @param frontier Work queues manager.
* @param curi CrawlURI to insert.
*/
protected void enqueue(final WorkQueueFrontier frontier,
protected synchronized long enqueue(final WorkQueueFrontier frontier,
CrawlURI curi) {
try {
insert(frontier, curi, false);
@@ -157,6 +152,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
}
count++;
enqueueCount++;
return count;
}
/**
@@ -169,7 +165,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
*
* @return topmost queue item, or null
*/
public CrawlURI peek(final WorkQueueFrontier frontier) {
public synchronized CrawlURI peek(final WorkQueueFrontier frontier) {
if(peekItem == null && count > 0) {
try {
peekItem = peekItem(frontier);
@@ -191,7 +187,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
*
* @param frontier Work queues manager.
*/
protected void dequeue(final WorkQueueFrontier frontier, CrawlURI expected) {
protected synchronized void dequeue(final WorkQueueFrontier frontier, CrawlURI expected) {
try {
deleteItem(frontier, peekItem);
} catch (IOException e) {
@@ -226,7 +222,16 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
return this.sessionBudget;
}
public void startActiveSession() {
/**
* Begin an 'active' session, which begins when a queue first offers a
* URI for crawling, and continues until it is deactivated (for example,
* for session-budget reasons).
*/
public void considerActive() {
if(active) {
return;
}
active=true;
expenditureAtLastActivation = totalExpenditure;
}
@@ -246,18 +251,27 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
}
/**
* Check whether queue has temporarily (session) or permanently (total)
* exceeded its budgets.
* Check whether queue has temporarily (session) exceeded its budget.
*
* @return true if queue is over either of its set budget(s)
*/
public boolean isOverBudget() {
public boolean isOverSessionBudget() {
// check whether session budget exceeded
// or totalExpenditure exceeds totalBudget
return (sessionBudget > 0 && (totalExpenditure - expenditureAtLastActivation) > sessionBudget)
|| (this.totalBudget >= 0 && this.totalExpenditure >= this.totalBudget);
return (sessionBudget > 0 && (totalExpenditure - expenditureAtLastActivation) > sessionBudget);
}
/**
* Check whether queue has permanently (total) exceeded its budget.
*
* @return true if queue is over either of its set budget(s)
*/
public boolean isOverTotalBudget() {
// check whether session budget exceeded
// or totalExpenditure exceeds totalBudget
return (this.totalBudget >= 0 && this.totalExpenditure >= this.totalBudget);
}
/**
* Return the tally of all expenditures on this queue
*
@@ -315,39 +329,6 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
return this.classKey;
}
/**
* Clear isHeld to false
*/
public void clearHeld() {
isHeld = false;
if(logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE,
"queue unheld: " + getClassKey());
}
}
/**
* Whether the queue is already in a lifecycle stage --
* such as ready, in-progress, snoozed -- and thus should
* not be redundantly inserted to readyClassQueues
*
* @return isHeld
*/
public boolean isHeld() {
return isHeld;
}
/**
* Set isHeld to true
*/
public void setHeld() {
isHeld = true;
if(logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE,
"queue held: " + getClassKey());
}
}
/**
* Forgive the peek, allowing a subsequent peek to
* return a different item.
@@ -404,7 +385,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
*
* @return Returns the count.
*/
public long getCount() {
public synchronized long getCount() {
return this.count;
}
@@ -464,46 +445,6 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
protected abstract CrawlURI peekItem(final WorkQueueFrontier frontier)
throws IOException;
/**
* Suspends this WorkQueue. Closes all connections to resources etc.
*
* @param frontier
* @throws IOException
*/
protected void suspend(final WorkQueueFrontier frontier) throws IOException {
}
/**
* Resumes this WorkQueue. Eventually opens connections to resources etc.
*
* @param frontier
* @throws IOException
*/
protected void resume(final WorkQueueFrontier frontier) throws IOException {
}
public void setActive(final WorkQueueFrontier frontier, final boolean b) {
if(active != b) {
active = b;
if(logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE,
(active ? "queue set active: " : "queue unset active: ") +
this.getClassKey());
}
try {
if(active) {
resume(frontier);
} else {
suspend(frontier);
}
} catch (IOException e) {
//FIXME better exception handling
e.printStackTrace();
throw new RuntimeException(e);
}
}
}
//
// Reporter
//
@@ -522,7 +463,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
reportTo(null,writer);
}
public Map<String, Object> shortReportMap() {
public synchronized Map<String, Object> shortReportMap() {
Map<String,Object> map = new LinkedHashMap<String, Object>();
map.put("queueName", classKey);
@@ -555,7 +496,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
return sessionBudget - (totalExpenditure-expenditureAtLastActivation);
}
public void shortReportLineTo(PrintWriter writer) {
public synchronized void shortReportLineTo(PrintWriter writer) {
// queue name
writer.print(classKey);
writer.print(" ");
@@ -616,7 +557,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
* @param writer
* @throws IOException
*/
public void reportTo(String name, PrintWriter writer) {
public synchronized void reportTo(String name, PrintWriter writer) {
// name is ignored: only one kind of report for now
writer.print("Queue ");
writer.print(classKey);
@@ -695,13 +636,6 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
return precedenceProvider.getPrecedence();
}
/**
* @return the onInactiveQueues
*/
public Set<Integer> getOnInactiveQueues() {
return onInactiveQueues;
}
/* (non-Javadoc)
* @see org.archive.modules.fetcher.FetchStats.HasFetchStats#tally(org.archive.modules.CrawlURI, org.archive.modules.fetcher.FetchStats.Stage)
*/
@@ -710,7 +644,26 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
precedenceProvider.tally(curi, stage);
}
public boolean isActive() {
return active;
public synchronized void noteDeactivated() {
active = false;
isManaged = true;
}
/**
* Whether the queue is already in a lifecycle stage --
* such as ready, in-progress, snoozed -- and thus should
* not be redundantly inserted to readyClassQueues
*
* @return isManaged
*/
public boolean isManaged() {
return isManaged;
}
/* (non-Javadoc)
* @see java.lang.Object#toString()
*/
public String toString() {
return super.toString()+"("+getClassKey()+")";
}
}
engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java
+134 -240
View File
@@ -32,12 +32,16 @@ import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.SortedMap;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
@@ -46,9 +50,6 @@ import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.regex.Pattern;
import org.apache.commons.collections.Bag;
import org.apache.commons.collections.BagUtils;
import org.apache.commons.collections.bag.HashBag;
import org.apache.commons.collections.iterators.ObjectArrayIterator;
import org.archive.crawler.datamodel.UriUniqFilter;
import org.archive.crawler.datamodel.UriUniqFilter.CrawlUriReceiver;
@@ -115,23 +116,6 @@ implements Closeable,
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.appCtx = (AbstractApplicationContext)applicationContext;
}
/**
* Whether queues should start INACTIVE (only becoming active
* when needed to keep the crawler busy), or if queues should
* start out ready (which means all nonempty queues are
* considered in a round-robin fashion)
*
* @return true if new queues should held inactive
*/
boolean holdQueues = true;
public boolean getHoldQueues() {
return holdQueues;
}
public void setHoldQueues(boolean holdQueues) {
this.holdQueues = holdQueues;
}
/** amount to replenish budget on each activation (duty cycle) */
{
@@ -208,8 +192,8 @@ implements Closeable,
protected BlockingQueue<String> readyClassQueues;
/** all per-class queues from whom a URI is outstanding */
protected Bag inProcessQueues =
BagUtils.synchronizedBag(new HashBag()); // of ClassKeyQueue
protected Set<WorkQueue> inProcessQueues =
Collections.newSetFromMap(new ConcurrentHashMap<WorkQueue, Boolean>()); // of ClassKeyQueue
/**
* All per-class queues held in snoozed state, sorted by wake time.
@@ -234,7 +218,6 @@ implements Closeable,
largestQueues.setMaxSize(count);
}
protected int highestPrecedenceWaiting = Integer.MAX_VALUE;
/** The UriUniqFilter to use, tracking those UURIs which are
@@ -285,6 +268,7 @@ implements Closeable,
*/
protected void initInternalQueues()
throws IOException, DatabaseException {
this.initOtherQueues();
if (workQueueDataOnDisk()
&& preparer.getQueueAssignmentPolicy().maximumNumberOfKeys() >= 0
&& preparer.getQueueAssignmentPolicy().maximumNumberOfKeys() <=
@@ -294,7 +278,6 @@ implements Closeable,
} else {
this.initAllQueues();
}
this.initOtherQueues();
}
/**
@@ -403,55 +386,26 @@ implements Closeable,
// assert Thread.currentThread() == managerThread;
WorkQueue wq = getQueueFor(curi);
int originalPrecedence = wq.getPrecedence();
wq.enqueue(this, curi);
// always take budgeting values from current curi
// (whose overlay settings should be active here)
wq.setSessionBudget(getBalanceReplenishAmount());
wq.setTotalBudget(getQueueTotalBudget());
// Update recovery log.
doJournalAdded(curi);
if(wq.isRetired()) {
return;
}
incrementQueuedUriCount();
if(wq.isHeld()) {
if(wq.isActive()) {
// queue active -- promote will be handled ok by normal cycling
// do nothing
} else {
// queue is already in a waiting inactive queue; update
synchronized(wq) {
int originalPrecedence = wq.getPrecedence();
wq.enqueue(this, curi);
// always take budgeting values from current curi
// (whose overlay settings should be active here)
wq.setSessionBudget(getBalanceReplenishAmount());
wq.setTotalBudget(getQueueTotalBudget());
if(!wq.isRetired()) {
incrementQueuedUriCount();
int currentPrecedence = wq.getPrecedence();
if(currentPrecedence < originalPrecedence ) {
// queue bumped up; adjust ordering
if(!wq.isManaged() || currentPrecedence < originalPrecedence) {
// queue newly filled or bumped up in precedence; ensure enqueuing
// at precedence level (perhaps duplicate; if so that's handled elsewhere)
deactivateQueue(wq);
// this intentionally places queue in duplicate inactiveQueue\
// only when it comes off the right queue will it activate;
// otherwise it reenqueues to right inactive queue, if not
// already there (see activateInactiveQueue())
if(logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE,
"queue re-deactivated to p" +currentPrecedence
+ ": " + wq.getClassKey());
}
} else {
// queue bumped down or stayed same;
// do nothing until it comes up
}
}
} else {
// begin juggling queue between internal ordering structures
wq.setHeld();
if(getHoldQueues()) {
deactivateQueue(wq);
} else {
wq.startActiveSession();
readyQueue(wq);
}
}
// Update recovery log.
doJournalAdded(curi);
largestQueues.update(wq.getClassKey(), wq.getCount());
}
@@ -459,11 +413,10 @@ implements Closeable,
* Put the given queue on the readyClassQueues queue
* @param wq
*/
private void readyQueue(WorkQueue wq) {
protected void readyQueue(WorkQueue wq) {
// assert Thread.currentThread() == managerThread;
try {
wq.setActive(this, true);
readyClassQueues.put(wq.getClassKey());
if(logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE,
@@ -477,20 +430,28 @@ implements Closeable,
}
}
// ConcurrentHashMap<String, String> inactiveByClass = new ConcurrentHashMap<String, String>();
/**
* Put the given queue on the inactiveQueues queue
* @param wq
*/
protected void deactivateQueue(WorkQueue wq) {
// assert Thread.currentThread() == managerThread;
int precedence = wq.getPrecedence();
if(!wq.getOnInactiveQueues().contains(precedence)) {
// not already on target, add
Queue<String> inactiveQueues =
getInactiveQueuesForPrecedence(precedence);
Queue<String> inactiveQueues =
getInactiveQueuesForPrecedence(precedence);
synchronized(wq) {
wq.noteDeactivated();
inProcessQueues.remove(wq);
if(wq.getCount()==0) {
System.err.println("deactivate empty queue?");
}
// String prev = inactiveByClass.put(wq.getClassKey(), wq.shortReportLine());
// if(prev!=null) {
// logger.log(Level.WARNING,"duplicate add: "+wq.getClassKey()+"\n"+wq.shortReportLegend()+"\n"+prev+wq.shortReportLine(), new Exception());
// }
inactiveQueues.add(wq.getClassKey());
wq.getOnInactiveQueues().add(precedence);
if(wq.getPrecedence() < highestPrecedenceWaiting ) {
highestPrecedenceWaiting = wq.getPrecedence();
}
@@ -499,13 +460,7 @@ implements Closeable,
"queue deactivated to p" + precedence
+ ": " + wq.getClassKey());
}
} else {
if(logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE,
"queue already p" + precedence+": " + wq.getClassKey());
}
}
wq.setActive(this, false);
}
/**
@@ -545,6 +500,7 @@ implements Closeable,
protected void retireQueue(WorkQueue wq) {
// assert Thread.currentThread() == managerThread;
inProcessQueues.remove(wq);
getRetiredQueues().add(wq.getClassKey());
decrementQueuedCount(wq.getCount());
wq.setRetired(true);
@@ -552,7 +508,6 @@ implements Closeable,
logger.log(Level.FINE,
"queue retired: " + wq.getClassKey());
}
wq.setActive(this, false);
}
/**
@@ -631,8 +586,7 @@ implements Closeable,
*
* @see org.archive.crawler.framework.Frontier#next()
*/
protected synchronized CrawlURI findEligibleURI() {
// assert Thread.currentThread() == managerThread;
protected CrawlURI findEligibleURI() {
// wake any snoozed queues
wakeQueues();
// consider rescheduled URIS
@@ -664,18 +618,38 @@ implements Closeable,
}
if(readyQ.getCount()==0) {
// readyQ is empty and ready: it's exhausted
// release held status, allowing any subsequent
// enqueues to again put queue in ready
readyQ.clearHeld();
readyQ.noteDeactivated();
readyQ = null;
continue;
}
if(!inProcessQueues.add(readyQ)) {
// double activation; discard this and move on
// (this guard allows other enqueuings to ready or
// the various inactive-by-precedence queues to
// sometimes redundantly enqueue a queue key)
readyQ = null;
continue;
}
// queue has gone 'in process'
readyQ.considerActive();
readyQ.setWakeTime(0); // clear obsolete wake time, if any
if (readyQ.isOverSessionBudget()) {
deactivateQueue(readyQ);
readyQ = null;
continue;
}
if (readyQ.isOverTotalBudget()) {
retireQueue(readyQ);
readyQ = null;
continue;
}
} while (readyQ == null);
if (readyQ == null) {
// no queues left in ready or readiable
break findauri;
}
assert !inProcessQueues.contains(readyQ) : "double activation";
returnauri: while(true) { // loop left by explicit return or break on empty
CrawlURI curi = null;
curi = readyQ.peek(this);
@@ -704,12 +678,12 @@ implements Closeable,
if (currentQueueKey.equals(curi.getClassKey())) {
// curi was in right queue, emit
noteAboutToEmit(curi, readyQ);
inProcessQueues.add(readyQ);
return curi;
}
// URI's assigned queue has changed since it
// was queued (eg because its IP has become
// known). Requeue to new queue.
// TODO: consider synchronization on readyQ
readyQ.dequeue(this,curi);
doJournalRelocated(curi);
curi.setClassKey(currentQueueKey);
@@ -720,7 +694,10 @@ implements Closeable,
// readyQ is empty and ready: it's exhausted
// release held status, allowing any subsequent
// enqueues to again put queue in ready
readyQ.clearHeld();
// FIXME: tiny window here where queue could
// receive new URI, be readied, fail not-in-process?
inProcessQueues.remove(readyQ);
readyQ.noteDeactivated();
readyQ = null;
continue findauri;
}
@@ -733,23 +710,15 @@ implements Closeable,
uriUniqFilter.requestFlush();
}
// never return null if there are any eligible inactives
if(getTotalEligibleInactiveQueues()>0) {
if(depthFindEligibleURI>1) {
logger.warning(
"FRONTIER.findEligibleURIs depth: "+ depthFindEligibleURI
+"\n"+shortReportLine());
}
if(depthFindEligibleURI>=5) {
logger.severe("RETURNING null");
return null;
}
// if truly nothing ready, wait a moment before returning null
// so that loop in surrounding next() has a chance of getting something
// next time
if(getTotalEligibleInactiveQueues()==0) {
try {
depthFindEligibleURI++;
return findEligibleURI();
} finally {
depthFindEligibleURI = 0;
}
Thread.sleep(1000);
} catch (InterruptedException e) {
//
}
}
// nothing eligible
@@ -780,79 +749,31 @@ implements Closeable,
/**
* Activate an inactive queue, if any are available.
*/
private void activateInactiveQueue() {
// assert Thread.currentThread() == managerThread;
protected boolean activateInactiveQueue() {
SortedMap<Integer,Queue<String>> inactiveQueuesByPrecedence =
getInactiveQueuesByPrecedence();
int targetPrecedence = highestPrecedenceWaiting;
Queue<String> inactiveQueues = inactiveQueuesByPrecedence.get(
targetPrecedence);
for( Entry<Integer, Queue<String>> entry : getInactiveQueuesByPrecedence().entrySet()) {
for (String key = entry.getValue().poll(); key!=null; key = entry.getValue().poll() ) {
// inactiveByClass.remove(key);
int expectedPrecedence = entry.getKey();
if(key!=null) {
WorkQueue candidateQ = (WorkQueue) this.allQueues.get(key);
if(candidateQ.getPrecedence() > expectedPrecedence) {
// queue demoted since placed; re-deactivate
deactivateQueue(candidateQ);
continue;
}
updateHighestWaiting(expectedPrecedence);
try {
readyClassQueues.put(key);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return true;
}
}
}
String key = inactiveQueues.poll();
assert key != null : "empty precedence queue in map";
if(inactiveQueues.isEmpty()) {
updateHighestWaiting(targetPrecedence+1);
}
WorkQueue candidateQ = (WorkQueue) this.allQueues.get(key);
assert candidateQ != null : "missing uri work queue";
boolean was = candidateQ.getOnInactiveQueues().remove(targetPrecedence);
assert was : "queue didn't know it was in "+targetPrecedence+" inactives";
if(candidateQ.isActive()) {
// queue had been multiply-scheduled due to changing precedence
// already active, so ignore this activation
if(logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE,
"queue activated+ignored/active from p" + targetPrecedence
+ ": " + candidateQ.getClassKey());
}
return;
}
if(candidateQ.getPrecedence() < targetPrecedence) {
// queue moved up; do nothing (already handled)
if(logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE,
"queue activated+ignored/higher from p"
+ targetPrecedence
+ ": " + candidateQ.getClassKey()
+ " ("+candidateQ.getPrecedence() + ") ");
}
return;
}
if(candidateQ.getPrecedence() > targetPrecedence) {
// queue moved down; deactivate to new level
if(logger.isLoggable(Level.FINE)) {
logger.log(Level.FINE,
"queue activated+deactivated from p" + targetPrecedence
+ ": " + candidateQ.getClassKey());
}
deactivateQueue(candidateQ);
return;
}
candidateQ.startActiveSession();
if (candidateQ.isOverBudget()) {
// if still over-budget after an activation & replenishing,
// retire
retireQueue(candidateQ);
return;
}
// long now = System.currentTimeMillis();
// long delay_ms = candidateQ.getWakeTime() - now;
// if (delay_ms > 0) {
// // queue still due for snoozing
// snoozeQueue(candidateQ, now, delay_ms);
// return;
// }
candidateQ.setWakeTime(0); // clear obsolete wake time, if any
readyQueue(candidateQ);
return false;
}
/**
@@ -879,7 +800,7 @@ implements Closeable,
*
* @param wq
*/
private void reenqueueQueue(WorkQueue wq) {
protected void reenqueueQueue(WorkQueue wq) {
//TODO:SPRINGY set overrides by queue?
getQueuePrecedencePolicy().queueReevaluate(wq);
if (logger.isLoggable(Level.FINE)) {
@@ -887,7 +808,6 @@ implements Closeable,
wq.getClassKey());
}
if(highestPrecedenceWaiting < wq.getPrecedence()
|| (wq.isOverBudget() && highestPrecedenceWaiting <= wq.getPrecedence())
|| wq.getPrecedence() >= getPrecedenceFloor()) {
// if still over budget, deactivate
deactivateQueue(wq);
@@ -911,27 +831,22 @@ implements Closeable,
* put all queues in slow-retry-snoozes back to busy-ness.
*/
public void forceWakeQueues() {
enqueueOrDo(new InEvent() {
@Override
public void process() {
Iterator<DelayedWorkQueue> iterSnoozed = snoozedClassQueues.iterator();
while(iterSnoozed.hasNext()) {
WorkQueue queue = iterSnoozed.next().getWorkQueue(WorkQueueFrontier.this);
queue.setWakeTime(0);
reenqueueQueue(queue);
iterSnoozed.remove();
}
Iterator<DelayedWorkQueue> iterOverflow = snoozedOverflow.values().iterator();
while(iterOverflow.hasNext()) {
WorkQueue queue = iterOverflow.next().getWorkQueue(WorkQueueFrontier.this);
queue.setWakeTime(0);
reenqueueQueue(queue);
iterOverflow.remove();
}
}
});
Iterator<DelayedWorkQueue> iterSnoozed = snoozedClassQueues.iterator();
while(iterSnoozed.hasNext()) {
WorkQueue queue = iterSnoozed.next().getWorkQueue(WorkQueueFrontier.this);
queue.setWakeTime(0);
reenqueueQueue(queue);
iterSnoozed.remove();
}
Iterator<DelayedWorkQueue> iterOverflow = snoozedOverflow.values().iterator();
while(iterOverflow.hasNext()) {
WorkQueue queue = iterOverflow.next().getWorkQueue(WorkQueueFrontier.this);
queue.setWakeTime(0);
reenqueueQueue(queue);
iterOverflow.remove();
}
}
/**
* Wake any queues sitting in the snoozed queue whose time has come.
*/
@@ -943,26 +858,15 @@ implements Closeable,
reenqueueQueue(queue);
}
// also consider overflow (usually empty)
long now = System.currentTimeMillis();
Iterator<DelayedWorkQueue> iter =
snoozedOverflow.values().iterator();
snoozedOverflow.headMap(System.currentTimeMillis()).values().iterator();
while(iter.hasNext()) {
DelayedWorkQueue dq = iter.next();
if(dq.getWakeTime()<=now) {
iter.remove();
snoozedOverflowCount.decrementAndGet();
WorkQueue queue = dq.getWorkQueue(this);
queue.setWakeTime(0);
reenqueueQueue(queue);
continue; // while
}
if(snoozedClassQueues.size()<MAX_SNOOZED_IN_MEMORY) {
iter.remove();
snoozedOverflowCount.decrementAndGet();
snoozedClassQueues.add(dq);
} else {
break; // while
}
iter.remove();
snoozedOverflowCount.decrementAndGet();
WorkQueue queue = dq.getWorkQueue(this);
queue.setWakeTime(0);
reenqueueQueue(queue);
}
}
@@ -996,7 +900,7 @@ implements Closeable,
wq.setTotalBudget(getQueueTotalBudget());
assert (wq.peek(this) == curi) : "unexpected peek " + wq;
inProcessQueues.remove(wq, 1);
int holderCost = curi.getHolderCost();
if (needsReenqueuing(curi)) {
@@ -1083,6 +987,7 @@ implements Closeable,
* @param delay_ms
*/
protected void handleQueue(WorkQueue wq, boolean forceRetire, long now, long delay_ms) {
inProcessQueues.remove(wq);
if(forceRetire) {
retireQueue(wq);
} else if (delay_ms > 0) {
@@ -1169,7 +1074,7 @@ implements Closeable,
}
int allCount = allQueues.size();
int inProcessCount = inProcessQueues.uniqueSet().size();
int inProcessCount = inProcessQueues.size();
int readyCount = readyClassQueues.size();
int snoozedCount = getSnoozedCount();
int activeCount = inProcessCount + readyCount + snoozedCount;
@@ -1177,8 +1082,6 @@ implements Closeable,
int ineligibleCount = getTotalIneligibleInactiveQueues();
int retiredCount = getRetiredQueues().size();
int exhaustedCount = allCount - activeCount - inactiveCount - retiredCount;
int inCount = inbound.size();
int outCount = outbound.size();
Map<String,Object> map = new LinkedHashMap<String, Object>();
map.put("totalQueues", allCount);
@@ -1191,8 +1094,6 @@ implements Closeable,
map.put("retiredQueues", retiredCount);
map.put("exhaustedQueues", exhaustedCount);
map.put("lastReachedState", lastReachedState);
map.put("inboundCount", inCount);
map.put("outboundCount", outCount);
return map;
}
@@ -1205,7 +1106,7 @@ implements Closeable,
return;
}
int allCount = allQueues.size();
int inProcessCount = inProcessQueues.uniqueSet().size();
int inProcessCount = inProcessQueues.size();
int readyCount = readyClassQueues.size();
int snoozedCount = getSnoozedCount();
int activeCount = inProcessCount + readyCount + snoozedCount;
@@ -1214,9 +1115,9 @@ implements Closeable,
int retiredCount = getRetiredQueues().size();
int exhaustedCount =
allCount - activeCount - inactiveCount - retiredCount;
int inCount = inbound.size();
int outCount = outbound.size();
State last = lastReachedState;
w.print(last);
w.print(" - ");
w.print(allCount);
w.print(" URI queues: ");
w.print(activeCount);
@@ -1234,8 +1135,7 @@ implements Closeable,
w.print(retiredCount);
w.print(" retired; ");
w.print(exhaustedCount);
w.print(" exhausted");
w.print(" ["+last+ ": "+inCount+" in, "+outCount+" out]");
w.print(" exhausted");
w.flush();
}
@@ -1314,7 +1214,6 @@ implements Closeable,
ArrayList<WorkQueue> inProcessQueuesCopy;
synchronized(this.inProcessQueues) {
// grab a copy that will be stable against mods for report duration
@SuppressWarnings("unchecked")
Collection<WorkQueue> inProcess = this.inProcessQueues;
inProcessQueuesCopy = new ArrayList<WorkQueue>(inProcess);
}
@@ -1390,7 +1289,7 @@ implements Closeable,
*/
private void standardReportTo(PrintWriter w) {
int allCount = allQueues.size();
int inProcessCount = inProcessQueues.uniqueSet().size();
int inProcessCount = inProcessQueues.size();
int readyCount = readyClassQueues.size();
int snoozedCount = getSnoozedCount();
int activeCount = inProcessCount + readyCount + snoozedCount;
@@ -1469,10 +1368,8 @@ implements Closeable,
w.print(exhaustedCount);
w.print("\n");
int inCount = inbound.size();
int outCount = outbound.size();
State last = lastReachedState;
w.print("\n Threadbound: "+last+ ": "+inCount+" in, "+outCount+" out");
w.print("\n Last state: "+last);
w.print("\n -----===== MANAGER THREAD =====-----\n");
ToeThread.reportThread(managerThread, w);
@@ -1481,7 +1378,6 @@ implements Closeable,
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, maxQueuesPerReportCategory);
appendQueueReports(w, "IN-PROCESS", copy.iterator(), copy.size(), maxQueuesPerReportCategory);
@@ -1627,7 +1523,7 @@ implements Closeable,
if(inProcessQueues==null || readyClassQueues==null || snoozedClassQueues==null) {
return 0;
}
int inProcessCount = inProcessQueues.uniqueSet().size();
int inProcessCount = inProcessQueues.size();
int readyCount = readyClassQueues.size();
int snoozedCount = getSnoozedCount();
int activeCount = inProcessCount + readyCount + snoozedCount;
@@ -1644,7 +1540,7 @@ implements Closeable,
if(inProcessQueues==null || readyClassQueues==null || snoozedClassQueues==null) {
return 0;
}
int inProcessCount = inProcessQueues.uniqueSet().size();
int inProcessCount = inProcessQueues.size();
int readyCount = readyClassQueues.size();
int snoozedCount = getSnoozedCount();
int activeCount = inProcessCount + readyCount + snoozedCount;
@@ -1664,7 +1560,6 @@ implements Closeable,
public boolean isEmpty() {
return queuedUriCount.get() == 0
&& (uriUniqFilter == null || uriUniqFilter.pending() == 0)
&& (inbound == null || inbound.isEmpty())
&& futureUriCount.get() == 0;
}
@@ -1675,5 +1570,4 @@ implements Closeable,
protected int getInProcessCount() {
return inProcessQueues.size();
}
} // TODO: slim class! Suspect it should be < 800 lines, shedding budgeting/reporting