2.x: fix SpscLAQ nepotism, FlowableRefCountTest.testRefCountAsync flaky (#5507)

Author: akarnokd

src/main/java/io/reactivex/internal/queue/SpscLinkedArrayQueue.java

@@ -114,8 +114,11 @@ private void soNext(AtomicReferenceArray<Object> curr, AtomicReferenceArray<Obje
         soElement(curr, calcDirectOffset(curr.length() - 1), next);
     }
     @SuppressWarnings("unchecked")
-    private AtomicReferenceArray<Object> lvNext(AtomicReferenceArray<Object> curr) {
-        return (AtomicReferenceArray<Object>)lvElement(curr, calcDirectOffset(curr.length() - 1));
+    private AtomicReferenceArray<Object> lvNextBufferAndUnlink(AtomicReferenceArray<Object> curr, int nextIndex) {
+        int nextOffset = calcDirectOffset(nextIndex);
+        AtomicReferenceArray<Object> nextBuffer = (AtomicReferenceArray<Object>)lvElement(curr, nextOffset);
+        soElement(curr, nextOffset, null); // Avoid GC nepotism
+        return nextBuffer;
     }
     /**
      * {@inheritDoc}
@@ -138,7 +141,7 @@ public T poll() {
             soConsumerIndex(index + 1);// this ensures correctness on 32bit platforms
             return (T) e;
         } else if (isNextBuffer) {
-            return newBufferPoll(lvNext(buffer), index, mask);
+            return newBufferPoll(lvNextBufferAndUnlink(buffer, mask + 1), index, mask);
         }
 
         return null;
@@ -164,7 +167,7 @@ public T peek() {
         final int offset = calcWrappedOffset(index, mask);
         final Object e = lvElement(buffer, offset);// LoadLoad
         if (e == HAS_NEXT) {
-            return newBufferPeek(lvNext(buffer), index, mask);
+            return newBufferPeek(lvNextBufferAndUnlink(buffer, mask + 1), index, mask);
         }
 
         return (T) e;

src/test/java/io/reactivex/internal/operators/flowable/FlowableRefCountTest.java

@@ -42,7 +42,7 @@ public class FlowableRefCountTest {
     public void testRefCountAsync() {
         final AtomicInteger subscribeCount = new AtomicInteger();
         final AtomicInteger nextCount = new AtomicInteger();
-        Flowable<Long> r = Flowable.interval(0, 5, TimeUnit.MILLISECONDS)
+        Flowable<Long> r = Flowable.interval(0, 20, TimeUnit.MILLISECONDS)
                 .doOnSubscribe(new Consumer<Subscription>() {
                     @Override
                     public void accept(Subscription s) {
@@ -67,12 +67,27 @@ public void accept(Long l) {
 
         Disposable s2 = r.subscribe();
 
-        // give time to emit
         try {
-            Thread.sleep(52);
+            Thread.sleep(10);
         } catch (InterruptedException e) {
         }
 
+        for (;;) {
+            int a = nextCount.get();
+            int b = receivedCount.get();
+            if (a > 10 && a < 20 && a == b) {
+                break;
+            }
+            if (a >= 20) {
+                break;
+            }
+            try {
+                Thread.sleep(20);
+            } catch (InterruptedException e) {
+            }
+        }
+        // give time to emit
+
         // now unsubscribe
         s2.dispose(); // unsubscribe s2 first as we're counting in 1 and there can be a race between unsubscribe and one subscriber getting a value but not the other
         s1.dispose();

src/test/java/io/reactivex/internal/queue/SimpleQueueTest.java

@@ -20,7 +20,7 @@
 
 import static org.junit.Assert.*;
 
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.*;
 
 import org.junit.Test;
 
@@ -155,4 +155,23 @@ public void run() {
         t1.join();
         t2.join();
     }
+
+    @Test
+    public void spscLinkedArrayQueueNoNepotism() {
+        SpscLinkedArrayQueue<Integer> q = new SpscLinkedArrayQueue<Integer>(16);
+
+        AtomicReferenceArray<Object> ara = q.producerBuffer;
+
+        for (int i = 0; i < 20; i++) {
+            q.offer(i);
+        }
+
+        assertNotNull(ara.get(16));
+
+        for (int i = 0; i < 20; i++) {
+            assertEquals(i, q.poll().intValue());
+        }
+
+        assertNull(ara.get(16));
+    }
 }