1.x: replay request coordination reduce overhead

This change will make sure whenever a new subscriber arrives, the
request coordination doesn't scan every other subscriber unnecessary.
(They will trigger a manageRequests individually anyway)

Author: akarnokd

src/main/java/rx/internal/operators/OperatorReplay.java

@@ -320,6 +320,11 @@ static final class ReplaySubscriber<T> extends Subscriber<T> implements Subscrip
         /** The upstream producer. */
         volatile Producer producer;
 
+        /** The queue that holds producers with request changes that need to be coordinated. */
+        List<InnerProducer<T>> coordinationQueue;
+        /** Indicate that all request amounts should be considered. */
+        boolean coordinateAll;
+        
         public ReplaySubscriber(AtomicReference<ReplaySubscriber<T>> current,
                 ReplayBuffer<T> buffer) {
             this.buffer = buffer;
@@ -435,7 +440,7 @@ public void setProducer(Producer p) {
                 throw new IllegalStateException("Only a single producer can be set on a Subscriber.");
             }
             producer = p;
-            manageRequests();
+            manageRequests(null);
             replay();
         }
 
@@ -478,70 +483,115 @@ public void onCompleted() {
         /**
          * Coordinates the request amounts of various child Subscribers.
          */
-        void manageRequests() {
+        void manageRequests(InnerProducer<T> inner) {
             // if the upstream has completed, no more requesting is possible
             if (isUnsubscribed()) {
                 return;
             }
             synchronized (this) {
                 if (emitting) {
+                    if (inner != null) {
+                        List<InnerProducer<T>> q = coordinationQueue;
+                        if (q == null) {
+                            q = new ArrayList<InnerProducer<T>>();
+                            coordinationQueue = q;
+                        }
+                        q.add(inner);
+                    } else {
+                        coordinateAll = true;
+                    }
                     missed = true;
                     return;
                 }
                 emitting = true;
             }
+            
+            long ri = maxChildRequested;
+            long maxTotalRequested;
+            
+            if (inner != null) {
+                maxTotalRequested = Math.max(ri, inner.totalRequested.get());
+            } else {
+                maxTotalRequested = ri;
+
+                @SuppressWarnings("unchecked")
+                InnerProducer<T>[] a = producers.get();
+                for (InnerProducer<T> rp : a) {
+                    maxTotalRequested = Math.max(maxTotalRequested, rp.totalRequested.get());
+                }
+                
+            }
+            makeRequest(maxTotalRequested, ri);
+            
             for (;;) {
                 // if the upstream has completed, no more requesting is possible
                 if (isUnsubscribed()) {
                     return;
                 }
 
-                @SuppressWarnings("unchecked")
-                InnerProducer<T>[] a = producers.get();
+                List<InnerProducer<T>> q;
+                boolean all;
+                synchronized (this) {
+                    if (!missed) {
+                        emitting = false;
+                        return;
+                    }
+                    missed = false;
+                    q = coordinationQueue;
+                    coordinationQueue = null;
+                    all = coordinateAll;
+                    coordinateAll = false;
+                }
 
-                long ri = maxChildRequested;
-                long maxTotalRequests = ri;
+                ri = maxChildRequested;
+                maxTotalRequested = ri;
 
-                for (InnerProducer<T> rp : a) {
-                    maxTotalRequests = Math.max(maxTotalRequests, rp.totalRequested.get());
+                if (q != null) {
+                    for (InnerProducer<T> rp : q) {
+                        maxTotalRequested = Math.max(maxTotalRequested, rp.totalRequested.get());
+                    }
+                } 
+                
+                if (all) {
+                    @SuppressWarnings("unchecked")
+                    InnerProducer<T>[] a = producers.get();
+                    for (InnerProducer<T> rp : a) {
+                        maxTotalRequested = Math.max(maxTotalRequested, rp.totalRequested.get());
+                    }
                 }
 
-                long ur = maxUpstreamRequested;
-                Producer p = producer;
+                makeRequest(maxTotalRequested, ri);
+            }
+        }
+        
+        void makeRequest(long maxTotalRequests, long previousTotalRequests) {
+            long ur = maxUpstreamRequested;
+            Producer p = producer;
 
-                long diff = maxTotalRequests - ri;
-                if (diff != 0) {
-                    maxChildRequested = maxTotalRequests;
-                    if (p != null) {
-                        if (ur != 0L) {
-                            maxUpstreamRequested = 0L;
-                            p.request(ur + diff);
-                        } else {
-                            p.request(diff);
-                        }
+            long diff = maxTotalRequests - previousTotalRequests;
+            if (diff != 0) {
+                maxChildRequested = maxTotalRequests;
+                if (p != null) {
+                    if (ur != 0L) {
+                        maxUpstreamRequested = 0L;
+                        p.request(ur + diff);
                     } else {
-                        // collect upstream request amounts until there is a producer for them
-                        long u = ur + diff;
-                        if (u < 0) {
-                            u = Long.MAX_VALUE;
-                        }
-                        maxUpstreamRequested = u;
+                        p.request(diff);
                     }
-                } else
-                // if there were outstanding upstream requests and we have a producer
-                if (ur != 0L && p != null) {
-                    maxUpstreamRequested = 0L;
-                    // fire the accumulated requests
-                    p.request(ur);
-                }
-                
-                synchronized (this) {
-                    if (!missed) {
-                        emitting = false;
-                        return;
+                } else {
+                    // collect upstream request amounts until there is a producer for them
+                    long u = ur + diff;
+                    if (u < 0) {
+                        u = Long.MAX_VALUE;
                     }
-                    missed = false;
+                    maxUpstreamRequested = u;
                 }
+            } else
+            // if there were outstanding upstream requests and we have a producer
+            if (ur != 0L && p != null) {
+                maxUpstreamRequested = 0L;
+                // fire the accumulated requests
+                p.request(ur);
             }
         }
 
@@ -631,7 +681,7 @@ public void request(long n) {
                     addTotalRequested(n);
                     // if successful, notify the parent dispacher this child can receive more
                     // elements
-                    parent.manageRequests();
+                    parent.manageRequests(this);
 
                     parent.buffer.replay(this);
                     return;
@@ -712,7 +762,7 @@ public void unsubscribe() {
                     // let's assume this child had 0 requested before the unsubscription while
                     // the others had non-zero. By removing this 'blocking' child, the others
                     // are now free to receive events
-                    parent.manageRequests();
+                    parent.manageRequests(this);
                 }
             }
         }