[HER-1546] Springify(5): Update checkpointing to work smoothly with spring-configured crawls

* BdbFrontier.java
    implement Checkpointable; save state via JSON or BDB; restore queues-of-queues on recovery
* WorkQueueFrontier.java
    (deactivateQueue) (retireQueue) protected
* BdbUriUniqFilter.java
    implement Checkpointable; save state via JSON or BDB; restore
* WorkQueue.java
    make peekItem transient -- should not be held through checkpoint
    make retired protected
* Frontier.java, AbstractFrontier.java
    remove/update obsolete code

engine/src/main/java/org/archive/crawler/framework/Frontier.java

@@ -257,14 +257,6 @@ public interface Frontier extends Lifecycle, MultiReporter {
      */
     public long disregardedUriCount();
 
-    /**
-     * Total number of bytes contained in all URIs that have been processed.
-     *
-     * @return The total amounts of bytes in all processed URIs.
-     * @deprecated misnomer; consult StatisticsTracker instead
-     */
-    public long totalBytesWritten();
-
     /**
      * Load URIs from a file, for scheduling and/or considered-included 
      * status (if from a recovery log). 

engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java

@@ -38,8 +38,6 @@ import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStreamReader;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
 import java.io.PrintWriter;
 import java.io.Serializable;
 import java.util.Collection;
@@ -55,7 +53,6 @@ import java.util.regex.Pattern;
 
 import org.apache.commons.httpclient.HttpStatus;
 import org.apache.commons.httpclient.URIException;
-import org.archive.checkpointing.CheckpointRecovery;
 import org.archive.crawler.event.CrawlStateEvent;
 import org.archive.crawler.framework.CrawlController;
 import org.archive.crawler.framework.Frontier;
@@ -71,11 +68,9 @@ import org.archive.modules.net.CrawlServer;
 import org.archive.modules.net.ServerCache;
 import org.archive.modules.seeds.SeedListener;
 import org.archive.modules.seeds.SeedModule;
-import org.archive.spring.ConfigPath;
 import org.archive.spring.HasKeyedProperties;
 import org.archive.spring.KeyedProperties;
 import org.archive.util.ArchiveUtils;
-import org.archive.util.IoUtils;
 import org.archive.util.iterator.LineReadingIterator;
 import org.archive.util.iterator.RegexLineIterator;
 import org.json.JSONException;
@@ -127,14 +122,6 @@ public abstract class AbstractFrontier
         kp.put("maxRetries",maxRetries);
     }
     
-    protected ConfigPath recoveryDir = new ConfigPath("recovery subdirectory","logs");
-    public ConfigPath getRecoveryDir() {
-        return recoveryDir;
-    }
-    public void setRecoveryDir(ConfigPath recoveryDir) {
-        this.recoveryDir = recoveryDir;
-    }
-    
     /**
      * Recover log on or off attribute.
      */
@@ -275,11 +262,7 @@ public abstract class AbstractFrontier
     /**
      * Used when bandwidth constraint are used.
      */
-    protected long totalProcessedBytes = 0;
-
-    protected long processedBytesAfterLastEmittedURI = 0;
-    
-    protected int lastMaxBandwidthKB = 0;
+    protected AtomicLong totalProcessedBytes = new AtomicLong(0);
 
     /**
      * Crawl replay logger.
@@ -829,11 +812,6 @@ public abstract class AbstractFrontier
         return disregardedUriCount.get();
     }
 
-    /** @deprecated misnomer; use StatisticsTracker figures instead */
-    public long totalBytesWritten() {
-        return totalProcessedBytes;
-    }
-    
     /**
      * When notified of a seed via the SeedListener interface, 
      * schedule it.
@@ -950,7 +928,7 @@ public abstract class AbstractFrontier
         DecideRule scope = (applyScope) ? getScope() : null;
         FrontierJournal newJournal = getFrontierJournal();
         Matcher m = Pattern.compile(acceptTags).matcher(""); 
-        BufferedReader br = IoUtils.getBufferedReader(source);
+        BufferedReader br = ArchiveUtils.getBufferedReader(source);
         String read;
         int lineCount = 0; 
         try {
@@ -1183,36 +1161,6 @@ public abstract class AbstractFrontier
     public void reportTo(PrintWriter writer) {
         reportTo(null, writer);
     }
-
-
-    private void writeObject(ObjectOutputStream out) 
-    throws IOException {
-        out.defaultWriteObject();
-        boolean recoveryLogEnabled = getRecoveryLogEnabled();
-        out.writeBoolean(recoveryLogEnabled);
-        if (recoveryLogEnabled) {
-            out.writeUTF(loggerModule.getPath().getFile().getAbsolutePath());
-        }
-    }
-    
-    
-    private void readObject(ObjectInputStream inp) 
-    throws IOException, ClassNotFoundException {
-        inp.defaultReadObject();
-        boolean recoveryLogEnabled = inp.readBoolean();
-        if (recoveryLogEnabled) {
-            String path = inp.readUTF();
-            if (inp instanceof CheckpointRecovery) {
-                CheckpointRecovery cr = (CheckpointRecovery)inp;
-                path = cr.translatePath(path);
-                new File(path).mkdirs();
-            }
-            initJournal(path);
-        }
-        targetState = State.PAUSE;
-        outbound = new ArrayBlockingQueue<CrawlURI>(outboundCapacity, true);
-        inbound = new ArrayBlockingQueue<InEvent>(inboundCapacity, true);
-    }
     
     /**
      * Arrange for the given InEvent to be done by the managerThread, via

engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java

@@ -18,12 +18,9 @@
  */
 package org.archive.crawler.frontier;
 
-import java.io.File;
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.Serializable;
-import java.util.Iterator;
-import java.util.List;
 import java.util.Queue;
 import java.util.SortedMap;
 import java.util.TreeMap;
@@ -38,14 +35,16 @@ import javax.management.openmbean.CompositeData;
 import org.apache.commons.collections.Closure;
 import org.archive.bdb.BdbModule;
 import org.archive.checkpointing.Checkpointable;
-import org.archive.checkpointing.RecoverAction;
+import org.archive.crawler.framework.Checkpoint;
 import org.archive.modules.CrawlURI;
 import org.archive.queue.StoredQueue;
 import org.archive.util.ArchiveUtils;
 import org.archive.util.Supplier;
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.annotation.Autowired;
 
-import com.sleepycat.collections.StoredIterator;
 import com.sleepycat.je.Database;
 import com.sleepycat.je.DatabaseException;
 
@@ -56,7 +55,7 @@ import com.sleepycat.je.DatabaseException;
  * @author Gordon Mohr
  */
 public class BdbFrontier extends WorkQueueFrontier 
-implements Serializable, Checkpointable {
+implements Serializable, Checkpointable, BeanNameAware {
     private static final long serialVersionUID = 1L;
 
     private static final Logger logger =
@@ -83,6 +82,11 @@ implements Serializable, Checkpointable {
         this.bdb = bdb;
     }
     
+    String beanName; 
+    public void setBeanName(String name) {
+        this.beanName = name;
+    }
+    
     boolean dumpPendingAtClose = false; 
     public boolean getDumpPendingAtClose() {
         return dumpPendingAtClose;
@@ -114,18 +118,16 @@ implements Serializable, Checkpointable {
      * @return the created BdbMultipleWorkQueues
      * @throws DatabaseException
      */
-    private BdbMultipleWorkQueues createMultipleWorkQueues(boolean recycle)
+    protected BdbMultipleWorkQueues createMultipleWorkQueues()
     throws DatabaseException {
         Database db;
-        if (recycle) {
-            db = bdb.getDatabase("pending");
-        } else {
-            BdbModule.BdbConfig dbConfig = new BdbModule.BdbConfig();
-            dbConfig.setAllowCreate(!recycle);
-            // Make database deferred write: URLs that are added then removed 
-            // before a page-out is required need never cause disk IO.
-            db = bdb.openManagedDatabase("pending", dbConfig, recycle);
-        }
+        boolean recycle = (recoveryCheckpoint != null);
+
+        BdbModule.BdbConfig dbConfig = new BdbModule.BdbConfig();
+        dbConfig.setAllowCreate(!recycle);
+        // Make database deferred write: URLs that are added then removed 
+        // before a page-out is required need never cause disk IO.
+        db = bdb.openManagedDatabase("pending", dbConfig, recycle);
         
         return new BdbMultipleWorkQueues(db, bdb.getClassCatalog());
     }
@@ -202,43 +204,85 @@ implements Serializable, Checkpointable {
         return true;
     }
 
-    
-    /**
-     * Constructor.
-     */
     public BdbFrontier() {
         super();
     }
     
-    public void checkpoint(File checkpointDir, List<RecoverAction> actions) 
-    throws IOException {
-        logger.fine("Started syncing already seen as part "
-            + "of checkpoint. Can take some time.");
+    public void startCheckpoint(Checkpoint checkpointInProgress) {}
+
+    public void doCheckpoint(Checkpoint checkpointInProgress) {
         // An explicit sync on the any deferred write dbs is needed to make the
         // db recoverable. Sync'ing the environment doesn't work.
-        if (this.pendingUris != null) {
-        	this.pendingUris.sync();
+        this.pendingUris.sync();
+        // object caches will be sync()d by BdbModule
+        
+        JSONObject json = new JSONObject();
+        try {
+            json.put("queuedUriCount", queuedUriCount.get());
+            json.put("succeededFetchCount", succeededFetchCount.get());
+            json.put("failedFetchCount", failedFetchCount.get());
+            json.put("disregardedUriCount", disregardedUriCount.get());
+            json.put("totalProcessedBytes", totalProcessedBytes.get());
+            checkpointInProgress.saveJson(beanName, json);
+        } catch (JSONException e) {
+            // impossible
+            throw new RuntimeException(e);
         }
-        logger.fine("Finished syncing already seen as part of checkpoint.");
     }
 
+    public void finishCheckpoint(Checkpoint checkpointInProgress) {}
+
+    Checkpoint recoveryCheckpoint;
+    @Autowired(required=false)
+    public void setRecoveryCheckpoint(Checkpoint checkpoint) {
+        this.recoveryCheckpoint = checkpoint; 
+    }
+    
     @Override
     protected void initAllQueues() throws DatabaseException {
-        this.allQueues = bdb.getObjectCache("allqueues", false, WorkQueue.class);
-        if (logger.isLoggable(Level.FINE)) {
-            Iterator<String> i = this.allQueues.keySet().iterator();
+        boolean isRecovery = (recoveryCheckpoint != null);
+        this.allQueues = bdb.getObjectCache("allqueues", isRecovery, WorkQueue.class);
+        if(isRecovery) {
+            JSONObject json = recoveryCheckpoint.loadJson(beanName);
             try {
-                for (; i.hasNext();) {
-                    logger.fine((String) i.next());
+                queuedUriCount.set(json.getLong("queuedUriCount"));
+                succeededFetchCount.set(json.getLong("succeededFetchCount"));
+                failedFetchCount.set(json.getLong("failedFetchCount"));
+                disregardedUriCount.set(json.getLong("disregardedUriCount"));
+                totalProcessedBytes.set(json.getLong("totalProcessedBytes"));
+            } catch (JSONException e) {
+                throw new RuntimeException(e);
+            }            
+            // restore WorkQueues to internal management queues
+            enqueueOrDo(new Recover());
+        }
+    }
+    
+    /**
+     * Frontier managerThread action to restore the placement of
+     * all queues to either the 'retired' collection or one of the
+     * inactive tiers (from which they will become ready/active as
+     * necessary). 
+     */
+    public class Recover extends InEvent {
+        @Override
+        public void process() {
+            // restore WorkQueues to internal management queues
+            for (String key : allQueues.keySet()) {
+                WorkQueue q = allQueues.get(key);
+                q.getOnInactiveQueues().clear();
+                q.setSessionBalance(0); 
+                if(q.isRetired()) {
+                    getRetiredQueues().add(key); 
+                } else {
+                    deactivateQueue(q);
                 }
-            } finally {
-                StoredIterator.close(i);
             }
         }
     }
     
     @Override
-    protected void initOtherQueues(boolean recycle) throws DatabaseException {
+    protected void initOtherQueues() throws DatabaseException {
         // small risk of OutOfMemoryError: if 'hold-queues' is false,
         // readyClassQueues may grow in size without bound
         readyClassQueues = new LinkedBlockingQueue<String>();
@@ -257,7 +301,7 @@ implements Serializable, Checkpointable {
         snoozedClassQueues = new DelayQueue<DelayedWorkQueue>();
         
         // initialize master map in which other queues live
-        this.pendingUris = createMultipleWorkQueues(recycle);
+        this.pendingUris = createMultipleWorkQueues();
     }
 
 

engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java

@@ -92,7 +92,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
     private long totalBudget = 0;
 
     /** The next item to be returned */
-    protected CrawlURI peekItem = null;
+    transient protected CrawlURI peekItem = null;
 
     /** Last URI enqueued */
     private String lastQueued;
@@ -109,7 +109,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup,
     /** Substats for all CrawlURIs in this group */
     protected FetchStats substats = new FetchStats();
 
-    private boolean retired;
+    protected boolean retired;
 
     public WorkQueue(final String pClassKey) {
         this.classKey = pClassKey;

engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java

@@ -236,14 +236,17 @@ ApplicationContextAware {
     }
     
     public void start() {
-        super.start();
+        if(isRunning()) {
+            return; 
+        }
         uriUniqFilter.setDestination(this);
-        
+        super.start();
         try {
-            initInternalQueues(false);
+            initInternalQueues();
         } catch (Exception e) {
             throw new IllegalStateException(e);
         }
+
     }
 
     /**
@@ -258,7 +261,7 @@ ApplicationContextAware {
      * @throws IOException
      * @throws DatabaseException
      */
-    protected void initInternalQueues(boolean recycle) 
+    protected void initInternalQueues() 
     throws IOException, DatabaseException {
         if (workQueueDataOnDisk()
                 && preparer.getQueueAssignmentPolicy().maximumNumberOfKeys() >= 0
@@ -269,7 +272,7 @@ ApplicationContextAware {
         } else {
             this.initAllQueues();
         }
-        this.initOtherQueues(recycle);
+        this.initOtherQueues();
     }
     
     /**
@@ -284,7 +287,7 @@ ApplicationContextAware {
      * way.
      * @throws DatabaseException
      */
-    protected abstract void initOtherQueues(boolean recycle) throws DatabaseException;
+    protected abstract void initOtherQueues() throws DatabaseException;
 
     
     
@@ -430,7 +433,7 @@ ApplicationContextAware {
      * Put the given queue on the inactiveQueues queue
      * @param wq
      */
-    private void deactivateQueue(WorkQueue wq) {
+    protected void deactivateQueue(WorkQueue wq) {
         assert Thread.currentThread() == managerThread;
         
         wq.setSessionBalance(0); // zero out session balance
@@ -492,7 +495,7 @@ ApplicationContextAware {
      * Put the given queue on the retiredQueues queue
      * @param wq
      */
-    private void retireQueue(WorkQueue wq) {
+    protected void retireQueue(WorkQueue wq) {
         assert Thread.currentThread() == managerThread;
 
         getRetiredQueues().add(wq.getClassKey());
@@ -928,7 +931,7 @@ ApplicationContextAware {
         log(curi);
 
         if (curi.isSuccess()) {
-            totalProcessedBytes += curi.getRecordedSize();
+            totalProcessedBytes.addAndGet(curi.getRecordedSize());
             incrementSucceededFetchCount();
             // Let everyone know in case they want to do something before we strip the curi.
             appCtx.publishEvent(

engine/src/main/java/org/archive/crawler/util/BdbUriUniqFilter.java

@@ -20,13 +20,16 @@ package org.archive.crawler.util;
 
 import java.io.File;
 import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
-import java.io.Serializable;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import org.archive.bdb.BdbModule;
+import org.archive.checkpointing.Checkpointable;
+import org.archive.crawler.framework.Checkpoint;
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.Lifecycle;
 
@@ -61,7 +64,7 @@ import com.sleepycat.je.OperationStatus;
  * @version $Date$, $Revision$
  */
 public class BdbUriUniqFilter extends SetBasedUriUniqFilter 
-implements Lifecycle, Serializable {
+implements Lifecycle, Checkpointable, BeanNameAware {
     private static final long serialVersionUID = -8099357538178524011L;
 
     private static Logger logger =
@@ -75,7 +78,7 @@ implements Lifecycle, Serializable {
     static protected DatabaseEntry ZERO_LENGTH_ENTRY = 
         new DatabaseEntry(new byte[0]);
     private static final String DB_NAME = "alreadySeenUrl";
-    protected long count = 0;
+    protected AtomicLong count = new AtomicLong(0);
     private long aggregatedLookupTime = 0;
     
     private static final String COLON_SLASH_SLASH = "://";
@@ -86,6 +89,11 @@ implements Lifecycle, Serializable {
         this.bdb = bdb;
     }
     
+    String beanName; 
+    public void setBeanName(String name) {
+        this.beanName = name;
+    }
+    
     public BdbUriUniqFilter() {
     }
     
@@ -94,13 +102,22 @@ implements Lifecycle, Serializable {
         if(isRunning()) {
             return; 
         }
+        boolean isRecovery = (recoveryCheckpoint != null);
         try {
             BdbModule.BdbConfig config = getDatabaseConfig();
-            config.setAllowCreate(true);
-            initialize(bdb.openManagedDatabase(DB_NAME, config, false));
+            config.setAllowCreate(!isRecovery);
+            initialize(bdb.openManagedDatabase(DB_NAME, config, isRecovery));
         } catch (DatabaseException e) {
             throw new IllegalStateException(e);
         }
+        if(isRecovery) {
+            JSONObject json = recoveryCheckpoint.loadJson(beanName);
+            try {
+                count.set(json.getLong("count"));
+            } catch (JSONException e) {
+                throw new RuntimeException(e);
+            }           
+        }
         isRunning = true; 
     }
     
@@ -212,8 +229,7 @@ implements Lifecycle, Serializable {
     
     public synchronized void close() {
         if (logger.isLoggable(Level.INFO)) {
-            logger.info("Count of alreadyseen on close "
-                    + Long.toString(count));
+            logger.info("Count of alreadyseen on close " + count.get());
         }
         Environment env = null;
         if (this.alreadySeen != null) {
@@ -314,10 +330,10 @@ implements Lifecycle, Serializable {
             logger.severe(e.getMessage());
         }
         if (status == OperationStatus.SUCCESS) {
-            count++;
+            count.incrementAndGet();
             if (logger.isLoggable(Level.INFO)) {
                 final int logAt = 10000;
-                if (count > 0 && ((count % logAt) == 0)) {
+                if (count.get() > 0 && ((count.get() % logAt) == 0)) {
                     logger.info("Average lookup " +
                         (aggregatedLookupTime / logAt) + "ms.");
                     aggregatedLookupTime = 0;
@@ -332,7 +348,7 @@ implements Lifecycle, Serializable {
     }
 
     protected long setCount() {
-        return count;
+        return count.get();
     }
 
     protected boolean setRemove(CharSequence uri) {
@@ -345,7 +361,7 @@ implements Lifecycle, Serializable {
             logger.severe(e.getMessage());
         }
         if (status == OperationStatus.SUCCESS) {
-            count--;
+            count.decrementAndGet();
             return true; // removed
         } else {
             return false; // not present
@@ -358,32 +374,23 @@ implements Lifecycle, Serializable {
         return 0;
     }
 
-    private void writeObject(ObjectOutputStream output) throws IOException {
-        // sync deferred-write database
+    // Checkpointable
+    // CrawlController's only interest is in knowing that a Checkpoint is
+    // being recovered
+    public void startCheckpoint(Checkpoint checkpointInProgress) {}
+    public void doCheckpoint(Checkpoint checkpointInProgress) throws IOException {
+        JSONObject json = new JSONObject();
         try {
-            alreadySeen.sync();
-        } catch (DatabaseException e) {
-            // TODO Auto-generated catch block
+            json.put("count", setCount());
+            checkpointInProgress.saveJson(beanName, json);
+        } catch (JSONException e) {
+            // impossible
             throw new RuntimeException(e);
         }
-        output.defaultWriteObject();
     }
-
-    private void readObject(ObjectInputStream input) 
-    throws IOException, ClassNotFoundException {
-        input.defaultReadObject();        
-
-        try {
-            BdbModule.BdbConfig config = getDatabaseConfig();
-            config.setAllowCreate(false);
-            reopen(bdb.getDatabase(DB_NAME));
-        } catch (DatabaseException e) {
-            IOException io = new IOException();
-            io.initCause(e);
-            throw io;
-        }
+    public void finishCheckpoint(Checkpoint checkpointInProgress) {}
+    Checkpoint recoveryCheckpoint;
+    public void setRecoveryCheckpoint(Checkpoint recoveryCheckpoint) {
+        this.recoveryCheckpoint = recoveryCheckpoint;
     }
-
-    
-
-}
+} //EOC