Cooperative notification on setting a promise's value

Author: cpettitt-linkedin

src/com/linkedin/parseq/promise/SettablePromiseImpl.java

@@ -19,12 +19,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * @author Chris Pettitt ([email protected])
@@ -36,11 +34,8 @@
 
   private final CountDownLatch _awaitLatch = new CountDownLatch(1);
 
-  private final Lock _lock = new ReentrantLock();
-  private List<PromiseListener<T>> _listeners = new ArrayList<PromiseListener<T>>();
-  private volatile T _value;
-  private volatile Throwable _error;
-  private volatile boolean _done;
+  private final AtomicReference<Result<T>> _result = new AtomicReference<Result<T>>();
+  private final ConcurrentLinkedQueue<PromiseListener<T>> _listeners = new ConcurrentLinkedQueue<PromiseListener<T>>();
 
   @Override
   public void done(final T value) throws PromiseResolvedException
@@ -57,30 +52,29 @@ public void fail(final Throwable error) throws PromiseResolvedException
   @Override
   public T get() throws PromiseException
   {
-    ensureDone();
-    if (_error != null)
+    Result<T> result = getResult();
+    if (result._error != null)
     {
-      throw new PromiseException(_error);
+      throw new PromiseException(result._error);
     }
-    return _value;
+    return result._value;
   }
 
   @Override
   public Throwable getError() throws PromiseUnresolvedException
   {
-    ensureDone();
-    return _error;
+    return getResult()._error;
   }
 
   @Override
   public T getOrDefault(final T defaultValue) throws PromiseUnresolvedException
   {
-    ensureDone();
-    if (_error != null)
+    Result<T> result = getResult();
+    if (result._error != null)
     {
       return defaultValue;
     }
-    return _value;
+    return result._value;
   }
 
   @Override
@@ -98,93 +92,92 @@ public boolean await(final long time, final TimeUnit unit) throws InterruptedExc
   @Override
   public void addListener(final PromiseListener<T> listener)
   {
-    _lock.lock();
-    try
-    {
-      if (!isDone())
-      {
-        _listeners.add(listener);
-        return;
-      }
-    }
-    finally
+    if (isDone())
     {
-      _lock.unlock();
+      listener.onResolved(this);
+      return;
     }
 
-    notifyListener(listener);
+    _listeners.add(listener);
+
+    if (isDone())
+      purgeListeners();
   }
 
   @Override
   public boolean isDone()
   {
-    return _done;
+    return _result.get() != null;
   }
 
   @Override
   public boolean isFailed()
   {
-    return isDone() && _error != null;
+    final Result<T> voe = _result.get();
+    return voe != null && voe._error != null;
   }
 
   private void doFinish(T value, Throwable error) throws PromiseResolvedException
   {
-    final List<PromiseListener<T>> listeners;
-
-    _lock.lock();
-    try
-    {
-      ensureNotDone();
-      _value = value;
-      _error = error;
-      listeners = _listeners;
-      _listeners = null;
-      _done = true;
-    }
-    finally
+    if (!_result.compareAndSet(null, new Result<T>(value, error)))
     {
-      _lock.unlock();
+      throw new PromiseResolvedException("Promise has already been satisfied");
     }
 
-    for (int i = listeners.size() - 1; i >= 0; i--)
-    {
-      notifyListener(listeners.get(i));
-    }
+    purgeListeners();
 
     _awaitLatch.countDown();
   }
 
-  private void notifyListener(final PromiseListener<T> listener)
+  private void purgeListeners()
   {
     // We intentionally catch Throwable around the listener invocation because
     // it will cause the notifier loop and subsequent count down in doFinish to
     // be skipped, which will certainly lead to bad behavior. It could be argued
     // that the catch should not apply for use of notifyListener from
     // addListener, but it seems better to err on the side of consistency and
     // least surprise.
-    try
-    {
-      listener.onResolved(this);
-    }
-    catch (Throwable e)
+    PromiseListener<T> listener;
+    while((listener = _listeners.poll()) != null)
     {
-      LOGGER.warn("An exception was thrown by listener: " + listener.getClass(), e);
+      try
+      {
+        listener.onResolved(this);
+      }
+      catch (Throwable e)
+      {
+        LOGGER.warn("An exception was thrown by listener: " + listener.getClass(), e);
+      }
     }
   }
 
-  private void ensureNotDone() throws PromiseResolvedException
+  private Result<T> getResult() throws PromiseUnresolvedException
   {
-    if (isDone())
+    final Result<T> voe = _result.get();
+    if (voe == null)
     {
-      throw new PromiseResolvedException("Promise has already been satisfied");
+      throw new PromiseUnresolvedException("Promise has not yet been satisfied");
     }
+    return voe;
   }
 
-  private void ensureDone() throws PromiseUnresolvedException
+  private static class Result<T>
   {
-    if (!isDone())
+    private final T _value;
+    private final Throwable _error;
+
+    public Result(T value, Throwable error)
     {
-      throw new PromiseUnresolvedException("Promise has not yet been satisfied");
+      if (error != null)
+      {
+        _value = null;
+        _error = error;
+      }
+      else
+      {
+        _value = value;
+        _error = null;
+      }
     }
   }
 }