Memory Leak of ChildContinuation introduced recently

[sunpeaksfivemile]

I am trying to upgrade from 1.3.1 to 1.3.3 and one of my applications now has a memory leak. I can reproduce the leak in 1.3.3 and 1.3.4 and can roll back to 1.3.1 to resolve the leak.

Here is a simplified code snippet which reproduces the memory leak. I have a sample project setup with gradle if its useful:

    fun CoroutineScope.triggerChannel(triggersPerSecond: Int) = produce(capacity = 3) {
        val start = System.nanoTime()
        val interval = TimeUnit.SECONDS.toNanos(1) / triggersPerSecond
     
        var iteration = 0L
        while(true) {
            val now = System.nanoTime()
            val trigger = start + interval * iteration
     
            if (now - trigger > interval * 2) {
                iteration++
                continue
            }
     
            if (now <= trigger) {
                delay(TimeUnit.NANOSECONDS.toMillis(trigger - now))
                continue
            }
     
            iteration++
            send(Unit)
        }
    }
     
    fun main() = runBlocking {
        val channel = triggerChannel(300)
        repeat(10) {
            launch {
                for (value in channel) {
                    delay(10)
                }
            }
        }
    }

Over time 'kotlinx.coroutines.CancellableContinuationImpl' and 'kotlinx.coroutines.ChildContinuation' are leaked until memory pressure starts to have my application spending nearly 100% of time in GC. After running for a few hours we have 20m of these ChildContinuations and another 20m of the CancellableContinuationImpl.

Heres a screenshot from a few minutes of running this sample code:

I found a forum post which seems to be a similar issue: https://discuss.kotlinlang.org/t/correct-layering-of-coroutine-scopes-to-prevent-leak/16493

[sunpeaksfivemile]

Here is a more simplified code sample which reproduces the leak:

fun CoroutineScope.triggerChannel() = produce(capacity = 3) {
    while(true) {
        delay(1)
        send(Unit)
    }
}

fun main() = runBlocking {
    val channel = triggerChannel()
    for (value in channel) {
        delay(1)
    }
}