svn commit: r1585196 - in /nutch/trunk: CHANGES.txt src/java/org/apache/nutch/fetcher/Fetcher.java

[jn...@apache.org]

Author: jnioche
Date: Sat Apr  5 19:54:47 2014
New Revision: 1585196

URL: http://svn.apache.org/r1585196
Log:
NUTCH-1747 Use AtomicInteger as semaphore in Fetcher

Modified:
    nutch/trunk/CHANGES.txt
    nutch/trunk/src/java/org/apache/nutch/fetcher/Fetcher.java

Modified: nutch/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/nutch/trunk/CHANGES.txt?rev=1585196&r1=1585195&r2=1585196&view=diff
==============================================================================
--- nutch/trunk/CHANGES.txt (original)
+++ nutch/trunk/CHANGES.txt Sat Apr  5 19:54:47 2014
@@ -2,6 +2,8 @@ Nutch Change Log
 
 Nutch Current Development
 
+* NUTCH-1747 Use AtomicInteger as semaphore in Fetcher (jnioche)
+
 * NUTCH-1735 code dedup fetcher queue redirects (snagel)
 
 * NUTCH-1745 Upgrade to ElasticSearch 1.1.0 (jnioche)

Modified: nutch/trunk/src/java/org/apache/nutch/fetcher/Fetcher.java
URL: http://svn.apache.org/viewvc/nutch/trunk/src/java/org/apache/nutch/fetcher/Fetcher.java?rev=1585196&r1=1585195&r2=1585196&view=diff
==============================================================================
--- nutch/trunk/src/java/org/apache/nutch/fetcher/Fetcher.java (original)
+++ nutch/trunk/src/java/org/apache/nutch/fetcher/Fetcher.java Sat Apr  5 19:54:47 2014
@@ -226,7 +226,7 @@ public class Fetcher extends Configured 
    */
   private static class FetchItemQueue {
     List<FetchItem> queue = Collections.synchronizedList(new LinkedList<FetchItem>());
-    Set<FetchItem>  inProgress = Collections.synchronizedSet(new HashSet<FetchItem>());
+    AtomicInteger  inProgress = new AtomicInteger();
     AtomicLong nextFetchTime = new AtomicLong();
     AtomicInteger exceptionCounter = new AtomicInteger();
     long crawlDelay;
@@ -254,7 +254,7 @@ public class Fetcher extends Configured 
     }
 
     public int getInProgressSize() {
-      return inProgress.size();
+      return inProgress.get();
     }
 
     public int incrementExceptionCounter() {
@@ -263,7 +263,7 @@ public class Fetcher extends Configured 
 
     public void finishFetchItem(FetchItem it, boolean asap) {
       if (it != null) {
-        inProgress.remove(it);
+        inProgress.decrementAndGet();
         setEndTime(System.currentTimeMillis(), asap);
       }
     }
@@ -275,18 +275,18 @@ public class Fetcher extends Configured 
 
     public void addInProgressFetchItem(FetchItem it) {
       if (it == null) return;
-      inProgress.add(it);
+      inProgress.incrementAndGet();
     }
 
     public FetchItem getFetchItem() {
-      if (inProgress.size() >= maxThreads) return null;
+      if (inProgress.get() >= maxThreads) return null;
       long now = System.currentTimeMillis();
       if (nextFetchTime.get() > now) return null;
       FetchItem it = null;
       if (queue.size() == 0) return null;
       try {
         it = queue.remove(0);
-        inProgress.add(it);
+        inProgress.incrementAndGet();
       } catch (Exception e) {
         LOG.error("Cannot remove FetchItem from queue or cannot add it to inProgress queue", e);
       }
@@ -295,7 +295,7 @@ public class Fetcher extends Configured 
 
     public synchronized void dump() {
       LOG.info("  maxThreads    = " + maxThreads);
-      LOG.info("  inProgress    = " + inProgress.size());
+      LOG.info("  inProgress    = " + inProgress.get());
       LOG.info("  crawlDelay    = " + crawlDelay);
       LOG.info("  minCrawlDelay = " + minCrawlDelay);
       LOG.info("  nextFetchTime = " + nextFetchTime.get());