From e1bbbf5a597540f64fce8630d32ebe14a6df02c6 Mon Sep 17 00:00:00 2001 From: gojomo Date: Thu, 3 Dec 2009 19:37:27 +0000 Subject: [PATCH] [HER-1704] WorkQueueFrontier.snoozedClassQueues not bounded in size, not reported in useful order * WorkQueueFrontier.java add snoozedOverflow map of (wakeTime)->(DelayedWorkQueue) for snoozes over a capped size maintain in parallel to snoozedClassQueues DelayQueue, maintain atomic size count in standard report, presort snoozedClassQueues before displaying * DelayedWorkQueue.java move to top-level so serialization doesn't pull out whole frontier * BdbFrontier.java initialize snoozedOverflow * BdbMultipleWorkQueues.java improve error message seen on problematic checkpoint-recovery --- .../archive/crawler/frontier/BdbFrontier.java | 9 +- .../frontier/BdbMultipleWorkQueues.java | 2 +- .../crawler/frontier/DelayedWorkQueue.java | 93 ++++++++ .../crawler/frontier/WorkQueueFrontier.java | 198 ++++++------------ 4 files changed, 164 insertions(+), 138 deletions(-) create mode 100644 engine/src/main/java/org/archive/crawler/frontier/DelayedWorkQueue.java 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 657cd8b7..45cb2283 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java @@ -290,11 +290,12 @@ implements Checkpointable, BeanNameAware { retiredQueues = bdb.getStoredQueue("retiredQueues", String.class, false); - // small risk of OutOfMemoryError: in large crawls with many - // unresponsive queues, an unbounded number of snoozed queues - // may exist + // primary snoozed queues snoozedClassQueues = new DelayQueue(); - + // just in case: overflow for extreme situations + snoozedOverflow = bdb.getStoredMap( + "snoozedOverflow", Long.class, DelayedWorkQueue.class, true, false); + this.futureUris = bdb.getStoredMap( "futureUris", Long.class, CrawlURI.class, true, recoveryCheckpoint!=null); diff --git a/engine/src/main/java/org/archive/crawler/frontier/BdbMultipleWorkQueues.java b/engine/src/main/java/org/archive/crawler/frontier/BdbMultipleWorkQueues.java index 0561b6f2..80d4f820 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/BdbMultipleWorkQueues.java +++ b/engine/src/main/java/org/archive/crawler/frontier/BdbMultipleWorkQueues.java @@ -337,7 +337,7 @@ public class BdbMultipleWorkQueues { } if (status!=OperationStatus.SUCCESS) { - LOGGER.severe("failed; "+status+ " "+curi); + LOGGER.severe("URI enqueueing failed; "+status+ " "+curi); } } diff --git a/engine/src/main/java/org/archive/crawler/frontier/DelayedWorkQueue.java b/engine/src/main/java/org/archive/crawler/frontier/DelayedWorkQueue.java new file mode 100644 index 00000000..fda85161 --- /dev/null +++ b/engine/src/main/java/org/archive/crawler/frontier/DelayedWorkQueue.java @@ -0,0 +1,93 @@ +/* + * This file is part of the Heritrix web crawler (crawler.archive.org). + * + * Licensed to the Internet Archive (IA) by one or more individual + * contributors. + * + * The IA licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.archive.crawler.frontier; + +import java.io.Serializable; +import java.util.concurrent.Delayed; +import java.util.concurrent.TimeUnit; + +/** + * A named WorkQueue wrapped with a wake time, perhaps referenced only + * by name. + * + * @contributor gojomo + */ +class DelayedWorkQueue implements Delayed, Serializable { + private static final long serialVersionUID = 1L; + + public String classKey; + public long wakeTime; + + /** + * Reference to the WorkQueue, perhaps saving a deserialization + * from allQueues. + */ + protected transient WorkQueue workQueue; + + public DelayedWorkQueue(WorkQueue queue) { + this.classKey = queue.getClassKey(); + this.wakeTime = queue.getWakeTime(); + this.workQueue = queue; + } + + // TODO: consider if this should be method on WorkQueueFrontier + public WorkQueue getWorkQueue(WorkQueueFrontier wqf) { + if (workQueue == null) { + // This is a recently deserialized DelayedWorkQueue instance + WorkQueue result = wqf.getQueueFor(classKey); + this.workQueue = result; + } + return workQueue; + } + + public long getDelay(TimeUnit unit) { + return unit.convert( + wakeTime - System.currentTimeMillis(), + TimeUnit.MILLISECONDS); + } + + public String getClassKey() { + return classKey; + } + + public long getWakeTime() { + return wakeTime; + } + + public void setWakeTime(long time) { + this.wakeTime = time; + } + + public int compareTo(Delayed obj) { + if (this == obj) { + return 0; // for exact identity only + } + DelayedWorkQueue other = (DelayedWorkQueue) obj; + if (wakeTime > other.getWakeTime()) { + return 1; + } + if (wakeTime < other.getWakeTime()) { + return -1; + } + // at this point, the ordering is arbitrary, but still + // must be consistent/stable over time + return this.classKey.compareTo(other.getClassKey()); + } + +} \ No newline at end of file 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 feb3c2a5..dd4412d9 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java @@ -30,9 +30,8 @@ import static org.archive.modules.fetcher.FetchStatusCodes.S_RUNTIME_EXCEPTION; import java.io.Closeable; import java.io.IOException; import java.io.PrintWriter; -import java.io.Serializable; -import java.lang.ref.SoftReference; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Iterator; import java.util.LinkedHashMap; @@ -43,6 +42,7 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.DelayQueue; import java.util.concurrent.Delayed; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.logging.Level; import java.util.logging.Logger; import java.util.regex.Pattern; @@ -50,6 +50,7 @@ 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; import org.archive.crawler.event.CrawlURIDispositionEvent; @@ -61,8 +62,6 @@ import org.archive.spring.KeyedProperties; import org.archive.util.ArchiveUtils; import org.archive.util.ObjectIdentityCache; import org.archive.util.ObjectIdentityMemCache; -import org.archive.util.Transform; -import org.archive.util.Transformer; import org.springframework.beans.BeansException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; @@ -212,7 +211,11 @@ implements Closeable, * All per-class queues held in snoozed state, sorted by wake time. */ transient protected DelayQueue snoozedClassQueues; + protected StoredSortedMap snoozedOverflow; + protected AtomicInteger snoozedOverflowCount = new AtomicInteger(0); + protected static int MAX_SNOOZED_IN_MEMORY = 5000; + /** URIs scheduled for reenqueuing at future date*/ protected StoredSortedMap futureUris; transient protected WorkQueue longestActiveQueue = null; @@ -884,10 +887,32 @@ implements Closeable, protected void wakeQueues() { DelayedWorkQueue waked; while((waked = snoozedClassQueues.poll())!=null) { - WorkQueue queue = waked.getWorkQueue(); + WorkQueue queue = waked.getWorkQueue(this); queue.setWakeTime(0); reenqueueQueue(queue); } + // also consider overflow (usually empty) + long now = System.currentTimeMillis(); + Iterator iter = + snoozedOverflow.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() snoozeToInactiveDelayMs && !inactiveQueues.isEmpty()) { -// deactivateQueue(wq); -// } else { - snoozedClassQueues.add(new DelayedWorkQueue(wq)); -// } + DelayedWorkQueue dq = new DelayedWorkQueue(wq); + if(snoozedClassQueues.size() inactiveQueues : getInactiveQueuesByPrecedence().values()) { @@ -1288,7 +1315,7 @@ implements Closeable, if(obj instanceof WorkQueue) { q = (WorkQueue)obj; } else if (obj instanceof DelayedWorkQueue) { - q = ((DelayedWorkQueue)obj).getWorkQueue(); + q = ((DelayedWorkQueue)obj).getWorkQueue(this); } else { try { q = this.allQueues.get((String)obj); @@ -1316,7 +1343,7 @@ implements Closeable, int allCount = allQueues.size(); int inProcessCount = inProcessQueues.uniqueSet().size(); int readyCount = readyClassQueues.size(); - int snoozedCount = snoozedClassQueues.size(); + int snoozedCount = getSnoozedCount(); int activeCount = inProcessCount + readyCount + snoozedCount; int inactiveCount = getTotalInactiveQueues(); int retiredCount = getRetiredQueues().size(); @@ -1413,16 +1440,10 @@ implements Closeable, this.readyClassQueues.size(), REPORT_MAX_QUEUES); w.print("\n -----===== SNOOZED QUEUES =====-----\n"); - Transformer ter = - new Transformer() { - public WorkQueue transform(DelayedWorkQueue dwq) { - return dwq.getWorkQueue(); - } - }; - Transform t = - new Transform(snoozedClassQueues, ter); - copy = extractSome(t, REPORT_MAX_QUEUES); - appendQueueReports(w, "SNOOZED", copy.iterator(), copy.size(), REPORT_MAX_QUEUES); + Object[] objs = snoozedClassQueues.toArray(); + DelayedWorkQueue[] qs = Arrays.copyOf(objs,objs.length,DelayedWorkQueue[].class); + Arrays.sort(qs); + appendQueueReports(w, "SNOOZED", new ObjectArrayIterator(qs), getSnoozedCount(), REPORT_MAX_QUEUES); w.print("\n -----===== INACTIVE QUEUES =====-----\n"); SortedMap> sortedInactives = getInactiveQueuesByPrecedence(); @@ -1478,22 +1499,28 @@ implements Closeable, int total, int max) { Object obj; WorkQueue q; - for(int count = 0; iterator.hasNext() && (count < max); count++) { + int count; + for(count = 0; iterator.hasNext() && (count < max); count++) { obj = iterator.next(); if (obj == null) { continue; } - q = (obj instanceof WorkQueue) - ? (WorkQueue)obj - : this.allQueues.get((String)obj); + if(obj instanceof WorkQueue) { + q = (WorkQueue)obj; + } else if (obj instanceof DelayedWorkQueue) { + q = (WorkQueue)((DelayedWorkQueue)obj).getWorkQueue(this); + } else { + q = this.allQueues.get((String)obj); + } if(q == null) { w.print("WARNING: No report for queue "+obj); } w.println(label+"#"+count+":"); q.reportTo(w); } - if(total > max) { - w.print("...and " + (total - max) + " more "+label+".\n"); + count++; + if(count < total) { + w.print("...and " + (total - count) + " more "+label+".\n"); } } @@ -1550,19 +1577,24 @@ implements Closeable, } int inProcessCount = inProcessQueues.uniqueSet().size(); int readyCount = readyClassQueues.size(); - int snoozedCount = snoozedClassQueues.size(); + int snoozedCount = getSnoozedCount(); int activeCount = inProcessCount + readyCount + snoozedCount; int inactiveCount = getTotalInactiveQueues(); int totalQueueCount = (activeCount+inactiveCount); return (totalQueueCount == 0) ? 0 : queuedUriCount.get() / totalQueueCount; } + + protected int getSnoozedCount() { + return snoozedClassQueues.size() + snoozedOverflowCount.get(); + } + public float congestionRatio() { if(inProcessQueues==null || readyClassQueues==null || snoozedClassQueues==null) { return 0; } int inProcessCount = inProcessQueues.uniqueSet().size(); int readyCount = readyClassQueues.size(); - int snoozedCount = snoozedClassQueues.size(); + int snoozedCount = getSnoozedCount(); int activeCount = inProcessCount + readyCount + snoozedCount; int eligibleInactiveCount = getTotalEligibleInactiveQueues(); return (float)(activeCount + eligibleInactiveCount) / (inProcessCount + snoozedCount); @@ -1592,105 +1624,5 @@ implements Closeable, return inProcessQueues.size(); } - class DelayedWorkQueue implements Delayed, Serializable { - - private static final long serialVersionUID = 1L; - - public String classKey; - public long wakeTime; - - /** - * Something can become a WorkQueue instance. This can be three things: - * - *
    - *
  1. null, if this DelayedWorkQueue instance was recently - * deserialized; - *
  2. A SoftReference<WorkQueue>, if the WorkQueue's waitTime - * exceeded SNOOZE_LONG_MS - *
  3. A hard WorkQueue reference, if the WorkQueue's waitTime did not - * exceed SNOOZE_LONG_MS. Idea here is that we thought we - * needed the WorkQueue soon and didn't want to risk losing the - * instance. - *
- * - * The {@link #getWorkQueue()} method figures out what to return in - * all three of the above cases. - */ - private transient Object workQueue; - - public DelayedWorkQueue(WorkQueue queue) { - this.classKey = queue.getClassKey(); - this.wakeTime = queue.getWakeTime(); - - this.workQueue = queue; - } - - private void setWorkQueue(WorkQueue queue) { - long wakeTime = queue.getWakeTime(); - long delay = wakeTime - System.currentTimeMillis(); - if (delay > getSnoozeLongMs()) { - this.workQueue = new SoftReference(queue); - } else { - this.workQueue = queue; - } - } - - - public WorkQueue getWorkQueue() { - if (workQueue == null) { - // This is a recently deserialized DelayedWorkQueue instance - WorkQueue result = getQueueFor(classKey); - setWorkQueue(result); - return result; - } - if (workQueue instanceof SoftReference) { - @SuppressWarnings("unchecked") - SoftReference ref = (SoftReference)workQueue; - WorkQueue result = ref.get(); - if (result == null) { - result = getQueueFor(classKey); - } - setWorkQueue(result); - return result; - } - return (WorkQueue)workQueue; - } - - public long getDelay(TimeUnit unit) { - return unit.convert( - wakeTime - System.currentTimeMillis(), - TimeUnit.MILLISECONDS); - } - - public String getClassKey() { - return classKey; - } - - public long getWakeTime() { - return wakeTime; - } - - public void setWakeTime(long time) { - this.wakeTime = time; - } - - public int compareTo(Delayed obj) { - if (this == obj) { - return 0; // for exact identity only - } - DelayedWorkQueue other = (DelayedWorkQueue) obj; - if (wakeTime > other.getWakeTime()) { - return 1; - } - if (wakeTime < other.getWakeTime()) { - return -1; - } - // at this point, the ordering is arbitrary, but still - // must be consistent/stable over time - return this.classKey.compareTo(other.getClassKey()); - } - - } - }