1.x: concat reduce overhead when streaming a source

[akarnokd]

This PR reduces the request tracking overhead of concat by tracking the produced item count in a plain field and subtracting it from the arbiter and requested values only before the inner source completes. So instead of N decrementAndGet call, we have 1 addAndGet(-N) per source.

I've added a perf class to measure the difference.

(Intel Celeron 1005M @ 2GHz, Windows 7 x64, Java 8u66)

The throughput increased considerably, although I would have expected more, especially in the 1M case where the subscription overhead doesn't matter.

I'll do further investigation on it and post a follow-up PR if this gets merged in the meantime.

[artem-zinnatullin]

👍

[zsxwing]

Call getAndAddRequest here? In case requested is changed to Long.MAX_VALUE at the same time.

[akarnokd]

In normal use, such change never happens; consumers either start with MAX_VALUE or some prefetch value, plus you have to be very precise to hit the narrow window between the two lines.

[davidmoten]

I'd prefer to see correctness. Speculating on the probability of a race
condition seems like dangerous territory to be entering.

On Sun, 31 Jan 2016 21:24 David Karnok [email protected] wrote:

In src/main/java/rx/internal/operators/OperatorConcat.java
#3589 (comment):

@@ -179,6 +187,14 @@ void subscribeNext() {
}
}
}
+

•    void produced(long c) {
  

•        arbiter.produced(c);
  

•        AtomicLong requestedField = requested;
  

•        if (requestedField.get() != Long.MAX_VALUE) {
  

•            requestedField.addAndGet(-c);
  

In normal use, such change never happens; consumers either start with
MAX_VALUE or some prefetch value, plus you have to be very precise to hit
the narrow window between the two lines.

—
Reply to this email directly or view it on GitHub
https://github.com/ReactiveX/RxJava/pull/3589/files#r51358378.

[akarnokd]

This line is correct as it is, what we are talking about is considering the case where one wants to subvert the optimization by racing with get().

[davidmoten]

I've a very narrow focus on this method so I could well be missing something but normally once requested is set to Long.MAX_VALUE we don't reduce its value and @zsxwing has indicated this possibility. I don't think it should be best endeavours as in I would expect to see a CAS loop here. Is this a bit fussy? Do we do this in lots of places already? (I thought a lot had been cleaned up).

[akarnokd]

The keyword is loop which can lenghten due to failed CAS in bounded mode or safepoint kicking in. AddAndGet is a one step atomic operation.

[zsxwing]

So no other place will update requested at the same time. Right?

[akarnokd]

There is a non-zero chance there, but it is very small. If it happens, all the effect it will have is to make the operator decrement the requested amount from then on, which adds ~40 cycle overhead after each source completes.

Updated the PR anyway to do a CAS loop on production to remove all worries about the case.

[zsxwing]

👍

[akarnokd]

@zsxwing, @davidmoten are you satisfied with the name BackpressureUtils.produced?

[davidmoten]

This path involves two AtomicLong.get(). Can't the whole block just be replaced by
long previous = BackpressureUtils.getAndAddRequest(requestedField, n);?

[akarnokd]

No, the getAndAddRequest doesn't check for max-value and always does the CAS. I'd leave this as is and have a separate PR where the getAndAddRequest is changed and all use places are verified/fixed.

[stevegury]

Just a few comments, otherwise 👍.

[akarnokd]

Changed to AtomicBoolean and moved c != 0 into parent because the parent was used for completing the inner anyway, I overlooked it at first.

[stevegury]

👍