Merge pull request #827 from sebastian-nagel/NUTCH-3067

NUTCH-3067 Improve performance of FetchItemQueues if error state is preserved

Author: sebastian-nagel

conf/nutch-default.xml

@@ -1173,6 +1173,17 @@
   </description>
 </property>
 
+<property>
+  <name>fetcher.exceptions.per.queue.clear.after</name>
+  <value>1800</value>
+  <description>Time in seconds after which exception counters in
+  queues can be cleared. This happens only if no items are queued in
+  this queue and the configured time span has elapsed in addition to
+  the crawl delay including the exponential backoff time, see
+  fetcher.exceptions.per.queue.delay.
+  </description>
+</property>
+
 <property>
   <name>fetcher.throughput.threshold.pages</name>
   <value>-1</value>
@@ -1187,7 +1198,11 @@
   <name>fetcher.throughput.threshold.retries</name>
   <value>5</value>
   <description>The number of times the fetcher.throughput.threshold.pages is allowed to be exceeded.
-  This settings prevents accidental slow downs from immediately killing the fetcher thread.
+  This settings prevents accidental slow downs from immediately shutting down the fetcher threads.
+  The throughput is checked approx. every second. Note: the number of retries should be lower
+  than the timeout defined by mapreduce.task.timeout and fetcher.threads.timeout.divisor (see there).
+  For the default values, the timeout is 300 seconds. Consequently, the number of retries should be
+  significantly lower.
   </description>
 </property>
 

src/java/org/apache/nutch/fetcher/FetchItemQueue.java

@@ -41,6 +41,8 @@ public class FetchItemQueue {
   private static final Logger LOG = LoggerFactory
       .getLogger(MethodHandles.lookup().lookupClass());
 
+  private static Text variableFetchDelayKey = new Text("_variableFetchDelay_");
+
   List<FetchItem> queue = Collections
       .synchronizedList(new LinkedList<FetchItem>());
   AtomicInteger inProgress = new AtomicInteger();
@@ -50,18 +52,20 @@ public class FetchItemQueue {
   long minCrawlDelay;
   int maxThreads;
   Text cookie;
-  Text variableFetchDelayKey = new Text("_variableFetchDelay_");
   boolean variableFetchDelaySet = false;
   // keep track of duplicates if fetcher.follow.outlinks.depth > 0. Some urls may 
   // not get followed due to hash collisions. Hashing is used to reduce memory
   // usage.
-  Set<Integer> alreadyFetched = new HashSet<>();
+  Set<Integer> alreadyFetched;
 
   public FetchItemQueue(Configuration conf, int maxThreads, long crawlDelay,
       long minCrawlDelay) {
     this.maxThreads = maxThreads;
     this.crawlDelay = crawlDelay;
     this.minCrawlDelay = minCrawlDelay;
+    if (conf.getInt("fetcher.follow.outlinks.depth", -1) > 0) {
+      alreadyFetched = new HashSet<>();
+    }
     // ready to start
     setEndTime(System.currentTimeMillis() - crawlDelay);
   }

src/java/org/apache/nutch/fetcher/FetchItemQueues.java

@@ -57,6 +57,7 @@ public class FetchItemQueues {
   long timelimit = -1;
   int maxExceptionsPerQueue = -1;
   long exceptionsPerQueueDelay = -1;
+  long exceptionsPerQueueClearAfter = 1800 * 1000L;
   boolean feederAlive = true;
   Configuration conf;
 
@@ -88,6 +89,8 @@ public FetchItemQueues(Configuration conf) {
         "fetcher.max.exceptions.per.queue", -1);
     this.exceptionsPerQueueDelay = (long) (conf
         .getFloat("fetcher.exceptions.per.queue.delay", .0f) * 1000);
+    this.exceptionsPerQueueClearAfter = (long) (conf
+        .getFloat("fetcher.exceptions.per.queue.clear.after", 1800.0f) * 1000);
 
     int dedupRedirMaxTime = conf.getInt("fetcher.redirect.dedupcache.seconds",
         -1);
@@ -179,25 +182,41 @@ public synchronized FetchItem getFetchItem() {
       it = queues.entrySet().iterator();
     }
 
+    boolean keepExceptionState = (maxExceptionsPerQueue > -1
+        || exceptionsPerQueueDelay > 0);
+
     while (it.hasNext()) {
       FetchItemQueue fiq = it.next().getValue();
 
       // reap empty queues which do not hold state required to ensure politeness
       if (fiq.getQueueSize() == 0 && fiq.getInProgressSize() == 0) {
         if (!feederAlive) {
-          // no more fetch items added
+          // no more fetch items added: queue can be safely removed
           it.remove();
-        } else if ((maxExceptionsPerQueue > -1 || exceptionsPerQueueDelay > 0)
-            && fiq.exceptionCounter.get() > 0) {
-          // keep queue because the exceptions counter is bound to it
-          // and is required to skip or delay items on this queue
-        } else if (fiq.nextFetchTime.get() > System.currentTimeMillis()) {
+          continue;
+        }
+
+        if (fiq.nextFetchTime.get() > System.currentTimeMillis()) {
           // keep queue to have it blocked in case new fetch items of this queue
           // are added by the QueueFeeder
-        } else {
-          // empty queue without state
-          it.remove();
+          continue;
         }
+
+        if (keepExceptionState && fiq.exceptionCounter.get() > 0) {
+          if ((fiq.nextFetchTime.get() + exceptionsPerQueueClearAfter) < System
+              .currentTimeMillis()) {
+            /*
+             * the time configured by fetcher.exceptions.per.queue.clear.after
+             * has passed in addition to the delay defined by the exponential
+             * backoff
+             */
+            it.remove();
+          }
+          continue;
+        }
+
+        // queue is empty and does not hold state required to ensure politeness
+        it.remove();
         continue;
       }
 
@@ -239,9 +258,9 @@ public synchronized int checkTimelimit() {
     return count;
   }
 
-  // empties the queues (used by timebomb and throughput threshold)
+  // empties the queues (used by fetcher timelimit and throughput threshold)
   public synchronized int emptyQueues() {
-    int count = 0;
+    int count = 0, queuesDropped = 0;
 
     for (String id : queues.keySet()) {
       FetchItemQueue fiq = queues.get(id);
@@ -251,8 +270,12 @@ public synchronized int emptyQueues() {
       int deleted = fiq.emptyQueue();
       totalSize.addAndGet(-deleted);
       count += deleted;
+      queuesDropped++;
     }
 
+    LOG.info("Emptied all queues: {} queues with {} items",
+        queuesDropped, count);
+
     return count;
   }
 
@@ -282,10 +305,18 @@ public synchronized int checkExceptionThreshold(String queueid,
     if (fiq == null) {
       return 0;
     }
+
     int excCount = fiq.incrementExceptionCounter();
+    if (maxExceptions != -1 && excCount >= maxExceptions) {
+      // too many exceptions for items in this queue - purge it
+      return purgeAndBlockQueue(queueid, fiq, excCount);
+    }
+
+    long nexFetchTime = 0;
     if (delay > 0) {
-      fiq.nextFetchTime.getAndAdd(delay);
+      nexFetchTime = fiq.nextFetchTime.addAndGet(delay);
       LOG.info("* queue: {} >> delayed next fetch by {} ms", queueid, delay);
+
     } else if (exceptionsPerQueueDelay > 0) {
       /*
        * Delay the next fetch by a time span growing exponentially with the
@@ -298,32 +329,41 @@ public synchronized int checkExceptionThreshold(String queueid,
         // double the initial delay with every observed exception
         exceptionDelay *= 2L << Math.min((excCount - 2), 31);
       }
-      fiq.nextFetchTime.getAndAdd(exceptionDelay);
+      nexFetchTime = fiq.nextFetchTime.addAndGet(exceptionDelay);
       LOG.info(
           "* queue: {} >> delayed next fetch by {} ms after {} exceptions in queue",
           queueid, exceptionDelay, excCount);
     }
-    if (maxExceptions != -1 && excCount >= maxExceptions) {
-      // too many exceptions for items in this queue - purge it
-      int deleted = fiq.emptyQueue();
-      if (deleted > 0) {
-        LOG.info(
-            "* queue: {} >> removed {} URLs from queue because {} exceptions occurred",
-            queueid, deleted, excCount);
-        totalSize.getAndAdd(-deleted);
-      }
-      if (feederAlive) {
-        LOG.info("* queue: {} >> blocked after {} exceptions", queueid,
-            excCount);
-        // keep queue IDs to ensure that these queues aren't created and filled
-        // again, see addFetchItem(FetchItem)
-        queuesMaxExceptions.add(queueid);
-      }
-      return deleted;
+
+    if (timelimit > 0 && nexFetchTime > timelimit) {
+      // the next fetch would happen after the fetcher timelimit
+      LOG.info(
+          "* queue: {} >> purging queue because next fetch scheduled after fetcher timelimit",
+          queueid);
+      return purgeAndBlockQueue(queueid, fiq, excCount);
     }
+
     return 0;
   }
 
+  private int purgeAndBlockQueue(String queueid, FetchItemQueue fiq,
+      int excCount) {
+    int deleted = fiq.emptyQueue();
+    if (deleted > 0) {
+      LOG.info(
+          "* queue: {} >> removed {} URLs from queue after {} exceptions occurred",
+          queueid, deleted, excCount);
+      totalSize.getAndAdd(-deleted);
+    }
+    if (feederAlive) {
+      LOG.info("* queue: {} >> blocked after {} exceptions", queueid, excCount);
+      // keep queue IDs to ensure that these queues aren't created and filled
+      // again, see addFetchItem(FetchItem)
+      queuesMaxExceptions.add(queueid);
+    }
+    return deleted;
+  }
+
   /**
    * Increment the exception counter of a queue in case of an exception e.g.
    * timeout; when higher than a given threshold simply empty the queue.

src/java/org/apache/nutch/fetcher/Fetcher.java

@@ -163,7 +163,7 @@ private void reportStatus(Context context, FetchItemQueues fetchQueues, int page
       StringBuilder status = new StringBuilder();
       Long elapsed = Long.valueOf((System.currentTimeMillis() - start) / 1000);
 
-      float avgPagesSec = (float) pages.get() / elapsed.floatValue();
+      float avgPagesSec = pages.get() / elapsed.floatValue();
       long avgBytesSec = (bytes.get() / 128l) / elapsed.longValue();
 
       status.append(activeThreads).append(" threads (")
@@ -212,11 +212,11 @@ public void run(Context innerContext)
         int timeoutDivisor = conf.getInt("fetcher.threads.timeout.divisor", 2);
         LOG.info("Fetcher: time-out divisor: {}", timeoutDivisor);
 
-        int queueDepthMuliplier = conf.getInt("fetcher.queue.depth.multiplier",
+        int queueDepthMultiplier = conf.getInt("fetcher.queue.depth.multiplier",
             50);
 
         feeder = new QueueFeeder(innerContext, fetchQueues,
-            threadCount * queueDepthMuliplier);
+            threadCount * queueDepthMultiplier);
 
         // the value of the time limit is either -1 or the time where it should
         // finish
@@ -317,12 +317,15 @@ public void run(Context innerContext)
             if (pagesLastSec < throughputThresholdPages) {
               throughputThresholdNumRetries++;
               LOG.warn(
-                  "{}: dropping below configured threshold of {} pages per second",
-                  throughputThresholdNumRetries, throughputThresholdPages);
+                  "{}: dropping below configured threshold of {} pages per second (current throughput: {} pages/sec.)",
+                  throughputThresholdNumRetries, throughputThresholdPages,
+                  pagesLastSec);
 
               // Quit if we dropped below threshold too many times
               if (throughputThresholdNumRetries == throughputThresholdMaxRetries) {
-                LOG.warn("Dropped below threshold too many times, killing!");
+                LOG.warn(
+                    "Dropped below threshold {} times, dropping fetch queues to shut down",
+                    throughputThresholdNumRetries);
 
                 // Disable the threshold checker
                 throughputThresholdPages = -1;
@@ -427,7 +430,8 @@ else if (bandwidthTargetCheckCounter == bandwidthTargetCheckEveryNSecs) {
            * we stop the fetching now.
            */
           if ((System.currentTimeMillis() - lastRequestStart.get()) > timeout) {
-            LOG.warn("Aborting with {} hung threads.", activeThreads);
+            LOG.warn("Aborting with {} hung threads{}.", activeThreads,
+                feeder.isAlive() ? " (queue feeder still alive)" : "");
             innerContext.getCounter("FetcherStatus", "hungThreads")
                 .increment(activeThreads.get());
             for (int i = 0; i < fetcherThreads.size(); i++) {