Operator Publish full rewrite

[akarnokd]

Complete rewrite of publish() and fixed bugs of #2810 and #2803.

[benjchristensen]

Wow. Will have to come back to this one :-) that's a pretty epic fix of my bad code!

[benjchristensen]

@davidmoten have you by chance tried this and been able to confirm it fixes your issues?

[akarnokd]

Ah, it was easy; you only need to dualize merge :).

[davidmoten]

@benjchristensen the original issue is covered by the rewrite, thanks @akarnokd . I've given the code a once over but not thorough, would take me a while and have little time at the moment.

[davidmoten]

@akarnokd I have had another brief foray into the code and I have every faith in your ability to create correct code but I'm not having a great time following it in the short slots of time I have between kid juggling. For longevity's sake I'm wondering if this beautiful looking bit of code could be bulked up with comments to lead me through and to lead future maintainers through. Of course if @benjchristensen or @zsxwing or another of the usual suspects think it's clear as day then further comments may be unnecessary.

[akarnokd]

Sure.

[akarnokd]

I've rebased, squashed and added comments to this PR.

[davidmoten]

In OperatorPublishTest.testPublish (old code, you made no changes), would be nice to have an assert between the calls to subscribe and the call to connect just to ensure that subscription has occured but nothing is emitted before the call to connect. This may be covered by other tests but not explicitly so I think could be worthwhile addition.

[akarnokd]

Thanks! Updated tests accordingly and fixed a case with clients starting with 0 requests.

[davidmoten]

Perhaps add a connect idempotency test?

[akarnokd]

Done.

[davidmoten]

This is one strategy I suppose and another might be request max but buffer responses for those subscribers that want less than max. I can imagine that either way somebody could suffer and I wonder if this strategy should be configurable?

[akarnokd]

You can only request the common minimum amount from upstream if the front-queue is bounded, even if some child subscribers request more. These may run ahead but then stay idle until the 'slower ones' catch up.

[davidmoten]

By the way thanks for the comments in the code, they are just what I need!

[akarnokd]

I'm preparing a JMH benchmark and it seems the current master fails with MissingBackpressureException on simple chains such as:

@Test
public void testObserveOn() {
    ConnectableObservable<Integer> co = Observable.range(0, 1000).publish();
    Observable<Integer> obs = co.observeOn(Schedulers.computation());
    for (int i = 0; i < 100; i++) {
        TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
        obs.subscribe(ts);
        Subscription s = co.connect();
        ts.awaitTerminalEvent(2, TimeUnit.SECONDS);
        ts.assertTerminalEvent();
        ts.assertNoErrors();
        assertEquals(1000, ts.getOnNextEvents().size());
        s.unsubscribe();
    }
}

I'll post and merge the PR for the perf test so it can be run independent of this PR.

[akarnokd]

I'll post its benchmark result but it takes a looong time. So far, I'm impressed (sync, 1 batch, 1 child, 1M elements):
master: 3.822 ops/s
this: 18.810 ops/s

[akarnokd]

Here is the benchmark result for this PR in JMH Comparison GUI format. This PR outperforms the current master by 1.5 - 4x most of the time. However, some async parameter combination timed out despite the high CPU usage; I'll look into those cases.

Edit:
The performance degradation appears to be due to the ReentrantLock unparking its successor in SubscriptionList, similar to #2857.

[benjchristensen]

By concurrent disconnect, I assume you mean concurrent subscribe/disconnect? If so then yes, the child subscribers may never receive anything if it is not re-connected. But that is exactly correct behavior if that's what happens.

Am I understanding your comment here correctly?

[akarnokd]

Yes, the behavior implemented and what you said is the correct one.

[benjchristensen]

Excellent code. Excellent comments. Very well done @akarnokd. This is a very hard piece of code. I can only imagine how much time you spent on this.

As best as I can tell it's all correct. It's complicated enough I may be missing something, but everything seems correct and matches what I wish I had coded this to be originally!