Optimisations with Observable.empty()

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.

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

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.

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.

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

@mttkay

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));
}

@davidmoten

@davidmoten merge can benefit from it, too.

@davidmoten

@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();
            }
        }

@zsxwing

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.

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.

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

@zsxwing

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?

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

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

Ta for improvement. The reason I have empty()s to merge is that this is a common pattern for me:

o.flatMap(t -> {
   if (cond(t)) 
       return someObservable(t);
  else 
       return Observable.empty(); 
  });

I'm experimenting now with substituting the special case where the flatMap function returns either 0 or 1 by doing

o.map(toOptional).filter(isPresent).map(toValue);

I'd hope it would be faster but not sure.

This PR is quite old. Can you create a reasonable benchmark to see how concatMap and flatMap behaves? In addition, I suggest we tackle this after both #2928 and #2960 are merged.

Yep I'll do benchmarks after those PRs have been merged.

This may be part of the operator-fusion optimizations in 2.x but I don't see this in 1.x due to non-technical reasons.

For a start: here is the baseline perf: #3754. I'm going to merge it so subsequent PRs can be benchmarked with it.

See #3759 for the optimized concats.

See #3761 for the optimized merges

Optimizations have been added and released with 1.1.4. Closing the issue. If you have more ideas where it's worth adding extra checks for empty, don't hesitate to open a new issue.

Navigation Menu

Repository navigation

Optimisations with Observable.empty() #1653

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Optimisations with Observable.empty() #1653

Description

Activity

davidmoten commented on Sep 1, 2014

davidmoten commented on Sep 1, 2014

mttkay commented on Sep 2, 2014

benjchristensen commented on Sep 2, 2014

davidmoten commented on Sep 2, 2014

zsxwing commented on Sep 3, 2014

zsxwing commented on Sep 3, 2014

davidmoten commented on Sep 3, 2014

benjchristensen commented on Oct 7, 2014

benjchristensen commented on Oct 7, 2014

davidmoten commented on Nov 14, 2014

akarnokd commented on Feb 5, 2015

davidmoten commented on Feb 5, 2015

akarnokd commented on May 20, 2015

davidmoten commented on May 21, 2015

akarnokd commented on Feb 9, 2016

akarnokd commented on Mar 13, 2016

akarnokd commented on Mar 14, 2016

akarnokd commented on Mar 14, 2016

akarnokd commented on May 5, 2016

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions