lee/heritrix3
1
0
Fork 0
mirror of https://github.com/internetarchive/heritrix3.git synced 2026-09-25 07:05:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

use "fair" ReentrantLock -- I stepped through heritrix that appeared to be deadlocked, found that the thread trying to shut down was being starved of the lock by another thread (the StarterRestarter) locking and unlocking it repeatedly, which can apparently happen because the lock was "unfair"

Browse Source
This commit is contained in:
Noah Levitt
2014-10-27 17:34:41 -07:00
parent d154cfbf36
commit 716e0cc7ef
1 changed files with 10 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
+10 -12
View File
@@ -107,7 +107,7 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
return isRunning;
}
private transient Lock lock = new ReentrantLock();
private transient Lock lock = new ReentrantLock(true);
private class StarterRestarter extends Thread {
public StarterRestarter(String name) {
@@ -121,17 +121,15 @@ public class AMQPUrlReceiver implements Lifecycle, ApplicationListener<CrawlStat
lock.lockInterruptibly();
if (!isRunning) {
// start up again
synchronized (AMQPUrlReceiver.this) {
try {
Consumer consumer = new UrlConsumer(channel());
channel().queueDeclare(getQueueName(), false, false, true, null);
channel().queueBind(getQueueName(), getExchange(), getQueueName());
channel().basicConsume(getQueueName(), false, consumer);
isRunning = true;
logger.info("started AMQP consumer uri=" + getAmqpUri() + " exchange=" + getExchange() + " queueName=" + getQueueName());
} catch (IOException e) {
logger.log(Level.SEVERE, "problem starting AMQP consumer (will try again after 30 seconds)", e);
}
try {
Consumer consumer = new UrlConsumer(channel());
channel().queueDeclare(getQueueName(), false, false, true, null);
channel().queueBind(getQueueName(), getExchange(), getQueueName());
channel().basicConsume(getQueueName(), false, consumer);
isRunning = true;
logger.info("started AMQP consumer uri=" + getAmqpUri() + " exchange=" + getExchange() + " queueName=" + getQueueName());
} catch (IOException e) {
logger.log(Level.SEVERE, "problem starting AMQP consumer (will try again after 30 seconds)", e);
}
}