Clarity about batching request

benjchristensen · benjchristensen · commit 593185aee666 · 2014-07-07T22:36:17.000-07:00
diff --git a/rxjava-core/src/main/java/rx/internal/operators/OperatorMerge.java b/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;
@@ -553,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
