Propagate cancellation on awaiting Completable and Listenable futures, provide ListenableFuture.asDeferred

Rationale: in most common use-cases awaiting the future is required to integrate with future-based API, current coroutine is the only user of this future and after its cancellation this future result is no longer needed. For non-cancelling await future.asDeferred().await() should be used

Fixes #515

Author: qwwdfsad

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-guava.txt

@@ -1,4 +1,5 @@
 public final class kotlinx/coroutines/experimental/guava/ListenableFutureKt {
+	public static final fun asDeferred (Lcom/google/common/util/concurrent/ListenableFuture;)Lkotlinx/coroutines/experimental/Deferred;
 	public static final fun asListenableFuture (Lkotlinx/coroutines/experimental/Deferred;)Lcom/google/common/util/concurrent/ListenableFuture;
 	public static final fun await (Lcom/google/common/util/concurrent/ListenableFuture;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
 	public static final fun future (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;)Lcom/google/common/util/concurrent/ListenableFuture;

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

@@ -149,17 +149,46 @@ private class DeferredListenableFuture<T>(
     override fun interruptTask() { deferred.cancel() }
 }
 
+/**
+ * Converts this listenable future to an instance of [Deferred].
+ * It is cancelled when the resulting deferred is cancelled.
+ */
+public fun <T> ListenableFuture<T>.asDeferred(): Deferred<T> {
+    // Fast path if already completed
+    if (isDone) {
+        return try {
+            @Suppress("UNCHECKED_CAST")
+            CompletableDeferred(get() as T)
+        } catch (e: Throwable) {
+            // unwrap original cause from ExecutionException
+            val original = (e as? ExecutionException)?.cause ?: e
+            CompletableDeferred<T>().also { it.completeExceptionally(original) }
+        }
+    }
+    val deferred = CompletableDeferred<T>()
+    Futures.addCallback(this, object : FutureCallback<T> {
+        override fun onSuccess(result: T?) {
+            deferred.complete(result!!)
+        }
+
+        override fun onFailure(t: Throwable) {
+            deferred.completeExceptionally(t)
+        }
+    }, MoreExecutors.directExecutor())
+
+    deferred.invokeOnCompletion { cancel(false) }
+    return deferred
+}
+
 /**
  * Awaits for completion of the future without blocking a thread.
  *
  * This suspending function is cancellable.
  * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
  * stops waiting for the future and immediately resumes with [CancellationException][kotlinx.coroutines.experimental.CancellationException].
  *
- * Note, that `ListenableFuture` does not support removal of installed listeners, so on cancellation of this wait
- * a few small objects will remain in the `ListenableFuture` list of listeners until the future completes. However, the
- * care is taken to clear the reference to the waiting coroutine itself, so that its memory can be released even if
- * the future never completes.
+ * This method is intended to be used with one-shot futures, so on coroutine cancellation future is cancelled as well.
+ * If cancelling given future is undesired, `future.asDeferred().await()` should be used instead.
  */
 public suspend fun <T> ListenableFuture<T>.await(): T {
     try {
@@ -172,6 +201,7 @@ public suspend fun <T> ListenableFuture<T>.await(): T {
         val callback = ContinuationCallback(cont)
         Futures.addCallback(this, callback, MoreExecutors.directExecutor())
         cont.invokeOnCancellation {
+            cancel(false)
             callback.cont = null // clear the reference to continuation from the future's callback
         }
     }

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

@@ -10,8 +10,10 @@ import kotlinx.coroutines.experimental.CancellationException
 import org.hamcrest.core.*
 import org.junit.*
 import org.junit.Assert.*
+import org.junit.Test
 import java.io.*
 import java.util.concurrent.*
+import kotlin.test.assertFailsWith
 
 class ListenableFutureTest : TestBase() {
     @Before
@@ -187,4 +189,90 @@ class ListenableFutureTest : TestBase() {
         yield() // yield main thread to job
         finish(6)
     }
+
+    @Test
+    fun testFutureCancellation() = runTest {
+        val future = awaitFutureWithCancel(true)
+        assertTrue(future.isCancelled)
+        assertFailsWith<CancellationException> { future.get() }
+        finish(4)
+    }
+
+    @Test
+    fun testNoFutureCancellation() = runTest {
+        val future = awaitFutureWithCancel(false)
+        assertFalse(future.isCancelled)
+        assertEquals(42, future.get())
+        finish(4)
+    }
+
+    @Test
+    fun testCompletedFutureAsDeferred() = runTest {
+        val future = SettableFuture.create<Int>()
+        val task = async {
+            expect(2)
+            assertEquals(42, future.asDeferred().await())
+            expect(4)
+        }
+
+        expect(1)
+        yield()
+        expect(3)
+        future.set(42)
+        task.join()
+        finish(5)
+    }
+
+    @Test
+    fun testFailedFutureAsDeferred() = runTest {
+        val future = SettableFuture.create<Int>().apply {
+            setException(TestException())
+        }
+        val deferred = future.asDeferred()
+        assertTrue(deferred.isCompletedExceptionally)
+        val completionException = deferred.getCompletionExceptionOrNull()!!
+        assertTrue(completionException is TestException)
+
+        try {
+            deferred.await()
+            expectUnreached()
+        } catch (e: Exception) {
+            assertTrue(e is TestException)
+        }
+    }
+
+    @Test
+    fun testThrowingFutureAsDeferred() = runTest {
+        val executor = MoreExecutors.listeningDecorator(ForkJoinPool.commonPool())
+        val future = executor.submit(Callable { throw TestException() })
+        val deferred = async {
+            future.asDeferred().await()
+        }
+
+        try {
+            deferred.await()
+            expectUnreached()
+        } catch (e: Exception) {
+            assertTrue(e is TestException)
+        }
+    }
+
+    private suspend fun CoroutineScope.awaitFutureWithCancel(cancellable: Boolean): ListenableFuture<Int> {
+        val latch = CountDownLatch(1)
+        val executor = MoreExecutors.listeningDecorator(ForkJoinPool.commonPool())
+        val future = executor.submit(Callable { latch.await(); 42 })
+        val deferred = async {
+            expect(2)
+            if (cancellable) future.await()
+            else future.asDeferred().await()
+        }
+        expect(1)
+        yield()
+        deferred.cancel()
+        expect(3)
+        latch.countDown()
+        return future
+    }
+
+    private class TestException : Exception()
 }

integration/kotlinx-coroutines-jdk8/src/future/Future.kt

@@ -190,11 +190,8 @@ public suspend fun <T> CompletableFuture<T>.await(): T =
  * This suspending function is cancellable.
  * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
  * stops waiting for the completion stage and immediately resumes with [CancellationException][kotlinx.coroutines.experimental.CancellationException].
- *
- * Note, that `CompletionStage` implementation does not support prompt removal of installed listeners, so on cancellation of this wait
- * a few small objects will remain in the `CompletionStage` stack of completion actions until it completes itself.
- * However, the care is taken to clear the reference to the waiting coroutine itself, so that its memory can be
- * released even if the completion stage never completes.
+ * This method is intended to be used with one-shot futures, so on coroutine cancellation completion stage is cancelled as well if it is instance of [CompletableFuture].
+ * If cancelling given stage is undesired, `stage.asDeferred().await()` should be used instead.
  */
 public suspend fun <T> CompletionStage<T>.await(): T {
     // fast path when CompletableFuture is already done (does not suspend)
@@ -211,7 +208,9 @@ public suspend fun <T> CompletionStage<T>.await(): T {
         val consumer = ContinuationConsumer(cont)
         whenComplete(consumer)
         cont.invokeOnCancellation {
-            consumer.cont = null // shall clear reference to continuation
+            // mayInterruptIfRunning is not used
+            (this as? CompletableFuture<T>)?.cancel(false)
+            consumer.cont = null // shall clear reference to continuation to aid GC
         }
     }
 }

integration/kotlinx-coroutines-jdk8/test/future/FutureTest.kt

@@ -9,12 +9,14 @@ import kotlinx.coroutines.experimental.CancellationException
 import org.hamcrest.core.*
 import org.junit.*
 import org.junit.Assert.*
+import org.junit.Test
 import java.util.concurrent.*
 import java.util.concurrent.atomic.*
 import java.util.concurrent.locks.*
 import java.util.function.*
 import kotlin.concurrent.*
 import kotlin.coroutines.experimental.*
+import kotlin.test.assertFailsWith
 
 class FutureTest : TestBase() {
     @Before
@@ -332,6 +334,42 @@ class FutureTest : TestBase() {
         assertEquals("value", result)
     }
 
+    @Test
+    fun testFutureCancellation() = runTest {
+        val future = awaitFutureWithCancel(true)
+        assertTrue(future.isCompletedExceptionally)
+        assertFailsWith<CancellationException> { future.get() }
+        finish(4)
+    }
+
+    @Test
+    fun testNoFutureCancellation() = runTest {
+        val future = awaitFutureWithCancel(false)
+        assertFalse(future.isCompletedExceptionally)
+        assertEquals(239, future.get())
+        finish(4)
+    }
+
+    private suspend fun CoroutineScope.awaitFutureWithCancel(cancellable: Boolean): CompletableFuture<Int> {
+        val latch = CountDownLatch(1)
+        val future = CompletableFuture.supplyAsync({
+            latch.await()
+            239
+        })
+
+        val deferred = async {
+            expect(2)
+            if (cancellable) future.await()
+            else future.asDeferred().await()
+        }
+        expect(1)
+        yield()
+        deferred.cancel()
+        expect(3)
+        latch.countDown()
+        return future
+    }
+
     class TestException(message: String) : Exception(message)
 
     private fun wrapContinuation(wrapper: (() -> Unit) -> Unit): CoroutineDispatcher = object : CoroutineDispatcher() {