Unlock TransactionConfidence event listeners, make sure MemoryPool is not locked when a tx is marked broadcast by a peer to avoid inversions via listeners.

Unfortunately this introduces some new FindBugs warnings because it doesn't understand the inside-out locking pattern used here, despite that it's correct.
Update issue 233.

core/src/main/java/com/google/bitcoin/core/MemoryPool.java

@@ -16,7 +16,7 @@
 
 package com.google.bitcoin.core;
 
-import com.google.common.base.Preconditions;
+import com.google.bitcoin.utils.Locks;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -27,6 +27,10 @@ import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.locks.ReentrantLock;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
 
 /**
  * <p>Tracks transactions that are being announced across the network. Typically one is created for you by a
@@ -41,6 +45,7 @@ import java.util.Set;
  */
 public class MemoryPool {
     private static final Logger log = LoggerFactory.getLogger(MemoryPool.class);
+    protected ReentrantLock lock = Locks.lock("mempool");
 
     // For each transaction we may have seen:
     //   - only its hash in an inv packet
@@ -108,7 +113,9 @@ public class MemoryPool {
      * which bothered to keep a reference. Typically, this is because the transaction does not involve any keys that
      * are relevant to any of our wallets.
      */
-    private synchronized void cleanPool() {
+    private void cleanPool() {
+        lock.lock();
+        try {
             Reference<? extends Transaction> ref;
             while ((ref = referenceQueue.poll()) != null) {
                 // Find which transaction got deleted by the GC.
@@ -116,12 +123,17 @@ public class MemoryPool {
                 // And remove the associated map entry so the other bits of memory can also be reclaimed.
                 memoryPool.remove(txRef.hash);
             }
+        } finally {
+            lock.unlock();
+        }
     }
 
     /**
      * Returns the number of peers that have seen the given hash recently.
      */
-    public synchronized int numBroadcastPeers(Sha256Hash txHash) {
+    public int numBroadcastPeers(Sha256Hash txHash) {
+        lock.lock();
+        try {
             cleanPool();
             Entry entry = memoryPool.get(txHash);
             if (entry == null) {
@@ -129,7 +141,7 @@ public class MemoryPool {
                 return 0;
             } else if (entry.tx == null) {
                 // We've seen at least one peer announce with an inv.
-            Preconditions.checkNotNull(entry.addresses);
+                checkNotNull(entry.addresses);
                 return entry.addresses.size();
             } else {
                 final Transaction tx = entry.tx.get();
@@ -140,10 +152,13 @@ public class MemoryPool {
                     memoryPool.remove(txHash);
                     return 0;
                 } else {
-                Preconditions.checkState(entry.addresses == null);
+                    checkState(entry.addresses == null);
                     return tx.getConfidence().numBroadcastPeers();
                 }
             }
+        } finally {
+            lock.unlock();
+        }
     }
 
     /**
@@ -152,14 +167,16 @@ public class MemoryPool {
      * @param byPeer The Peer that received it.
      * @return An object that is semantically the same TX but may be a different object instance.
      */
-    public synchronized Transaction seen(Transaction tx, PeerAddress byPeer) {
+    public Transaction seen(Transaction tx, PeerAddress byPeer) {
+        lock.lock();
+        try {
             cleanPool();
             Entry entry = memoryPool.get(tx.getHash());
             if (entry != null) {
                 // This TX or its hash have been previously announced.
                 if (entry.tx != null) {
                     // We already downloaded it.
-                Preconditions.checkState(entry.addresses == null);
+                    checkState(entry.addresses == null);
                     // We only want one canonical object instance for a transaction no matter how many times it is
                     // deserialized.
                     Transaction transaction = entry.tx.get();
@@ -174,13 +191,16 @@ public class MemoryPool {
                         log.info("{}: Provided with a transaction downloaded before: [{}] {}",
                                 new Object[]{byPeer, tx.getConfidence().numBroadcastPeers(), tx.getHash()});
                     }
-                tx.getConfidence().markBroadcastBy(byPeer);
+                    markBroadcast(byPeer, tx);
                     return tx;
                 } else {
                     // We received a transaction that we have previously seen announced but not downloaded until now.
-                Preconditions.checkNotNull(entry.addresses);
+                    checkNotNull(entry.addresses);
                     entry.tx = new WeakTransactionReference(tx, referenceQueue);
-                // Copy the previously announced peers into the confidence and then clear it out.
+                    // Copy the previously announced peers into the confidence and then clear it out. Unlock here
+                    // because markBroadcastBy can trigger event listeners and thus inversions.
+                    lock.unlock();
+                    try {
                         TransactionConfidence confidence = tx.getConfidence();
                         for (PeerAddress a : entry.addresses) {
                             confidence.markBroadcastBy(a);
@@ -188,6 +208,9 @@ public class MemoryPool {
                         entry.addresses = null;
                         log.debug("{}: Adding tx [{}] {} to the memory pool",
                                 new Object[]{byPeer, confidence.numBroadcastPeers(), tx.getHashAsString()});
+                    } finally {
+                        lock.lock();
+                    }
                     return tx;
                 }
             } else {
@@ -198,25 +221,30 @@ public class MemoryPool {
                 entry = new Entry();
                 entry.tx = new WeakTransactionReference(tx, referenceQueue);
                 memoryPool.put(tx.getHash(), entry);
-            tx.getConfidence().markBroadcastBy(byPeer);
+                markBroadcast(byPeer, tx);
                 return tx;
             }
+        } finally {
+            lock.unlock();
+        }
     }
 
     /**
      * Called by peers when they see a transaction advertised in an "inv" message. It either will increase the
      * confidence of the pre-existing transaction or will just keep a record of the address for future usage.
      */
-    public synchronized void seen(Sha256Hash hash, PeerAddress byPeer) {
+    public void seen(Sha256Hash hash, PeerAddress byPeer) {
+        lock.lock();
+        try {
             cleanPool();
             Entry entry = memoryPool.get(hash);
             if (entry != null) {
                 // This TX or its hash have been previously announced.
                 if (entry.tx != null) {
-                Preconditions.checkState(entry.addresses == null);
+                    checkState(entry.addresses == null);
                     Transaction tx = entry.tx.get();
                     if (tx != null) {
-                    tx.getConfidence().markBroadcastBy(byPeer);
+                        markBroadcast(byPeer, tx);
                         log.debug("{}: Announced transaction we have seen before [{}] {}",
                                 new Object[]{byPeer, tx.getConfidence().numBroadcastPeers(), tx.getHashAsString()});
                     } else {
@@ -224,7 +252,7 @@ public class MemoryPool {
                         // nothing found it interesting enough to keep around. So do nothing.
                     }
                 } else {
-                Preconditions.checkNotNull(entry.addresses);
+                    checkNotNull(entry.addresses);
                     entry.addresses.add(byPeer);
                     log.debug("{}: Announced transaction we have seen announced before [{}] {}",
                             new Object[]{byPeer, entry.addresses.size(), hash});
@@ -238,6 +266,21 @@ public class MemoryPool {
                 memoryPool.put(hash, entry);
                 log.info("{}: Announced new transaction [1] {}", byPeer, hash);
             }
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private void markBroadcast(PeerAddress byPeer, Transaction tx) {
+        // Marking a TX as broadcast by a peer can run event listeners that might call back into Peer or PeerGroup.
+        // Thus we unlock ourselves here to avoid potential inversions.
+        checkState(lock.isLocked());
+        lock.unlock();
+        try {
+            tx.getConfidence().markBroadcastBy(byPeer);
+        } finally {
+            lock.lock();
+        }
     }
 
     /**
@@ -245,14 +288,19 @@ public class MemoryPool {
      * we only saw advertisements for it yet or it has been downloaded but garbage collected due to nowhere else
      * holding a reference to it.
      */
-    public synchronized Transaction get(Sha256Hash hash) {
+    public Transaction get(Sha256Hash hash) {
+        lock.lock();
+        try {
             Entry entry = memoryPool.get(hash);
             if (entry == null) return null;  // Unknown.
             if (entry.tx == null) return null;  // Seen but only in advertisements.
             if (entry.tx.get() == null) return null;  // Was downloaded but garbage collected.
             Transaction tx = entry.tx.get();
-        Preconditions.checkNotNull(tx);
+            checkNotNull(tx);
             return tx;
+        } finally {
+            lock.unlock();
+        }
     }
 
     /**
@@ -260,8 +308,13 @@ public class MemoryPool {
      * was broadcast, downloaded and nothing kept a reference to it will eventually be cleared out by the garbage
      * collector and wasSeen() will return false - it does not keep a permanent record of every hash ever broadcast.
      */
-    public synchronized boolean maybeWasSeen(Sha256Hash hash) {
+    public boolean maybeWasSeen(Sha256Hash hash) {
+        lock.lock();
+        try {
             Entry entry = memoryPool.get(hash);
             return entry != null;
+        } finally {
+            lock.unlock();
+        }
     }
 }

core/src/main/java/com/google/bitcoin/core/TransactionConfidence.java

@@ -65,7 +65,7 @@ public class TransactionConfidence implements Serializable {
     /** The Transaction that this confidence object is associated with. */
     private Transaction transaction;
     // Lazily created listeners array.
-    private transient ArrayList<Listener> listeners;
+    private transient CopyOnWriteArrayList<Listener> listeners;
 
     // The depth of the transaction on the best chain in blocks. An unconfirmed block has depth 0.
     private int depth;
@@ -152,6 +152,7 @@ public class TransactionConfidence implements Serializable {
     public TransactionConfidence(Transaction tx) {
         // Assume a default number of peers for our set.
         broadcastBy = new CopyOnWriteArrayList<PeerAddress>();
+        listeners = new CopyOnWriteArrayList<Listener>();
         transaction = tx;
     }
 
@@ -178,18 +179,13 @@ public class TransactionConfidence implements Serializable {
      * {@link BlockChainListener}, attach it to a {@link BlockChain} and then use the getters on the
      * confidence object to determine the new depth.</p>
      */
-    public synchronized void addEventListener(Listener listener) {
+    public void addEventListener(Listener listener) {
         Preconditions.checkNotNull(listener);
-        if (listeners == null)
-            listeners = new ArrayList<Listener>(2);
-        // Dedupe registrations. This makes the wallet code simpler.
-        if (!listeners.contains(listener))
-            listeners.add(listener);
+        listeners.addIfAbsent(listener);
     }
 
-    public synchronized void removeEventListener(Listener listener) {
+    public void removeEventListener(Listener listener) {
         Preconditions.checkNotNull(listener);
-        Preconditions.checkNotNull(listeners);
         listeners.remove(listener);
     }
 
@@ -225,11 +221,13 @@ public class TransactionConfidence implements Serializable {
      * Called by other objects in the system, like a {@link Wallet}, when new information about the confidence of a 
      * transaction becomes available.
      */
-    public synchronized void setConfidenceType(ConfidenceType confidenceType) {
+    public void setConfidenceType(ConfidenceType confidenceType) {
         // Don't inform the event listeners if the confidence didn't really change.
+        synchronized (this) {
             if (confidenceType == this.confidenceType)
                 return;
             this.confidenceType = confidenceType;
+        }
         runListeners();
     }
 
@@ -261,7 +259,7 @@ public class TransactionConfidence implements Serializable {
     }
 
     /**
-     * Returns a synchronized set of {@link PeerAddress}es that announced the transaction.
+     * Returns a snapshot of {@link PeerAddress}es that announced the transaction.
      */
     public ListIterator<PeerAddress> getBroadcastBy() {
         return broadcastBy.listIterator();
@@ -310,13 +308,18 @@ public class TransactionConfidence implements Serializable {
      * Updates the internal counter that tracks how deeply buried the block is.
      * Work is the value of block.getWork().
      */
-    public synchronized void notifyWorkDone(Block block) throws VerificationException {
+    public void notifyWorkDone(Block block) throws VerificationException {
+        boolean notify = false;
+        synchronized (this) {
             if (getConfidenceType() == ConfidenceType.BUILDING) {
                 this.depth++;
                 this.workDone = this.workDone.add(block.getWork());
-            runListeners();
+                notify = true;
             }
         }
+        if (notify)
+            runListeners();
+    }
 
     /**
      * Depth in the chain is an approximation of how much time has elapsed since the transaction has been confirmed. On
@@ -405,13 +408,9 @@ public class TransactionConfidence implements Serializable {
     }
 
     private void runListeners() {
-        EventListenerInvoker.invoke(listeners, new EventListenerInvoker<Listener>() {
-            @Override
-            public void invoke(Listener listener) {
+        for (Listener listener : listeners)
             listener.onConfidenceChanged(transaction);
     }
-        });
-    }
 
     /**
      * The source of a transaction tries to identify where it came from originally. For instance, did we download it

examples/src/main/java/com/google/bitcoin/examples/PingService.java

@@ -120,6 +120,7 @@ public class PingService {
                         ": " + tx);
                 tx.getConfidence().addEventListener(new TransactionConfidence.Listener() {
                     public void onConfidenceChanged(Transaction tx2) {
+                        // Must be thread safe.
                         if (tx2.getConfidence().getConfidenceType() == TransactionConfidence.ConfidenceType.BUILDING) {
                             // Coins were confirmed (appeared in a block).
                             tx2.getConfidence().removeEventListener(this);