--- jsr166/src/jsr166y/LinkedTransferQueue.java	2009/07/27 03:22:39	1.30
+++ jsr166/src/jsr166y/LinkedTransferQueue.java	2009/08/04 20:32:16	1.44
@@ -10,14 +10,15 @@ import java.util.concurrent.*;
 
 import java.util.AbstractQueue;
 import java.util.Collection;
+import java.util.ConcurrentModificationException;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
+import java.util.Queue;
 import java.util.concurrent.locks.LockSupport;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 
 /**
- * An unbounded {@linkplain TransferQueue} based on linked nodes.
+ * An unbounded {@link TransferQueue} based on linked nodes.
  * This queue orders elements FIFO (first-in-first-out) with respect
  * to any given producer.  The <em>head</em> of the queue is that
  * element that has been on the queue the longest time for some
@@ -115,17 +116,46 @@ public class LinkedTransferQueue<E> exte
             this.isData = isData;
         }
 
-        @SuppressWarnings("rawtypes")
-        static final AtomicReferenceFieldUpdater<Node, Node>
-            nextUpdater = AtomicReferenceFieldUpdater.newUpdater
-            (Node.class, Node.class, "next");
+        // Unsafe mechanics
+
+        private static final sun.misc.Unsafe UNSAFE = getUnsafe();
+        private static final long nextOffset =
+            objectFieldOffset(UNSAFE, "next", Node.class);
 
         final boolean casNext(Node<E> cmp, Node<E> val) {
-            return nextUpdater.compareAndSet(this, cmp, val);
+            return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
         }
 
         final void clearNext() {
-            nextUpdater.lazySet(this, this);
+            UNSAFE.putOrderedObject(this, nextOffset, this);
+        }
+
+        /**
+         * Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
+         * Replace with a simple call to Unsafe.getUnsafe when integrating
+         * into a jdk.
+         *
+         * @return a sun.misc.Unsafe
+         */
+        private static sun.misc.Unsafe getUnsafe() {
+            try {
+                return sun.misc.Unsafe.getUnsafe();
+            } catch (SecurityException se) {
+                try {
+                    return java.security.AccessController.doPrivileged
+                        (new java.security
+                         .PrivilegedExceptionAction<sun.misc.Unsafe>() {
+                            public sun.misc.Unsafe run() throws Exception {
+                                java.lang.reflect.Field f = sun.misc
+                                    .Unsafe.class.getDeclaredField("theUnsafe");
+                                f.setAccessible(true);
+                                return (sun.misc.Unsafe) f.get(null);
+                            }});
+                } catch (java.security.PrivilegedActionException e) {
+                    throw new RuntimeException("Could not initialize intrinsics",
+                                               e.getCause());
+                }
+            }
         }
 
         private static final long serialVersionUID = -3375979862319811754L;
@@ -189,7 +219,7 @@ public class LinkedTransferQueue<E> exte
             Node<E> t = tail.get();
             Node<E> h = head.get();
 
-            if (t != null && (t == h || t.isData == isData)) {
+            if (t == h || t.isData == isData) {
                 if (s == null)
                     s = new Node<E>(e, isData);
                 Node<E> last = t.next;
@@ -201,9 +231,7 @@ public class LinkedTransferQueue<E> exte
                     tail.compareAndSet(t, s);
                     return awaitFulfill(t, s, e, mode, nanos);
                 }
-            }
-
-            else if (h != null) {
+            } else {
                 Node<E> first = h.next;
                 if (t == tail.get() && first != null &&
                     advanceHead(h, first)) {
@@ -231,7 +259,7 @@ public class LinkedTransferQueue<E> exte
             Node<E> t = tail.get();
             Node<E> h = head.get();
 
-            if (t != null && (t == h || t.isData == isData)) {
+            if (t == h || t.isData == isData) {
                 Node<E> last = t.next;
                 if (t == tail.get()) {
                     if (last != null)
@@ -239,8 +267,7 @@ public class LinkedTransferQueue<E> exte
                     else
                         return null;
                 }
-            }
-            else if (h != null) {
+            } else {
                 Node<E> first = h.next;
                 if (t == tail.get() &&
                     first != null &&
@@ -264,7 +291,7 @@ public class LinkedTransferQueue<E> exte
      * @param e the comparison value for checking match
      * @param mode mode
      * @param nanos timeout value
-     * @return matched item, or s if cancelled
+     * @return matched item, or null if cancelled
      */
     private E awaitFulfill(Node<E> pred, Node<E> s, E e,
                            int mode, long nanos) {
@@ -302,9 +329,9 @@ public class LinkedTransferQueue<E> exte
             }
             if (spins < 0) {
                 Node<E> h = head.get(); // only spin if at head
-                spins = ((h != null && h.next == s) ?
-                         ((mode == TIMEOUT) ?
-                          maxTimedSpins : maxUntimedSpins) : 0);
+                spins = ((h.next != s) ? 0 :
+                         (mode == TIMEOUT) ? maxTimedSpins :
+                         maxUntimedSpins);
             }
             if (spins > 0)
                 --spins;
@@ -330,7 +357,7 @@ public class LinkedTransferQueue<E> exte
         for (;;) {
             Node<E> h = head.get();
             Node<E> first = h.next;
-            if (first != null && first.next == first) { // help advance
+            if (first != null && first.get() == first) { // help advance
                 advanceHead(h, first);
                 continue;
             }
@@ -443,29 +470,35 @@ public class LinkedTransferQueue<E> exte
     }
 
     /**
-     * @throws InterruptedException {@inheritDoc}
-     * @throws NullPointerException {@inheritDoc}
+     * Inserts the specified element at the tail of this queue.
+     * As the queue is unbounded, this method will never block.
+     *
+     * @throws NullPointerException if the specified element is null
      */
-    public void put(E e) throws InterruptedException {
-        if (e == null) throw new NullPointerException();
-        if (Thread.interrupted()) throw new InterruptedException();
-        xfer(e, NOWAIT, 0);
+    public void put(E e) {
+        offer(e);
     }
 
     /**
-     * @throws InterruptedException {@inheritDoc}
-     * @throws NullPointerException {@inheritDoc}
+     * Inserts the specified element at the tail of this queue.
+     * As the queue is unbounded, this method will never block or
+     * return {@code false}.
+     *
+     * @return {@code true} (as specified by
+     *  {@link BlockingQueue#offer(Object,long,TimeUnit) BlockingQueue.offer})
+     * @throws NullPointerException if the specified element is null
      */
-    public boolean offer(E e, long timeout, TimeUnit unit)
-        throws InterruptedException {
-        if (e == null) throw new NullPointerException();
-        if (Thread.interrupted()) throw new InterruptedException();
-        xfer(e, NOWAIT, 0);
-        return true;
+    public boolean offer(E e, long timeout, TimeUnit unit) {
+        return offer(e);
     }
 
     /**
-     * @throws NullPointerException {@inheritDoc}
+     * Inserts the specified element at the tail of this queue.
+     * As the queue is unbounded, this method will never return {@code false}.
+     *
+     * @return {@code true} (as specified by
+     *         {@link BlockingQueue#offer(Object) BlockingQueue.offer})
+     * @throws NullPointerException if the specified element is null
      */
     public boolean offer(E e) {
         if (e == null) throw new NullPointerException();
@@ -474,17 +507,42 @@ public class LinkedTransferQueue<E> exte
     }
 
     /**
-     * @throws NullPointerException {@inheritDoc}
+     * Inserts the specified element at the tail of this queue.
+     * As the queue is unbounded, this method will never throw
+     * {@link IllegalStateException} or return {@code false}.
+     *
+     * @return {@code true} (as specified by {@link Collection#add})
+     * @throws NullPointerException if the specified element is null
      */
     public boolean add(E e) {
+        return offer(e);
+    }
+
+    /**
+     * Transfers the element to a waiting consumer immediately, if possible.
+     *
+     * <p>More precisely, transfers the specified element immediately
+     * if there exists a consumer already waiting to receive it (in
+     * {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
+     * otherwise returning {@code false} without enqueuing the element.
+     *
+     * @throws NullPointerException if the specified element is null
+     */
+    public boolean tryTransfer(E e) {
         if (e == null) throw new NullPointerException();
-        xfer(e, NOWAIT, 0);
-        return true;
+        return fulfill(e) != null;
     }
 
     /**
-     * @throws InterruptedException {@inheritDoc}
-     * @throws NullPointerException {@inheritDoc}
+     * Transfers the element to a consumer, waiting if necessary to do so.
+     *
+     * <p>More precisely, transfers the specified element immediately
+     * if there exists a consumer already waiting to receive it (in
+     * {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
+     * else inserts the specified element at the tail of this queue
+     * and waits until the element is received by a consumer.
+     *
+     * @throws NullPointerException if the specified element is null
      */
     public void transfer(E e) throws InterruptedException {
         if (e == null) throw new NullPointerException();
@@ -495,8 +553,18 @@ public class LinkedTransferQueue<E> exte
     }
 
     /**
-     * @throws InterruptedException {@inheritDoc}
-     * @throws NullPointerException {@inheritDoc}
+     * Transfers the element to a consumer if it is possible to do so
+     * before the timeout elapses.
+     *
+     * <p>More precisely, transfers the specified element immediately
+     * if there exists a consumer already waiting to receive it (in
+     * {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
+     * else inserts the specified element at the tail of this queue
+     * and waits until the element is received by a consumer,
+     * returning {@code false} if the specified wait time elapses
+     * before the element can be transferred.
+     *
+     * @throws NullPointerException if the specified element is null
      */
     public boolean tryTransfer(E e, long timeout, TimeUnit unit)
         throws InterruptedException {
@@ -508,32 +576,18 @@ public class LinkedTransferQueue<E> exte
         throw new InterruptedException();
     }
 
-    /**
-     * @throws NullPointerException {@inheritDoc}
-     */
-    public boolean tryTransfer(E e) {
-        if (e == null) throw new NullPointerException();
-        return fulfill(e) != null;
-    }
-
-    /**
-     * @throws InterruptedException {@inheritDoc}
-     */
     public E take() throws InterruptedException {
-        Object e = xfer(null, WAIT, 0);
+        E e = xfer(null, WAIT, 0);
         if (e != null)
-            return (E) e;
+            return e;
         Thread.interrupted();
         throw new InterruptedException();
     }
 
-    /**
-     * @throws InterruptedException {@inheritDoc}
-     */
     public E poll(long timeout, TimeUnit unit) throws InterruptedException {
-        Object e = xfer(null, TIMEOUT, unit.toNanos(timeout));
+        E e = xfer(null, TIMEOUT, unit.toNanos(timeout));
         if (e != null || !Thread.interrupted())
-            return (E) e;
+            return e;
         throw new InterruptedException();
     }
 
@@ -586,28 +640,38 @@ public class LinkedTransferQueue<E> exte
         for (;;) {
             Node<E> t = tail.get();
             Node<E> h = head.get();
-            if (h != null && t != null) {
-                Node<E> last = t.next;
-                Node<E> first = h.next;
-                if (t == tail.get()) {
-                    if (last != null)
-                        tail.compareAndSet(t, last);
-                    else if (first != null) {
-                        Object x = first.get();
-                        if (x == first)
-                            advanceHead(h, first);
-                        else
-                            return h;
-                    }
+            Node<E> last = t.next;
+            Node<E> first = h.next;
+            if (t == tail.get()) {
+                if (last != null)
+                    tail.compareAndSet(t, last);
+                else if (first != null) {
+                    Object x = first.get();
+                    if (x == first)
+                        advanceHead(h, first);
                     else
                         return h;
                 }
+                else
+                    return h;
             }
             reclean();
         }
     }
 
-
+    /**
+     * Returns an iterator over the elements in this queue in proper
+     * sequence, from head to tail.
+     *
+     * <p>The returned iterator is a "weakly consistent" iterator that
+     * will never throw
+     * {@link ConcurrentModificationException ConcurrentModificationException},
+     * and guarantees to traverse elements as they existed upon
+     * construction of the iterator, and may (but is not guaranteed
+     * to) reflect any modifications subsequent to construction.
+     *
+     * @return an iterator over the elements in this queue in proper sequence
+     */
     public Iterator<E> iterator() {
         return new Itr();
     }
@@ -622,42 +686,39 @@ public class LinkedTransferQueue<E> exte
     class Itr implements Iterator<E> {
         Node<E> next;        // node to return next
         Node<E> pnext;       // predecessor of next
-        Node<E> snext;       // successor of next
         Node<E> curr;        // last returned node, for remove()
         Node<E> pcurr;       // predecessor of curr, for remove()
-        E nextItem;        // Cache of next item, once committed to in next
+        E nextItem;          // Cache of next item, once committed to in next
 
         Itr() {
-            findNext();
+            advance();
         }
 
         /**
-         * Ensures next points to next valid node, or null if none.
+         * Moves to next valid node and returns item to return for
+         * next(), or null if no such.
          */
-        void findNext() {
+        private E advance() {
+            pcurr = pnext;
+            curr = next;
+            E item = nextItem;
+
             for (;;) {
-                Node<E> pred = pnext;
-                Node<E> q = next;
-                if (pred == null || pred == q) {
-                    pred = traversalHead();
-                    q = pred.next;
-                }
-                if (q == null || !q.isData) {
+                pnext = (next == null) ? traversalHead() : next;
+                next = pnext.next;
+                if (next == pnext) {
                     next = null;
-                    return;
+                    continue;  // restart
                 }
-                Object x = q.get();
-                Node<E> s = q.next;
-                if (x != null && q != x && q != s) {
+                if (next == null)
+                    break;
+                Object x = next.get();
+                if (x != null && x != next) {
                     nextItem = (E) x;
-                    snext = s;
-                    pnext = pred;
-                    next = q;
-                    return;
+                    break;
                 }
-                pnext = q;
-                next = s;
             }
+            return item;
         }
 
         public boolean hasNext() {
@@ -665,14 +726,9 @@ public class LinkedTransferQueue<E> exte
         }
 
         public E next() {
-            if (next == null) throw new NoSuchElementException();
-            pcurr = pnext;
-            curr = next;
-            pnext = next;
-            next = snext;
-            E x = nextItem;
-            findNext();
-            return x;
+            if (next == null)
+                throw new NoSuchElementException();
+            return advance();
         }
 
         public void remove() {
@@ -701,6 +757,11 @@ public class LinkedTransferQueue<E> exte
         }
     }
 
+    /**
+     * Returns {@code true} if this queue contains no elements.
+     *
+     * @return {@code true} if this queue contains no elements
+     */
     public boolean isEmpty() {
         for (;;) {
             Node<E> h = traversalHead();
@@ -742,34 +803,57 @@ public class LinkedTransferQueue<E> exte
      * @return the number of elements in this queue
      */
     public int size() {
-        int count = 0;
-        Node<E> h = traversalHead();
-        for (Node<E> p = h.next; p != null && p.isData; p = p.next) {
-            Object x = p.get();
-            if (x != null && x != p) {
-                if (++count == Integer.MAX_VALUE) // saturated
+        for (;;) {
+            int count = 0;
+            Node<E> pred = traversalHead();
+            for (;;) {
+                Node<E> q = pred.next;
+                if (q == pred) // restart
                     break;
+                if (q == null || !q.isData)
+                    return count;
+                Object x = q.get();
+                if (x != null && x != q) {
+                    if (++count == Integer.MAX_VALUE) // saturated
+                        return count;
+                }
+                pred = q;
             }
         }
-        return count;
     }
 
     public int getWaitingConsumerCount() {
-        int count = 0;
-        Node<E> h = traversalHead();
-        for (Node<E> p = h.next; p != null && !p.isData; p = p.next) {
-            if (p.get() == null) {
-                if (++count == Integer.MAX_VALUE)
+        // converse of size -- count valid non-data nodes
+        for (;;) {
+            int count = 0;
+            Node<E> pred = traversalHead();
+            for (;;) {
+                Node<E> q = pred.next;
+                if (q == pred) // restart
                     break;
+                if (q == null || q.isData)
+                    return count;
+                Object x = q.get();
+                if (x == null) {
+                    if (++count == Integer.MAX_VALUE) // saturated
+                        return count;
+                }
+                pred = q;
             }
         }
-        return count;
-    }
-
-    public int remainingCapacity() {
-        return Integer.MAX_VALUE;
     }
 
+    /**
+     * Removes a single instance of the specified element from this queue,
+     * if it is present.  More formally, removes an element {@code e} such
+     * that {@code o.equals(e)}, if this queue contains one or more such
+     * elements.
+     * Returns {@code true} if this queue contained the specified element
+     * (or equivalently, if this queue changed as a result of the call).
+     *
+     * @param o element to be removed from this queue, if present
+     * @return {@code true} if this queue changed as a result of the call
+     */
     public boolean remove(Object o) {
         if (o == null)
             return false;
@@ -777,10 +861,10 @@ public class LinkedTransferQueue<E> exte
             Node<E> pred = traversalHead();
             for (;;) {
                 Node<E> q = pred.next;
-                if (q == null || !q.isData)
-                    return false;
                 if (q == pred) // restart
                     break;
+                if (q == null || !q.isData)
+                    return false;
                 Object x = q.get();
                 if (x != null && x != q && o.equals(x) &&
                     q.compareAndSet(x, q)) {
@@ -793,6 +877,17 @@ public class LinkedTransferQueue<E> exte
     }
 
     /**
+     * Always returns {@code Integer.MAX_VALUE} because a
+     * {@code LinkedTransferQueue} is not capacity constrained.
+     *
+     * @return {@code Integer.MAX_VALUE} (as specified by
+     *         {@link BlockingQueue#remainingCapacity()})
+     */
+    public int remainingCapacity() {
+        return Integer.MAX_VALUE;
+    }
+
+    /**
      * Save the state to a stream (that is, serialize it).
      *
      * @serialData All of the elements (each an {@code E}) in
@@ -842,13 +937,15 @@ public class LinkedTransferQueue<E> exte
 
     private static final sun.misc.Unsafe UNSAFE = getUnsafe();
     private static final long headOffset =
-        objectFieldOffset("head", LinkedTransferQueue.class);
+        objectFieldOffset(UNSAFE, "head", LinkedTransferQueue.class);
     private static final long tailOffset =
-        objectFieldOffset("tail", LinkedTransferQueue.class);
+        objectFieldOffset(UNSAFE, "tail", LinkedTransferQueue.class);
     private static final long cleanMeOffset =
-        objectFieldOffset("cleanMe", LinkedTransferQueue.class);
+        objectFieldOffset(UNSAFE, "cleanMe", LinkedTransferQueue.class);
+
 
-    private static long objectFieldOffset(String field, Class<?> klazz) {
+    static long objectFieldOffset(sun.misc.Unsafe UNSAFE,
+                                  String field, Class<?> klazz) {
         try {
             return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
         } catch (NoSuchFieldException e) {