diff --git a/engine/src/main/java/org/archive/crawler/framework/Frontier.java b/engine/src/main/java/org/archive/crawler/framework/Frontier.java index 814b3474..16a523b2 100644 --- a/engine/src/main/java/org/archive/crawler/framework/Frontier.java +++ b/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). diff --git a/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java b/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java index 13826412..1333002a 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java +++ b/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(outboundCapacity, true); - inbound = new ArrayBlockingQueue(inboundCapacity, true); - } /** * Arrange for the given InEvent to be done by the managerThread, via 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 dd9fe17c..e163f7fe 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java +++ b/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 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 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(); @@ -257,7 +301,7 @@ implements Serializable, Checkpointable { snoozedClassQueues = new DelayQueue(); // initialize master map in which other queues live - this.pendingUris = createMultipleWorkQueues(recycle); + this.pendingUris = createMultipleWorkQueues(); } diff --git a/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java b/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java index c03dc2cd..f9fa6b54 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java +++ b/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; 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 e97653c7..a5c41545 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java +++ b/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( diff --git a/engine/src/main/java/org/archive/crawler/util/BdbUriUniqFilter.java b/engine/src/main/java/org/archive/crawler/util/BdbUriUniqFilter.java index bbe648d8..e657e59c 100644 --- a/engine/src/main/java/org/archive/crawler/util/BdbUriUniqFilter.java +++ b/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; } - - - -} \ No newline at end of file +} //EOC \ No newline at end of file