Author: ab
Date: Tue Dec  1 14:50:15 2009
New Revision: 885776

URL: http://svn.apache.org/viewvc?rev=885776&view=rev
Log:
NUTCH-770 Timebomb for Fetcher.

Modified:
    lucene/nutch/trunk/conf/nutch-default.xml
    lucene/nutch/trunk/src/java/org/apache/nutch/fetcher/Fetcher.java

Modified: lucene/nutch/trunk/conf/nutch-default.xml
URL: 
http://svn.apache.org/viewvc/lucene/nutch/trunk/conf/nutch-default.xml?rev=885776&r1=885775&r2=885776&view=diff
==============================================================================
--- lucene/nutch/trunk/conf/nutch-default.xml (original)
+++ lucene/nutch/trunk/conf/nutch-default.xml Tue Dec  1 14:50:15 2009
@@ -601,6 +601,15 @@
   <description>If true, fetcher will store content.</description>
 </property>
 
+<property>
+  <name>fetcher.timelimit.mins</name>
+  <value>-1</value>
+  <description>This is the number of minutes allocated to the fetching.
+  Once this value is reached, any remaining entry from the input URL list is 
skipped 
+  and all active queues are emptied. The default value of -1 deactivates the 
time limit.
+  </description>
+</property>
+
 <!-- indexer properties -->
 
 <property>
@@ -1277,4 +1286,14 @@
   </description>
 </property>
 
+<!-- solr index properties -->
+<property>
+  <name>solrindex.mapping.file</name>
+  <value>solrindex-mapping.xml</value>
+  <description>
+  Defines the name of the file that will be used in the mapping of internal
+  nutch field names to solr index fields as specified in the target Solr 
schema.
+  </description>
+</property>
+
 </configuration>

Modified: lucene/nutch/trunk/src/java/org/apache/nutch/fetcher/Fetcher.java
URL: 
http://svn.apache.org/viewvc/lucene/nutch/trunk/src/java/org/apache/nutch/fetcher/Fetcher.java?rev=885776&r1=885775&r2=885776&view=diff
==============================================================================
--- lucene/nutch/trunk/src/java/org/apache/nutch/fetcher/Fetcher.java (original)
+++ lucene/nutch/trunk/src/java/org/apache/nutch/fetcher/Fetcher.java Tue Dec  
1 14:50:15 2009
@@ -222,6 +222,12 @@
       setEndTime(System.currentTimeMillis() - crawlDelay);
     }
     
+    public synchronized int emptyQueue() {
+      int presize = queue.size();
+      queue.clear();
+      return presize;
+    }
+    
     public int getQueueSize() {
       return queue.size();
     }
@@ -299,6 +305,7 @@
     boolean byIP;
     long crawlDelay;
     long minCrawlDelay;
+    long timelimit = -1;
     Configuration conf;    
     
     public FetchItemQueues(Configuration conf) {
@@ -308,6 +315,7 @@
       this.byIP = conf.getBoolean("fetcher.threads.per.host.by.ip", false);
       this.crawlDelay = (long) (conf.getFloat("fetcher.server.delay", 1.0f) * 
1000);
       this.minCrawlDelay = (long) (conf.getFloat("fetcher.server.min.delay", 
0.0f) * 1000);
+      this.timelimit = conf.getLong("fetcher.timelimit.mins", -1);
     }
     
     public int getTotalSize() {
@@ -371,6 +379,29 @@
       return null;
     }
     
+    // called only once the feeder has stopped
+    public synchronized int checkTimelimit() {
+      int count = 0;
+      if (System.currentTimeMillis() >= timelimit && timelimit != -1) {
+        // emptying the queues
+        for (String id : queues.keySet()) {
+          FetchItemQueue fiq = queues.get(id);
+          if (fiq.getQueueSize() == 0) continue;
+          LOG.info("* queue: " + id + " >> timelimit! ");
+          int deleted = fiq.emptyQueue();
+          for (int i = 0; i < deleted; i++) {
+            totalSize.decrementAndGet();
+          }
+          count += deleted;
+        }
+        // there might also be a case where totalsize !=0 but number of queues
+        // == 0
+        // in which case we simply force it to 0 to avoid blocking
+        if (totalSize.get() != 0 && queues.size() == 0) totalSize.set(0);
+      }
+      return count;
+    }
+    
     public synchronized void dump() {
       for (String id : queues.keySet()) {
         FetchItemQueue fiq = queues.get(id);
@@ -389,6 +420,7 @@
     private RecordReader<Text, CrawlDatum> reader;
     private FetchItemQueues queues;
     private int size;
+    private long timelimit = -1;
     
     public QueueFeeder(RecordReader<Text, CrawlDatum> reader,
         FetchItemQueues queues, int size) {
@@ -399,11 +431,29 @@
       this.setName("QueueFeeder");
     }
     
+    public void setTimeLimit(long tl) {
+      timelimit = tl;
+    }
+
     public void run() {
       boolean hasMore = true;
       int cnt = 0;
-      
+      int timelimitcount = 0;
       while (hasMore) {
+        if (System.currentTimeMillis() >= timelimit && timelimit != -1) {
+          // enough .. lets' simply
+          // read all the entries from the input without processing them
+          try {
+            Text url = new Text();
+            CrawlDatum datum = new CrawlDatum();
+            hasMore = reader.next(url, datum);
+            timelimitcount++;
+          } catch (IOException e) {
+            LOG.fatal("QueueFeeder error reading input, record " + cnt, e);
+            return;
+          }
+          continue;
+        }
         int feed = size - queues.getTotalSize();
         if (feed <= 0) {
           // queues are full - spin-wait until they have some free space
@@ -430,7 +480,8 @@
           }
         }
       }
-      LOG.info("QueueFeeder finished: total " + cnt + " records.");
+      LOG.info("QueueFeeder finished: total " + cnt + " records + hit by time 
limit :"
+          + timelimitcount);
     }
   }
   
@@ -899,6 +950,10 @@
 
     feeder = new QueueFeeder(input, fetchQueues, threadCount * 50);
     //feeder.setPriority((Thread.MAX_PRIORITY + Thread.NORM_PRIORITY) / 2);
+    
+    // the value of the time limit is either -1 or the time where it should 
finish
+    long timelimit = getConf().getLong("fetcher.timelimit.mins", -1);
+    if (timelimit != -1) feeder.setTimeLimit(timelimit);
     feeder.start();
 
     // set non-blocking & no-robots mode for HTTP protocol plugins.
@@ -924,6 +979,14 @@
       if (!feeder.isAlive() && fetchQueues.getTotalSize() < 5) {
         fetchQueues.dump();
       }
+      
+      // check timelimit
+      if (!feeder.isAlive()) {
+        int hitByTimeLimit = fetchQueues.checkTimelimit();
+        if (hitByTimeLimit != 0) reporter.incrCounter("FetcherStatus",
+            "hitByTimeLimit", hitByTimeLimit);
+      }
+      
       // some requests seem to hang, despite all intentions
       if ((System.currentTimeMillis() - lastRequestStart.get()) > timeout) {
         if (LOG.isWarnEnabled()) {
@@ -947,6 +1010,16 @@
       LOG.info("Fetcher: segment: " + segment);
     }
 
+    // set the actual time for the timelimit relative
+    // to the beginning of the whole job and not of a specific task
+    // otherwise it keeps trying again if a task fails
+    long timelimit = getConf().getLong("fetcher.timelimit.mins", -1);
+    if (timelimit != -1) {
+      timelimit = System.currentTimeMillis() + (timelimit * 60 * 1000);
+      LOG.info("Fetcher Timelimit set for : " + timelimit);
+      getConf().setLong("fetcher.timelimit.mins", timelimit);
+    }
+        
     JobConf job = new NutchJob(getConf());
     job.setJobName("fetch " + segment);
 


Reply via email to