{"action":"created","comment":{"url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/comments/32750138","id":32750138,"diff_hunk":"@@ -71,715 +72,878 @@\n         }\n         return (OperatorMerge<T>)HolderNoDelay.INSTANCE;\n     }\n-    /*\n-     * benjchristensen => This class is complex and I'm not a fan of it despite writing it. I want to give some background\n-     * as to why for anyone who wants to try and help improve it.\n-     * \n-     * One of my first implementations that added backpressure support (Producer.request) was fairly elegant and used a simple\n-     * queue draining approach. It was simple to understand as all onNext were added to their queues, then a single winner\n-     * would drain the queues, similar to observeOn. It killed the Netflix API when I canaried it. There were two problems:\n-     * (1) performance and (2) object allocation overhead causing massive GC pressure. Remember that merge is one of the most\n-     * used operators (mostly due to flatmap) and is therefore critical to and a limiter of performance in any application.\n-     * \n-     * All subsequent work on this class and the various fast-paths and branches within it have been to achieve the needed functionality\n-     * while reducing or eliminating object allocation and keeping performance acceptable.\n-     * \n-     * This has meant adopting strategies such as:\n-     * \n-     * - ring buffers instead of growable queues\n-     * - object pooling\n-     * - skipping request logic when downstream does not need backpressure\n-     * - ScalarValueQueue for optimizing synchronous single-value Observables\n-     * - adopting data structures that use Unsafe (and gating them based on environment so non-Oracle JVMs still work)\n-     * \n-     * It has definitely increased the complexity and maintenance cost of this class, but the performance gains have been significant.\n-     * \n-     * The biggest cost of the increased complexity is concurrency bugs and reasoning through what's going on.\n-     * \n-     * I'd love to have contributions that improve this class, but keep in mind the performance and GC pressure.\n-     * The benchmarks I use are in the JMH OperatorMergePerf class. GC memory pressure is tested using Java Flight Recorder\n-     * to track object allocation.\n+    /**\n+     * Creates a new instance of the operator with the given delayError and maxConcurrency settings.\n+     * @param delayErrors\n+     * @param maxConcurrent the maximum number of concurrent subscriptions or Integer.MAX_VALUE for unlimited\n+     * @return\n      */\n-\n-    private OperatorMerge() {\n-        this.delayErrors = false;\n+    public static <T> OperatorMerge<T> instance(boolean delayErrors, int maxConcurrent) {\n+        if (maxConcurrent == Integer.MAX_VALUE) {\n+            return instance(delayErrors);\n+        }\n+        return new OperatorMerge<T>(delayErrors, maxConcurrent);\n     }\n \n-    private OperatorMerge(boolean delayErrors) {\n+    final boolean delayErrors;\n+    final int maxConcurrent;\n+\n+    private OperatorMerge(boolean delayErrors, int maxConcurrent) {\n         this.delayErrors = delayErrors;\n+        this.maxConcurrent = maxConcurrent;\n     }\n \n-    private final boolean delayErrors;\n-\n     @Override\n     public Subscriber<Observable<? extends T>> call(final Subscriber<? super T> child) {\n-        return new MergeSubscriber<T>(child, delayErrors);\n-\n+        MergeSubscriber<T> subscriber = new MergeSubscriber<T>(child, delayErrors, maxConcurrent);\n+        MergeProducer<T> producer = new MergeProducer<T>(subscriber);\n+        subscriber.producer = producer;\n+        producer.lazySet(0); // make sure subscriber.producer is visible\n+        \n+        child.add(subscriber);\n+        child.setProducer(producer);\n+        \n+        return subscriber;\n     }\n \n-    private static final class MergeSubscriber<T> extends Subscriber<Observable<? extends T>> {\n-        final NotificationLite<T> on = NotificationLite.instance();\n-        final Subscriber<? super T> actual;\n-        private final MergeProducer<T> mergeProducer;\n-        private int wip;\n-        private boolean completed;\n-        private final boolean delayErrors;\n-        private ConcurrentLinkedQueue<Throwable> exceptions;\n-\n-        private volatile SubscriptionIndexedRingBuffer<InnerSubscriber<T>> childrenSubscribers;\n-\n-        private volatile RxRingBuffer scalarValueQueue = null;\n-\n-        /* protected by lock on MergeSubscriber instance */\n-        private int missedEmitting = 0;\n-        private boolean emitLock = false;\n+    static final class MergeProducer<T> extends AtomicLong implements Producer {\n+        /** */\n+        private static final long serialVersionUID = -1214379189873595503L;\n \n-        /**\n-         * Using synchronized(this) for `emitLock` instead of ReentrantLock or AtomicInteger is faster when there is no contention.\n-         * \n-         * <pre> {@code\n-         * Using ReentrantLock:\n-         * r.o.OperatorMergePerf.merge1SyncStreamOfN           1000  thrpt         5    44185.294     1295.565    ops/s\n-         * \n-         * Using synchronized(this):\n-         * r.o.OperatorMergePerf.merge1SyncStreamOfN           1000  thrpt         5    79715.981     3704.486    ops/s\n-         * \n-         * Still slower though than allowing concurrency:\n-         * r.o.OperatorMergePerf.merge1SyncStreamOfN           1000  thrpt         5   149331.046     4851.290    ops/s\n-         * } </pre>\n-         */\n-\n-        public MergeSubscriber(Subscriber<? super T> actual, boolean delayErrors) {\n-            super(actual);\n-            this.actual = actual;\n-            this.mergeProducer = new MergeProducer<T>(this);\n-            this.delayErrors = delayErrors;\n-            // decoupled the subscription chain because we need to decouple and control backpressure\n-            actual.add(this);\n-            actual.setProducer(mergeProducer);\n-        }\n-\n-        @Override\n-        public void onStart() {\n-            // we request backpressure so we can handle long-running Observables that are enqueueing, such as flatMap use cases\n-            // we decouple the Producer chain while keeping the Subscription chain together (perf benefit) via super(actual)\n-            request(RxRingBuffer.SIZE);\n+        final MergeSubscriber<T> subscriber;\n+        /** Indicates the producer mode: -1 no request, 0 unbounded fast, 1 bounded. */\n+        volatile int mode;\n+        @SuppressWarnings(\"rawtypes\")\n+        static final AtomicIntegerFieldUpdater<MergeProducer> MODE\n+        = AtomicIntegerFieldUpdater.newUpdater(MergeProducer.class, \"mode\");\n+        \n+        public MergeProducer(MergeSubscriber<T> subscriber) {\n+            this.subscriber = subscriber;\n+            this.mode = -1;\n         }\n-\n-        /*\n-         * This is expected to be executed sequentially as per the Rx contract or it will not work.\n-         */\n+        \n         @Override\n-        public void onNext(Observable<? extends T> t) {\n-            if (t instanceof ScalarSynchronousObservable) {\n-                ScalarSynchronousObservable<? extends T> t2 = (ScalarSynchronousObservable<? extends T>)t;\n-                handleScalarSynchronousObservable(t2);\n-            } else {\n-                if (t == null || isUnsubscribed()) {\n+        public void request(long n) {\n+            if (n >= 0) {\n+                int m = mode;\n+                // if in unbounded mode, quit\n+                if (m == 0) {\n                     return;\n                 }\n-                synchronized (this) {\n-                    // synchronized here because `wip` can be concurrently changed by children Observables\n-                    wip++;\n+                // if bounded mode, update request count and emit\n+                if (m == 1) {\n+                    BackpressureUtils.getAndAddRequest(this, n);\n+                    subscriber.emit();\n+                    return;\n+                }\n+                // mode has not been set\n+                if (n == Long.MAX_VALUE) {\n+                    // try setting mode to unbounded\n+                    if (MODE.compareAndSet(this, -1, 0)) {\n+                        subscriber.requestAll();\n+                        return;\n+                    }\n+                } else {\n+                    if (MODE.compareAndSet(this, -1, 1)) {\n+                        BackpressureUtils.getAndAddRequest(this, n);\n+                        subscriber.requestDefault();\n+                        return;\n+                    }\n                 }\n-                handleNewSource(t);\n+                // mode has changed concurrently\n+                // recheck state\n+                m = mode;\n+                // if in unbounded mode, quit\n+                if (m == 0) {\n+                    return;\n+                } else\n+                // if bounded mode, update request count and emit\n+                if (m == 1) {\n+                    BackpressureUtils.getAndAddRequest(this, n);\n+                    subscriber.emit();\n+                    return;\n+                }\n+                throw new IllegalStateException(\"Shouldn't be in mode: \" + m);\n             }\n         }\n-\n-        private void handleNewSource(Observable<? extends T> t) {\n-            if (childrenSubscribers == null) {\n-                // lazily create this only if we receive Observables we need to subscribe to\n-                childrenSubscribers = new SubscriptionIndexedRingBuffer<InnerSubscriber<T>>();\n-                add(childrenSubscribers);\n-            }\n-            MergeProducer<T> producerIfNeeded = null;\n-            // if we have received a request then we need to respect it, otherwise we fast-path\n-            if (mergeProducer.requested != Long.MAX_VALUE) {\n-                /**\n-                 * <pre> {@code\n-                 * With this optimization:\n-                 * \n-                 * r.o.OperatorMergePerf.merge1SyncStreamOfN      1000  thrpt         5    57100.080     4686.331    ops/s\n-                 * r.o.OperatorMergePerf.merge1SyncStreamOfN   1000000  thrpt         5       60.875        1.622    ops/s\n-                 *  \n-                 * Without this optimization:\n-                 * \n-                 * r.o.OperatorMergePerf.merge1SyncStreamOfN      1000  thrpt         5    29863.945     1858.002    ops/s\n-                 * r.o.OperatorMergePerf.merge1SyncStreamOfN   1000000  thrpt         5       30.516        1.087    ops/s\n-                 * } </pre>\n-                 */\n-                producerIfNeeded = mergeProducer;\n-            }\n-            InnerSubscriber<T> i = new InnerSubscriber<T>(this, producerIfNeeded);\n-            i.sindex = childrenSubscribers.add(i);\n-            t.unsafeSubscribe(i);\n-            if (!isUnsubscribed()) {\n-                request(1);\n-            }\n+        public long produced(int n) {\n+            return addAndGet(-n);\n         }\n+    }\n+    \n+    /**\n+     * The subscriber that observes Observables. \n+     * @param <T> the value type\n+     */\n+    static final class MergeSubscriber<T> extends Subscriber<Observable<? extends T>> {\n+        final Subscriber<? super T> child;\n+        final boolean delayErrors;\n+        final int maxConcurrent;\n+        \n+        MergeProducer<T> producer;\n+        \n+        /** Upstream might not produce scalar values at all so avoid creating a queue for it upfront. */\n+        volatile RxRingBuffer scalarQueue;\n+        /** Tracks the active subscriptions to sources. */\n+        volatile CompositeSubscription subscriptions;\n+        /** Due to the emission loop, we need to store errors somewhere if !delayErrors. */\n+        volatile ConcurrentLinkedQueue<Throwable> errors;\n+        \n+        final NotificationLite<T> nl;\n+        \n+        volatile boolean done;\n+        \n+        /** Guarded by this. */\n+        boolean emitting;\n+        /** Guarded by this. */\n+        boolean missed;\n+        \n+        final Object innerGuard;\n+        /** Copy-on-write array, guarded by innerGuard. */\n+        volatile InnerSubscriber<?>[] innerSubscribers;\n+        \n+        /** Used to generate unique InnerSubscriber IDs. Modified from onNext only. */\n+        long uniqueId;\n+        \n+        /** Which was the last InnerSubscriber that emitted? Accessed if emitting == true. */\n+        long lastId;\n+        /** What was its index in the innerSubscribers array? Accessed if emitting == true. */\n+        int lastIndex;\n+        \n+        /** An empty array to avoid creating new empty arrays in removeInner. */ \n+        static final InnerSubscriber<?>[] EMPTY = new InnerSubscriber<?>[0];\n \n-        private void handleScalarSynchronousObservable(ScalarSynchronousObservable<? extends T> t) {\n-            // fast-path for scalar, synchronous values such as Observable.from(int)\n-            /**\n-             * Without this optimization:\n-             * \n-             * <pre> {@code\n-             * Benchmark                                          (size)   Mode   Samples        Score  Score error    Units\n-             * r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1         1  thrpt         5  2,418,452.409   130572.665    ops/s\n-             * r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1      1000  thrpt         5     5,690.456       94.958    ops/s\n-             * r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5          takes too long\n-             * \n-             * With this optimization:\n-             * \n-             * r.o.OperatorMergePerf.merge1SyncStreamOfN               1  thrpt         5  5,475,300.198   156741.334    ops/s\n-             * r.o.OperatorMergePerf.merge1SyncStreamOfN            1000  thrpt         5    68,932.278     1311.023    ops/s\n-             * r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5       64.405        0.611    ops/s\n-             * } </pre>\n-             * \n-             */\n-            if (mergeProducer.requested == Long.MAX_VALUE) {\n-                handleScalarSynchronousObservableWithoutRequestLimits(t);\n-            } else {\n-                handleScalarSynchronousObservableWithRequestLimits(t);\n+        /** In fast mode, this counts the number of active subscriptions, including the one to upstream. */\n+        volatile int fastWip;\n+        @SuppressWarnings(\"rawtypes\")\n+        static final AtomicIntegerFieldUpdater<MergeSubscriber> FAST_WIP\n+        = AtomicIntegerFieldUpdater.newUpdater(MergeSubscriber.class, \"fastWip\");\n+        \n+        volatile boolean fastMode;\n+        \n+        /** Guarded by this. */\n+        Object[] fastArray;\n+        /** Guarded by this. */\n+        int fastSize;\n+        \n+        /** Fast, unbounded emission report contents of 'errors'. */\n+        static final Object REPORT_ERROR = new Object();\n+        /** Fast, unbounded emission complete. */\n+        static final Object REPORT_COMPLETED = new Object();\n+        \n+        public MergeSubscriber(Subscriber<? super T> child, boolean delayErrors, int maxConcurrent) {\n+            this.child = child;\n+            this.delayErrors = delayErrors;\n+            this.maxConcurrent = maxConcurrent;\n+            this.nl = NotificationLite.instance();\n+            this.innerGuard = new Object();\n+            this.innerSubscribers = EMPTY;\n+            this.request(0);\n+        }\n+        \n+        Queue<Throwable> getOrCreateErrorQueue() {\n+            ConcurrentLinkedQueue<Throwable> q = errors;\n+            if (q == null) {\n+                synchronized (this) {\n+                    q = errors;\n+                    if (q == null) {\n+                        q = new ConcurrentLinkedQueue<Throwable>();\n+                        errors = q;\n+                    }\n+                }\n             }\n+            return q;\n         }\n-\n-        private void handleScalarSynchronousObservableWithoutRequestLimits(ScalarSynchronousObservable<? extends T> t) {\n-            T value = t.get();\n-            if (getEmitLock()) {\n-                boolean moreToDrain;\n-                try {\n-                    actual.onNext(value);\n-                } finally {\n-                    moreToDrain = releaseEmitLock();\n+        CompositeSubscription getOrCreateComposite() {\n+            CompositeSubscription c = subscriptions;\n+            if (c == null) {\n+                boolean shouldAdd = false;\n+                synchronized (this) {\n+                    c = subscriptions;\n+                    if (c == null) {\n+                        c = new CompositeSubscription();\n+                        subscriptions = c;\n+                        shouldAdd = true;\n+                    }\n                 }\n-                if (moreToDrain) {\n-                    drainQueuesIfNeeded();\n+                if (shouldAdd) {\n+                    add(c);\n                 }\n-                request(1);\n+            }\n+            return c;\n+        }\n+        \n+        /** \n+         * In case the downstream requests less than Long.MAX_VALUE initially, do the standard\n+         * queued processing.\n+         */\n+        void requestDefault() {\n+            request(Math.min(maxConcurrent, RxRingBuffer.SIZE));\n+        }\n+        /**\n+         * In case the downstream requests Long.MAX_VALUE, perform a simplified merging without\n+         * using queues at all.\n+         */\n+        void requestAll() {\n+            FAST_WIP.lazySet(this, 1);\n+            fastMode = true;\n+            request(maxConcurrent == Integer.MAX_VALUE ? Long.MAX_VALUE : maxConcurrent);\n+        }\n+        \n+        @Override\n+        public void onNext(Observable<? extends T> t) {\n+            if (t == null) {\n                 return;\n+            }\n+            if (fastMode) {\n+                if (t.getClass() == ScalarSynchronousObservable.class) {\n+                    T v = ((ScalarSynchronousObservable<? extends T>)t).get();\n+                    fastEmitNext(v);\n+                    if (maxConcurrent != Integer.MAX_VALUE) {\n+                        request(1);\n+                    }\n+                } else {\n+                    FastInnerSubscriber<T> fastInner = new FastInnerSubscriber<T>(this);\n+                    getOrCreateComposite().add(fastInner);\n+                    FAST_WIP.getAndIncrement(this);\n+                    t.unsafeSubscribe(fastInner);\n+                }\n             } else {\n-                try {\n-                    getOrCreateScalarValueQueue().onNext(value);\n-                } catch (MissingBackpressureException e) {\n-                    onError(e);\n+                if (t.getClass() == ScalarSynchronousObservable.class) {\n+                    tryEmitScalar(((ScalarSynchronousObservable<? extends T>)t).get());\n+                } else {\n+                    InnerSubscriber<T> inner = new InnerSubscriber<T>(this, uniqueId++);\n+                    addInner(inner);\n+                    t.unsafeSubscribe(inner);\n+                    emit();\n                 }\n-                return;\n             }\n         }\n-\n-        private void handleScalarSynchronousObservableWithRequestLimits(ScalarSynchronousObservable<? extends T> t) {\n-            if (getEmitLock()) {\n-                boolean emitted = false;\n-                boolean moreToDrain;\n-                boolean isReturn = false;\n+        \n+        void innerError(Throwable e, FastInnerSubscriber<T> fastInner) {\n+            getOrCreateErrorQueue().offer(e);\n+            subscriptions.remove(fastInner);\n+            int n = FAST_WIP.decrementAndGet(this);\n+            // if we don't delay errors or this was the very last event\n+            if (!delayErrors || n == 0) {\n                 try {\n-                    long r = mergeProducer.requested;\n-                    if (r > 0) {\n-                        emitted = true;\n-                        actual.onNext(t.get());\n-                        MergeProducer.REQUESTED.decrementAndGet(mergeProducer);\n-                        // we handle this Observable without ever incrementing the wip or touching other machinery so just return here\n-                        isReturn = true;\n-                    }\n+                    fastEmitError();\n                 } finally {\n-                    moreToDrain = releaseEmitLock();\n-                }\n-                if (moreToDrain) {\n-                    drainQueuesIfNeeded();\n+                    unsubscribe();\n                 }\n-                if (emitted) {\n-                    request(1);\n-                }\n-                if (isReturn) {\n-                    return;\n-                }\n-            }\n-\n-            // if we didn't return above we need to enqueue\n-            // enqueue the values for later delivery\n-            try {\n-                getOrCreateScalarValueQueue().onNext(t.get());\n-            } catch (MissingBackpressureException e) {\n-                onError(e);\n+            } else {\n+                request(1);\n             }\n         }\n-\n-        private RxRingBuffer getOrCreateScalarValueQueue() {\n-            RxRingBuffer svq = scalarValueQueue;\n-            if (svq == null) {\n-                svq = RxRingBuffer.getSpscInstance();\n-                scalarValueQueue = svq;\n+        private void reportError() {\n+            List<Throwable> list = new ArrayList<Throwable>(errors);\n+            if (list.size() == 1) {\n+                child.onError(list.get(0));\n+            } else {\n+                child.onError(new CompositeException(list));\n             }\n-            return svq;\n         }\n-\n-        private synchronized boolean releaseEmitLock() {\n-            emitLock = false;\n-            if (missedEmitting == 0) {\n-                return false;\n+        \n+        \n+        void innerComplete(FastInnerSubscriber<T> fastInner) {\n+            subscriptions.remove(fastInner);\n+            if (FAST_WIP.decrementAndGet(this) == 0) {\n+                Queue<Throwable> e = errors;\n+                if (e == null || e.isEmpty()) {\n+                    fastEmitCompleted();\n+                } else {\n+                    fastEmitError();\n+                }\n             } else {\n-                return true;\n+                request(1);\n             }\n         }\n-\n-        private synchronized boolean getEmitLock() {\n-            if (emitLock) {\n-                missedEmitting++;\n-                return false;\n+        \n+        @Override\n+        public void onError(Throwable e) {\n+            getOrCreateErrorQueue().offer(e);\n+            if (fastMode) {\n+                int n = FAST_WIP.decrementAndGet(this);\n+                // if we don't delay errors or this was the very last event\n+                if (!delayErrors || n == 0) {\n+                    fastEmitError();\n+                }\n             } else {\n-                emitLock = true;\n-                missedEmitting = 0;\n-                return true;\n+                done = true;\n+                emit();\n             }\n         }\n-\n-        private boolean drainQueuesIfNeeded() {\n-            while (true) {\n-                if (getEmitLock()) {\n-                    int emitted = 0;\n-                    boolean moreToDrain;\n-                    try {\n-                        emitted = drainScalarValueQueue();\n-                        drainChildrenQueues();\n-                    } finally {\n-                        moreToDrain = releaseEmitLock();\n-                    }\n-                    // request outside of lock\n-                    if (emitted > 0) {\n-                        request(emitted);\n-                    }\n-                    if (!moreToDrain) {\n-                        return true;\n+        @Override\n+        public void onCompleted() {\n+            if (fastMode) {\n+                if (FAST_WIP.decrementAndGet(this) == 0) {\n+                    Queue<Throwable> e = errors;\n+                    if (e == null || e.isEmpty()) {\n+                        fastEmitCompleted();\n+                    } else {\n+                        reportError();\n                     }\n-                    // otherwise we'll loop and get whatever was added\n-                } else {\n-                    return false;\n                 }\n+            } else {\n+                done = true;\n+                emit();\n             }\n         }\n-\n-        int lastDrainedIndex = 0;\n-\n-        /**\n-         * ONLY call when holding the EmitLock.\n-         */\n-        private void drainChildrenQueues() {\n-            if (childrenSubscribers != null) {\n-                lastDrainedIndex = childrenSubscribers.forEach(DRAIN_ACTION, lastDrainedIndex);\n+        \n+        void addInner(InnerSubscriber<T> inner) {\n+            getOrCreateComposite().add(inner);\n+            synchronized (innerGuard) {\n+                InnerSubscriber<?>[] a = innerSubscribers;\n+                int n = a.length;\n+                InnerSubscriber<?>[] b = new InnerSubscriber<?>[n + 1];\n+                System.arraycopy(a, 0, b, 0, n);\n+                b[n] = inner;\n+                innerSubscribers = b;\n             }\n         }\n-\n+        void removeInner(InnerSubscriber<T> inner) {\n+            inner.queue.release();\n+            // subscription is non-null here because the very first addInner will create it before\n+            // this can be called\n+            subscriptions.remove(inner);\n+            synchronized (innerGuard) {\n+                InnerSubscriber<?>[] a = innerSubscribers;\n+                int n = a.length;\n+                int j = -1;\n+                // locate the inner\n+                for (int i = 0; i < n; i++) {\n+                    if (inner.equals(a[i])) {\n+                        j = i;\n+                        break;\n+                    }\n+                }\n+                if (j < 0) {\n+                    return;\n+                }\n+                if (n == 1) {\n+                    innerSubscribers = EMPTY;\n+                    return;\n+                }\n+                InnerSubscriber<?>[] b = new InnerSubscriber<?>[n - 1];\n+                System.arraycopy(a, 0, b, 0, j);\n+                System.arraycopy(a, j + 1, b, j, n - j - 1);\n+                innerSubscribers = b;\n+            }\n+        }\n+        \n         /**\n-         * ONLY call when holding the EmitLock.\n+         * Tries a fast-path emission of a scalar value or posts it to the scalar queue\n+         * for processing later.\n+         * @param value\n          */\n-        private int drainScalarValueQueue() {\n-            RxRingBuffer svq = scalarValueQueue;\n-            if (svq != null) {\n-                long r = mergeProducer.requested;\n-                int emittedWhileDraining = 0;\n-                if (r < 0) {\n-                    // drain it all\n-                    Object o = null;\n-                    while ((o = svq.poll()) != null) {\n-                        on.accept(actual, o);\n-                        emittedWhileDraining++;\n+        void tryEmitScalar(T value) {\n+            boolean success = false;\n+            synchronized (this) {\n+                // if nobody is emitting and child has available requests\n+                if (!emitting && producer.get() > 0) {\n+                    emitting = true;\n+                    success = true;\n+                }\n+            }\n+            if (success) {\n+                boolean skipFinal = false;\n+                try {\n+                    try {\n+                        child.onNext(value);\n+                    } catch (Throwable t) {\n+                        skipFinal = true;\n+                        Exceptions.throwIfFatal(t);\n+                        try {\n+                            child.onError(t);\n+                        } finally {\n+                            unsubscribe();\n+                        }\n+                        return;\n                     }\n-                } else if (r > 0) {\n-                    // drain what was requested\n-                    long toEmit = r;\n-                    for (int i = 0; i < toEmit; i++) {\n-                        Object o = svq.poll();\n-                        if (o == null) {\n-                            break;\n-                        } else {\n-                            on.accept(actual, o);\n-                            emittedWhileDraining++;\n+                    producer.produced(1);\n+                    request(1);\n+                    // check if some state changed while emitting\n+                    synchronized (this) {\n+                        skipFinal = true;\n+                        if (!missed) {\n+                            emitting = false;\n+                            return;\n                         }\n+                        missed = false;\n                     }\n-                    // decrement the number we emitted from outstanding requests\n-                    MergeProducer.REQUESTED.getAndAdd(mergeProducer, -emittedWhileDraining);\n-                }\n-                return emittedWhileDraining;\n-            }\n-            return 0;\n-        }\n-\n-        final Func1<InnerSubscriber<T>, Boolean> DRAIN_ACTION = new Func1<InnerSubscriber<T>, Boolean>() {\n-\n-            @Override\n-            public Boolean call(InnerSubscriber<T> s) {\n-                if (s.q != null) {\n-                    long r = mergeProducer.requested;\n-                    int emitted = s.drainQueue();\n-                    if (emitted > 0) {\n-                        s.requestMore(emitted);\n+                } finally {\n+                    if (!skipFinal) {\n+                        synchronized (this) {\n+                            emitting = false;\n+                        }\n                     }\n-                    if (emitted == r) {\n-                        // we emitted as many as were requested so stop the forEach loop\n-                        return Boolean.FALSE;\n+                }\n+                // in case there was some activity, continue with the regular emission loop\n+                emitLoop();\n+            } else {\n+                // either someone else is emitting or there was no available request\n+                // queue up the scalar\n+                RxRingBuffer svq = scalarQueue;\n+                if (svq == null) {\n+                    svq = RxRingBuffer.getSpscInstance();","path":"src/main/java/rx/internal/operators/OperatorMerge.java","position":840,"original_position":749,"commit_id":"b883f6429411cdb22facc81054b63d48a143d763","original_commit_id":"8a6d506bb199d439d72a9b884cdda624d178a4db","user":{"login":"akarnokd","id":1269832,"avatar_url":"https://avatars.githubusercontent.com/u/1269832?v=3","gravatar_id":"","url":"https://api.github.com/users/akarnokd","html_url":"https://github.com/akarnokd","followers_url":"https://api.github.com/users/akarnokd/followers","following_url":"https://api.github.com/users/akarnokd/following{/other_user}","gists_url":"https://api.github.com/users/akarnokd/gists{/gist_id}","starred_url":"https://api.github.com/users/akarnokd/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akarnokd/subscriptions","organizations_url":"https://api.github.com/users/akarnokd/orgs","repos_url":"https://api.github.com/users/akarnokd/repos","events_url":"https://api.github.com/users/akarnokd/events{/privacy}","received_events_url":"https://api.github.com/users/akarnokd/received_events","type":"User","site_admin":false},"body":"Fine.","created_at":"2015-06-18T16:35:44Z","updated_at":"2015-06-18T16:35:44Z","html_url":"https://github.com/ReactiveX/RxJava/pull/2928#discussion_r32750138","pull_request_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928","_links":{"self":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/comments/32750138"},"html":{"href":"https://github.com/ReactiveX/RxJava/pull/2928#discussion_r32750138"},"pull_request":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928"}}},"pull_request":{"url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928","id":34495543,"html_url":"https://github.com/ReactiveX/RxJava/pull/2928","diff_url":"https://github.com/ReactiveX/RxJava/pull/2928.diff","patch_url":"https://github.com/ReactiveX/RxJava/pull/2928.patch","issue_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2928","number":2928,"state":"open","locked":false,"title":"Merge and MergeMaxConcurrent unified and rewritten","user":{"login":"akarnokd","id":1269832,"avatar_url":"https://avatars.githubusercontent.com/u/1269832?v=3","gravatar_id":"","url":"https://api.github.com/users/akarnokd","html_url":"https://github.com/akarnokd","followers_url":"https://api.github.com/users/akarnokd/followers","following_url":"https://api.github.com/users/akarnokd/following{/other_user}","gists_url":"https://api.github.com/users/akarnokd/gists{/gist_id}","starred_url":"https://api.github.com/users/akarnokd/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akarnokd/subscriptions","organizations_url":"https://api.github.com/users/akarnokd/orgs","repos_url":"https://api.github.com/users/akarnokd/repos","events_url":"https://api.github.com/users/akarnokd/events{/privacy}","received_events_url":"https://api.github.com/users/akarnokd/received_events","type":"User","site_admin":false},"body":"I've rewritten ```merge``` and ```mergeMaxConcurrent``` together from scratch.\r\n\r\n  - Added experimental ```maxConcurrent``` overloads to the array-merge and the mergeDelayError\r\n  - Fixed ```OperatorMapNotification``` beause it disregarded backpressure on its terminal emissions.\r\n  - While debugging some merge test failures, I've encountered a bug in OperatorPublish with the same underlying logical error (see comment) and fixed it here.\r\n  - Accidentally introduced a few tabs, therefore, I went in and replaced all tabs with spaces.\r\n  - I've changed ```OperatorMergeTest.shouldNotCompleteIfThereArePendingScalarSynchronousEmissionsWhenTheLastInnerSubscriberCompletes``` because it is legal to get onCompleted without any request.\r\n","created_at":"2015-04-30T20:02:47Z","updated_at":"2015-06-18T16:35:44Z","closed_at":null,"merged_at":null,"merge_commit_sha":"28015cc3dd5468c6ac15c94d38124287ed7e7c1f","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928/commits","review_comments_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928/comments","review_comment_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls/comments{/number}","comments_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/2928/comments","statuses_url":"https://api.github.com/repos/ReactiveX/RxJava/statuses/b883f6429411cdb22facc81054b63d48a143d763","head":{"label":"akarnokd:OperatorMergeFullRewrite","ref":"OperatorMergeFullRewrite","sha":"b883f6429411cdb22facc81054b63d48a143d763","user":{"login":"akarnokd","id":1269832,"avatar_url":"https://avatars.githubusercontent.com/u/1269832?v=3","gravatar_id":"","url":"https://api.github.com/users/akarnokd","html_url":"https://github.com/akarnokd","followers_url":"https://api.github.com/users/akarnokd/followers","following_url":"https://api.github.com/users/akarnokd/following{/other_user}","gists_url":"https://api.github.com/users/akarnokd/gists{/gist_id}","starred_url":"https://api.github.com/users/akarnokd/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akarnokd/subscriptions","organizations_url":"https://api.github.com/users/akarnokd/orgs","repos_url":"https://api.github.com/users/akarnokd/repos","events_url":"https://api.github.com/users/akarnokd/events{/privacy}","received_events_url":"https://api.github.com/users/akarnokd/received_events","type":"User","site_admin":false},"repo":{"id":14518033,"name":"RxJava","full_name":"akarnokd/RxJava","owner":{"login":"akarnokd","id":1269832,"avatar_url":"https://avatars.githubusercontent.com/u/1269832?v=3","gravatar_id":"","url":"https://api.github.com/users/akarnokd","html_url":"https://github.com/akarnokd","followers_url":"https://api.github.com/users/akarnokd/followers","following_url":"https://api.github.com/users/akarnokd/following{/other_user}","gists_url":"https://api.github.com/users/akarnokd/gists{/gist_id}","starred_url":"https://api.github.com/users/akarnokd/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akarnokd/subscriptions","organizations_url":"https://api.github.com/users/akarnokd/orgs","repos_url":"https://api.github.com/users/akarnokd/repos","events_url":"https://api.github.com/users/akarnokd/events{/privacy}","received_events_url":"https://api.github.com/users/akarnokd/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/akarnokd/RxJava","description":"RxJava - a library for composing asynchronous and event-based programs using observable sequences for the Java VM.","fork":true,"url":"https://api.github.com/repos/akarnokd/RxJava","forks_url":"https://api.github.com/repos/akarnokd/RxJava/forks","keys_url":"https://api.github.com/repos/akarnokd/RxJava/keys{/key_id}","collaborators_url":"https://api.github.com/repos/akarnokd/RxJava/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/akarnokd/RxJava/teams","hooks_url":"https://api.github.com/repos/akarnokd/RxJava/hooks","issue_events_url":"https://api.github.com/repos/akarnokd/RxJava/issues/events{/number}","events_url":"https://api.github.com/repos/akarnokd/RxJava/events","assignees_url":"https://api.github.com/repos/akarnokd/RxJava/assignees{/user}","branches_url":"https://api.github.com/repos/akarnokd/RxJava/branches{/branch}","tags_url":"https://api.github.com/repos/akarnokd/RxJava/tags","blobs_url":"https://api.github.com/repos/akarnokd/RxJava/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/akarnokd/RxJava/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/akarnokd/RxJava/git/refs{/sha}","trees_url":"https://api.github.com/repos/akarnokd/RxJava/git/trees{/sha}","statuses_url":"https://api.github.com/repos/akarnokd/RxJava/statuses/{sha}","languages_url":"https://api.github.com/repos/akarnokd/RxJava/languages","stargazers_url":"https://api.github.com/repos/akarnokd/RxJava/stargazers","contributors_url":"https://api.github.com/repos/akarnokd/RxJava/contributors","subscribers_url":"https://api.github.com/repos/akarnokd/RxJava/subscribers","subscription_url":"https://api.github.com/repos/akarnokd/RxJava/subscription","commits_url":"https://api.github.com/repos/akarnokd/RxJava/commits{/sha}","git_commits_url":"https://api.github.com/repos/akarnokd/RxJava/git/commits{/sha}","comments_url":"https://api.github.com/repos/akarnokd/RxJava/comments{/number}","issue_comment_url":"https://api.github.com/repos/akarnokd/RxJava/issues/comments{/number}","contents_url":"https://api.github.com/repos/akarnokd/RxJava/contents/{+path}","compare_url":"https://api.github.com/repos/akarnokd/RxJava/compare/{base}...{head}","merges_url":"https://api.github.com/repos/akarnokd/RxJava/merges","archive_url":"https://api.github.com/repos/akarnokd/RxJava/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/akarnokd/RxJava/downloads","issues_url":"https://api.github.com/repos/akarnokd/RxJava/issues{/number}","pulls_url":"https://api.github.com/repos/akarnokd/RxJava/pulls{/number}","milestones_url":"https://api.github.com/repos/akarnokd/RxJava/milestones{/number}","notifications_url":"https://api.github.com/repos/akarnokd/RxJava/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/akarnokd/RxJava/labels{/name}","releases_url":"https://api.github.com/repos/akarnokd/RxJava/releases{/id}","created_at":"2013-11-19T08:04:52Z","updated_at":"2014-07-09T07:02:30Z","pushed_at":"2015-06-18T12:24:20Z","git_url":"git://github.com/akarnokd/RxJava.git","ssh_url":"git@github.com:akarnokd/RxJava.git","clone_url":"https://github.com/akarnokd/RxJava.git","svn_url":"https://github.com/akarnokd/RxJava","homepage":"","size":174977,"stargazers_count":1,"watchers_count":1,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":1,"default_branch":"master"}},"base":{"label":"ReactiveX:1.x","ref":"1.x","sha":"476ce0a05095ff2e559be524b0c9ab7c935a76a8","user":{"login":"ReactiveX","id":6407041,"avatar_url":"https://avatars.githubusercontent.com/u/6407041?v=3","gravatar_id":"","url":"https://api.github.com/users/ReactiveX","html_url":"https://github.com/ReactiveX","followers_url":"https://api.github.com/users/ReactiveX/followers","following_url":"https://api.github.com/users/ReactiveX/following{/other_user}","gists_url":"https://api.github.com/users/ReactiveX/gists{/gist_id}","starred_url":"https://api.github.com/users/ReactiveX/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ReactiveX/subscriptions","organizations_url":"https://api.github.com/users/ReactiveX/orgs","repos_url":"https://api.github.com/users/ReactiveX/repos","events_url":"https://api.github.com/users/ReactiveX/events{/privacy}","received_events_url":"https://api.github.com/users/ReactiveX/received_events","type":"Organization","site_admin":false},"repo":{"id":7508411,"name":"RxJava","full_name":"ReactiveX/RxJava","owner":{"login":"ReactiveX","id":6407041,"avatar_url":"https://avatars.githubusercontent.com/u/6407041?v=3","gravatar_id":"","url":"https://api.github.com/users/ReactiveX","html_url":"https://github.com/ReactiveX","followers_url":"https://api.github.com/users/ReactiveX/followers","following_url":"https://api.github.com/users/ReactiveX/following{/other_user}","gists_url":"https://api.github.com/users/ReactiveX/gists{/gist_id}","starred_url":"https://api.github.com/users/ReactiveX/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ReactiveX/subscriptions","organizations_url":"https://api.github.com/users/ReactiveX/orgs","repos_url":"https://api.github.com/users/ReactiveX/repos","events_url":"https://api.github.com/users/ReactiveX/events{/privacy}","received_events_url":"https://api.github.com/users/ReactiveX/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/ReactiveX/RxJava","description":"RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.","fork":false,"url":"https://api.github.com/repos/ReactiveX/RxJava","forks_url":"https://api.github.com/repos/ReactiveX/RxJava/forks","keys_url":"https://api.github.com/repos/ReactiveX/RxJava/keys{/key_id}","collaborators_url":"https://api.github.com/repos/ReactiveX/RxJava/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/ReactiveX/RxJava/teams","hooks_url":"https://api.github.com/repos/ReactiveX/RxJava/hooks","issue_events_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/events{/number}","events_url":"https://api.github.com/repos/ReactiveX/RxJava/events","assignees_url":"https://api.github.com/repos/ReactiveX/RxJava/assignees{/user}","branches_url":"https://api.github.com/repos/ReactiveX/RxJava/branches{/branch}","tags_url":"https://api.github.com/repos/ReactiveX/RxJava/tags","blobs_url":"https://api.github.com/repos/ReactiveX/RxJava/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/ReactiveX/RxJava/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/ReactiveX/RxJava/git/refs{/sha}","trees_url":"https://api.github.com/repos/ReactiveX/RxJava/git/trees{/sha}","statuses_url":"https://api.github.com/repos/ReactiveX/RxJava/statuses/{sha}","languages_url":"https://api.github.com/repos/ReactiveX/RxJava/languages","stargazers_url":"https://api.github.com/repos/ReactiveX/RxJava/stargazers","contributors_url":"https://api.github.com/repos/ReactiveX/RxJava/contributors","subscribers_url":"https://api.github.com/repos/ReactiveX/RxJava/subscribers","subscription_url":"https://api.github.com/repos/ReactiveX/RxJava/subscription","commits_url":"https://api.github.com/repos/ReactiveX/RxJava/commits{/sha}","git_commits_url":"https://api.github.com/repos/ReactiveX/RxJava/git/commits{/sha}","comments_url":"https://api.github.com/repos/ReactiveX/RxJava/comments{/number}","issue_comment_url":"https://api.github.com/repos/ReactiveX/RxJava/issues/comments{/number}","contents_url":"https://api.github.com/repos/ReactiveX/RxJava/contents/{+path}","compare_url":"https://api.github.com/repos/ReactiveX/RxJava/compare/{base}...{head}","merges_url":"https://api.github.com/repos/ReactiveX/RxJava/merges","archive_url":"https://api.github.com/repos/ReactiveX/RxJava/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/ReactiveX/RxJava/downloads","issues_url":"https://api.github.com/repos/ReactiveX/RxJava/issues{/number}","pulls_url":"https://api.github.com/repos/ReactiveX/RxJava/pulls{/number}","milestones_url":"https://api.github.com/repos/ReactiveX/RxJava/milestones{/number}","notifications_url":"https://api.github.com/repos/ReactiveX/RxJava/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/ReactiveX/RxJava/labels{/name}","releases_url":"https://api.github.com/repos/ReactiveX/RxJava/releases{/id}","created_at":"2013-01-08T20:11:48Z","updated_at":"2015-06-18T12:54:39Z","pushed_at":"2015-06-18T12:35:23Z","git_url":"git://github.com/ReactiveX/RxJava.git","ssh_url":"git@github.com:ReactiveX/RxJava.git","clone_url":"https://github.com/ReactiveX/RxJava.git","svn_url":"https://github.com/ReactiveX/RxJava","homepage":"","size":520514,"stargazers_count":6112,"watchers_count":6112,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":true,"forks_count":839,"mirror_url":null,"open_issues_count":135,"forks":839,"open_issues":135,"watchers":6112,"default_branch":"1.x"}},"_links":{"self":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928"},"html":{"href":"https://github.com/ReactiveX/RxJava/pull/2928"},"issue":{"href":"https://api.github.com/repos/ReactiveX/RxJava/issues/2928"},"comments":{"href":"https://api.github.com/repos/ReactiveX/RxJava/issues/2928/comments"},"review_comments":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928/comments"},"review_comment":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/ReactiveX/RxJava/pulls/2928/commits"},"statuses":{"href":"https://api.github.com/repos/ReactiveX/RxJava/statuses/b883f6429411cdb22facc81054b63d48a143d763"}}}}