fix OperatorConcat race condition where request lost

[davidmoten]

If a request arrives while ConcatInnerSubscriber is being constructed (L199-201) then that additional request does not induce a request from the subscriber and could stall the stream. The fix is not to pass requested value in the constructor but to call ConcatInnerSubscriber.requestMore with requested only once currentSubscriber is set.

[akarnokd]

This doesn't really help much (the constructor version is behaviorally equivalent) because there is still a small window between requesting the requested and setting the current. Swapping the requestMore and set doesn't help either because then a concurrent get will request r0 + 2n: r0 + n because requested is updated and n again because the reference is non-null. You need my ProducerArbiter, however, If you can wait a few hours, I'll post a PR with all my useful producers put into rx.internal.producers since making them 'official' takes several days of back-and-forth.

[benjchristensen]

Sounds like you want to fix this issue using #2963 ?

[davidmoten]

Yeah my focus was very narrow. Similar situation to retry so 'ProducerArbiter` will be the go. Those producers sound great, thanks.

[akarnokd]

The new producers have been merged into 1.x.

[akarnokd]

Could you also see if #2876 passes as well?

[davidmoten]

Ok, I've introduced the joys of ProducerArbiter to OperatorConcat. Tests pass and I've added the test from #2876 which doesn't fail after running for tens of minutes. There is one catch though, I changed the #2876 test to use Observable.from(Arrays.asList(t)) instead of Observable.just(t) so would be backpressure aware. Using ProducerArbiter means that sources that emit more than requested like Observable.just may induce an IllegalStateException from ProducerArbiter.

I'm supposing at some stage in the future (2.0?) we will be strict about our sources but till then perhaps we should allow a ProducerArbiter to be created in non-strict mode so that if a source emits more than requested it is forgiving about it?

[akarnokd]

Could you change this to 2 seconds?

[akarnokd]

No, because it would derail the state validation.

[davidmoten]

Changed the test duration to 2 seconds and added error message to ProducerArbiter so users will have a bit more of an idea what went wrong if the produced-more-than-requested IllegalStateException occcurs.

[davidmoten]

Unrelated CI failure, issue #2870 covers it.

rx.internal.operators.OperatorSerializeTest > testMultiThreadedWithNPEinMiddle FAILED
    java.lang.AssertionError
        at org.junit.Assert.fail(Assert.java:92)
        at org.junit.Assert.assertTrue(Assert.java:43)
        at org.junit.Assert.assertTrue(Assert.java:54)
        at rx.internal.operators.OperatorSerializeTest.testMultiThreadedWithNPEinMiddle(OperatorSerializeTest.java:134)

[akarnokd]

Maybe it is time to make just() do backpressure properly.

Would you like to pursue a fix for testMultiThreadedWithNPEinMiddle? I think if the exception itself is created way upfront and not new'd in place, the chance of it delayed quite to the end should decrease significantly.

[davidmoten]

Sure, I can look at it.

By the way I've removed unused imports from OperatorConcatTest and also changed the onStart so that it only requests as many inner observables as it needs. Do you think this is ok?

[davidmoten]

I'd love to see just support backpressure though I assume we will take a biggish performance hit if we do.

[akarnokd]

I think requesting 2 was no accident: while working on the current, the next can be 'created' asynchronously. Otherwise, once the current is processed, you need to wait more for the next Observable source to arrive.

[davidmoten]

The only catch is that by requesting 2 we make the assumption that creating this spare in advance doesn't require significant resources. If an unsubscribe occurs while the first inner is emitting then the act of creating the second is wasted (and may have been demanding).

[davidmoten]

There's a compromise either way for sure. Perhaps we could request 2 inners if Long.MAX_VALUE is requested and otherwise 1.

[akarnokd]

I'd say, let's have it 2 on default and have another PR which offers an overload to concat() and concatMap() where the 'prefetch' amount can be specified.

[davidmoten]

Sounds good. We can discuss further in another PR.

[akarnokd]

Looks okay to me now. Thanks!