From 716e0cc7ef9bbabdef6b4ed6b87f1e97041f42f1 Mon Sep 17 00:00:00 2001
From: Noah Levitt <nlevitt@archive.org>
Date: Mon, 27 Oct 2014 17:34:41 -0700
Subject: [PATCH] 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"

---
 .../crawler/frontier/AMQPUrlReceiver.java     | 22 +++++++++----------
 1 file changed, 10 insertions(+), 12 deletions(-)

diff --git a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
index 3dac0194..faec6615 100644
--- a/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
+++ b/contrib/src/main/java/org/archive/crawler/frontier/AMQPUrlReceiver.java
@@ -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);
                         }
                     }