SPY-167: Avoid deadlock on notify listeners.

Motivation
----------
Because of how the addListener and notifyListener works, it could be that
one deadlocks each other because they are waiting on the same locks from
different threads.

Modifications
-------------
The listeners are copied before notified, moving it out of the synch block.

Result
------
Since notify is not synched anymore, the lock should be not happening anymore.

Note that also the listeners array is set to empty which avoid notifying listeners
more than once potentially.

Thanks to @adamhonen for pointing out the fix.

Change-Id: I7ebda58c2eadd62d8885e3eeac79d5a971e07979
Reviewed-on: http://review.couchbase.org/36219
Reviewed-by: Matt Ingenthron <matt@couchbase.com>
Tested-by: Michael Nitschinger <michael.nitschinger@couchbase.com>

Author: daschl

src/main/java/net/spy/memcached/internal/AbstractListenableFuture.java

@@ -1,6 +1,6 @@
 /**
  * Copyright (C) 2006-2009 Dustin Sallings
- * Copyright (C) 2009-2013 Couchbase, Inc.
+ * Copyright (C) 2009-2014 Couchbase, Inc.
  *
  * Permission is hereby granted, free of charge, to any person obtaining a copy
  * of this software and associated documentation files (the "Software"), to deal
@@ -23,11 +23,12 @@
 
 package net.spy.memcached.internal;
 
+import net.spy.memcached.compat.SpyObject;
+
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
-import net.spy.memcached.compat.SpyObject;
 
 /**
  * The {@link AbstractListenableFuture} implements common functionality shared
@@ -55,7 +56,12 @@ public abstract class AbstractListenableFuture
    */
   private List<GenericCompletionListener<? extends Future<T>>> listeners;
 
-  public AbstractListenableFuture(ExecutorService executor) {
+  /**
+   * Creates a new {@link AbstractListenableFuture}.
+   *
+   * @param executor the executor in which the callbacks will be executed in.
+   */
+  protected AbstractListenableFuture(ExecutorService executor) {
     service = executor;
     listeners = new ArrayList<GenericCompletionListener<? extends Future<T>>>();
   }
@@ -79,24 +85,19 @@ protected ExecutorService executor() {
    * @return the current future to allow chaining.
    */
   protected Future<T> addToListeners(
-    GenericCompletionListener<? extends Future<T>> listener) {
+    final GenericCompletionListener<? extends Future<T>> listener) {
     if (listener == null) {
       throw new IllegalArgumentException("The listener can't be null.");
     }
 
-    if(isDone()) {
-      notifyListener(executor(), this, listener);
-      return this;
+    synchronized(this) {
+      listeners.add(listener);
     }
 
-    synchronized(this) {
-      if (!isDone()) {
-        listeners.add(listener);
-        return this;
-      }
+    if(isDone()) {
+      notifyListeners();
     }
 
-    notifyListener(executor(), this, listener);
     return this;
   }
 
@@ -117,7 +118,7 @@ public void run() {
         } catch(Throwable t) {
           getLogger().warn(
             "Exception thrown wile executing " + listener.getClass().getName()
-            + ".operationComplete()", t);
+              + ".operationComplete()", t);
         }
       }
     });
@@ -139,9 +140,15 @@ protected void notifyListeners() {
    *
    * @param future the future to pass on to the listeners.
    */
-  protected synchronized void notifyListeners(final Future<?> future) {
+  protected void notifyListeners(final Future<?> future) {
+    final List<GenericCompletionListener<? extends Future<T>>> copy =
+      new ArrayList<GenericCompletionListener<? extends Future<T>>>();
+    synchronized(this) {
+      copy.addAll(listeners);
+      listeners = new ArrayList<GenericCompletionListener<? extends Future<T>>>();
+    }
     for (GenericCompletionListener<? extends Future<? super T>> listener
-      : listeners) {
+      : copy) {
       notifyListener(executor(), future, listener);
     }
   }
@@ -158,16 +165,11 @@ protected Future<T> removeFromListeners(
       throw new IllegalArgumentException("The listener can't be null.");
     }
 
-    if(isDone()) {
-      return this;
-    }
-
-    synchronized(this) {
-      if (!isDone()) {
+    if (!isDone()) {
+      synchronized(this) {
         listeners.remove(listener);
       }
     }
-
     return this;
   }
-}
+}