(Updated): timeout() operator leaking ScheduledFutureTasks on Scheduler

[daschl]

Hi folks,

(update, see below, looks like it's the timeout operator)

I'm raising a ticket earlier than later, because I suspect we have a leak somewhere in the Scheduler logic. It could by in my code too, but from the characteristics it looks more like something rx internal.

I've been able to generate some OOM scenarios where GC takes all the time to clean things up and it's not able to, which pointed me to a leak. A heap dump showed the leak suspect is the ScheduledFutureTask as part of the computation scheduler:

You can see a full GC kicking in without actually making old gen smaller:

The code path uses both .observeOn() once to move it onto a scheduler and also blocks .single() at the very end. So I'll start looking in those places. But maybe @akarnokd or @benjchristensen you have an idea where I can start debugging?

[daschl]

One thing that is suspicious is that my rx computation pools should be very silent, because I'm using observeOn my own scheduler (a spinoff of the computation one because it does not allow thread naming manually).

So while they are quiet (see threads) they have lots in their work queue.

[daschl]

@akarnokd any idea what a ScheduledAction "null" is?

Looks like it's carrying a Timeout operation which makes sense to be done on the Rx one because I forgot to move it to our computation scheduler. Probably the leak is in the timeout operator?

Quick update: yup that looks like to be the cause, when I remove the timeout operator and do a manual full GC it goes back to steady, nothing leaking there.

[akarnokd]

ScheduledAction extends AtomicReference so unless it has run already, it contains a null value. I did this for the free atomic operator(s).

In addition, there is a shortcoming in Java 6 ScheduledThreadPool where cancelled far-future tasks remain in the pool's queue until their time has actually passed, at when they get just evicted. Java 7 introduced a setRemoveOnCancelPolicy() method to get rid of them ASAP (defaults to false for compatibility reasons). (Historical note: in my RxJava predecessor, I had this method called reflectively when a scheduled pool was instantiated.)

[daschl]

@akarnokd see my last update, it seems to be in timeout() .. when I remove it, no leak.

[daschl]

Here is with the timeout operator removed:

[benjchristensen]

Is the use case able to be represented in a simple unit test so we can reproduce and fix?

If not then is it a high volume of timeouts (that don't trigger) causing it?

[daschl]

I'm trying to get it into something testable, although since I'm not familiar with the low level internals I don't know how we can prove that it is not leaking. I'll try to get something that at least shows it in a profiler when running it.

Btw note the code that triggers it is very simple: https://github.com/couchbase/couchbase-java-client/blob/master/src/main/java/com/couchbase/client/java/CouchbaseBucket.java#L288

Could it also be interplay between timeout() and the blocking observable conversion that is not doing something that the timeout operator expects?

[daschl]

Okay with something simple like

while(true) {
   Observable
        .timer(5, TimeUnit.MICROSECONDS)
        .timeout(1, TimeUnit.SECONDS)
        .toBlocking()
        .single();
}

Or chaining in a different scheduler before the timeout its not reproducible, no leaks there. One difference as well is that the code is observing a AsyncSubject, I don't know if that makes a difference in this case though.

[daschl]

@akarnokd @benjchristensen Wow, actually it is reproducible with that code. It shows up when you increase the timeout a lot. Could it be that with the current code the task only vanishes once the timeout is over and the actual onComplete does not cancel it?

while(true) {
   Observable
        .timer(5, TimeUnit.MICROSECONDS)
        .timeout(1, TimeUnit.SECONDS)
        .toBlocking()
        .single();
}

while(true) {
   Observable
        .timer(5, TimeUnit.MICROSECONDS)
        .timeout(100, TimeUnit.SECONDS)
        .toBlocking()
        .single();
}

[akarnokd]

We've fixed the task retention problem for JDK 6 in 1.0.5. Does the problem still happen for you?