diff --git a/engine/src/main/java/org/archive/crawler/framework/CrawlController.java b/engine/src/main/java/org/archive/crawler/framework/CrawlController.java index bbc87936..b2706952 100644 --- a/engine/src/main/java/org/archive/crawler/framework/CrawlController.java +++ b/engine/src/main/java/org/archive/crawler/framework/CrawlController.java @@ -253,6 +253,7 @@ implements Serializable, this.checkpointerPeriod = checkpointerPeriod; } + protected CrawlerLoggerModule loggerModule; public CrawlerLoggerModule getLoggerModule() { return this.loggerModule; @@ -294,7 +295,7 @@ implements Serializable, * Knows if checkpoint in progress and what name of checkpoint is. Also runs * checkpoints. */ - private Checkpointer checkpointer; + protected Checkpointer checkpointer; public CrawlController() { } @@ -313,7 +314,7 @@ implements Serializable, } this.checkpointer = new Checkpointer( this, getCheckpointsDir().getFile()); - + sExit = CrawlStatus.FINISHED_ABNORMAL; // force creation of DNS Cache now -- avoids CacheCleaner in toe-threads group @@ -335,6 +336,7 @@ implements Serializable, public void stop() { // TODO: more stop/cleanup? isRunning = false; + this.checkpointer.cleanup(); } /** @@ -391,12 +393,7 @@ implements Serializable, protected void completeStop() { LOGGER.fine("Entered complete stop."); - loggerModule.closeLogFiles(); this.reserveMemory = null; - if (this.checkpointer != null) { - this.checkpointer.cleanup(); - this.checkpointer = null; - } if (this.toePool != null) { this.toePool.cleanup(); } @@ -443,16 +440,8 @@ implements Serializable, if (this.checkpointer.isCheckpointing()) { throw new IllegalStateException("Checkpoint already running."); } - this.checkpointer.checkpoint(); + this.checkpointer.checkpoint(appCtx,checkpointsDir.getFile()); } - - /** - * @return True if checkpointing. - */ - public boolean isCheckpointing() { - return this.state == State.CHECKPOINTING; - } - /** * Operator requested for crawl to stop. 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 a824bb00..13826412 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java @@ -656,14 +656,10 @@ public abstract class AbstractFrontier requestState(State.RUN); } - synchronized public void terminate() { requestState(State.FINISH); // TODO: move this recover-cleanup to manager thread? - if (this.recover != null) { - this.recover.close(); - this.recover = null; - } + ArchiveUtils.closeQuietly(this.recover); } /** 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 00d3ebb8..18626e3b 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,10 @@ */ 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,9 +36,10 @@ 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.Checkpointer.Checkpoint; import org.archive.modules.CrawlURI; import org.archive.queue.StoredQueue; +import org.archive.util.ArchiveUtils; import org.archive.util.Supplier; import org.springframework.beans.factory.annotation.Autowired; @@ -182,6 +181,7 @@ implements Serializable, Checkpointable { } protected void closeQueue() { + // before closing/releasing, dump if requested if (getDumpPendingAtClose()) { try { dumpAllPendingToLog(); @@ -189,10 +189,7 @@ implements Serializable, Checkpointable { logger.log(Level.WARNING, "dump pending problem", e); } } - if (this.pendingUris != null) { - this.pendingUris.close(); - this.pendingUris = null; - } + ArchiveUtils.closeQuietly(pendingUris); } protected BdbMultipleWorkQueues getWorkQueues() { @@ -223,7 +220,6 @@ implements Serializable, Checkpointable { logger.fine("Finished syncing already seen as part of checkpoint."); } - @Override protected void initAllQueues() throws DatabaseException { this.allQueues = bdb.getObjectCache("allqueues", false, WorkQueue.class); diff --git a/engine/src/main/java/org/archive/crawler/frontier/BdbMultipleWorkQueues.java b/engine/src/main/java/org/archive/crawler/frontier/BdbMultipleWorkQueues.java index 4a7ca64b..56d328b7 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/BdbMultipleWorkQueues.java +++ b/engine/src/main/java/org/archive/crawler/frontier/BdbMultipleWorkQueues.java @@ -553,5 +553,6 @@ public class BdbMultipleWorkQueues { CrawlURI item = (CrawlURI) crawlUriBinding.entryToObject(value); c.execute(item); } + cursor.close(); } } 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 529f0809..e97653c7 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java @@ -286,23 +286,25 @@ ApplicationContextAware { */ protected abstract void initOtherQueues(boolean recycle) throws DatabaseException; + + /* (non-Javadoc) - * @see org.archive.crawler.frontier.AbstractFrontier#crawlEnded(java.lang.String) + * @see org.archive.crawler.frontier.AbstractFrontier#stop() + */ + @Override + public void stop() { + super.stop(); + // also release resources and trigger end-of-frontier actions + close(); + } + + /** + * Release resources only needed when running */ public void close() { - // Cleanup. CrawlJobs persist after crawl has finished so undo any - // references. - if (this.uriUniqFilter != null) { - this.uriUniqFilter.close(); - } - - try { - closeQueue(); - } catch (IOException e) { - logger.log(Level.WARNING,"closeQueue problem",e); - } - - this.allQueues.close(); + ArchiveUtils.closeQuietly(uriUniqFilter); + closeQueue(); + ArchiveUtils.closeQuietly(allQueues); } /** @@ -1460,7 +1462,7 @@ ApplicationContextAware { } } - protected abstract void closeQueue() throws IOException; + protected abstract void closeQueue(); /** * Returns true if the WorkQueue implementation of this