Fix the bug that cache doesn't unsubscribe the source Observable when th...

[zsxwing]

...e source is terminated

Fix #2191

[zsxwing]

non-determistic failures

rx.BackpressureTests > testOnBackpressureDrop FAILED
    java.lang.AssertionError
        at org.junit.Assert.fail(Assert.java:92)
        at org.junit.Assert.assertTrue(Assert.java:43)
        at org.junit.Assert.assertTrue(Assert.java:54)
        at rx.BackpressureTests.testOnBackpressureDrop(BackpressureTests.java:423)

rx.internal.operators.OperatorRetryTest > testRetryWithBackpressure FAILED
    java.lang.Exception: test timed out after 10000 milliseconds

[akarnokd]

Drop: does it fail locally if run in a loop? Based on the test, it is possible a bogged down system can make it fail because everyone is slow enough so the drop is never triggered.

Retry: how many seconds does this single test take on your computer?

[zsxwing]

Retry: how many seconds does this single test take on your computer?

It's fine in my computer. The error was reported by Travis CI.

[akarnokd]

It has 10 seconds timeout. If it takes 6 on your computer, then it is possible the Travis server is just slow this time.

[zsxwing]

Misread your comment. These two test cases are fine in my computer. I just copied the errors from Travis CI log.

[akarnokd]

I managed to run the original code in #2191 and putting the sleep after the thread.start will use only a handful of io() threads. It confirms my suspicion that the original code simply floods the system with threaded activities and does not give them enough time to finish.

If the upstream completes, it should let any downstream object go (i.e., a completed ReplaySubject will throw away all of its subscribers). I've changed the code to allocate a large object and a memory dump right after the thread has completed shows no object retention. I don't see any issues here so could you elaborate on this issue/change (or on #2455)?

[zsxwing]

If the upstream completes, it should let any downstream object go (i.e., a completed ReplaySubject will throw away all of its subscribers).

I'm confused. This is not a memory leak bug. Instead, it's a bug that OnSubscribeCache doesn't unsubscribe the source after onCompleted or onError. See the unit test testUnsubscribeSource in this PR.

Because unsubscribe won't be called, the following codes in CachedThreadScheduler won't be called:

        @Override
        public void unsubscribe() {
            if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
                // unsubscribe should be idempotent, so only do this once
                CachedWorkerPool.INSTANCE.release(threadWorker);
            }
            innerSubscription.unsubscribe();
        }

Then, the worker won't be put back into the expiringWorkerQueue. That's why Schedulers.io() will create so many threads in #2191.

[akarnokd]

Thanks, that makes sense now.