Memory Leak in CoroutineDispatcher.asScheduler() and DispatcherScheduler

[lengfeld]

Describe the bug

The following code produces a memory leak. E.g. after some time the code crashes with a out of Memory exception:

        [...]
        val buggyScheduler =  Dispatchers.Default.asScheduler()
        val workingScheduler =  Schedulers.computation()
        
        val s = if (BUG_ENALBED) buggyScheduler else workingScheduler

        disposable = Flowable.interval(1, 1, TimeUnit.SECONDS, Schedulers.computation())
            .doOnNext { Log.d(TAG, "timer tick every 1  second here; create new Flowable") }
            .switchMap {
                val bigObject = ByteArray(10 * 1024 * 1024) // 10 MiB

                Flowable.interval(5, 1, TimeUnit.MINUTES, s)
                    .doOnNext {
                        // The object "bigObject" is captured in this lambda.
                        Log.d(TAG, "Inner Flowable has fired! size=${bigObject.size}")
                    }
            }
            // The following line is never executed. The upstream Flowable is replaced every one second.
            // So it has never a chance to emit it's first element, because it's set to five minutes.
            //
            // What is the bug?
            // The upstream flowable is replaced. Therefore it's disposed. But the implementation in
            // RxScheduler.java does _not_ removed the scheduled Runnable/coroutine from the task queue
            // even so they are disposed.
            // So every second a new runnable/coroutine is added to the task queue that first runs after 5 minutes.
            // These runnables/coroutines queue up instead of being garbage collected. If the runnables
            // contains/reference big objects you have a memory leak that is visible.
            .doOnNext { Log.d(TAG, "Never executed") }
            .subscribe()

The bug is file RxScheduler.kt in line 135/136

    if (delayMillis <= 0) {
        toSchedule.run()
    } else {
        @Suppress("INVISIBLE_MEMBER", "INVISIBLE_REFERENCE") // do not remove the INVISIBLE_REFERENCE suppression: required in K2
        ctx.delay.invokeOnTimeout(delayMillis, toSchedule, ctx).let { handle = it }
    }

The return value of invokeOnTimeout, a Disposable, is not used to remove the runnable from the task queue when the outer disposable is disposed.

Provide a Reproducer

The full reproducer is in this repository: https://github.com/lengfeld/android-app-kotlinx-coroutines-rx3-leak-reproducer/tree/main?tab=readme-ov-file

Thanks for looking into this.

[dkhalanskyjb]

Hi! Thanks, yes, there is indeed a leak. The root cause analysis is a bit off, but the problem is clear, here's the proposed fix: #4616

[lengfeld]

@dkhalanskyjb Thanks for the quick fix.