Netflix/RxJava — pull request #643 by akarnokd (2013-12-19 22:36:35+00:00)
Title: Fixed OutOfMemoryError with CPU scheduler in recursive mode.

Found an [issue in Rx.NET](https://rx.codeplex.com/discussions/431558) regarding an out-of-memory situation due to long chained subscriptions when using recursive scheduling.

The following test program crashes or just hangs indefinitely:

```java
public class RunRecursiveScheduling {
/**
* Generates an observable sequence by iterating a state from an initial
* state until the condition returns false.
*/
public static OnSubscribeFunc generate(
final TState initialState,
final Func1 condition,
final Func1 iterate,
final Func1 resultSelector,
final Scheduler scheduler) {
return new OnSubscribeFunc() {
@Override
public Subscription onSubscribe(final Observer observer) {
return scheduler.schedule(initialState, new Func2() {
@Override
public Subscription call(Scheduler s, TState state) {
boolean hasNext;
try {
hasNext = condition.call(state);
} catch (Throwable t) {
observer.onError(t);
return Subscriptions.empty();
}
if (hasNext) {
R result;
try {
result = resultSelector.call(state);
} catch (Throwable t) {
observer.onError(t);
return Subscriptions.empty();
}
observer.onNext(result);

TState nextState;
try {
nextState = iterate.call(state);
} catch (Throwable t) {
observer.onError(t);
return Subscriptions.empty();
}

return s.schedule(nextState, this);
}
observer.onCompleted();
return Subscriptions.empty();
}
});
}
};
}

public static void main(String[] args) throws Exception {
Observable source = Observable.create(generate(
0, a -> true, a -> a + 1, a -> a, Schedulers.threadPoolForComputation()));

CountDownLatch latch = new CountDownLatch(1);
Subscription s = source.subscribe(v -> {
if (v % 100_000 == 0) {
System.out.println(v);
}
if (v >= 10_000_000) {
latch.countDown();
}
},
e -> { e.printStackTrace(); latch.countDown(); },
latch::countDown);

latch.await();

System.out.println("Wait done.");

s.unsubscribe();

System.out.println("Unsubscribe done.");
}
}
```

The issue lies in the fact that CPU scheduler (and perhaps the others) are not really reentrant, therefore, they create new composite subscriptions whenever a recursive scheduling happens. So instead of showing a simple subscription to the outside world, a chain of subscription is extended on every recursive schedule call.

A working solution is to have a different scheduler shown to the ```Func2``` than the actual CPU scheduler. This new scheduler, called ```ReentrantScheduler```, maintains internal subscriptions which get replaced if a recursive scheduling is executed.

I found three issues with this new approach and the old tests:
- In ```SchedulersTest.testRecursiveScheduler2```, the logic relied on the expectation that the call on L338 is executed at least once if the outer subscription is unsubscribed. The new logic stops the schedule chain instantly, so it is very unlikely the call on L338 gets executed after this.
- In ```OperationObserveOn.Observation``` tries to solve the underlying issue as well with its own subscription replacer logic. I haven't changed the operator as it might not work with non CPU scheduler after that.
- The ```ReentrantSchehduler``` maintains two subscriptions. One for the result of the schedule calls, and one for the the ```DiscardableAction```s. The former subscriptions can be swapped out without unsubscribing the previous completed schedule, but I'm not sure about the ```DiscardableAction```s; if I unsubscribe them, tests hang because it basically cancels itself and no further task is executed. This might be a conceptual error in ```ReentrantScheduler```; perhaps it should not return the entire composite on each schedule call, but rather return the content of the ```actionSub``` only.