Optimisations with Observable.empty()

[davidmoten]

I notice that calling observable.concat(Observable.empty()) invokes back-pressure. I'd find it useful if when I used Observable.empty() that optimisations came into play such as ensuring that observable.concat(Observable.empty()) simply returned observable.

To achieve this I'd make a private constant EMPTY = from(new ArrayList()) in Observable so that calling Observable.empty() always returned this object cast into the appropriate generic type. Then I'd use a simple object reference equality test to determine if an optimisation could be made.

Do you think this proposal has legs? If so, I'll knock up a PR. I would seek to optimise more than just the concat operator , it would include merge and possibly others which I can think about if the time comes.

[davidmoten]

In this vein might be good to get from(List) return EMPTY whenever the list is empty. That would enable the optimisations more often.

[davidmoten]

This is inaccurate "I notice that calling observable.concat(Observable.empty()) invokes back-pressure". It was actually a combination of operators including concat. Regardless I think the optimisations would be useful.

[mttkay]

I stumbled upon your report while also looking at Observable.empty(). I wonder why it instantiates a new ArrayList every single time instead of using Collections.emptyList()? It's the canonical way to represent the list "null object" in Java and it's more memory efficient too, since it doesn't allocate an empty array of capacity 16 that will never end up having items.

[benjchristensen]

No idea why we aren't using Collections.emptyList() ... we should be. Just a simple oversight by my or someone else who has contributed.

[davidmoten]

Thanks @mttkay, that looks good. @benjchristensen what about more optimisations that would

• collapse the call stack
• avoid unnecessary invocations of backpressure slow path

For example, concat is now:

public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2) {
    return concat(just(t1, t2));
}

but could be:

public final static <T> Observable<T> concat(Observable<? extends T> t1, Observable<? extends T> t2) {
    if (t1 == EMPTY) 
        return t2;
    else if (t2 == EMPTY)
        return t1;
    else 
        return concat(just(t1, t2));
}

[zsxwing]

@davidmoten merge can benefit from it, too.

[zsxwing]

@davidmoten what about adding the empty check here, too: https://github.com/ReactiveX/RxJava/blob/1.x/src/main/java/rx/internal/operators/OperatorConcat.java#L126

        @Override
        public void onNext(Observable<? extends T> t) {
            if(t==EMPTY) return; // check EMPTY
            queue.add(nl.next(t));
            if (WIP_UPDATER.getAndIncrement(this) == 0) {
                subscribeNext();
            }
        }

[davidmoten]

there are a LOT of operators that can benefit from this. I would hazard a guess that most can use this optimisation at the high level within the Observable class and yes good point @zsxwing we could dig into operators for more optimisations. I'll proceed with the PR if @benjchristensen gives it the tick.

[benjchristensen]

If you want to submit a PR with this work done I'll accept it. If we can use Collections.emptyList() somewhere that we are currently allocating that will always be better.

[benjchristensen]

Put on 1.0.x as this is not a blocker for 1.0, but whenever you submit the PR I'll do it.

[davidmoten]

I'm starting work on this PR now and I'd like to get the ok on some detail.

I'd propose adding this to Observable just after the init of the hook:

private static final Observable<?> EMPTY = from(Collections.emptyList());

Observable.empty() would be changed to:

    @SuppressWarnings("unchecked")
    public final static <T> Observable<T> empty() {
        return (Observable<T>) EMPTY;
    }

As an example of how it would be used the static method Observable.merge would be changed to:

    @SuppressWarnings("unchecked")
    public final static <T> Observable<T> merge(Observable<? extends T> t1, Observable<? extends T> t2) {
        if (t1 == EMPTY) 
            return (Observable<T>) t2;
        else if (t2 == EMPTY)
            return (Observable<T>) t1;
        else 
            return merge(from(Arrays.asList(t1, t2)));
    }

Note that I've confirmed in the debugger that this optimisation avoids triggering backpressure if say o.mergeWith(Observable.empty()) is called. I'll get a unit test going for this as well.

A large number of methods will use this optimisation so I'd like to get my exact approach approved before submitting.

The final thing to note is that optimisations outside of the Observable class would do a reference equality check against Observable.empty(). So in the example from OperatorConcat by @zsxwing above:

        @Override
        public void onNext(Observable<? extends T> t) {
            if(t==Observable.empty()) return; // check EMPTY
            queue.add(nl.next(t));
            if (WIP_UPDATER.getAndIncrement(this) == 0) {
                subscribeNext();
            }
        }

I'd like to limit the scope of my PR to optimisations in Observable.java only. Optimisations in Operators would wait for this base PR to be accepted.

How does that sound?

[akarnokd]

Why do you have so many empty() to merge?

Edit: At least I looked at the implementation and improved it in #2622.