Skip to content

Commit

Permalink
Remove Request Batching in Merge
Browse files Browse the repository at this point in the history 
Removing the batching until we can find a correct way to do it.

The performance impact of this change is seen here:

Benchmark                                          (size)   Mode   Samples          1.x    No Request Batching
r.o.OperatorMergePerf.merge1SyncStreamOfN               1  thrpt         5  4585554.607    4666745.314 102%
r.o.OperatorMergePerf.merge1SyncStreamOfN            1000  thrpt         5    51273.033      39922.246 78%
r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5       47.515         37.634 79%
r.o.OperatorMergePerf.mergeNAsyncStreamsOfN             1  thrpt         5    90901.735      93454.726 103%
r.o.OperatorMergePerf.mergeNAsyncStreamsOfN          1000  thrpt         5        5.407          4.910 91%
r.o.OperatorMergePerf.mergeNSyncStreamsOf1              1  thrpt         5  4181618.767    4173322.551 100%
r.o.OperatorMergePerf.mergeNSyncStreamsOf1            100  thrpt         5   422193.599     408972.130 97%
r.o.OperatorMergePerf.mergeNSyncStreamsOf1           1000  thrpt         5    36886.812      36448.978 99%
r.o.OperatorMergePerf.mergeNSyncStreamsOfN              1  thrpt         5  4815945.720    4887943.643 101%
r.o.OperatorMergePerf.mergeNSyncStreamsOfN           1000  thrpt         5       43.926         39.027 89%
r.o.OperatorMergePerf.mergeTwoAsyncStreamsOfN           1  thrpt         5    72578.046      70412.656 97%
r.o.OperatorMergePerf.mergeTwoAsyncStreamsOfN        1000  thrpt         5     3260.024       3064.403 94%
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1         1  thrpt         5  4678858.201    4808504.588 103%
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1      1000  thrpt         5    34407.547      36364.476 106%
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1   1000000  thrpt         5       31.312         32.261 103%
  • Loading branch information
@benjchristensen
benjchristensen committed Dec 12, 2014
1 parent dd73c15 commit 162c31c
Show file tree
Hide file tree
Showing 2 changed files with 97 additions and 40 deletions.
45 changes: 6 additions & 39 deletions src/main/java/rx/internal/operators/OperatorMerge.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -186,8 +186,9 @@ private void handleNewSource(Observable<? extends T> t) {
InnerSubscriber<T> i = new InnerSubscriber<T>(this, producerIfNeeded);
i.sindex = childrenSubscribers.add(i);
t.unsafeSubscribe(i);
if (!isUnsubscribed())
if (!isUnsubscribed()) {
request(1);
}
}

private void handleScalarSynchronousObservable(ScalarSynchronousObservable<? extends T> t) {
Expand Down Expand Up @@ -382,19 +383,8 @@ private int drainScalarValueQueue() {
public Boolean call(InnerSubscriber<T> s) {
if (s.q != null) {
long r = mergeProducer.requested;
int emitted = 0;
emitted += s.drainQueue();
int emitted = s.drainQueue();
if (emitted > 0) {
/*
* `s.emitted` is not volatile (because of performance impact of making it so shown by JMH tests)
* but `emitted` can ONLY be touched by the thread holding the `emitLock` which we're currently inside.
*
* Entering and leaving the emitLock flushes all values so this is visible to us.
*/
emitted += s.emitted;
// TODO we may want to store this in s.emitted and only request if above batch
// reset this since we have requested them all
s.emitted = 0;
s.requestMore(emitted);
}
if (emitted == r) {
Expand Down Expand Up @@ -542,9 +532,6 @@ private static final class InnerSubscriber<T> extends Subscriber<T> {
static final AtomicIntegerFieldUpdater<InnerSubscriber> ONCE_TERMINATED = AtomicIntegerFieldUpdater.newUpdater(InnerSubscriber.class, "terminated");

private final RxRingBuffer q = RxRingBuffer.getSpmcInstance();
/* protected by emitLock */
int emitted = 0;
final int THRESHOLD = (int) (q.capacity() * 0.7);

public InnerSubscriber(MergeSubscriber<T> parent, MergeProducer<T> producer) {
this.parentSubscriber = parent;
Expand Down Expand Up @@ -618,6 +605,7 @@ private void emit(T t, boolean complete) {
* putting in the queue, it attempts to get the lock. We are optimizing for the non-contended case.
*/
if (parentSubscriber.getEmitLock()) {
long emitted = 0;
enqueue = false;
try {
// drain the queue if there is anything in it before emitting the current value
Expand Down Expand Up @@ -660,30 +648,9 @@ private void emit(T t, boolean complete) {
} finally {
drain = parentSubscriber.releaseEmitLock();
}
if (emitted > THRESHOLD) {
// this is for batching requests when we're in a use case that isn't queueing, always fast-pathing the onNext
/**
* <pre> {@code
* Without this batching:
*
* Benchmark (size) Mode Samples Score Score error Units
* r.o.OperatorMergePerf.merge1SyncStreamOfN 1 thrpt 5 5060743.715 100445.513 ops/s
* r.o.OperatorMergePerf.merge1SyncStreamOfN 1000 thrpt 5 36606.582 1610.582 ops/s
* r.o.OperatorMergePerf.merge1SyncStreamOfN 1000000 thrpt 5 38.476 0.973 ops/s
*
* With this batching:
*
* Benchmark (size) Mode Samples Score Score error Units
* r.o.OperatorMergePerf.merge1SyncStreamOfN 1 thrpt 5 5367945.738 262740.137 ops/s
* r.o.OperatorMergePerf.merge1SyncStreamOfN 1000 thrpt 5 62703.930 8496.036 ops/s
* r.o.OperatorMergePerf.merge1SyncStreamOfN 1000000 thrpt 5 72.711 3.746 ops/s
*} </pre>
*/
// request upstream what we just emitted
if(emitted > 0) {
request(emitted);
// we are modifying this outside of the emit lock ... but this can be considered a "lazySet"
// and it will be flushed before anything else touches it because the emitLock will be obtained
// before any other usage of it
emitted = 0;
}
}
if (enqueue) {
Expand Down
92 changes: 91 additions & 1 deletion src/test/java/rx/internal/operators/OperatorMergeTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
Expand All @@ -40,9 +41,14 @@
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;

import rx.*;
import rx.Notification;
import rx.Observable;
import rx.Observable.OnSubscribe;
import rx.Observer;
import rx.Scheduler;
import rx.Scheduler.Worker;
import rx.Subscriber;
import rx.Subscription;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func1;
Expand Down Expand Up @@ -1105,4 +1111,88 @@ public void shouldNotReceivedDelayedErrorWhileThereAreStillNormalEmissionsInTheQ
subscriber.assertReceivedOnNext(asList(1, 2, 3, 4));
assertEquals(asList(exception), subscriber.getOnErrorEvents());
}

@Test
public void testMergeKeepsRequesting() throws InterruptedException {
//for (int i = 0; i < 5000; i++) {
//System.out.println(i + ".......................................................................");
final CountDownLatch latch = new CountDownLatch(1);
final ConcurrentLinkedQueue<String> messages = new ConcurrentLinkedQueue<String>();

Observable.range(1, 2)
// produce many integers per second
.flatMap(new Func1<Integer, Observable<Integer>>() {
@Override
public Observable<Integer> call(final Integer number) {
return Observable.range(1, Integer.MAX_VALUE)
.doOnRequest(new Action1<Long>() {

@Override
public void call(Long n) {
messages.add(">>>>>>>> A requested[" + number + "]: " + n);
}

})
// pause a bit
.doOnNext(pauseForMs(3))
// buffer on backpressure
.onBackpressureBuffer()
// do in parallel
.subscribeOn(Schedulers.computation())
.doOnRequest(new Action1<Long>() {

@Override
public void call(Long n) {
messages.add(">>>>>>>> B requested[" + number + "]: " + n);
}

});
}

})
// take a number bigger than 2* RxRingBuffer.SIZE (used by OperatorMerge)
.take(RxRingBuffer.SIZE * 2 + 1)
// log count
.doOnNext(printCount())
// release latch
.doOnCompleted(new Action0() {
@Override
public void call() {
latch.countDown();
}
}).subscribe();
boolean a = latch.await(2, TimeUnit.SECONDS);
if (!a) {
for (String s : messages) {
System.out.println("DEBUG => " + s);
}
}
assertTrue(a);
//}
}

private static Action1<Integer> printCount() {
return new Action1<Integer>() {
long count;

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

private static Action1<Integer> pauseForMs(final long time) {
return new Action1<Integer>() {
@Override
public void call(Integer s) {
try {
Thread.sleep(time);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
};
}
}

0 comments on commit 162c31c

Please sign in to comment.