Operator Repeat and other operator fixes

[akarnokd]

Reimplemented repeat() which exposed some other problems.

• Added QueueDrain based on this.
• Added overload of repeat(long n) to repeat a source n times (which is not the same as repeat().take(n) generally). Note that Ben's prototype was not working correctly because it run the re-subscription code in a loop: if subscribeOn was used, it was constantly resubscribing before the old even finished.
• For some reason, observeOn+take didn't work as expected, causing repeat tests to fail as well. I couldn't determine the cause. I've noticed that take was unable to unsubscribe the upstream as there was nothing in its cs.
• To fix the problem above, I've reimplemented observeOn as well. It does not use recursive scheduling but the QueueDrain.
• This exposed another problem with GroupBy: in some cases, childObserver.onCompleted was sent out twice, breaking merge/flatMap. I've added a once check, but I'm not 100% certain there isn't anything else wrong with the GroupBy's staggered behavior.
• Unfortunately, OperationParallelMergeTest is broken with this PR. I don't quite understand why it doesn't work nor have a clue how to fix it or the new observeOn. Maybe the parallel tests relied on thread timing and not expiring too fast so they could continue one them.

Performance

from+repeat: 3.8MOps/s
from+repeat+observeOn: 1.5MOps/s
range: 31.8MOps/s
from+observeOn+repeat: 15 kOps/s

[cloudbees-pull-request-builder]

RxJava-pull-requests #733 FAILURE
Looks like there's a problem with this pull request

[benjchristensen]

I've noticed that take was unable to unsubscribe the upstream as there was nothing in its cs.

What is the unit test for this scenario?

[benjchristensen]

What do you think about this accepting Notification instead of Action0? The reason is that we now have 2 wrappers ... the Notification and Action0 around a type T and 2 object allocations for each onNext.

The use of Action0 is definitely more generic, but as we've seen by your CompositeSubscription changes, we're at the point where we're moving away from generic to achieve performance and memory gains, and this class will be involved in very high throughput scenarios.

[akarnokd]

Possible. I can't think of any scenario right now where Action0 is necessary. I'll update the code.

[benjchristensen]

Unfortunately, OperationParallelMergeTest is broken with this PR

It breaks because observeOn is now hopping threads as it always schedulers on the outer Scheduler.

The fix I pasted above resolves the OperationParallelMergeTest tests.

[benjchristensen]

I like how the QueueDrain class abstracts this away. The concern as stated in another comment is the double wrapping of Action0 and Notification we now have.

[akarnokd]

Will lock into the specialization.

[akarnokd]

The test case with take that caused problems was something like this:

Observable.range(1, 20000000).observeOn(Schedulers.newThread()).take(10).subscribe(System.out::println);

It prints 1..10, but then it doesn't stop and spins up a lot of threads. Basically, the OperationRepeatTest failed for me with timeouts.

[benjchristensen]

I merged repeat based on the discussion above. I'll await your work on observeOn.

[benjchristensen]

then it doesn't stop and spins up a lot of threads

Isn't that because we haven't re-implemented observeOn yet?

This works:

        Observable.range(1, 20000000).take(10).toBlockingObservable().forEach(new Action1<Integer>() {

            @Override
            public void call(Integer t1) {
                System.out.println(t1);
            }

        });

[benjchristensen]

Do you intend on submitting updated code soon, or should I merge the observeOn changes as discussed above?

[akarnokd]

I skip on this one.

[benjchristensen]

Okay, thanks @akarnokd

[benjchristensen]

The test case with take that caused problems was something like this:

Is the take issue resolved for you now based on #833 ?

[akarnokd]

The issue was with observeOn+take together. This fixes take, but observeOn in its current form doesn't work. This spins:

Observable.range(1, 20000000)
                .observeOn(Schedulers.newThread())
                .take(10).toBlockingObservable().forEach(System.out::println);

I guess I need to revisit the issue once ObserveOn has been updated.

[benjchristensen]

That code works on the new observeOn implementation I just submitted: #835

[akarnokd]

Excellent. I'm looking at the new observeOn right now for speedup options.

[benjchristensen]

Thanks @akarnokd I would appreciate that. My brain is done for the night!