Unused objects not collected in Flow.transform

[Vladiatro]

I produce chunks of data with flow {} and then transform them into smaller parts as in the following example:

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.transform
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

fun main() {
    runBlocking {
        generate().flattenToBooleans().collect()
    }
}

fun generate() = flow {
    emit((1..99999).toMutableList())
    delay(99999)
}

fun Flow<MutableList<Int>>.flattenToBooleans() = transform { list ->
    list.forEach { emit(it == 99999) }
}

suspend fun Flow<Boolean>.collect() = collect { isLast ->
    if (isLast) {
        println("the last has been collected")
    }
}

I expect the initial list of 99999 integers to be removed by the GC after it is transformed to booleans, but for some reason, it is still kept in the memory after "the last has been collected" is printed:

If I manually clear the list, the objects are removed and everything works normally:

// ...
fun Flow<MutableList<Int>>.flattenToBooleans() = transform { list ->
    list.forEach { emit(it == 99999) }
    list.clear()
}
// ...

If I send another list of integers in flow {}, the first one is expectedly removed from the heap, but the new one is still present somewhere in another place:

I expect the processed elements to be removed by the GC as soon as possible when using transform to free up the space for the other coroutines. This is the code example where commenting out list.clear() makes the program run out of the heap space with -Xmx500m:

fun main() {
    runBlocking {
        repeat(3) {
            launch {
                generate().flattenToBooleans().collect()
            }
            delay(500)
        }
    }
}

fun generate() = flow {
    emit((1..9999999).toMutableList())
    delay(99999)
}

fun Flow<MutableList<Int>>.flattenToBooleans() = transform { list ->
    list.forEach { emit(it == 9999999) }
    list.clear() // comment this line to make the code fail with -Xmx500m
}

suspend fun Flow<Boolean>.collect() = collect { isLast ->
    if (isLast) {
        println("last collected")
    }
}

Kotlin 1.6.10, coroutines 1.6.0, azul-15.0.6 JVM on an ARM macOS.

[qwwdfsad]

Thanks for the detailed report!

Self-contained reproducer:

@Test
fun foo() = runBlocking {
    val flow = flow {
        emit(listOf(239))
        expect(2)
        hang {}
    }
    val j = flow.transform { l -> l.forEach { _ -> emit(42) } }.onEach { expect (1) }.launchIn(this)
    yield()
    expect(3)
    FieldWalker.assertReachableCount(0, j) { it == 239 }
    j.cancelAndJoin()
    finish(4)
}