diff --git a/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java b/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java index ca05212a..59ec3128 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java @@ -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 inbound; - public BlockingQueue getInbound() { - return inbound; - } - public void setInbound(BlockingQueue inbound) { - this.inbound = inbound; - } - - /** outbound URIs */ - protected BlockingQueue outbound; - public BlockingQueue getOutbound() { - return outbound; - } - public void setOutbound(BlockingQueue 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(200, true); - } - if(inbound==null) { - inbound = new ArrayBlockingQueue(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) { diff --git a/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java b/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java index dee0c3a3..f4db3e07 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java @@ -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(); - inactiveQueuesByPrecedence = new TreeMap>(); + inactiveQueuesByPrecedence = new ConcurrentSkipListMap>(); - retiredQueues = bdb.getStoredQueue("retiredQueues", String.class, false); + retiredQueues = bdb.getStoredQueue("retiredQueues", String.class); // primary snoozed queues snoozedClassQueues = new DelayQueue(); @@ -326,7 +323,7 @@ implements Checkpointable, BeanNameAware { */ @Override Queue 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("<< 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> 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 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); + } + } } diff --git a/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java b/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java index 9f05309a..3417d1fb 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java +++ b/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java @@ -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 onInactiveQueues = new HashSet(); - + /** 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 shortReportMap() { + public synchronized Map shortReportMap() { Map map = new LinkedHashMap(); 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 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()+")"; } } diff --git a/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java b/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java index 8ca4a72a..e714c629 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java @@ -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 readyClassQueues; /** all per-class queues from whom a URI is outstanding */ - protected Bag inProcessQueues = - BagUtils.synchronizedBag(new HashBag()); // of ClassKeyQueue + protected Set inProcessQueues = + Collections.newSetFromMap(new ConcurrentHashMap()); // 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 inactiveByClass = new ConcurrentHashMap(); /** * 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 inactiveQueues = - getInactiveQueuesForPrecedence(precedence); + + Queue 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> inactiveQueuesByPrecedence = - getInactiveQueuesByPrecedence(); - - int targetPrecedence = highestPrecedenceWaiting; - Queue inactiveQueues = inactiveQueuesByPrecedence.get( - targetPrecedence); + for( Entry> 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 iterSnoozed = snoozedClassQueues.iterator(); - while(iterSnoozed.hasNext()) { - WorkQueue queue = iterSnoozed.next().getWorkQueue(WorkQueueFrontier.this); - queue.setWakeTime(0); - reenqueueQueue(queue); - iterSnoozed.remove(); - } - Iterator iterOverflow = snoozedOverflow.values().iterator(); - while(iterOverflow.hasNext()) { - WorkQueue queue = iterOverflow.next().getWorkQueue(WorkQueueFrontier.this); - queue.setWakeTime(0); - reenqueueQueue(queue); - iterOverflow.remove(); - } - } - }); + Iterator iterSnoozed = snoozedClassQueues.iterator(); + while(iterSnoozed.hasNext()) { + WorkQueue queue = iterSnoozed.next().getWorkQueue(WorkQueueFrontier.this); + queue.setWakeTime(0); + reenqueueQueue(queue); + iterSnoozed.remove(); + } + Iterator 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 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() 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 map = new LinkedHashMap(); 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 inProcessQueuesCopy; synchronized(this.inProcessQueues) { // grab a copy that will be stable against mods for report duration - @SuppressWarnings("unchecked") Collection inProcess = this.inProcessQueues; inProcessQueuesCopy = new ArrayList(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 inProcess = inProcessQueues; ArrayList 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 \ No newline at end of file