package com.github.davidmoten.rx2;

import java.util.concurrent.atomic.AtomicReference;

import com.github.davidmoten.guavamini.Preconditions;

public class MpscQueue<T> {

    private final AtomicReference<Node<T>> head = new AtomicReference<Node<T>>();
    private final AtomicReference<Boolean> tailSet = new AtomicReference<Boolean>(false);
    private Node<T> tail;

    private static class Node<T> {
        T value;
        Node<T> next;

        public Node(T value) {
            this.value = value;
        }
    }

    public void push(T value) {
        Preconditions.checkNotNull(value, "cannot push a null value");
        // CAS loop
        Node<T> node = new Node<T>(value);
        while (true) {
            Node<T> h = head.get();
            if (head.compareAndSet(h, node)) {
                h.next = node;
                if (tailSet.compareAndSet(false, true)) {
                    tail = h;
                }
                break;
            }
        }
    }

    public T poll() {
        if (tail == null) {
            return null;
        } else {
            Node<T> tempTail = tail;
            Node<T> next = tempTail.next;
            if (next == null) {
                tail = null;
                tailSet.set(false);
            }
            T v = tempTail.value;
            // help gc including prevent nepotism
            tempTail.value = null;
            return v;
        }
    }

}

@@ -144,16 +144,22 @@
 %0Apublic 
+final 
 class Mp
@@ -339,24 +339,40 @@
 ean%3E(false);
+%0A%0A    // mutable
 %0A    private
@@ -406,16 +406,22 @@
  static 
+final 
 class No
@@ -428,16 +428,37 @@
 de%3CT%3E %7B%0A
+%0A        // mutable%0A%0A
         
@@ -497,23 +497,16 @@
         
-public 
 Node(T v
@@ -586,32 +586,162 @@
 push(T value) %7B%0A
+        // performs one volatile read and two CAS operations per call to this method%0A        // (under contention can be higher)%0A%0A
         Precondi
@@ -799,28 +799,8 @@
 %22);%0A
-        // CAS loop%0A
     
@@ -1029,24 +1029,341 @@
 e, true)) %7B%0A
+                    // don't mind that this is not instantly visible to poll() because%0A                    // in reactive operators the push is always followed by a drain that%0A                    // forces a poll which will be ordered after this assignment (so this%0A                    // assignment will be visible)%0A
             
@@ -1470,24 +1470,103 @@
  T poll() %7B%0A
+        // performs one volatile write only when reading the end of the queue%0A%0A
         if (
@@ -1647,20 +1647,16 @@
 %3CT%3E temp
-Tail
  = tail;
@@ -1687,20 +1687,16 @@
 t = temp
-Tail
 .next;%0A 
@@ -1787,24 +1787,74 @@
 set(false);%0A
+            %7D else %7B%0A                tail = next;%0A
             
@@ -1873,28 +1873,24 @@
   T v = temp
-Tail
 .value;%0A    
@@ -1955,12 +1955,8 @@
 temp
-Tail
 .val

test dev