[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
This commit is contained in:
gojomo committed 2010-09-03 02:14:41 +00:00
1 parent d12d078a07
commit de551ced58
5 files changed
+69 -71

No files matched your search

@@ -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();
}
@@ -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.
@@ -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<CrawlURI> dispositionPending = new ThreadLocal<CrawlURI>();
/* (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
@@ -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)