diff --git a/sdk/common/src/main/java/io/opentelemetry/sdk/common/CompletableResultCode.java b/sdk/common/src/main/java/io/opentelemetry/sdk/common/CompletableResultCode.java index 600ec73a48e..9810d0869c6 100644 --- a/sdk/common/src/main/java/io/opentelemetry/sdk/common/CompletableResultCode.java +++ b/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 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 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; } diff --git a/sdk/common/src/test/java/io/opentelemetry/sdk/common/CompletableResultCodeTest.java b/sdk/common/src/test/java/io/opentelemetry/sdk/common/CompletableResultCodeTest.java index b83e3e03ec7..f0c8deaccf4 100644 --- a/sdk/common/src/test/java/io/opentelemetry/sdk/common/CompletableResultCodeTest.java +++ b/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"); + } }