Dispose waiting node when Deferred.await is cancelled

This is part of the problem that is shown in issue #893. The other
part of the problem (what happens during race) is addressed by PR #896

Fixes #893

Author: elizarov

common/kotlinx-coroutines-core-common/src/JobSupport.kt

@@ -1106,7 +1106,7 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
          */
         val cont = AwaitContinuation(uCont.intercepted(), this)
         cont.initCancellability()
-        invokeOnCompletion(ResumeAwaitOnCompletion(this, cont).asHandler)
+        cont.disposeOnCancellation(invokeOnCompletion(ResumeAwaitOnCompletion(this, cont).asHandler))
         cont.getResult()
     }
 

core/kotlinx-coroutines-core/test/CancelledAwaitStressTest.kt

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import org.junit.*
+
+class CancelledAwaitStressTest : TestBase() {
+    private val n = 1000 * stressTestMultiplier
+
+    /**
+     * Tests that memory does not leak from cancelled [Deferred.await]
+     */
+    @Test
+    fun testCancelledAwait() = runTest {
+        val d = async {
+            delay(Long.MAX_VALUE)
+        }
+        repeat(n) {
+            val waiter = launch(start = CoroutineStart.UNDISPATCHED) {
+                val a = ByteArray(10000000) // allocate 10M of memory here
+                d.await()
+                keepMe(a) // make sure it is kept in state machine
+            }
+            waiter.cancel() // cancel await
+            yield() // complete the waiter job, release its memory
+        }
+        d.cancel() // done test
+    }
+
+    /**
+     * Tests that memory does not leak from cancelled [Job.join]
+     */
+    @Test
+    fun testCancelledJoin() = runTest {
+        val j = launch {
+            delay(Long.MAX_VALUE)
+        }
+        repeat(n) {
+            val joiner = launch(start = CoroutineStart.UNDISPATCHED) {
+                val a = ByteArray(10000000) // allocate 10M of memory here
+                j.join()
+                keepMe(a) // make sure it is kept in state machine
+            }
+            joiner.cancel() // cancel join
+            yield() // complete the joiner job, release its memory
+        }
+        j.cancel() // done test
+    }
+
+    private fun keepMe(a: ByteArray) {
+        // does nothing, makes sure the variable is kept in state-machine
+    }
+}