From 3b932325391936309506db85aa08f4fdbbf9598d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?D=C3=A1vid=20Karnok?= Date: Wed, 13 Jan 2016 23:47:58 +0100 Subject: [PATCH] 1.x: overhead reduction in range() and merge() operators --- .../internal/operators/OnSubscribeRange.java | 110 +++++++++--------- .../rx/internal/operators/OperatorMerge.java | 22 +++- 2 files changed, 76 insertions(+), 56 deletions(-) diff --git a/src/main/java/rx/internal/operators/OnSubscribeRange.java b/src/main/java/rx/internal/operators/OnSubscribeRange.java index c7631b2cb9..8f17303a2d 100644 --- a/src/main/java/rx/internal/operators/OnSubscribeRange.java +++ b/src/main/java/rx/internal/operators/OnSubscribeRange.java @@ -25,108 +25,110 @@ */ public final class OnSubscribeRange implements OnSubscribe { - private final int start; - private final int end; + private final int startIndex; + private final int endIndex; public OnSubscribeRange(int start, int end) { - this.start = start; - this.end = end; + this.startIndex = start; + this.endIndex = end; } @Override - public void call(final Subscriber o) { - o.setProducer(new RangeProducer(o, start, end)); + public void call(final Subscriber childSubscriber) { + childSubscriber.setProducer(new RangeProducer(childSubscriber, startIndex, endIndex)); } private static final class RangeProducer extends AtomicLong implements Producer { /** */ private static final long serialVersionUID = 4114392207069098388L; - private final Subscriber o; - private final int end; - private long index; + private final Subscriber childSubscriber; + private final int endOfRange; + private long currentIndex; - RangeProducer(Subscriber o, int start, int end) { - this.o = o; - this.index = start; - this.end = end; + RangeProducer(Subscriber childSubscriber, int startIndex, int endIndex) { + this.childSubscriber = childSubscriber; + this.currentIndex = startIndex; + this.endOfRange = endIndex; } @Override - public void request(long n) { + public void request(long requestedAmount) { if (get() == Long.MAX_VALUE) { // already started with fast-path return; } - if (n == Long.MAX_VALUE && compareAndSet(0L, Long.MAX_VALUE)) { + if (requestedAmount == Long.MAX_VALUE && compareAndSet(0L, Long.MAX_VALUE)) { // fast-path without backpressure fastpath(); - } else if (n > 0L) { - long c = BackpressureUtils.getAndAddRequest(this, n); + } else if (requestedAmount > 0L) { + long c = BackpressureUtils.getAndAddRequest(this, requestedAmount); if (c == 0L) { // backpressure is requested - slowpath(n); + slowpath(requestedAmount); } } } /** - * + * Emits as many values as requested or remaining from the range, whichever is smaller. */ - void slowpath(long r) { - long idx = index; - while (true) { - /* - * This complicated logic is done to avoid touching the volatile `index` and `requested` values - * during the loop itself. If they are touched during the loop the performance is impacted significantly. - */ - long fs = end - idx + 1; - long e = Math.min(fs, r); - final boolean complete = fs <= r; - - fs = e + idx; - final Subscriber o = this.o; + void slowpath(long requestedAmount) { + long emitted = 0L; + long endIndex = endOfRange + 1L; + long index = currentIndex; + + final Subscriber childSubscriber = this.childSubscriber; + + for (;;) { - for (long i = idx; i != fs; i++) { - if (o.isUnsubscribed()) { + while (emitted != requestedAmount && index != endIndex) { + if (childSubscriber.isUnsubscribed()) { return; } - o.onNext((int) i); + + childSubscriber.onNext((int)index); + + index++; + emitted++; } - if (complete) { - if (o.isUnsubscribed()) { - return; - } - o.onCompleted(); + if (childSubscriber.isUnsubscribed()) { return; } - idx = fs; - index = fs; - - r = addAndGet(-e); - if (r == 0L) { - // we're done emitting the number requested so return + if (index == endIndex) { + childSubscriber.onCompleted(); return; } + + requestedAmount = get(); + + if (requestedAmount == emitted) { + currentIndex = index; + requestedAmount = addAndGet(-emitted); + if (requestedAmount == 0L) { + break; + } + emitted = 0L; + } } } /** - * + * Emits all remaining values without decrementing the requested amount. */ void fastpath() { - final long end = this.end + 1L; - final Subscriber o = this.o; - for (long i = index; i != end; i++) { - if (o.isUnsubscribed()) { + final long endIndex = this.endOfRange + 1L; + final Subscriber childSubscriber = this.childSubscriber; + for (long index = currentIndex; index != endIndex; index++) { + if (childSubscriber.isUnsubscribed()) { return; } - o.onNext((int) i); + childSubscriber.onNext((int) index); } - if (!o.isUnsubscribed()) { - o.onCompleted(); + if (!childSubscriber.isUnsubscribed()) { + childSubscriber.onCompleted(); } } } diff --git a/src/main/java/rx/internal/operators/OperatorMerge.java b/src/main/java/rx/internal/operators/OperatorMerge.java index 56a7058d26..bb68edcbfe 100644 --- a/src/main/java/rx/internal/operators/OperatorMerge.java +++ b/src/main/java/rx/internal/operators/OperatorMerge.java @@ -177,6 +177,10 @@ static final class MergeSubscriber extends Subscriber /** An empty array to avoid creating new empty arrays in removeInner. */ static final InnerSubscriber[] EMPTY = new InnerSubscriber[0]; + final int scalarEmissionLimit; + + int scalarEmissionCount; + public MergeSubscriber(Subscriber child, boolean delayErrors, int maxConcurrent) { this.child = child; this.delayErrors = delayErrors; @@ -184,7 +188,13 @@ public MergeSubscriber(Subscriber child, boolean delayErrors, int max this.nl = NotificationLite.instance(); this.innerGuard = new Object(); this.innerSubscribers = EMPTY; - request(maxConcurrent == Integer.MAX_VALUE ? Long.MAX_VALUE : maxConcurrent); + if (maxConcurrent == Integer.MAX_VALUE) { + scalarEmissionLimit = Integer.MAX_VALUE; + request(Long.MAX_VALUE); + } else { + scalarEmissionLimit = Math.max(1, maxConcurrent >> 1); + request(maxConcurrent); + } } Queue getOrCreateErrorQueue() { @@ -488,7 +498,15 @@ protected void emitScalar(T value, long r) { if (r != Long.MAX_VALUE) { producer.produced(1); } - this.requestMore(1); + + int produced = scalarEmissionCount + 1; + if (produced == scalarEmissionLimit) { + scalarEmissionCount = 0; + this.requestMore(produced); + } else { + scalarEmissionCount = produced; + } + // check if some state changed while emitting synchronized (this) { skipFinal = true;