From de551ced586756830df3e676f8e38357b3f36833 Mon Sep 17 00:00:00 2001 From: gojomo Date: Fri, 3 Sep 2010 02:14:41 +0000 Subject: [PATCH] [HER-1801] H3: Unreleased lock causes deadlock when checkpointing * Frontier.java add beginDisposition(), endDisposition() methods * AbstractFrontier.java implement beginDisposition(), endDisposition() methods to maintain dispositions-in-progress lock * BdbFrontier.java acquire exclusive dispositionInProgressLock before allowing checkpoint to begin; release at completion of checkpoint * ToeThread.java call beginDisposition before dispositionChain/finished(); call endDisposition after finished() and on exceptions * DispositionChain.java shed all prior relation to locking/checkpointing --- .../archive/crawler/framework/Frontier.java | 17 +++++ .../archive/crawler/framework/ToeThread.java | 17 +++-- .../crawler/frontier/AbstractFrontier.java | 36 ++++++++++- .../archive/crawler/frontier/BdbFrontier.java | 8 ++- .../org/archive/modules/DispositionChain.java | 62 +------------------ 5 files changed, 69 insertions(+), 71 deletions(-) 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 dfeea33e..b4ff0a4a 100644 --- a/engine/src/main/java/org/archive/crawler/framework/Frontier.java +++ b/engine/src/main/java/org/archive/crawler/framework/Frontier.java @@ -520,4 +520,21 @@ public interface Frontier extends Lifecycle, MultiReporter { FINISH // end and cleanup; may not return to any other state after // this state is requested/reached } + + /** + * Inform frontier that a block of processing that should complete atomically + * with respect to checkpoints is about to begin. Callers should ensure an + * endDisposition() call soon follows; a mismatch risks freezing the frontier + * if a checkpoint is requested. + * @param curi + */ + public void beginDisposition(CrawlURI curi); + + /** + * Inform frontier the processing signalled by an earlier pending + * beginDisposition() call has finished. Implementors should be resilient + * against extra endDisposition calls, as callers dealing with exceptional + * conditions need to be free to call this 'just in case'. + */ + public void endDisposition(); } diff --git a/engine/src/main/java/org/archive/crawler/framework/ToeThread.java b/engine/src/main/java/org/archive/crawler/framework/ToeThread.java index c30081c7..5c447eb0 100644 --- a/engine/src/main/java/org/archive/crawler/framework/ToeThread.java +++ b/engine/src/main/java/org/archive/crawler/framework/ToeThread.java @@ -123,7 +123,7 @@ implements MultiReporter, ProgressStatisticsReporter, String name = controller.getMetadata().getJobName(); logger.fine(getName()+" started for order '"+name+"'"); Recorder.setHttpRecorder(httpRecorder); - + try { while ( true ) { ArchiveUtils.continueCheck(); @@ -132,6 +132,7 @@ implements MultiReporter, ProgressStatisticsReporter, CrawlURI curi = controller.getFrontier().next(); + synchronized(this) { ArchiveUtils.continueCheck(); setCurrentCuri(curi); @@ -144,9 +145,9 @@ implements MultiReporter, ProgressStatisticsReporter, KeyedProperties.loadOverridesFrom(curi); controller.getFetchChain().process(curi,this); - // TODO: insert barrier here to support checkpointing - // only checkpoint when all URIs have finished - // disposition chain + + controller.getFrontier().beginDisposition(curi); + controller.getDispositionChain().process(curi,this); } catch (RuntimeExceptionWrapper e) { @@ -175,6 +176,7 @@ implements MultiReporter, ProgressStatisticsReporter, synchronized(this) { controller.getFrontier().finished(currentCuri); + controller.getFrontier().endDisposition(); setCurrentCuri(null); } @@ -186,13 +188,16 @@ implements MultiReporter, ProgressStatisticsReporter, } } catch (InterruptedException e) { // thread interrupted, ok to end - logger.log(Level.FINE,this.getName()+ " ended with Interruption"); + logger.log(Level.INFO,this.getName()+ " ended with Interruption"); } catch (Exception e) { // everything else (including interruption) logger.log(Level.SEVERE,"Fatal exception in "+getName(),e); } catch (OutOfMemoryError err) { seriousError(err); - } + } finally { + controller.getFrontier().endDisposition(); + + } setCurrentCuri(null); // Do cleanup so that objects can be GC. 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 b5f8a427..bd6a41dd 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java @@ -1134,8 +1134,8 @@ public abstract class AbstractFrontier // loaded in FetchHTTP on expectation that we're to go around // again. If no rfc2617 loaded, we should not be here. boolean loaded = curi.hasRfc2617CredentialAvatar(); - if (!loaded && logger.isLoggable(Level.INFO)) { - logger.info("Have 401 but no creds loaded " + curi); + if (!loaded && logger.isLoggable(Level.FINE)) { + logger.fine("Have 401 but no creds loaded " + curi); } return loaded; case S_DEFERRED: @@ -1311,4 +1311,34 @@ public abstract class AbstractFrontier } } } -} + + /** lock allowing steps of outside processing that need to complete + * all-or-nothing to signal their in-progress status */ + protected ReentrantReadWriteLock dispositionInProgressLock = + new ReentrantReadWriteLock(true); + /** remembers a disposition-in-progress, so that extra endDisposition() + * calls are harmless */ + protected ThreadLocal dispositionPending = new ThreadLocal(); + + /* (non-Javadoc) + * @see org.archive.crawler.framework.Frontier#beginDisposition(org.archive.modules.CrawlURI) + */ + @Override + public void beginDisposition(CrawlURI curi) { + dispositionPending.set(curi); + dispositionInProgressLock.readLock().lock(); + } + + /* (non-Javadoc) + * @see org.archive.crawler.framework.Frontier#endDisposition() + */ + @Override + public void endDisposition() { + // avoid a mismatched unlock; allows callers to be less complicated, + // calling endDisposition 'just in case' a begin happened + if(dispositionPending.get()!=null) { + dispositionInProgressLock.readLock().unlock(); + dispositionPending.set(null); + } + } +} //EOC 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 815aea6c..2b410f48 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java @@ -219,7 +219,9 @@ implements Checkpointable, BeanNameAware { super(); } - public void startCheckpoint(Checkpoint checkpointInProgress) {} + public void startCheckpoint(Checkpoint checkpointInProgress) { + dispositionInProgressLock.writeLock().lock(); + } public void doCheckpoint(Checkpoint checkpointInProgress) { // An explicit sync on the any deferred write dbs is needed to make the @@ -242,7 +244,9 @@ implements Checkpointable, BeanNameAware { } } - public void finishCheckpoint(Checkpoint checkpointInProgress) {} + public void finishCheckpoint(Checkpoint checkpointInProgress) { + dispositionInProgressLock.writeLock().unlock(); + } Checkpoint recoveryCheckpoint; @Autowired(required=false) diff --git a/modules/src/main/java/org/archive/modules/DispositionChain.java b/modules/src/main/java/org/archive/modules/DispositionChain.java index 606ea464..a12ab13f 100644 --- a/modules/src/main/java/org/archive/modules/DispositionChain.java +++ b/modules/src/main/java/org/archive/modules/DispositionChain.java @@ -1,64 +1,6 @@ package org.archive.modules; -import java.io.IOException; -/* - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Licensed to the Internet Archive (IA) by one or more individual - * contributors. - * - * The IA licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -import java.util.concurrent.locks.ReentrantReadWriteLock; - -import org.archive.checkpointing.Checkpoint; -import org.archive.checkpointing.Checkpointable; -import org.springframework.beans.factory.annotation.Autowired; - -public class DispositionChain extends ProcessorChain implements Checkpointable { - protected ReentrantReadWriteLock dispositionInProgressLock = - new ReentrantReadWriteLock(true); - - - public void startCheckpoint(Checkpoint checkpointInProgress) { - dispositionInProgressLock.writeLock().lock(); - } - - public void doCheckpoint(Checkpoint checkpointInProgress) throws IOException { - // do nothing; this class only participates in checkpointing - // via the startCheckpoint/finishCheckpoint locking - } - - public void finishCheckpoint(Checkpoint checkpointInProgress) { - dispositionInProgressLock.writeLock().unlock(); - } - - @Autowired(required=false) - public void setRecoveryCheckpoint(Checkpoint checkpoint) { - // do nothing - } - - @Override - public void process(CrawlURI curi, ChainStatusReceiver thread) throws InterruptedException { - dispositionInProgressLock.readLock().lock(); - try { - super.process(curi, thread); - } finally { - dispositionInProgressLock.readLock().unlock(); - } - } - - +public class DispositionChain extends ProcessorChain { + }