Do not throw from JobListenableFuture.isCancelled (#2498)

* Fix toString representation of JobListenableFuture
* Do not throw from JobListenableFuture.isCancelled. This properly handles ExecutionException that can be thrown from getUninterruptibly.

Fixes #2421

Co-authored-by: Vsevolod Tolstopyatov <[email protected]>

Author: vadimsemenov

integration/kotlinx-coroutines-guava/src/ListenableFuture.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
  */
 
 package kotlinx.coroutines.guava
@@ -302,7 +302,8 @@ private class ListenableFutureCoroutine<T>(
 ) : AbstractCoroutine<T>(context) {
 
     // JobListenableFuture propagates external cancellation to `this` coroutine. See JobListenableFuture.
-    @JvmField val future = JobListenableFuture<T>(this)
+    @JvmField
+    val future = JobListenableFuture<T>(this)
 
     override fun onCompleted(value: T) {
         future.complete(value)
@@ -347,6 +348,17 @@ private class JobListenableFuture<T>(private val jobToCancel: Job): ListenableFu
      */
     private val auxFuture = SettableFuture.create<Any>()
 
+    /**
+     * `true` if [auxFuture.get][ListenableFuture.get] throws [ExecutionException].
+     *
+     * Note: this is eventually consistent with the state of [auxFuture].
+     *
+     * Unfortunately, there's no API to figure out if [ListenableFuture] throws [ExecutionException]
+     * apart from calling [ListenableFuture.get] on it. To avoid unnecessary [ExecutionException] allocation
+     * we use this field as an optimization.
+     */
+    private var auxFutureIsFailed: Boolean = false
+
     /**
      * When the attached coroutine [isCompleted][Job.isCompleted] successfully
      * its outcome should be passed to this method.
@@ -366,7 +378,8 @@ private class JobListenableFuture<T>(private val jobToCancel: Job): ListenableFu
     // CancellationException is wrapped into `Cancelled` to preserve original cause and message.
     // All the other exceptions are delegated to SettableFuture.setException.
     fun completeExceptionallyOrCancel(t: Throwable): Boolean =
-        if (t is CancellationException) auxFuture.set(Cancelled(t)) else auxFuture.setException(t)
+        if (t is CancellationException) auxFuture.set(Cancelled(t))
+        else auxFuture.setException(t).also { if (it) auxFutureIsFailed = true }
 
     /**
      * Returns cancellation _in the sense of [Future]_. This is _not_ equivalent to
@@ -385,7 +398,16 @@ private class JobListenableFuture<T>(private val jobToCancel: Job): ListenableFu
         // this Future hasn't itself been successfully cancelled, the Future will return
         // isCancelled() == false. This is the only discovered way to reconcile the two different
         // cancellation contracts.
-        return auxFuture.isCancelled || (isDone && Uninterruptibles.getUninterruptibly(auxFuture) is Cancelled)
+        return auxFuture.isCancelled || isDone && !auxFutureIsFailed && try {
+            Uninterruptibles.getUninterruptibly(auxFuture) is Cancelled
+        } catch (e: CancellationException) {
+            // `auxFuture` got cancelled right after `auxFuture.isCancelled` returned false.
+            true
+        } catch (e: ExecutionException) {
+            // `auxFutureIsFailed` hasn't been updated yet.
+            auxFutureIsFailed = true
+            false
+        }
     }
 
     /**
@@ -455,7 +477,7 @@ private class JobListenableFuture<T>(private val jobToCancel: Job): ListenableFu
             try {
                 when (val result = Uninterruptibles.getUninterruptibly(auxFuture)) {
                     is Cancelled -> append("CANCELLED, cause=[${result.exception}]")
-                    else -> append("SUCCESS, result=[$result")
+                    else -> append("SUCCESS, result=[$result]")
                 }
             } catch (e: CancellationException) {
                 // `this` future was cancelled by `Future.cancel`. In this case there's no cause or message.
@@ -469,6 +491,7 @@ private class JobListenableFuture<T>(private val jobToCancel: Job): ListenableFu
         } else {
             append("PENDING, delegate=[$auxFuture]")
         }
+        append(']')
     }
 }
 

integration/kotlinx-coroutines-guava/test/ListenableFutureTest.kt

@@ -680,6 +680,50 @@ class ListenableFutureTest : TestBase() {
         finish(5)
     }
 
+    @Test
+    fun testFutureCompletedExceptionally() = runTest {
+        val testException = TestException()
+        // NonCancellable to not propagate error to this scope.
+        val future = future(context = NonCancellable) {
+            throw testException
+        }
+        yield()
+        assertTrue(future.isDone)
+        assertFalse(future.isCancelled)
+        val thrown = assertFailsWith<ExecutionException> { future.get() }
+        assertEquals(testException, thrown.cause)
+    }
+
+    @Test
+    fun testAsListenableFutureCompletedExceptionally() = runTest {
+        val testException = TestException()
+        val deferred = CompletableDeferred<String>().apply {
+            completeExceptionally(testException)
+        }
+        val asListenableFuture = deferred.asListenableFuture()
+        assertTrue(asListenableFuture.isDone)
+        assertFalse(asListenableFuture.isCancelled)
+        val thrown = assertFailsWith<ExecutionException> { asListenableFuture.get() }
+        assertEquals(testException, thrown.cause)
+    }
+
+    @Test
+    fun stressTestJobListenableFutureIsCancelledDoesNotThrow() = runTest {
+        repeat(1000) {
+            val deferred = CompletableDeferred<String>()
+            val asListenableFuture = deferred.asListenableFuture()
+            // We heed two threads to test a race condition.
+            withContext(Dispatchers.Default) {
+                val cancellationJob = launch {
+                    asListenableFuture.cancel(false)
+                }
+                while (!cancellationJob.isCompleted) {
+                    asListenableFuture.isCancelled // Shouldn't throw.
+                }
+            }
+        }
+    }
+
     private inline fun <reified T: Throwable> ListenableFuture<*>.checkFutureException() {
         val e = assertFailsWith<ExecutionException> { get() }
         val cause = e.cause!!

integration/kotlinx-coroutines-guava/test/ListenableFutureToStringTest.kt

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2016-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.guava
+
+import kotlinx.coroutines.*
+import org.junit.Test
+import kotlin.test.*
+
+class ListenableFutureToStringTest : TestBase() {
+    @Test
+    fun testSuccessfulFuture() = runTest {
+        val deferred = CompletableDeferred("OK")
+        val succeededFuture = deferred.asListenableFuture()
+        val toString = succeededFuture.toString()
+        assertTrue(message = "Unexpected format: $toString") {
+            toString.matches(Regex("""kotlinx\.coroutines\.guava\.JobListenableFuture@[^\[]*\[status=SUCCESS, result=\[OK]]"""))
+        }
+    }
+
+    @Test
+    fun testFailedFuture() = runTest {
+        val exception = TestRuntimeException("test")
+        val deferred = CompletableDeferred<String>().apply {
+            completeExceptionally(exception)
+        }
+        val failedFuture = deferred.asListenableFuture()
+        val toString = failedFuture.toString()
+        assertTrue(message = "Unexpected format: $toString") {
+            toString.matches(Regex("""kotlinx\.coroutines\.guava\.JobListenableFuture@[^\[]*\[status=FAILURE, cause=\[$exception]]"""))
+        }
+    }
+
+    @Test
+    fun testPendingFuture() = runTest {
+        val deferred = CompletableDeferred<String>()
+        val pendingFuture = deferred.asListenableFuture()
+        val toString = pendingFuture.toString()
+        assertTrue(message = "Unexpected format: $toString") {
+            toString.matches(Regex("""kotlinx\.coroutines\.guava\.JobListenableFuture@[^\[]*\[status=PENDING, delegate=\[.*]]"""))
+        }
+    }
+
+    @Test
+    fun testCancelledCoroutineAsListenableFuture() = runTest {
+        val exception = CancellationException("test")
+        val deferred = CompletableDeferred<String>().apply {
+            cancel(exception)
+        }
+        val cancelledFuture = deferred.asListenableFuture()
+        val toString = cancelledFuture.toString()
+        assertTrue(message = "Unexpected format: $toString") {
+            toString.matches(Regex("""kotlinx\.coroutines\.guava\.JobListenableFuture@[^\[]*\[status=CANCELLED, cause=\[$exception]]"""))
+        }
+    }
+
+    @Test
+    fun testCancelledFuture() = runTest {
+        val deferred = CompletableDeferred<String>()
+        val cancelledFuture = deferred.asListenableFuture().apply {
+            cancel(false)
+        }
+        val toString = cancelledFuture.toString()
+        assertTrue(message = "Unexpected format: $toString") {
+            toString.matches(Regex("""kotlinx\.coroutines\.guava\.JobListenableFuture@[^\[]*\[status=CANCELLED]"""))
+        }
+    }
+}