Fix stack overflow when calling asDeferred on a Future many times

Fixes #4156

kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

@@ -454,7 +454,6 @@ public final class kotlinx/coroutines/JobKt {
 	public static synthetic fun cancelChildren$default (Lkotlinx/coroutines/Job;Ljava/lang/Throwable;ILjava/lang/Object;)V
 	public static synthetic fun cancelChildren$default (Lkotlinx/coroutines/Job;Ljava/util/concurrent/CancellationException;ILjava/lang/Object;)V
 	public static final fun cancelFutureOnCancellation (Lkotlinx/coroutines/CancellableContinuation;Ljava/util/concurrent/Future;)V
-	public static final fun cancelFutureOnCompletion (Lkotlinx/coroutines/Job;Ljava/util/concurrent/Future;)Lkotlinx/coroutines/DisposableHandle;
 	public static final fun ensureActive (Lkotlin/coroutines/CoroutineContext;)V
 	public static final fun ensureActive (Lkotlinx/coroutines/Job;)V
 	public static final fun getJob (Lkotlin/coroutines/CoroutineContext;)Lkotlinx/coroutines/Job;

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

@@ -138,7 +138,7 @@ public fun <T> CompletionStage<T>.asDeferred(): Deferred<T> {
             handleCoroutineException(EmptyCoroutineContext, e)
         }
     }
-    result.cancelFutureOnCompletion(future)
+    result.invokeOnCompletion(handler = CancelFutureOnCompletion(future))
     return result
 }
 
@@ -190,3 +190,18 @@ private class ContinuationHandler<T>(
         }
     }
 }
+
+private class CancelFutureOnCompletion(
+    private val future: Future<*>
+) : JobNode() {
+    override fun invoke(cause: Throwable?) {
+        // Don't interrupt when cancelling future on completion, because no one is going to reset this
+        // interruption flag and it will cause spurious failures elsewhere.
+        // We do not cancel the future if it's already completed in some way,
+        // because `cancel` on a completed future won't change the state but is not guaranteed to behave well
+        // on reentrancy. See https://github.com/Kotlin/kotlinx.coroutines/issues/4156
+        if (cause != null && !future.isDone) future.cancel(false)
+    }
+
+    override val onCancelling get() = false
+}

kotlinx-coroutines-core/jvm/src/Future.kt

@@ -9,42 +9,17 @@ import java.util.concurrent.*
  * Cancels a specified [future] when this job is cancelled.
  * This is a shortcut for the following code with slightly more efficient implementation (one fewer object created).
  * ```
- * invokeOnCompletion { if (it != null) future.cancel(false) }
- * ```
- *
- * @suppress **This an internal API and should not be used from general code.**
- */
-@InternalCoroutinesApi
-public fun Job.cancelFutureOnCompletion(future: Future<*>): DisposableHandle =
-    invokeOnCompletion(handler = CancelFutureOnCompletion(future))
-
-/**
- * Cancels a specified [future] when this job is cancelled.
- * This is a shortcut for the following code with slightly more efficient implementation (one fewer object created).
- * ```
- * invokeOnCancellation { if (it != null) future.cancel(false) }
+ * invokeOnCancellation { if (it != null && !future.isDone) future.cancel(false) }
  * ```
  */
 public fun CancellableContinuation<*>.cancelFutureOnCancellation(future: Future<*>): Unit =
     invokeOnCancellation(handler = CancelFutureOnCancel(future))
 
-private class CancelFutureOnCompletion(
-    private val future: Future<*>
-) : JobNode() {
-    override fun invoke(cause: Throwable?) {
-        // Don't interrupt when cancelling future on completion, because no one is going to reset this
-        // interruption flag and it will cause spurious failures elsewhere
-        if (cause != null) future.cancel(false)
-    }
-
-    override val onCancelling get() = false
-}
-
 private class CancelFutureOnCancel(private val future: Future<*>) : CancelHandler {
     override fun invoke(cause: Throwable?) {
         // Don't interrupt when cancelling future on completion, because no one is going to reset this
         // interruption flag and it will cause spurious failures elsewhere
-        if (cause != null)  future.cancel(false)
+        if (cause != null && !future.isDone) future.cancel(false)
     }
     override fun toString() = "CancelFutureOnCancel[$future]"
 }

kotlinx-coroutines-core/jvm/test/jdk8/future/FutureTest.kt

@@ -595,4 +595,14 @@ class FutureTest : TestBase() {
             GlobalScope.future<Unit>(start = CoroutineStart.LAZY) {  }
         }
     }
+
+    @Test
+    fun testStackOverflowOnExceptionalCompletion() = runTest {
+        val future = CompletableFuture<Unit>()
+        val didRun = AtomicBoolean(false)
+        future.whenComplete { _, _ -> didRun.set(true) }
+        val deferreds = (0 until 100000L).map { future.asDeferred() }
+        future.completeExceptionally(RuntimeException())
+        assertTrue(deferreds.all { it.isCompleted })
+    }
 }