Functional now ... it seems.

Author: benjchristensen

rxjava-core/src/main/java/rx/observers/SerializedObserver.java

@@ -1,5 +1,8 @@
 package rx.observers;
 
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -14,6 +17,11 @@ public SerializedObserver(Observer<T> s) {
         this.s = s;
     }
 
+    final AtomicInteger received = new AtomicInteger();
+    final AtomicInteger counter = new AtomicInteger();
+    final AtomicInteger offered = new AtomicInteger();
+    static AtomicInteger decremented = new AtomicInteger();
+
     @Override
     public void onCompleted() {
         State current = null;
@@ -26,7 +34,6 @@ public void onCompleted() {
             }
             newState = current.complete();
         } while (!state.compareAndSet(current, newState));
-        System.out.println("********** onCompleted");
         terminateIfNecessary(newState);
     }
 
@@ -42,43 +49,54 @@ public void onError(Throwable e) {
             }
             newState = current.error(e);
         } while (!state.compareAndSet(current, newState));
-        System.out.println("********** onError");
         terminateIfNecessary(newState);
     }
 
+    AtomicInteger conc = new AtomicInteger();
+    AtomicInteger lost = new AtomicInteger();
+    Set<Object> items = Collections.synchronizedSet(new HashSet<Object>());
+
     @SuppressWarnings("unchecked")
     @Override
     public void onNext(T t) {
         State current = null;
         State newState = null;
+
+        int contention = 0;
+        State orig = null;
         do {
             current = state.get();
+            if (orig == null) {
+                orig = current;
+            }
             if (current.isTerminated()) {
                 // already received terminal state
                 return;
             }
             newState = current.offerItem(t);
+            contention++;
         } while (!state.compareAndSet(current, newState));
 
+        do {
+            current = state.get();
+            newState = current.startProcessing();
+        } while (!state.compareAndSet(current, newState));
         if (newState.shouldProcess()) {
-            // this thread wins and will emit then drain queue if it concurrently gets added to
-            s.onNext(t);
-
-            // drain queue if exists
-            // we do "if" instead of "while" so we don't starve one thread
+            // drain queue 
             Object[] items = newState.queue;
             for (int i = 0; i < items.length; i++) {
                 s.onNext((T) items[i]);
+                counter.incrementAndGet();
             }
 
             // finish processing to let this thread move on
             do {
                 current = state.get();
-                newState = current.finishProcessing(items.length + 1); // the + 1 is for the first onNext of itself
+                newState = current.finishProcessing(items.length);
             } while (!state.compareAndSet(current, newState));
-            System.out.println("********** finishProcessing");
-            terminateIfNecessary(newState);
+
         }
+        terminateIfNecessary(newState);
     }
 
     @SuppressWarnings("unchecked")
@@ -89,6 +107,7 @@ private void terminateIfNecessary(State current) {
                 current = state.get();
                 newState = current.startTermination();
             } while (!state.compareAndSet(current, newState));
+
             if (newState.shouldProcess()) {
                 // drain any items left
                 for (int i = 0; i < newState.queue.length; i++) {
@@ -114,9 +133,9 @@ public static class State {
         final Throwable onError;
 
         private final static Object[] EMPTY = new Object[0];
+        private final static Object[] PROCESS_SELF = new Object[1];
 
         private final static State NON_TERMINATED_EMPTY = new State(false, false, 0, false, null, EMPTY);
-        private final static State NON_TERMINATED_PROCESS_SELF = new State(true, true, 1, false, null, EMPTY);
 
         public State(boolean shouldProcess, boolean isSomeoneProcessing, int queueSize, boolean onComplete, Throwable onError, Object[] queue) {
             this.shouldProcess = shouldProcess;
@@ -145,59 +164,59 @@ public State complete() {
 
         public State error(Throwable e) {
             // immediately empty the queue and emit error as soon as possible
-            return new State(false, isSomeoneProcessing, queueSize, onComplete, e, EMPTY);
+            return new State(false, isSomeoneProcessing, 0, onComplete, e, EMPTY);
         }
 
         public State startTermination() {
             if (isSomeoneProcessing) {
-                System.out.println("start terminate and DO NOT process => queue size: " + (queueSize + 1));
                 return new State(false, isSomeoneProcessing, queueSize, onComplete, onError, queue);
             } else {
-                System.out.println("start terminate and process => queue size: " + (queueSize + 1));
-                return new State(true, isSomeoneProcessing, queueSize, onComplete, onError, queue);
+                return new State(true, true, queueSize, onComplete, onError, queue);
             }
         }
 
-        AtomicInteger max = new AtomicInteger();
-
         public State offerItem(Object item) {
-            if (queueSize == 0) {
-                // no concurrent requests so don't queue, we'll process immediately
-                if (isTerminated()) {
-                    // return count of 0 meaning don't emit as we are terminated
-                    return new State(false, false, 0, onComplete, onError, EMPTY);
-                } else {
-                    return NON_TERMINATED_PROCESS_SELF;
-                }
+            if (isTerminated()) {
+                // return count of 0 meaning don't emit as we are terminated
+                return new State(false, isSomeoneProcessing, 0, onComplete, onError, EMPTY);
             } else {
-                // there are items queued so we need to queue
                 int idx = queue.length;
                 Object[] newQueue = new Object[idx + 1];
                 System.arraycopy(queue, 0, newQueue, 0, idx);
                 newQueue[idx] = item;
 
-                if (isSomeoneProcessing) {
-                    // we just add to queue
-                    return new State(false, isSomeoneProcessing, queueSize + 1, onComplete, onError, newQueue);
-                } else {
-                    // we add to queue and claim work
-                    return new State(false, true, queueSize + 1, onComplete, onError, newQueue);
-                }
+                // we just add to queue
+                return new State(false, isSomeoneProcessing, queueSize + 1, onComplete, onError, newQueue);
+            }
+        }
+
+        public State startProcessing() {
+            if (isSomeoneProcessing) {
+                return new State(false, true, queueSize, onComplete, onError, queue);
+            } else {
+                return new State(true, true, queueSize, onComplete, onError, queue);
             }
         }
 
         public State finishProcessing(int numOnNextSent) {
-            int numOnNextFromQueue = numOnNextSent - 1; // we remove the "self" onNext as it doesn't affect the queue
-            int size = queueSize - numOnNextFromQueue;
-            System.out.println("finishProcessing => queue size: " + size + "   after processing: " + numOnNextSent);
-            if (size > 1 || isTerminated()) {
-                Object[] newQueue = new Object[queue.length - numOnNextFromQueue];
-                System.arraycopy(queue, numOnNextFromQueue, newQueue, 0, newQueue.length);
+            int size = queueSize - numOnNextSent;
+            if (size > 0 || isTerminated()) {
+                // if size == 0 but we are terminated then it's an empty queue
+                Object[] newQueue = EMPTY;
+                if (size > 0) {
+                    newQueue = new Object[queue.length - numOnNextSent];
+                    System.arraycopy(queue, numOnNextSent, newQueue, 0, newQueue.length);
+                }
                 return new State(false, false, size, onComplete, onError, newQueue);
             } else {
                 return NON_TERMINATED_EMPTY;
             }
         }
 
+        @Override
+        public String toString() {
+            return "State => shouldProcess: " + shouldProcess + " processing: " + isSomeoneProcessing + " queueSize: " + queueSize + " queue: " + queue.length + " terminated: " + isTerminated();
+        }
+
     }
 }

rxjava-core/src/perf/java/rx/operators/OperatorSerializePerformance.java

@@ -147,7 +147,7 @@ public void call(Integer t1) {
 
         IntegerSumObserver o = new IntegerSumObserver();
         s.subscribe(o);
-        System.out.println("sum : " + o.sum);
+//        System.out.println("sum : " + o.sum);
 
         return o.sum;
     }

rxjava-core/src/perf/java/rx/operators/OperatorSynchronizePerformance.java

@@ -14,9 +14,9 @@
 import rx.schedulers.Schedulers;
 
 public class OperatorSynchronizePerformance extends AbstractPerformanceTester {
-    //    static int reps = Integer.MAX_VALUE / 1024;
+    static int reps = Integer.MAX_VALUE / 1024;
 
-    static int reps = 1000; // timeTwoStreamsIntervals
+    //    static int reps = 1000; // timeTwoStreamsIntervals
 
     OperatorSynchronizePerformance() {
         super(reps);
@@ -30,7 +30,7 @@ public static void main(String args[]) {
 
                 @Override
                 public void call() {
-                    spt.timeTwoStreamsIntervals();
+                    spt.timeTwoStreams();
                 }
             });
         } catch (Exception e) {

rxjava-core/src/test/java/rx/observers/SerializedObserverTest.java

@@ -19,6 +19,7 @@
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -92,8 +93,8 @@ public void testMultiThreadedBasic() {
         assertEquals(1, busyObserver.maxConcurrentThreads.get());
     }
 
-    @Test
-    public void testMultiThreadedWithNPE() {
+    @Test(timeout=1000)
+    public void testMultiThreadedWithNPE() throws InterruptedException {
         Subscription s = mock(Subscription.class);
         TestMultiThreadedObservable onSubscribe = new TestMultiThreadedObservable(s, "one", "two", "three", null);
         Observable<String> w = Observable.create(onSubscribe);
@@ -103,8 +104,9 @@ public void testMultiThreadedWithNPE() {
 
         w.subscribe(aw);
         onSubscribe.waitToFinish();
+        busyObserver.terminalEvent.await();
 
-        System.out.println("maxConcurrentThreads: " + onSubscribe.maxConcurrentThreads.get());
+        System.out.println("OnSubscribe maxConcurrentThreads: " + onSubscribe.maxConcurrentThreads.get() + "  Observer maxConcurrentThreads: " + busyObserver.maxConcurrentThreads.get());
 
         // we can't know how many onNext calls will occur since they each run on a separate thread
         // that depends on thread scheduling so 0, 1, 2 and 3 are all valid options
@@ -286,7 +288,7 @@ public static class OnNextThread implements Runnable {
         @Override
         public void run() {
             for (int i = 0; i < numStringsToSend; i++) {
-                Observer.onNext("aString");
+                Observer.onNext(Thread.currentThread().getId() + "-" + i);
             }
         }
     }
@@ -296,12 +298,12 @@ public void run() {
      */
     public static class CompletionThread implements Runnable {
 
-        private final Observer<String> Observer;
+        private final Observer<String> observer;
         private final TestConcurrencyObserverEvent event;
         private final Future<?>[] waitOnThese;
 
         CompletionThread(Observer<String> Observer, TestConcurrencyObserverEvent event, Future<?>... waitOnThese) {
-            this.Observer = Observer;
+            this.observer = Observer;
             this.event = event;
             this.waitOnThese = waitOnThese;
         }
@@ -321,9 +323,9 @@ public void run() {
 
             /* send the event */
             if (event == TestConcurrencyObserverEvent.onError) {
-                Observer.onError(new RuntimeException("mocked exception"));
+                observer.onError(new RuntimeException("mocked exception"));
             } else if (event == TestConcurrencyObserverEvent.onCompleted) {
-                Observer.onCompleted();
+                observer.onCompleted();
 
             } else {
                 throw new IllegalArgumentException("Expecting either onError or onCompleted");
@@ -566,6 +568,7 @@ private static class BusyObserver extends Subscriber<String> {
         AtomicInteger onNextCount = new AtomicInteger();
         AtomicInteger threadsRunning = new AtomicInteger();
         AtomicInteger maxConcurrentThreads = new AtomicInteger();
+        final CountDownLatch terminalEvent = new CountDownLatch(1);
 
         @Override
         public void onCompleted() {
@@ -575,17 +578,20 @@ public void onCompleted() {
             } finally {
                 captureMaxThreads();
                 threadsRunning.decrementAndGet();
+                terminalEvent.countDown();
             }
         }
 
         @Override
         public void onError(Throwable e) {
+            System.out.println(">>>>>>>>>>>>>>>>>>>> onError received: " + e);
             threadsRunning.incrementAndGet();
             try {
                 onError = true;
             } finally {
                 captureMaxThreads();
                 threadsRunning.decrementAndGet();
+                terminalEvent.countDown();
             }
         }