Android HandlerThreadScheduler fix (closes #1241)

[vinc3m1]

Some events are not delivered due to race condition around the check to innerSubscription.isUnsubscribed() in the posted runnable. It's currently possible for the Handler thread to be delayed enough that innerSubscription becomes unsubscribed, and causing the action not to run.

After cross-checking the code with NewThreadScheduler, it seems that the expected behavior is to continue to run the action even if the inner/parent subscription is unsubscribed, and only stop when the outer subscription is unsubscribed (which is already happening with handler.removeCallbacks(runnable)).

This fixes #1241.

[cloudbees-pull-request-builder]

RxJava-pull-requests #1143 SUCCESS
This pull request looks good

[benjchristensen]

Thanks @vinc3m1 for debugging and fixing this. It looks like a correct change to me and the comparison with NewThreadScheduler seems legit.

@zsxwing Since you've been involved with the Android code and schedulers, can you review this as well?

@akarnokd Since you are involved in a review of schedulers right now, can you also glance at this?

[zsxwing]

                    if (isUnsubscribed()) {
                         return;
                     }

I think the above codes is necessary, or worker.unsubscribe can not cancel the scheduled actions.

However, I think a better approach is

            if (innerSubscription.isUnsubscribed()) {
                // don't schedule, we are unsubscribed
                return Subscriptions.empty();
            }
            handler.postDelayed(runnable, unit.toMillis(delayTime));
            Subscription s = Subscriptions.create(new Action0() {

                @Override
                public void call() {
                    handler.removeCallbacks(runnable);
                }

            });
            innerSubscription.add(s);
            return s;

[vinc3m1]

+1 to what @zsxwing proposed. The unsubscribed check should happen before the schedule, but not inside the runnable. I can make the appropriate changes to the PR.

[cloudbees-pull-request-builder]

RxJava-pull-requests #1150 SUCCESS
This pull request looks good

[zsxwing]

Could you add the returned Subscription to innerSubscription? innerSubscription also needs to be a CompositeSubscription

[vinc3m1]

Hm, wouldn't that cause the original problem to occur again? If innerSubscription gets unsubscribed, it would cancel the runnable, effectively doing the same thing as before this change.

[zsxwing]

I think that's the expected behavior. If innerSubscription gets unsubscribed, all scheduled actions should be cancelled. You can take a look at NewThreadScheduler.NewThreadWorker. Could you test with NewThreadScheduler to see if there is same issue?

[vinc3m1]

I will try to run a test tomorrow but it's rather hard to reproduce as it requires the new thread to be slightly blocked. But I read the code from NewThreadExecutor and it doesn't cancel any scheduled tasks that I can tell. It does call executor.shutdown() but that still executes all previously submitted tasks.

[akarnokd]

NewThreadScheduler uses executor.shutdown() which allows already submitted tasks to run but prevents new tasks being scheduled. I think this is the wrong behavior there and I should have used executor.shutdownNow() instead; a fix is underway. Since the Handler scheduler can't be stopped and thus stopping all tasks, you need to keep track of the worker's submitted tasks.

[zsxwing]

Sorry, I should have typed EventLoopsScheduler.EventLoopWorker

[akarnokd]

I've looked at #1241 and it seems this is the artifact of the observeOn. What happens is that the onError(e) is scheduled to be executed but right after it, a SafeSubscriber is immediately unsubscribing the entire chain and thus cancelling the delivery of the scheduled error. By removing the isUnsubscribed check from the runnable, it is still likely removeCallbacks succeeds and the action is never run.

A possible fix would be for the observeOn to schedule the unsubscription of the inner worker instead of calling it immediately. This allows delivering all undelayed events to be run before the unsubscription. However, I don't see the full effect of this change right now.

[zsxwing]

What about if creating a new Subscription rather than chaining them here: https://github.com/Netflix/RxJava/blob/e6ae50b279b224d0560772e147e48397623b1693/rxjava-core/src/main/java/rx/operators/OperatorObserveOn.java#L68 ?

[akarnokd]

Here is the revised ObserveOn that should let undelayed actions before the unsubscribe run before letting the unsubscribe happen. I haven't done this in PR because of it would conflict with #1246.

[vinc3m1]

@akarnokd's solution looks like it should work. Should I close this PR for now, or wait until you open a new one after the conflict is resolved?

[akarnokd]

Since this PR just makes the issue happen less likely, I suppose this can be closed. Let's discuss further options in #1241