Merge pull request ReactiveX#1414 from benjchristensen/merge-fixes

Merge Fixes

Author: benjchristensen

rxjava-core/src/main/java/rx/internal/operators/OperatorMerge.java

@@ -449,6 +449,7 @@ private static final class InnerSubscriber<T> extends Subscriber<T> {
         private boolean mayNeedToDrain = false;
         /* 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;
@@ -535,7 +536,9 @@ private void emit(T t, boolean complete) {
                             emitted++;
                         }
                     } else {
-                        if (producer.requested > 0) {
+                        // this needs to check q.count() as draining above may not have drained the full queue
+                        // perf tests show this to be okay, though different queue implementations could perform poorly with this
+                        if (producer.requested > 0 && q.count() == 0) {
                             if (complete) {
                                 parentSubscriber.completeInner(this);
                             } else {
@@ -551,8 +554,25 @@ private void emit(T t, boolean complete) {
                 } finally {
                     drain = parentSubscriber.releaseEmitLock();
                 }
-                if (emitted > 256) {
+                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(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

rxjava-core/src/test/java/rx/BackpressureTests.java

@@ -180,7 +180,7 @@ public Observable<Integer> call(Integer i) {
     }
 
     @Ignore
-    @Test(timeout = 1000)
+    @Test(timeout = 2000)
     public void testZipSync() {
         int NUM = (int) ((int) RxRingBuffer.SIZE * 4.1);
         AtomicInteger c1 = new AtomicInteger();
@@ -208,7 +208,7 @@ public Integer call(Integer t1, Integer t2) {
     }
 
     @Ignore
-    @Test(timeout = 1000)
+    @Test(timeout = 2000)
     public void testZipAsync() {
         int NUM = (int) ((int) RxRingBuffer.SIZE * 2.1);
         AtomicInteger c1 = new AtomicInteger();
@@ -337,7 +337,7 @@ public void onNext(Integer t) {
         assertEquals(20, batches.get());
     }
 
-    @Test(timeout = 1000)
+    @Test(timeout = 2000)
     public void testUserSubscriberUsingRequestAsync() throws InterruptedException {
         AtomicInteger c = new AtomicInteger();
         final AtomicInteger totalReceived = new AtomicInteger();
@@ -380,7 +380,7 @@ public void onNext(Integer t) {
         assertEquals(20, batches.get());
     }
 
-    @Test(timeout = 1000)
+    @Test(timeout = 2000)
     public void testFirehoseFailsAsExpected() {
         AtomicInteger c = new AtomicInteger();
         TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
@@ -391,7 +391,7 @@ public void testFirehoseFailsAsExpected() {
         assertTrue(ts.getOnErrorEvents().get(0) instanceof MissingBackpressureException);
     }
 
-    @Test(timeout = 1000)
+    @Test(timeout = 2000)
     public void testOnBackpressureDrop() {
         int NUM = (int) ((int) RxRingBuffer.SIZE * 1.1); // > 1 so that take doesn't prevent buffer overflow
         AtomicInteger c = new AtomicInteger();
@@ -405,7 +405,7 @@ public void testOnBackpressureDrop() {
         assertTrue(NUM < ts.getOnNextEvents().get(NUM - 1).intValue());
     }
 
-    @Test(timeout = 1000)
+    @Test(timeout = 2000)
     public void testOnBackpressureBuffer() {
         int NUM = (int) ((int) RxRingBuffer.SIZE * 1.1); // > 1 so that take doesn't prevent buffer overflow
         AtomicInteger c = new AtomicInteger();

rxjava-core/src/test/java/rx/internal/operators/OperatorMergeTest.java

@@ -669,7 +669,7 @@ public void onNext(Integer t) {
                 if (t < 100)
                     try {
                         // force a slow consumer
-                        Thread.sleep(2);
+                        Thread.sleep(1);
                     } catch (InterruptedException e) {
                         e.printStackTrace();
                     }