1.x: fix counted buffer and window backpressure

src/main/java/rx/Observable.java

@@ -9744,8 +9744,8 @@ public final <TClosing> Observable<Observable<T>> window(Func0<? extends Observa
      * <img width="640" height="400" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/window3.png" alt="">
      * <dl>
      *  <dt><b>Backpressure Support:</b></dt>
-     *  <dd>The operator honors backpressure on its outer subscriber, ignores backpressure in its inner Observables 
-     *  but each of them will emit at most {@code count} elements.</dd>
+     *  <dd>The operator honors backpressure of its inner and outer subscribers, however, the inner Observable uses an
+     *  unbounded buffer that may hold at most {@code count} elements.</dd>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>This version of {@code window} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
@@ -9754,6 +9754,7 @@ public final <TClosing> Observable<Observable<T>> window(Func0<? extends Observa
      *            the maximum size of each window before it should be emitted
      * @return an Observable that emits connected, non-overlapping windows, each containing at most
      *         {@code count} items from the source Observable
+     * @throws IllegalArgumentException if either count is non-positive
      * @see <a href="http://reactivex.io/documentation/operators/window.html">ReactiveX operators documentation: Window</a>
      */
     public final Observable<Observable<T>> window(int count) {
@@ -9769,8 +9770,8 @@ public final Observable<Observable<T>> window(int count) {
      * <img width="640" height="365" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/window4.png" alt="">
      * <dl>
      *  <dt><b>Backpressure Support:</b></dt>
-     *  <dd>The operator honors backpressure on its outer subscriber, ignores backpressure in its inner Observables 
-     *  but each of them will emit at most {@code count} elements.</dd>
+     *  <dd>The operator honors backpressure of its inner and outer subscribers, however, the inner Observable uses an
+     *  unbounded buffer that may hold at most {@code count} elements.</dd>
      *  <dt><b>Scheduler:</b></dt>
      *  <dd>This version of {@code window} does not operate by default on a particular {@link Scheduler}.</dd>
      * </dl>
@@ -9782,9 +9783,16 @@ public final Observable<Observable<T>> window(int count) {
      *            {@code count} are equal this is the same operation as {@link #window(int)}.
      * @return an Observable that emits windows every {@code skip} items containing at most {@code count} items
      *         from the source Observable
+     * @throws IllegalArgumentException if either count or skip is non-positive
      * @see <a href="http://reactivex.io/documentation/operators/window.html">ReactiveX operators documentation: Window</a>
      */
     public final Observable<Observable<T>> window(int count, int skip) {
+        if (count <= 0) {
+            throw new IllegalArgumentException("count > 0 required but it was " + count);
+        }
+        if (skip <= 0) {
+            throw new IllegalArgumentException("skip > 0 required but it was " + skip);
+        }
         return lift(new OperatorWindowWithSize<T>(count, skip));
     }
 

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

@@ -15,8 +15,10 @@
  */
 package rx.internal.operators;
 
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+import java.util.Queue;
+import java.util.concurrent.atomic.*;
+
+import rx.Subscriber;
 
 /**
  * Utility functions for use with backpressure.
@@ -32,6 +34,8 @@ private BackpressureUtils() {
      * addition once the addition is successful (uses CAS semantics). If
      * overflows then sets {@code requested} field to {@code Long.MAX_VALUE}.
      * 
+     * @param <T> the type of the target object on which the field updater operates
+     * 
      * @param requested
      *            atomic field updater for a request count
      * @param object
@@ -103,6 +107,208 @@ public static long addCap(long a, long b) {
         return u;
     }
 
+    /**
+     * Masks the most significant bit, i.e., 0x8000_0000_0000_0000L.
+     */
+    static final long COMPLETED_MASK = Long.MIN_VALUE;
+    /**
+     * Masks the request amount bits, i.e., 0x7FFF_FFFF_FFFF_FFFF.
+     */
+    static final long REQUESTED_MASK = Long.MAX_VALUE;
+
+    /**
+     * Signals the completion of the main sequence and switches to post-completion replay mode.
+     * 
+     * <p>
+     * Don't modify the queue after calling this method!
+     * 
+     * <p>
+     * Post-completion backpressure handles the case when a source produces values based on
+     * requests when it is active but more values are available even after its completion.
+     * In this case, the onCompleted() can't just emit the contents of the queue but has to
+     * coordinate with the requested amounts. This requires two distinct modes: active and
+     * completed. In active mode, requests flow through and the queue is not accessed but
+     * in completed mode, requests no-longer reach the upstream but help in draining the queue.
+     * <p>
+     * The algorithm utilizes the most significant bit (bit 63) of a long value (AtomicLong) since
+     * request amount only goes up to Long.MAX_VALUE (bits 0-62) and negative values aren't
+     * allowed.
+     * 
+     * @param <T> the value type to emit
+     * @param requested the holder of current requested amount
+     * @param queue the queue holding values to be emitted after completion
+     * @param actual the subscriber to receive the values
+     */
+    public static <T> void postCompleteDone(AtomicLong requested, Queue<T> queue, Subscriber<? super T> actual) {
+        for (;;) {
+            long r = requested.get();
+
+            // switch to completed mode only once
+            if ((r & COMPLETED_MASK) != 0L) {
+                return;
+            }
+
+            //
+            long u = r | COMPLETED_MASK;
+
+            if (requested.compareAndSet(r, u)) {
+                // if we successfully switched to post-complete mode and there 
+                // are requests available start draining the queue
+                if (r != 0L) {
+                    // if the switch happened when there was outstanding requests, start draining
+                    postCompleteDrain(requested, queue, actual);
+                }
+                return;
+            }
+        }
+    }
+
+    /**
+     * Accumulates requests (validated) and handles the completed mode draining of the queue based on the requests.
+     * 
+     * <p>
+     * Post-completion backpressure handles the case when a source produces values based on
+     * requests when it is active but more values are available even after its completion.
+     * In this case, the onCompleted() can't just emit the contents of the queue but has to
+     * coordinate with the requested amounts. This requires two distinct modes: active and
+     * completed. In active mode, requests flow through and the queue is not accessed but
+     * in completed mode, requests no-longer reach the upstream but help in draining the queue.
+     * 
+     * @param <T> the value type to emit
+     * @param requested the holder of current requested amount
+     * @param n the value requested;
+     * @param queue the queue holding values to be emitted after completion
+     * @param actual the subscriber to receive the values
+     * @return true if in the active mode and the request amount of n can be relayed to upstream, false if
+     * in the post-completed mode and the queue is draining.
+     */
+    public static <T> boolean postCompleteRequest(AtomicLong requested, long n, Queue<T> queue, Subscriber<? super T> actual) {
+        if (n < 0L) {
+            throw new IllegalArgumentException("n >= 0 required but it was " + n);
+        }
+        if (n == 0) {
+            return (requested.get() & COMPLETED_MASK) == 0;
+        }
+
+        for (;;) {
+            long r = requested.get();
+
+            // mask of the completed flag
+            long c = r & COMPLETED_MASK;
+            // mask of the requested amount
+            long u = r & REQUESTED_MASK;
+
+            // add the current requested amount and the new requested amount
+            // cap at Long.MAX_VALUE;
+            long v = addCap(u, n);
+
+            // restore the completed flag
+            v |= c;
+
+            if (requested.compareAndSet(r, v)) {
+                // if there was no outstanding request before and in
+                // the post-completed state, start draining
+                if (r == COMPLETED_MASK) {
+                    postCompleteDrain(requested, queue, actual);
+                    return false;
+                }
+                // returns true for active mode and false if the completed flag was set
+                return c == 0L;
+            }
+        }
+    }
+
+    /**
+     * Drains the queue based on the outstanding requests in post-completed mode (only!).
+     * 
+     * @param <T> the value type to emit
+     * @param requested the holder of current requested amount
+     * @param queue the queue holding values to be emitted after completion
+     * @param actual the subscriber to receive the values
+     */
+    static <T> void postCompleteDrain(AtomicLong requested, Queue<T> queue, Subscriber<? super T> subscriber) {
+
+        long r = requested.get();
+        /*
+         * Since we are supposed to be in the post-complete state, 
+         * requested will have its top bit set.
+         * To allow direct comparison, we start with an emission value which has also
+         * this flag set, then increment it as usual.
+         * Since COMPLETED_MASK is essentially Long.MIN_VALUE, 
+         * there won't be any overflow or sign flip.
+         */
+        long e = COMPLETED_MASK;
+
+        for (;;) {
+
+            /*
+             * This is an improved queue-drain algorithm with a specialization 
+             * in which we know the queue won't change anymore (i.e., done is always true
+             * when looking at the classical algorithm and there is no error).
+             * 
+             * Note that we don't check for cancellation or emptyness upfront for two reasons:
+             * 1) if e != r, the loop will do this and we quit appropriately
+             * 2) if e == r, then either there was no outstanding requests or we emitted the requested amount
+             *    and the execution simply falls to the e == r check below which checks for emptyness anyway.
+             */
+
+            while (e != r) {
+                if (subscriber.isUnsubscribed()) {
+                    return;
+                }
+
+                T v = queue.poll();
+
+                if (v == null) {
+                    subscriber.onCompleted();
+                    return;
+                }
+
+                subscriber.onNext(v);
+
+                e++;
+            }
+
+            /*
+             * If the emission count reaches the requested amount the same time the queue becomes empty
+             * this will make sure the subscriber is completed immediately instead of on the next request.
+             * This is also true if there are no outstanding requests (this the while loop doesn't run)
+             * and the queue is empty from the start.
+             */
+            if (e == r) {
+                if (subscriber.isUnsubscribed()) {
+                    return;
+                }
+                if (queue.isEmpty()) {
+                    subscriber.onCompleted();
+                    return;
+                }
+            }
+
+            /*
+             * Fast flow: see if more requests have arrived in the meantime.
+             * This avoids an atomic add (~40 cycles) and resumes the emission immediately.
+             */
+            r = requested.get();
+
+            if (r == e) {
+                /* 
+                 * Atomically decrement the requested amount by the emission amount.
+                 * We can't use the full emission value because of the completed flag,
+                 * however, due to two's complement representation, the flag on requested
+                 * is preserved.
+                 */
+                r = requested.addAndGet(-(e & REQUESTED_MASK));
+                // The requested amount actually reached zero, quit
+                if (r == COMPLETED_MASK) {
+                    return;
+                }
+                // reset the emission count
+                e = COMPLETED_MASK;
+            }
+        }
+    }
+
     /**
      * Atomically subtracts a value from the requested amount unless it's at Long.MAX_VALUE.
      * @param requested the requested amount holder