From 1ce23ecbcdc4c01907db22eecfab9653df3514cf Mon Sep 17 00:00:00 2001 From: gojomo Date: Tue, 21 Sep 2010 10:55:07 +0000 Subject: [PATCH] [HER-747] Ensure 'retired' status for queues is operator-controllable [HER-1768] Changing queue totalBudget not retiring queue as expected * CrawlURI.java (includesRetireDirective) internalize to CrawlURI * AbstractFrontier.java allow re-enqueuing of untried URI with retire-driective clean up terminology/method-names * WorkQueue.java, WorkQueueFrontier.java, BdbFrontier.java adjust budget/sessionBalance handling to always count up, always consult latest values, always refresh values from latest URI (processFinish) reorganize to reduce redundancy * DispositionProcessor.java offer forceRetire setting * QuotaEnforcer.java set failed-status *or* forceRetire, not both [HER-1823] H3: adding SheetAssociations mid-crawl won't affect already queued CrawlURIs (sheet liveness) * SheetsOverlayManager.java (applyOverlaysTo) rename, unify, reapply whenever called to ensure liveness * CandidatesProcessor.java adjust to rename --- .../crawler/frontier/AbstractFrontier.java | 21 +- .../archive/crawler/frontier/BdbFrontier.java | 1 - .../archive/crawler/frontier/WorkQueue.java | 127 ++++++----- .../crawler/frontier/WorkQueueFrontier.java | 197 +++++++----------- .../postprocessor/CandidatesProcessor.java | 4 +- .../postprocessor/DispositionProcessor.java | 21 ++ .../crawler/prefetch/QuotaEnforcer.java | 5 +- .../crawler/spring/SheetOverlaysManager.java | 56 +++-- .../java/org/archive/modules/CrawlURI.java | 5 + 9 files changed, 215 insertions(+), 222 deletions(-) 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 8eeb783b..705eed9e 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java @@ -21,6 +21,7 @@ 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; @@ -494,6 +495,9 @@ public abstract class AbstractFrontier CrawlURI retval = outbound.take(); + // TODO: consider optimizations avoiding this recalc of + // overrides when not necessary + sheetOverlaysManager.applyOverlaysTo(retval); // // TODO: consider if following necessary for maintaining throughput // if(outbound.size()<=1) { // doOrEnqueue(NOOP); @@ -567,7 +571,7 @@ public abstract class AbstractFrontier * @see org.archive.crawler.framework.Frontier#schedule(org.archive.modules.CrawlURI) */ public void schedule(CrawlURI curi) { - sheetOverlaysManager.applyOverridesTo(curi); + sheetOverlaysManager.applyOverlaysTo(curi); if(curi.getClassKey()==null) { // remedial processing try { @@ -590,7 +594,7 @@ public abstract class AbstractFrontier * @param caUri CrawlURI. */ public void receive(CrawlURI curi) { - sheetOverlaysManager.applyOverridesTo(curi); + sheetOverlaysManager.applyOverlaysTo(curi); // prefer doing asap if already in manager thread doOrEnqueue(new ScheduleAlways(curi)); } @@ -956,7 +960,7 @@ public abstract class AbstractFrontier String uriHopsViaString = read.substring(3).trim(); CrawlURI curi = CrawlURI.fromHopsViaString(uriHopsViaString); if(scope!=null) { - sheetOverlaysManager.applyOverridesTo(curi); + sheetOverlaysManager.applyOverlaysTo(curi); try { KeyedProperties.loadOverridesFrom(curi); if(!scope.accepts(curi)) { @@ -1111,14 +1115,15 @@ public abstract class AbstractFrontier } /** - * Checks if a recently completed CrawlURI that did not finish successfully - * needs to be retried (processed again after some time elapses) + * Checks if a recently processed CrawlURI that did not finish successfully + * needs to be reenqueued (and thus possibly, processed again after some + * time elapses) * * @param curi * The CrawlURI to check * @return True if we need to retry. */ - protected boolean needsRetrying(CrawlURI curi) { + protected boolean needsReenqueuing(CrawlURI curi) { if (overMaxRetries(curi)) { return false; } @@ -1144,6 +1149,10 @@ public abstract class AbstractFrontier // TODO: consider if any others (S_TIMEOUT in some cases?) deserve // retry return true; + case S_UNATTEMPTED: + if(curi.includesRetireDirective()) { + return true; + } // otherwise, fall-through: no status is an error without queue-directive default: return false; } 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 2b410f48..f6e13775 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java @@ -288,7 +288,6 @@ implements Checkpointable, BeanNameAware { for (String key : allQueues.keySet()) { WorkQueue q = allQueues.get(key); q.getOnInactiveQueues().clear(); - q.setSessionBalance(0); if(q.isRetired()) { getRetiredQueues().add(key); } else { 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 fbf97339..9f05309a 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java +++ b/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java @@ -58,56 +58,59 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, protected final String classKey; /** whether queue is active (ready/in-process/snoozed) or on a waiting queue */ - private boolean active = true; + protected boolean active = true; /** Total number of stored items */ - private long count = 0; + protected long count = 0; /** Total number of items ever enqueued */ - private long enqueueCount = 0; + protected long enqueueCount = 0; /** Whether queue is already in lifecycle stage */ - private boolean isHeld = false; + protected boolean isHeld = false; /** Time to wake, if snoozed */ - private long wakeTime = 0; + protected long wakeTime = 0; /** assigned precedence */ - private PrecedenceProvider precedenceProvider = new SimplePrecedenceProvider(1); + protected PrecedenceProvider precedenceProvider = new SimplePrecedenceProvider(1); /** set of by-precedence inactive-queues on which WorkQueue is waiting */ - private Set onInactiveQueues = new HashSet(); + protected Set onInactiveQueues = new HashSet(); - /** Running 'budget' indicating whether queue should stay active */ - private int sessionBalance = 0; + /** Per-session 'budget' controlling activity duration */ + protected int sessionBudget = 0; /** Cost of the last item to be charged against queue */ - private int lastCost = 0; + protected int lastCost = 0; /** Total number of items charged against queue; with totalExpenditure * can be used to calculate 'average cost'. */ - private long costCount = 0; + protected long costCount = 0; /** Running tally of total expenditures on this queue */ - private long totalExpenditure = 0; + protected long totalExpenditure = 0; + /** Record of expenditures at last activation (session start) */ + protected long expenditureAtLastActivation = 0; + /** Total to spend on this queue over its lifetime */ - private long totalBudget = 0; + protected long totalBudget = 0; /** The next item to be returned */ transient protected CrawlURI peekItem = null; /** Last URI enqueued */ - private String lastQueued; + protected String lastQueued; /** Last URI peeked */ - private String lastPeeked; + protected String lastPeeked; /** time of last dequeue (disposition of some URI) **/ - private long lastDequeueTime; + protected long lastDequeueTime; /** count of errors encountered */ - private long errorCount = 0; + protected long errorCount = 0; /** Substats for all CrawlURIs in this group */ protected FetchStats substats = new FetchStats(); @@ -202,12 +205,16 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, } /** - * Set the session 'activity budget balance' to the given value + * Set the session 'activity budget' to the given value. Automatically + * reset continually as new CrawlURIs are enqueued; a direct change + * here by operator will not persist. Instead, change the 'balanceReplenishAmount' + * (or overlay its value with a URI/queue-specific value) to affect this + * value. * * @param balance to use */ - public void setSessionBalance(int balance) { - this.sessionBalance = balance; + void setSessionBudget(int budget) { + this.sessionBudget = budget; } /** @@ -215,30 +222,39 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, * * @return session balance */ - public int getSessionBalance() { - return this.sessionBalance; + public int getSessionBudget() { + return this.sessionBudget; } + public void startActiveSession() { + expenditureAtLastActivation = totalExpenditure; + } + /** * Set the total expenditure level allowable before queue is * considered inherently 'over-budget'. * + * Automatically reset continually as new CrawlURIs are enqueued; a direct change + * here by operator will not persist. Instead, change the 'queueTotalBudget' + * (or overlay its value with a URI/queue-specific value) to affect this + * value. + * * @param budget */ - public void setTotalBudget(long budget) { + void setTotalBudget(long budget) { this.totalBudget = budget; } /** - * Check whether queue has temporarily or permanently exceeded - * its budget. + * Check whether queue has temporarily (session) or permanently (total) + * exceeded its budgets. * - * @return true if queue is over its set budget(s) + * @return true if queue is over either of its set budget(s) */ public boolean isOverBudget() { - // check whether running balance is depleted + // check whether session budget exceeded // or totalExpenditure exceeds totalBudget - return this.sessionBalance <= 0 + return (sessionBudget > 0 && (totalExpenditure - expenditureAtLastActivation) > sessionBudget) || (this.totalBudget >= 0 && this.totalExpenditure >= this.totalBudget); } @@ -252,49 +268,28 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, } /** - * Increase the internal running budget to be used before - * deactivating the queue + * Decrease the internal running budget by the given amount. (Use + * negative value to effect 'refund'/undo.) * - * @param amount amount to increment - * @return updated budget value - */ - public int incrementSessionBalance(int amount) { - this.sessionBalance = this.sessionBalance + amount; - return this.sessionBalance; - } - - /** - * Decrease the internal running budget by the given amount. * @param amount tp decrement * @return updated budget value */ - public int expend(int amount) { - this.sessionBalance = this.sessionBalance - amount; + public void expend(int amount) { this.totalExpenditure = this.totalExpenditure + amount; - this.lastCost = amount; - this.costCount++; - return this.sessionBalance; + if(amount >= 0) { + this.lastCost = amount; + this.costCount++; + } else { + this.costCount--; + } } - /** - * A URI should not have been charged against queue (eg - * it was disregarded); return the amount expended - * @param amount to return - * @return updated budget value - */ - public int refund(int amount) { - this.sessionBalance = this.sessionBalance + amount; - this.totalExpenditure = this.totalExpenditure - amount; - this.costCount--; - return this.sessionBalance; - } /** * Note an error and assess an extra penalty. * @param penalty additional amount to deduct */ public void noteError(int penalty) { - this.sessionBalance = this.sessionBalance - penalty; this.totalExpenditure = this.totalExpenditure + penalty; errorCount++; } @@ -395,12 +390,10 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, * @param frontier Work queues manager. * @param curi CrawlURI to update. */ - public void update(final WorkQueueFrontier frontier, CrawlURI curi) { + protected void update(final WorkQueueFrontier frontier, CrawlURI curi) { try { insert(frontier, curi, true); } catch (IOException e) { - //FIXME better exception handling - e.printStackTrace(); throw new RuntimeException(e); } } @@ -536,7 +529,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, map.put("precedence", getPrecedence()); map.put("itemCount", count); map.put("enqueueCount", enqueueCount); - map.put("sessionBalance", sessionBalance); + map.put("sessionBalance", getSessionBalance()); map.put("lastCost", lastCost); map.put("averageCost", (double) totalExpenditure / costCount); if (lastDequeueTime != 0) { @@ -558,6 +551,10 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, return map; } + protected long getSessionBalance() { + return sessionBudget - (totalExpenditure-expenditureAtLastActivation); + } + public void shortReportLineTo(PrintWriter writer) { // queue name writer.print(classKey); @@ -571,7 +568,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, // enqueue count writer.print(Long.toString(enqueueCount)); writer.print(" "); - writer.print(sessionBalance); + writer.print(getSessionBalance()); writer.print(" "); writer.print(lastCost); writer.print("("); @@ -643,7 +640,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, writer.print(Long.toString(totalBudget)); writer.print(")\n"); writer.print(" active balance: "); - writer.print(sessionBalance); + writer.print(getSessionBalance()); writer.print("\n last(avg) cost: "); writer.print(lastCost); writer.print("("); @@ -669,7 +666,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, * * @param b new value for retired status */ - public void setRetired(boolean b) { + void setRetired(boolean b) { this.retired = b; } 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 971462a7..3b260175 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java @@ -23,7 +23,6 @@ import static org.archive.crawler.event.CrawlURIDispositionEvent.Disposition.DEF import static org.archive.crawler.event.CrawlURIDispositionEvent.Disposition.DISREGARDED; import static org.archive.crawler.event.CrawlURIDispositionEvent.Disposition.FAILED; import static org.archive.crawler.event.CrawlURIDispositionEvent.Disposition.SUCCEEDED; -import static org.archive.modules.CoreAttributeConstants.A_FORCE_RETIRE; import static org.archive.modules.fetcher.FetchStatusCodes.S_DEFERRED; import static org.archive.modules.fetcher.FetchStatusCodes.S_RUNTIME_EXCEPTION; @@ -215,7 +214,7 @@ implements Closeable, protected AtomicInteger snoozedOverflowCount = new AtomicInteger(0); protected static int MAX_SNOOZED_IN_MEMORY = 5000; - /** URIs scheduled for reenqueuing at future date*/ + /** URIs scheduled to be re-enqueued at future date */ protected StoredSortedMap futureUris; transient protected WorkQueue longestActiveQueue = null; @@ -346,7 +345,7 @@ implements Closeable, */ @Override public void schedule(CrawlURI curi) { - sheetOverlaysManager.applyOverridesTo(curi); + sheetOverlaysManager.applyOverlaysTo(curi); try { KeyedProperties.loadOverridesFrom(curi); if(curi.getClassKey()==null) { @@ -391,6 +390,11 @@ implements Closeable, 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); @@ -428,7 +432,7 @@ implements Closeable, if(getHoldQueues()) { deactivateQueue(wq); } else { - replenishSessionBalance(wq); + wq.startActiveSession(); readyQueue(wq); } } @@ -468,7 +472,6 @@ implements Closeable, protected void deactivateQueue(WorkQueue wq) { assert Thread.currentThread() == managerThread; - wq.setSessionBalance(0); // zero out session balance int precedence = wq.getPrecedence(); if(!wq.getOnInactiveQueues().contains(precedence)) { // not already on target, add @@ -548,7 +551,11 @@ implements Closeable, abstract Queue getRetiredQueues(); /** - * Accomodate any changes in settings. + * Accommodate any changes in retirement-determining settings (like + * total-budget or force-retire changes/overlays. + * + * (Essentially, exists to be called from tools like the UI + * Scripting Console when the operator knows it's necessary.) */ public void reconsiderRetiredQueues() { @@ -558,7 +565,7 @@ implements Closeable, // be re-retired; if not, they'll get a chance to become // active under the new rules. - // TODO: Only do this when necessary. + // TODO: Do this automatically, only when necessary. String key = getRetiredQueues().poll(); while (key != null) { @@ -800,7 +807,7 @@ implements Closeable, deactivateQueue(candidateQ); return; } - replenishSessionBalance(candidateQ); + candidateQ.startActiveSession(); if (candidateQ.isOverBudget()) { // if still over-budget after an activation & replenishing, // retire @@ -836,44 +843,6 @@ implements Closeable, highestPrecedenceWaiting = Integer.MAX_VALUE; } - /** - * Replenish the budget of the given queue by the appropriate amount. - * - * @param queue queue to replenish - */ - private void replenishSessionBalance(WorkQueue queue) { - assert queue.peekItem == null : "unexpected peekItem set"; - // get a CrawlURI for override context purposes - CrawlURI contextUri = queue.peek(this); - if(contextUri == null) { - // use globals TODO: fix problems this will cause if - // global total budget < override on empty queue - queue.setSessionBalance(getBalanceReplenishAmount()); - queue.setTotalBudget(getQueueTotalBudget()); - return; - } - // TODO: consider confusing cross-effects of this and IP-based politeness - - contextUri.setOverlayMapsSource(sheetOverlaysManager); - try { - // TODO:SPRINGY set override - KeyedProperties.loadOverridesFrom(contextUri); - - //queue.setSessionBalance(contextUri.get(this, BALANCE_REPLENISH_AMOUNT)); - queue.setSessionBalance(getBalanceReplenishAmount()); - - // reset total budget (it may have changed) - // TODO: is this the best way to be sensitive to potential mid-crawl changes - // TODO:SPRINGY set override - //long totalBudget = contextUri.get(this, QUEUE_TOTAL_BUDGET); - long totalBudget = getQueueTotalBudget(); - queue.setTotalBudget(totalBudget); - queue.unpeek(contextUri); // don't insist on that URI being next released - } finally { - KeyedProperties.clearOverridesFrom(contextUri); - } - } - /** * Enqueue the given queue to either readyClassQueues or inactiveQueues, * as appropriate. @@ -975,7 +944,11 @@ implements Closeable, * and other related URIs may become eligible for release * via the next next() call, as a result of finished(). * - * (non-Javadoc) + * TODO: make as many decisions about what happens to the CrawlURI + * (success, failure, retry) and queue (retire, snooze, ready) as + * possible elsewhere, such as in DispositionProcessor. Then, break + * this into simple branches or focused methods for each case. + * * @see org.archive.crawler.framework.Frontier#finished(org.archive.modules.CrawlURI) */ protected void processFinish(CrawlURI curi) { @@ -985,42 +958,31 @@ implements Closeable, curi.incrementFetchAttempts(); logNonfatalErrors(curi); + WorkQueue wq = (WorkQueue) curi.getHolder(); + // always refresh budgeting values from current curi + // (whose overlay settings should be active here) + wq.setSessionBudget(getBalanceReplenishAmount()); + wq.setTotalBudget(getQueueTotalBudget()); + assert (wq.peek(this) == curi) : "unexpected peek " + wq; inProcessQueues.remove(wq, 1); + int holderCost = curi.getHolderCost(); - if(includesRetireDirective(curi)) { - // CrawlURI is marked to trigger retirement of its queue - curi.processingCleanup(); + if (needsReenqueuing(curi)) { + // codes/errors which don't consume the URI, leaving it atop queue + if(curi.getFetchStatus()!=S_DEFERRED) { + wq.expend(holderCost); // all retries but DEFERRED cost + } + long delay_ms = retryDelayFor(curi) * 1000; + curi.processingCleanup(); // lose state that shouldn't burden retry wq.unpeek(curi); wq.update(this, curi); // rewrite any changes - retireQueue(wq); - return; - } - - if (needsRetrying(curi)) { - // Consider errors which can be retried, leaving uri atop queue - if(curi.getFetchStatus()!=S_DEFERRED) { - wq.expend(curi.getHolderCost()); // all retries but DEFERRED cost - } - long delay_sec = retryDelayFor(curi); - curi.processingCleanup(); // lose state that shouldn't burden retry - - wq.unpeek(curi); - // TODO: consider if this should happen automatically inside unpeek() - wq.update(this, curi); // rewrite any changes - if (delay_sec > 0) { - long delay_ms = delay_sec * 1000; - snoozeQueue(wq, now, delay_ms); - } else { - reenqueueQueue(wq); - } - - // Let everyone interested know that it will be retried. - appCtx.publishEvent( - new CrawlURIDispositionEvent(this,curi,DEFERRED_FOR_RETRY)); + handleQueue(wq,curi.includesRetireDirective(),now,delay_ms); + appCtx.publishEvent(new CrawlURIDispositionEvent(this,curi,DEFERRED_FOR_RETRY)); doJournalReenqueued(curi); - return; + + return; // no further dequeueing, logging, rescheduling to occur } // Curi will definitely be disposed of without retry, so remove from queue @@ -1028,56 +990,45 @@ implements Closeable, decrementQueuedCount(1); log(curi); + if (curi.isSuccess()) { - totalProcessedBytes.addAndGet(curi.getRecordedSize()); + // codes deemed 'success' incrementSucceededFetchCount(); - // Let everyone know in case they want to do something before we strip the curi. - appCtx.publishEvent( - new CrawlURIDispositionEvent(this,curi,SUCCEEDED)); + totalProcessedBytes.addAndGet(curi.getRecordedSize()); + appCtx.publishEvent(new CrawlURIDispositionEvent(this,curi,SUCCEEDED)); doJournalFinishedSuccess(curi); - wq.expend(curi.getHolderCost()); // successes cost + } else if (isDisregarded(curi)) { - // Check for codes that mean that while we the crawler did - // manage to schedule it, it must be disregarded for some reason. + // codes meaning 'undo' (even though URI was enqueued, + // we now want to disregard it from normal success/failure tallies) + // (eg robots-excluded, operator-changed-scope, etc) incrementDisregardedUriCount(); - // Let interested listeners know of disregard disposition. - appCtx.publishEvent( - new CrawlURIDispositionEvent(this,curi,DISREGARDED)); + appCtx.publishEvent(new CrawlURIDispositionEvent(this,curi,DISREGARDED)); + holderCost = 0; // no charge for disregarded URIs + // TODO: consider reinstating forget-URI capability, so URI could be + // re-enqueued if discovered again doJournalDisregarded(curi); + + } else { + // codes meaning 'failure' + incrementFailedFetchCount(); + appCtx.publishEvent(new CrawlURIDispositionEvent(this,curi,FAILED)); // if exception, also send to crawlErrors if (curi.getFetchStatus() == S_RUNTIME_EXCEPTION) { Object[] array = { curi }; loggerModule.getRuntimeErrors().log(Level.WARNING, curi.getUURI() .toString(), array); - } - // TODO: consider reinstating forget-uri - } else { - // In that case FAILURE, note & log - //Let interested listeners know of failed disposition. - appCtx.publishEvent( - new CrawlURIDispositionEvent(this,curi,FAILED)); - // if exception, also send to crawlErrors - if (curi.getFetchStatus() == S_RUNTIME_EXCEPTION) { - Object[] array = { curi }; - this.loggerModule.getRuntimeErrors().log(Level.WARNING, curi.getUURI() - .toString(), array); - } - incrementFailedFetchCount(); - // let queue note error - //TODO:SPRINGY set overrides by curi or wq? - assert KeyedProperties.overridesActiveFrom(curi); - + } + // charge queue any extra error penalty wq.noteError(getErrorPenaltyAmount()); doJournalFinishedFailure(curi); - wq.expend(curi.getHolderCost()); // failures cost + } + wq.expend(holderCost); // successes & failures charge cost to queue + long delay_ms = curi.getPolitenessDelay(); - if (delay_ms > 0) { - snoozeQueue(wq,now,delay_ms); - } else { - reenqueueQueue(wq); - } + handleQueue(wq,curi.includesRetireDirective(),now,delay_ms); if(curi.getRescheduleTime()>0) { // marked up for forced-revisit at a set time @@ -1090,10 +1041,24 @@ implements Closeable, curi.processingCleanup(); } } - - protected boolean includesRetireDirective(CrawlURI curi) { - return curi.containsDataKey(A_FORCE_RETIRE) - && (Boolean)curi.getData().get(A_FORCE_RETIRE); + + /** + * Send an active queue to its next state, based on the supplied + * parameters. + * + * @param wq + * @param forceRetire + * @param now + * @param delay_ms + */ + protected void handleQueue(WorkQueue wq, boolean forceRetire, long now, long delay_ms) { + if(forceRetire) { + retireQueue(wq); + } else if (delay_ms > 0) { + snoozeQueue(wq, now, delay_ms); + } else { + reenqueueQueue(wq); + } } /** @@ -1594,7 +1559,7 @@ implements Closeable, public void considerIncluded(CrawlURI curi) { this.uriUniqFilter.note(curi.getCanonicalString()); - sheetOverlaysManager.applyOverridesTo(curi); + sheetOverlaysManager.applyOverlaysTo(curi); try { KeyedProperties.loadOverridesFrom(curi); curi.setClassKey(getClassKey(curi)); @@ -1675,5 +1640,5 @@ implements Closeable, return inProcessQueues.size(); } -} +} // TODO: slim class! Suspect it should be < 800 lines, shedding budgeting/reporting diff --git a/engine/src/main/java/org/archive/crawler/postprocessor/CandidatesProcessor.java b/engine/src/main/java/org/archive/crawler/postprocessor/CandidatesProcessor.java index 0381c84a..a2746e93 100644 --- a/engine/src/main/java/org/archive/crawler/postprocessor/CandidatesProcessor.java +++ b/engine/src/main/java/org/archive/crawler/postprocessor/CandidatesProcessor.java @@ -136,7 +136,7 @@ public class CandidatesProcessor extends Processor { if (curi.hasPrerequisiteUri() && curi.getFetchStatus() == S_DEFERRED) { CrawlURI prereq = curi.getPrerequisiteUri(); prereq.setFullVia(curi); - sheetOverlaysManager.applyOverridesTo(prereq); + sheetOverlaysManager.applyOverlaysTo(prereq); try { KeyedProperties.clearOverridesFrom(curi); KeyedProperties.loadOverridesFrom(prereq); @@ -172,7 +172,7 @@ public class CandidatesProcessor extends Processor { wref.getDestination().toString()); continue; } - sheetOverlaysManager.applyOverridesTo(candidate); + sheetOverlaysManager.applyOverlaysTo(candidate); try { KeyedProperties.clearOverridesFrom(curi); KeyedProperties.loadOverridesFrom(candidate); diff --git a/engine/src/main/java/org/archive/crawler/postprocessor/DispositionProcessor.java b/engine/src/main/java/org/archive/crawler/postprocessor/DispositionProcessor.java index 3c86a89f..049a250e 100644 --- a/engine/src/main/java/org/archive/crawler/postprocessor/DispositionProcessor.java +++ b/engine/src/main/java/org/archive/crawler/postprocessor/DispositionProcessor.java @@ -130,6 +130,22 @@ public class DispositionProcessor extends Processor { kp.put("maxPerHostBandwidthUsageKbSec",max); } + /** + * Whether to set a CrawlURI's force-retired directive, retiring + * its queue when it finishes. Mainly intended for URI-specific + * overlay settings; setting true globally will just retire all queues + * after they offer one URI, rapidly ending a crawl. + */ + { + setForceRetire(false); + } + public boolean getForceRetire() { + return (Boolean) kp.get("forceRetire"); + } + public void setForceRetire(boolean force) { + kp.put("forceRetire",force); + } + /** * Auto-discovered module providing configured (or overridden) * User-Agent value and RobotsHonoringPolicy @@ -195,6 +211,11 @@ public class DispositionProcessor extends Processor { // set politeness delay curi.setPolitenessDelay(politenessDelayFor(curi)); + // consider operator-set force-retire + if (getForceRetire()) { + curi.setForceRetire(true); + } + // TODO: set other disposition decisions // success, failure, retry(retry-delay) } diff --git a/engine/src/main/java/org/archive/crawler/prefetch/QuotaEnforcer.java b/engine/src/main/java/org/archive/crawler/prefetch/QuotaEnforcer.java index e8f467f1..e403c5d1 100644 --- a/engine/src/main/java/org/archive/crawler/prefetch/QuotaEnforcer.java +++ b/engine/src/main/java/org/archive/crawler/prefetch/QuotaEnforcer.java @@ -380,10 +380,13 @@ public class QuotaEnforcer extends Processor { protected boolean applyQuota(CrawlURI curi, String key, long actual) { long quota = (Long)kp.get(key); if (quota >= 0 && actual >= quota) { - curi.setFetchStatus(S_BLOCKED_BY_QUOTA); curi.getAnnotations().add("Q:"+key); if (getForceRetire()) { + // retire queue without disposing URI curi.setForceRetire(true); + } else { + // overquota without retire option means treat as if 'failed' + curi.setFetchStatus(S_BLOCKED_BY_QUOTA); } return true; } diff --git a/engine/src/main/java/org/archive/crawler/spring/SheetOverlaysManager.java b/engine/src/main/java/org/archive/crawler/spring/SheetOverlaysManager.java index a1154b1b..d21b0d51 100644 --- a/engine/src/main/java/org/archive/crawler/spring/SheetOverlaysManager.java +++ b/engine/src/main/java/org/archive/crawler/spring/SheetOverlaysManager.java @@ -52,7 +52,6 @@ import org.springframework.context.event.ContextRefreshedEvent; * * @contributor gojomo */ -@SuppressWarnings("unchecked") public class SheetOverlaysManager implements BeanFactoryAware, OverlayMapsSource, ApplicationListener { private static final Logger logger = Logger.getLogger(SheetOverlaysManager.class.getName()); @@ -110,7 +109,6 @@ BeanFactoryAware, OverlayMapsSource, ApplicationListener { * from the set of all DecideRuledSheetAssociation instances. * @param ruleSheets */ - @SuppressWarnings("unchecked") @Autowired(required=false) public void addRuleAssociations(Set associations) { // always keep sorted by order @@ -164,32 +162,6 @@ BeanFactoryAware, OverlayMapsSource, ApplicationListener { } } - /** - * Apply the proper overlays (by Sheet beanName) to the given CrawlURI. - * - * TODO: add guard against redundant application more than once? - * TODO: add mechanism for reapplying overlays after settings change? - * @param curi - */ - public void applyOverlays(CrawlURI curi) { - // apply SURT-based overlays - String effectiveSurt = SurtPrefixSet.getCandidateSurt(curi.getPolicyBasisUURI()); - List foundPrefixes = PrefixFinder.findKeys(sheetNamesBySurt, effectiveSurt); - for(String prefix : foundPrefixes) { - for(String name : sheetNamesBySurt.get(prefix)) { - curi.getOverlayNames().push(name); - } - } - // apply deciderule-based overlays - for(DecideRuledSheetAssociation assoc : ruleAssociations) { - if(assoc.getRules().accepts(curi)) { - curi.getOverlayNames().addAll(assoc.getTargetSheetNames()); - } - } - // even if no overlays set, let creation of empty list signal - // step has occurred -- helps ensure overlays added once-only - curi.getOverlayNames(); - } /** * Retrieve the named overlay Map. @@ -327,10 +299,32 @@ BeanFactoryAware, OverlayMapsSource, ApplicationListener { return sheet; } - public void applyOverridesTo(CrawlURI curi) { + /** + * Apply the proper overlays (by Sheet beanName) to the given CrawlURI, + * according to configured associations. + * + * TODO: add guard against redundant application more than once? + * TODO: add mechanism for reapplying overlays after settings change? + * @param curi + */ + public void applyOverlaysTo(CrawlURI curi) { curi.setOverlayMapsSource(this); - if(!curi.haveOverlayNamesBeenSet()) { - applyOverlays(curi); + // apply SURT-based overlays + String effectiveSurt = SurtPrefixSet.getCandidateSurt(curi.getPolicyBasisUURI()); + List foundPrefixes = PrefixFinder.findKeys(sheetNamesBySurt, effectiveSurt); + for(String prefix : foundPrefixes) { + for(String name : sheetNamesBySurt.get(prefix)) { + curi.getOverlayNames().push(name); + } } + // apply deciderule-based overlays + for(DecideRuledSheetAssociation assoc : ruleAssociations) { + if(assoc.getRules().accepts(curi)) { + curi.getOverlayNames().addAll(assoc.getTargetSheetNames()); + } + } + // even if no overlays set, let creation of empty list signal + // step has occurred -- helps ensure overlays added once-only + curi.getOverlayNames(); } } \ No newline at end of file diff --git a/modules/src/main/java/org/archive/modules/CrawlURI.java b/modules/src/main/java/org/archive/modules/CrawlURI.java index 2dd153d9..14e4fab7 100644 --- a/modules/src/main/java/org/archive/modules/CrawlURI.java +++ b/modules/src/main/java/org/archive/modules/CrawlURI.java @@ -1863,4 +1863,9 @@ implements MultiReporter, Serializable, OverlayContext { // deferrals from now don't count against future try resetDeferrals(); } + + public boolean includesRetireDirective() { + return containsDataKey(A_FORCE_RETIRE) + && (Boolean)getData().get(A_FORCE_RETIRE); + } }