[Android] Problems with HandlerThreadScheduler since 0.19.1

[mttkay]

We've put our app in beta recently, running RxJava 0.19.1. We're suddenly seeing problems with scheduling on the main thread, where notifications arrive even after unsubscribing from the observable. I suspect RxJava itself might be the cause here since the scheduler implementation changed in 0.19. I wanted to double check if anyone else is seeing this happen?

More specifically, what happens is that in a subscriber from which we unsubscribe in Fragment#onDestroyView, getView() returns null, crashing the app. This should never happen when unsubscribing before the views get destroyed, unless messages for some reason keep propagating to the subscriber. There had even been a fix committed recently that eagerly removes all pending Runnables from the handler's message looper upon unsubscription, so I wonder how this can even happen.

Unfortunately, these things are incredibly difficult to debug as they are usually timing issues. Some code extracts:

public class PlaylistTagsFragment extends Fragment ... {
    @Override
    public void onViewCreated(View view, Bundle savedInstanceState) {
        super.onViewCreated(view, savedInstanceState);
        ...

        viewSubscriptions = new CompositeSubscription();
        viewSubscriptions.add(allTagsObservable.subscribe(new TagsSubscriber()));
        viewSubscriptions.add(recentTagsObservable.subscribe(new RecentsSubscriber()));
    }

    @Override
    public void onDestroyView() {
        viewSubscriptions.unsubscribe();
        super.onDestroyView();
    }

    private final class RecentsSubscriber extends DefaultSubscriber<PlaylistTagsCollection> {

        @Override
        public void onNext(PlaylistTagsCollection tags) {
            if (!tags.getCollection().isEmpty()) {
                // crashes here
                getView().findViewById(R.id.recent_tags_container).setVisibility(VISIBLE);
                displayRecentTags(tags);
            }
        }
    }

}

[mttkay]

So, one thing I discovered is that in this particular case, the action we execute is actually quite fast for small amounts of data and we forgot to actually schedule it on a background thread (so we didn't notice it was running on the main UI thread). i.e. we didn't subscribeOn anything when firing the observable, but we did observeOn the main thread.

I wonder if there's a timing issue or other glitch when observing something on the same thread you're already on? I can't reproduce it locally, unfortunately.

[dpsm]

@mttkay I saw your description above which made me curious. After looking into OperatorObserveOn and It's interaction with HandlerThreadScheduler here is what I found:

1. Both ObserveOnSubscriber.onNext(..) and ScheduledUnsubscribe.unsubscribe() end up posting a Runnable in the HandlerThreadScheduler's Handler.

So with the code below:

final Subscription subscription = Observable.from(new Object())
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1() {
@OverRide
public void call(final Object o) {
Log.d("HERE", ":(");
}
});
subscription.unsubscribe();

If ObserveOnSubscriber.onNext(..) is called and an InnerHandlerThreadScheduler.schedule(..) call puts a ScheduledAction in the Loopers queue, when you call subscription.unsubscribe(); it will also queue another Runnable behind the one which will call onNext(..) on your subscriber.

So you end up with something like this in the Looper's MessageQueue
HEAD <== ObserveOnSubscriber.pollQueue(..) <== InnerHandlerThreadScheduler.unsubscribe() <== TAIL

Note that that happens since even though you are calling unsubscribe from the main thread, ScheduledUnsubscribe.unsubscribe() queues the subscribe() for the worker; see below:

@OverRide
public void unsubscribe() {
if (ONCE_UPDATER.getAndSet(this, 1) == 0) {
worker.schedule(new Action0() {
@OverRide
public void call() {
worker.unsubscribe();
}
});
}
}

So when your call to unsubscribe() returns within onDestroyView() all pending Runnable instances in the queue will still be executed. It seems to me that within OperatorObserveOn.pollQueue() there should be a check for scheduledUnsubscribe.isUnsubscribed()?

I am going to send a pull request with the fix for it today. Let me know your thoughts!?

[dpsm]

@mttkay there is a fix at #1409

[akarnokd]

The problem is that an unsubscribe() call from downstream is not distinguished from upstream. The former should unsubscribe the worker and drop every queued event whereas an upstream unsubscribe - usually called due to an onError case - should let queued events including the final onError to be delivered before the worker is shut down. The problem is in L75: don't share the child subscriber's internal composite, but rather use child.add(new ObserveOnSubscriber(...)) near L56.

[mttkay]

@dpsm good find! I wasn't aware of OperatorObserveOn scheduling unsubscribe. This is highly problematic actually, since code like this will not work as expected:

void onDestroyView() {
  subscription.unsubscribe(); // is supposed to detach listeners from views
  super.onDestroyView();
}

since the call to super, detaching all the views, will now overtake the scheduled unsubscription. This is counter intuitive, as you would think that unsubscribing is synchronous and immediately makes an attempt at stopping the observable and detaching subscribers.

This also means that removing posted runnables in the subscription block is less effective than it could be.

@akarnokd what do you mean with up- and downstream?

[dpsm]

@mttkay we have a fix to be merged soon that should implement the correct behaviour. See my comment above referring to the fix.

[dpsm]

@benjchristensen the fix for it is merged. @mttkay please verify with version 0.19.4

[mttkay]

Thanks so much guys, I will see to release a new beta running on 0.19.4
ASAP. It usually takes a few days for crash reports to come in, but I'll
report back in case the problem should persist.

[benjchristensen]

Closing out. Please reopen or start a new issue if the fix did not work.