Run CompletableResultCode completion action outside the lock

sdk/common/src/main/java/io/opentelemetry/sdk/common/CompletableResultCode.java

@@ -8,12 +8,15 @@
 import io.opentelemetry.api.internal.GuardedBy;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 import javax.annotation.Nullable;
 
 /**
@@ -24,6 +27,9 @@
  * convey a result at a later time. CompletableResultCode facilitates this.
  */
 public final class CompletableResultCode {
+
+  private static final Logger logger = Logger.getLogger(CompletableResultCode.class.getName());
+
   /** Returns a {@link CompletableResultCode} that has been completed successfully. */
   public static CompletableResultCode ofSuccess() {
     return SUCCESS;
@@ -100,15 +106,7 @@ public CompletableResultCode() {}
 
   /** Complete this {@link CompletableResultCode} successfully if it is not already completed. */
   public CompletableResultCode succeed() {
-    synchronized (lock) {
-      if (succeeded == null) {
-        succeeded = true;
-        for (Runnable action : completionActions) {
-          action.run();
-        }
-      }
-    }
-    return this;
+    return complete(/* success= */ true, /* throwable= */ null);
   }
 
   /**
@@ -131,18 +129,48 @@ public CompletableResultCode failExceptionally(@Nullable Throwable throwable) {
   }
 
   private CompletableResultCode failInternal(@Nullable Throwable throwable) {
+    return complete(/* success= */ false, throwable);
+  }
+
+  private CompletableResultCode complete(boolean success, @Nullable Throwable throwable) {
+    List<Runnable> actions = null;
     synchronized (lock) {
       if (succeeded == null) {
-        succeeded = false;
-        this.throwable = throwable;
-        for (Runnable action : completionActions) {
-          action.run();
+        succeeded = success;
+        if (!success) {
+          this.throwable = throwable;
+        }
+        if (!completionActions.isEmpty()) {
+          actions = new ArrayList<>(completionActions);
+          completionActions.clear();
         }
       }
     }
+    // Completion actions run without the lock held. Running them under the lock lets an action
+    // which completes another result deadlock through lock inversion.
+    if (actions != null) {
+      runActions(actions);
+    }
     return this;
   }
 
+  private static void runActions(List<Runnable> actions) {
+    RuntimeException firstException = null;
+    for (Runnable action : actions) {
+      try {
+        action.run();
+      } catch (RuntimeException e) {
+        logger.log(Level.WARNING, "Exception thrown by completion action.", e);
+        if (firstException == null) {
+          firstException = e;
+        }
+      }
+    }
+    if (firstException != null) {
+      throw firstException;
+    }
+  }
+
   /**
    * Obtain the current state of completion. Generally call once completion is achieved via the
    * {@link #whenComplete(Runnable)} method.
@@ -173,7 +201,9 @@ public Throwable getFailureThrowable() {
   }
 
   /**
-   * Perform an action on completion. Actions are guaranteed to be called only once.
+   * Perform an action on completion. Actions are guaranteed to be called only once. Actions are not
+   * invoked while internal locks are held. Every action runs even if an earlier one throws. Each
+   * exception is logged, and the first one is rethrown after all the actions have executed.
    *
    * @param action the action to perform
    * @return this completable result so that it may be further composed
@@ -188,7 +218,7 @@ public CompletableResultCode whenComplete(Runnable action) {
       }
     }
     if (runNow) {
-      action.run();
+      runActions(Collections.singletonList(action));
     }
     return this;
   }

sdk/common/src/test/java/io/opentelemetry/sdk/common/CompletableResultCodeTest.java

@@ -6,6 +6,8 @@
 package io.opentelemetry.sdk.common;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.awaitility.Awaitility.await;
 
 import com.google.common.util.concurrent.Uninterruptibles;
@@ -14,6 +16,7 @@
 import java.util.Collections;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import org.junit.jupiter.api.Test;
 
@@ -234,4 +237,97 @@ void joinInterrupted() {
     assertThat(result.isSuccess()).isFalse();
     assertThat(result.isDone()).isFalse();
   }
+
+  @Test
+  void callbackCompletionAnotherResultDoesNotDeadlock() throws InterruptedException {
+    CompletableResultCode first = new CompletableResultCode();
+    CompletableResultCode second = new CompletableResultCode();
+    CountDownLatch callbackEntered = new CountDownLatch(2);
+    CountDownLatch release = new CountDownLatch(1);
+
+    first.whenComplete(
+        () -> {
+          callbackEntered.countDown();
+          Uninterruptibles.awaitUninterruptibly(release);
+          second.succeed();
+        });
+
+    second.whenComplete(
+        () -> {
+          callbackEntered.countDown();
+          Uninterruptibles.awaitUninterruptibly(release);
+          first.succeed();
+        });
+
+    Thread firstThread = new Thread(first::succeed, "complete-first");
+    Thread secondThread = new Thread(second::succeed, "complete-second");
+    firstThread.setDaemon(true);
+    secondThread.setDaemon(true);
+    firstThread.start();
+    secondThread.start();
+
+    assertThat(callbackEntered.await(10, TimeUnit.SECONDS)).isTrue();
+    release.countDown();
+
+    firstThread.join(10_000);
+    secondThread.join(10_000);
+
+    assertThat(firstThread.isAlive()).isFalse();
+    assertThat(secondThread.isAlive()).isFalse();
+    assertThat(first.isSuccess()).isTrue();
+    assertThat(second.isSuccess()).isTrue();
+  }
+
+  @Test
+  void completionActionExceptionDoesNotAbortLaterActions() {
+    CompletableResultCode result = new CompletableResultCode();
+    AtomicBoolean actionInvoked = new AtomicBoolean();
+
+    result.whenComplete(
+        () -> {
+          throw new RuntimeException("callback failure");
+        });
+    result.whenComplete(() -> actionInvoked.set(true));
+    assertThatCode(result::succeed)
+        .isInstanceOf(RuntimeException.class)
+        .hasMessage("callback failure");
+
+    assertThat(actionInvoked).isTrue();
+    assertThat(result.isSuccess()).isTrue();
+  }
+
+  @Test
+  void completionActionExceptionDoesNotPreventOfAllCompletion() {
+    CompletableResultCode source = new CompletableResultCode();
+    CompletableResultCode other = new CompletableResultCode();
+
+    // Registered before ofAll so that it runs before ofAll's bookkeeping action.
+    source.whenComplete(
+        () -> {
+          throw new RuntimeException("callback failure");
+        });
+    CompletableResultCode all = CompletableResultCode.ofAll(Arrays.asList(source, other));
+    other.succeed();
+
+    assertThatThrownBy(source::succeed)
+        .isInstanceOf(RuntimeException.class)
+        .hasMessage("callback failure");
+
+    assertThat(all.isDone()).isTrue();
+    assertThat(all.isSuccess()).isTrue();
+  }
+
+  @Test
+  void completionActionExceptionDoesNotEscapeWhenAlreadyComplete() {
+    CompletableResultCode result = new CompletableResultCode().succeed();
+
+    assertThatCode(
+            () ->
+                result.whenComplete(
+                    () -> {
+                      throw new RuntimeException("callback failure");
+                    }))
+        .isInstanceOf(RuntimeException.class)
+        .hasMessage("callback failure");
+  }
 }