1 /*

     2  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     3  *

     4  * This code is free software; you can redistribute it and/or modify it

     5  * under the terms of the GNU General Public License version 2 only, as

     6  * published by the Free Software Foundation.  Sun designates this

     7  * particular file as subject to the "Classpath" exception as provided

     8  * by Sun in the LICENSE file that accompanied this code.

     9  *

    10  * This code is distributed in the hope that it will be useful, but WITHOUT

    11  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    12  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    13  * version 2 for more details (a copy is included in the LICENSE file that

    14  * accompanied this code).

    15  *

    16  * You should have received a copy of the GNU General Public License version

    17  * 2 along with this work; if not, write to the Free Software Foundation,

    18  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    19  *

    20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,

    21  * CA 95054 USA or visit www.sun.com if you need additional information or

    22  * have any questions.

    23  */

    24 

    25 /*

    26  * This file is available under and governed by the GNU General Public

    27  * License version 2 only, as published by the Free Software Foundation.

    28  * However, the following notice accompanied the original version of this

    29  * file:

    30  *

    31  * Written by Doug Lea with assistance from members of JCP JSR-166

    32  * Expert Group and released to the public domain, as explained at

    33  * http://creativecommons.org/licenses/publicdomain

    34  */

    35 

    36 package java.util.concurrent;

    37 import java.util.*;

    38 import java.util.concurrent.atomic.*;

    39 

    40 

    41 /**

    42  * An unbounded thread-safe {@linkplain Queue queue} based on linked nodes.

    43  * This queue orders elements FIFO (first-in-first-out).

    44  * The <em>head</em> of the queue is that element that has been on the

    45  * queue the longest time.

    46  * The <em>tail</em> of the queue is that element that has been on the

    47  * queue the shortest time. New elements

    48  * are inserted at the tail of the queue, and the queue retrieval

    49  * operations obtain elements at the head of the queue.

    50  * A <tt>ConcurrentLinkedQueue</tt> is an appropriate choice when

    51  * many threads will share access to a common collection.

    52  * This queue does not permit <tt>null</tt> elements.

    53  *

    54  * <p>This implementation employs an efficient &quot;wait-free&quot;

    55  * algorithm based on one described in <a

    56  * href="http://www.cs.rochester.edu/u/michael/PODC96.html"> Simple,

    57  * Fast, and Practical Non-Blocking and Blocking Concurrent Queue

    58  * Algorithms</a> by Maged M. Michael and Michael L. Scott.

    59  *

    60  * <p>Beware that, unlike in most collections, the <tt>size</tt> method

    61  * is <em>NOT</em> a constant-time operation. Because of the

    62  * asynchronous nature of these queues, determining the current number

    63  * of elements requires a traversal of the elements.

    64  *

    65  * <p>This class and its iterator implement all of the

    66  * <em>optional</em> methods of the {@link Collection} and {@link

    67  * Iterator} interfaces.

    68  *

    69  * <p>Memory consistency effects: As with other concurrent

    70  * collections, actions in a thread prior to placing an object into a

    71  * {@code ConcurrentLinkedQueue}

    72  * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>

    73  * actions subsequent to the access or removal of that element from

    74  * the {@code ConcurrentLinkedQueue} in another thread.

    75  *

    76  * <p>This class is a member of the

    77  * <a href="{@docRoot}/../technotes/guides/collections/index.html">

    78  * Java Collections Framework</a>.

    79  *

    80  * @since 1.5

    81  * @author Doug Lea

    82  * @param <E> the type of elements held in this collection

    83  *

    84  */

    85 public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>

    86         implements Queue<E>, java.io.Serializable {

    87     private static final long serialVersionUID = 196745693267521676L;

    88 

    89     /*

    90      * This is a straight adaptation of Michael & Scott algorithm.

    91      * For explanation, read the paper.  The only (minor) algorithmic

    92      * difference is that this version supports lazy deletion of

    93      * internal nodes (method remove(Object)) -- remove CAS'es item

    94      * fields to null. The normal queue operations unlink but then

    95      * pass over nodes with null item fields. Similarly, iteration

    96      * methods ignore those with nulls.

    97      *

    98      * Also note that like most non-blocking algorithms in this

    99      * package, this implementation relies on the fact that in garbage

   100      * collected systems, there is no possibility of ABA problems due

   101      * to recycled nodes, so there is no need to use "counted

   102      * pointers" or related techniques seen in versions used in

   103      * non-GC'ed settings.

   104      */

   105 

   106     private static class Node<E> {

   107         private volatile E item;

   108         private volatile Node<E> next;

   109 

   110         private static final

   111             AtomicReferenceFieldUpdater<Node, Node>

   112             nextUpdater =

   113             AtomicReferenceFieldUpdater.newUpdater

   114             (Node.class, Node.class, "next");

   115         private static final

   116             AtomicReferenceFieldUpdater<Node, Object>

   117             itemUpdater =

   118             AtomicReferenceFieldUpdater.newUpdater

   119             (Node.class, Object.class, "item");

   120 

   121         Node(E x) { item = x; }

   122 

   123         Node(E x, Node<E> n) { item = x; next = n; }

   124 

   125         E getItem() {

   126             return item;

   127         }

   128 

   129         boolean casItem(E cmp, E val) {

   130             return itemUpdater.compareAndSet(this, cmp, val);

   131         }

   132 

   133         void setItem(E val) {

   134             itemUpdater.set(this, val);

   135         }

   136 

   137         Node<E> getNext() {

   138             return next;

   139         }

   140 

   141         boolean casNext(Node<E> cmp, Node<E> val) {

   142             return nextUpdater.compareAndSet(this, cmp, val);

   143         }

   144 

   145         void setNext(Node<E> val) {

   146             nextUpdater.set(this, val);

   147         }

   148 

   149     }

   150 

   151     private static final

   152         AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>

   153         tailUpdater =

   154         AtomicReferenceFieldUpdater.newUpdater

   155         (ConcurrentLinkedQueue.class, Node.class, "tail");

   156     private static final

   157         AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>

   158         headUpdater =

   159         AtomicReferenceFieldUpdater.newUpdater

   160         (ConcurrentLinkedQueue.class,  Node.class, "head");

   161 

   162     private boolean casTail(Node<E> cmp, Node<E> val) {

   163         return tailUpdater.compareAndSet(this, cmp, val);

   164     }

   165 

   166     private boolean casHead(Node<E> cmp, Node<E> val) {

   167         return headUpdater.compareAndSet(this, cmp, val);

   168     }

   169 

   170 

   171     /**

   172      * Pointer to header node, initialized to a dummy node.  The first

   173      * actual node is at head.getNext().

   174      */

   175     private transient volatile Node<E> head = new Node<E>(null, null);

   176 

   177     /** Pointer to last node on list **/

   178     private transient volatile Node<E> tail = head;

   179 

   180 

   181     /**

   182      * Creates a <tt>ConcurrentLinkedQueue</tt> that is initially empty.

   183      */

   184     public ConcurrentLinkedQueue() {}

   185 

   186     /**

   187      * Creates a <tt>ConcurrentLinkedQueue</tt>

   188      * initially containing the elements of the given collection,

   189      * added in traversal order of the collection's iterator.

   190      * @param c the collection of elements to initially contain

   191      * @throws NullPointerException if the specified collection or any

   192      *         of its elements are null

   193      */

   194     public ConcurrentLinkedQueue(Collection<? extends E> c) {

   195         for (Iterator<? extends E> it = c.iterator(); it.hasNext();)

   196             add(it.next());

   197     }

   198 

   199     // Have to override just to update the javadoc

   200 

   201     /**

   202      * Inserts the specified element at the tail of this queue.

   203      *

   204      * @return <tt>true</tt> (as specified by {@link Collection#add})

   205      * @throws NullPointerException if the specified element is null

   206      */

   207     public boolean add(E e) {

   208         return offer(e);

   209     }

   210 

   211     /**

   212      * Inserts the specified element at the tail of this queue.

   213      *

   214      * @return <tt>true</tt> (as specified by {@link Queue#offer})

   215      * @throws NullPointerException if the specified element is null

   216      */

   217     public boolean offer(E e) {

   218         if (e == null) throw new NullPointerException();

   219         Node<E> n = new Node<E>(e, null);

   220         for (;;) {

   221             Node<E> t = tail;

   222             Node<E> s = t.getNext();

   223             if (t == tail) {

   224                 if (s == null) {

   225                     if (t.casNext(s, n)) {

   226                         casTail(t, n);

   227                         return true;

   228                     }

   229                 } else {

   230                     casTail(t, s);

   231                 }

   232             }

   233         }

   234     }

   235 

   236     public E poll() {

   237         for (;;) {

   238             Node<E> h = head;

   239             Node<E> t = tail;

   240             Node<E> first = h.getNext();

   241             if (h == head) {

   242                 if (h == t) {

   243                     if (first == null)

   244                         return null;

   245                     else

   246                         casTail(t, first);

   247                 } else if (casHead(h, first)) {

   248                     E item = first.getItem();

   249                     if (item != null) {

   250                         first.setItem(null);

   251                         return item;

   252                     }

   253                     // else skip over deleted item, continue loop,

   254                 }

   255             }

   256         }

   257     }

   258 

   259     public E peek() { // same as poll except don't remove item

   260         for (;;) {

   261             Node<E> h = head;

   262             Node<E> t = tail;

   263             Node<E> first = h.getNext();

   264             if (h == head) {

   265                 if (h == t) {

   266                     if (first == null)

   267                         return null;

   268                     else

   269                         casTail(t, first);

   270                 } else {

   271                     E item = first.getItem();

   272                     if (item != null)

   273                         return item;

   274                     else // remove deleted node and continue

   275                         casHead(h, first);

   276                 }

   277             }

   278         }

   279     }

   280 

   281     /**

   282      * Returns the first actual (non-header) node on list.  This is yet

   283      * another variant of poll/peek; here returning out the first

   284      * node, not element (so we cannot collapse with peek() without

   285      * introducing race.)

   286      */

   287     Node<E> first() {

   288         for (;;) {

   289             Node<E> h = head;

   290             Node<E> t = tail;

   291             Node<E> first = h.getNext();

   292             if (h == head) {

   293                 if (h == t) {

   294                     if (first == null)

   295                         return null;

   296                     else

   297                         casTail(t, first);

   298                 } else {

   299                     if (first.getItem() != null)

   300                         return first;

   301                     else // remove deleted node and continue

   302                         casHead(h, first);

   303                 }

   304             }

   305         }

   306     }

   307 

   308 

   309     /**

   310      * Returns <tt>true</tt> if this queue contains no elements.

   311      *

   312      * @return <tt>true</tt> if this queue contains no elements

   313      */

   314     public boolean isEmpty() {

   315         return first() == null;

   316     }

   317 

   318     /**

   319      * Returns the number of elements in this queue.  If this queue

   320      * contains more than <tt>Integer.MAX_VALUE</tt> elements, returns

   321      * <tt>Integer.MAX_VALUE</tt>.

   322      *

   323      * <p>Beware that, unlike in most collections, this method is

   324      * <em>NOT</em> a constant-time operation. Because of the

   325      * asynchronous nature of these queues, determining the current

   326      * number of elements requires an O(n) traversal.

   327      *

   328      * @return the number of elements in this queue

   329      */

   330     public int size() {

   331         int count = 0;

   332         for (Node<E> p = first(); p != null; p = p.getNext()) {

   333             if (p.getItem() != null) {

   334                 // Collections.size() spec says to max out

   335                 if (++count == Integer.MAX_VALUE)

   336                     break;

   337             }

   338         }

   339         return count;

   340     }

   341 

   342     /**

   343      * Returns <tt>true</tt> if this queue contains the specified element.

   344      * More formally, returns <tt>true</tt> if and only if this queue contains

   345      * at least one element <tt>e</tt> such that <tt>o.equals(e)</tt>.

   346      *

   347      * @param o object to be checked for containment in this queue

   348      * @return <tt>true</tt> if this queue contains the specified element

   349      */

   350     public boolean contains(Object o) {

   351         if (o == null) return false;

   352         for (Node<E> p = first(); p != null; p = p.getNext()) {

   353             E item = p.getItem();

   354             if (item != null &&

   355                 o.equals(item))

   356                 return true;

   357         }

   358         return false;

   359     }

   360 

   361     /**

   362      * Removes a single instance of the specified element from this queue,

   363      * if it is present.  More formally, removes an element <tt>e</tt> such

   364      * that <tt>o.equals(e)</tt>, if this queue contains one or more such

   365      * elements.

   366      * Returns <tt>true</tt> if this queue contained the specified element

   367      * (or equivalently, if this queue changed as a result of the call).

   368      *

   369      * @param o element to be removed from this queue, if present

   370      * @return <tt>true</tt> if this queue changed as a result of the call

   371      */

   372     public boolean remove(Object o) {

   373         if (o == null) return false;

   374         for (Node<E> p = first(); p != null; p = p.getNext()) {

   375             E item = p.getItem();

   376             if (item != null &&

   377                 o.equals(item) &&

   378                 p.casItem(item, null))

   379                 return true;

   380         }

   381         return false;

   382     }

   383 

   384     /**

   385      * Returns an array containing all of the elements in this queue, in

   386      * proper sequence.

   387      *

   388      * <p>The returned array will be "safe" in that no references to it are

   389      * maintained by this queue.  (In other words, this method must allocate

   390      * a new array).  The caller is thus free to modify the returned array.

   391      *

   392      * <p>This method acts as bridge between array-based and collection-based

   393      * APIs.

   394      *

   395      * @return an array containing all of the elements in this queue

   396      */

   397     public Object[] toArray() {

   398         // Use ArrayList to deal with resizing.

   399         ArrayList<E> al = new ArrayList<E>();

   400         for (Node<E> p = first(); p != null; p = p.getNext()) {

   401             E item = p.getItem();

   402             if (item != null)

   403                 al.add(item);

   404         }

   405         return al.toArray();

   406     }

   407 

   408     /**

   409      * Returns an array containing all of the elements in this queue, in

   410      * proper sequence; the runtime type of the returned array is that of

   411      * the specified array.  If the queue fits in the specified array, it

   412      * is returned therein.  Otherwise, a new array is allocated with the

   413      * runtime type of the specified array and the size of this queue.

   414      *

   415      * <p>If this queue fits in the specified array with room to spare

   416      * (i.e., the array has more elements than this queue), the element in

   417      * the array immediately following the end of the queue is set to

   418      * <tt>null</tt>.

   419      *

   420      * <p>Like the {@link #toArray()} method, this method acts as bridge between

   421      * array-based and collection-based APIs.  Further, this method allows

   422      * precise control over the runtime type of the output array, and may,

   423      * under certain circumstances, be used to save allocation costs.

   424      *

   425      * <p>Suppose <tt>x</tt> is a queue known to contain only strings.

   426      * The following code can be used to dump the queue into a newly

   427      * allocated array of <tt>String</tt>:

   428      *

   429      * <pre>

   430      *     String[] y = x.toArray(new String[0]);</pre>

   431      *

   432      * Note that <tt>toArray(new Object[0])</tt> is identical in function to

   433      * <tt>toArray()</tt>.

   434      *

   435      * @param a the array into which the elements of the queue are to

   436      *          be stored, if it is big enough; otherwise, a new array of the

   437      *          same runtime type is allocated for this purpose

   438      * @return an array containing all of the elements in this queue

   439      * @throws ArrayStoreException if the runtime type of the specified array

   440      *         is not a supertype of the runtime type of every element in

   441      *         this queue

   442      * @throws NullPointerException if the specified array is null

   443      */

   444     public <T> T[] toArray(T[] a) {

   445         // try to use sent-in array

   446         int k = 0;

   447         Node<E> p;

   448         for (p = first(); p != null && k < a.length; p = p.getNext()) {

   449             E item = p.getItem();

   450             if (item != null)

   451                 a[k++] = (T)item;

   452         }

   453         if (p == null) {

   454             if (k < a.length)

   455                 a[k] = null;

   456             return a;

   457         }

   458 

   459         // If won't fit, use ArrayList version

   460         ArrayList<E> al = new ArrayList<E>();

   461         for (Node<E> q = first(); q != null; q = q.getNext()) {

   462             E item = q.getItem();

   463             if (item != null)

   464                 al.add(item);

   465         }

   466         return al.toArray(a);

   467     }

   468 

   469     /**

   470      * Returns an iterator over the elements in this queue in proper sequence.

   471      * The returned iterator is a "weakly consistent" iterator that

   472      * will never throw {@link ConcurrentModificationException},

   473      * and guarantees to traverse elements as they existed upon

   474      * construction of the iterator, and may (but is not guaranteed to)

   475      * reflect any modifications subsequent to construction.

   476      *

   477      * @return an iterator over the elements in this queue in proper sequence

   478      */

   479     public Iterator<E> iterator() {

   480         return new Itr();

   481     }

   482 

   483     private class Itr implements Iterator<E> {

   484         /**

   485          * Next node to return item for.

   486          */

   487         private Node<E> nextNode;

   488 

   489         /**

   490          * nextItem holds on to item fields because once we claim

   491          * that an element exists in hasNext(), we must return it in

   492          * the following next() call even if it was in the process of

   493          * being removed when hasNext() was called.

   494          */

   495         private E nextItem;

   496 

   497         /**

   498          * Node of the last returned item, to support remove.

   499          */

   500         private Node<E> lastRet;

   501 

   502         Itr() {

   503             advance();

   504         }

   505 

   506         /**

   507          * Moves to next valid node and returns item to return for

   508          * next(), or null if no such.

   509          */

   510         private E advance() {

   511             lastRet = nextNode;

   512             E x = nextItem;

   513 

   514             Node<E> p = (nextNode == null)? first() : nextNode.getNext();

   515             for (;;) {

   516                 if (p == null) {

   517                     nextNode = null;

   518                     nextItem = null;

   519                     return x;

   520                 }

   521                 E item = p.getItem();

   522                 if (item != null) {

   523                     nextNode = p;

   524                     nextItem = item;

   525                     return x;

   526                 } else // skip over nulls

   527                     p = p.getNext();

   528             }

   529         }

   530 

   531         public boolean hasNext() {

   532             return nextNode != null;

   533         }

   534 

   535         public E next() {

   536             if (nextNode == null) throw new NoSuchElementException();

   537             return advance();

   538         }

   539 

   540         public void remove() {

   541             Node<E> l = lastRet;

   542             if (l == null) throw new IllegalStateException();

   543             // rely on a future traversal to relink.

   544             l.setItem(null);

   545             lastRet = null;

   546         }

   547     }

   548 

   549     /**

   550      * Save the state to a stream (that is, serialize it).

   551      *

   552      * @serialData All of the elements (each an <tt>E</tt>) in

   553      * the proper order, followed by a null

   554      * @param s the stream

   555      */

   556     private void writeObject(java.io.ObjectOutputStream s)

   557         throws java.io.IOException {

   558 

   559         // Write out any hidden stuff

   560         s.defaultWriteObject();

   561 

   562         // Write out all elements in the proper order.

   563         for (Node<E> p = first(); p != null; p = p.getNext()) {

   564             Object item = p.getItem();

   565             if (item != null)

   566                 s.writeObject(item);

   567         }

   568 

   569         // Use trailing null as sentinel

   570         s.writeObject(null);

   571     }

   572 

   573     /**

   574      * Reconstitute the Queue instance from a stream (that is,

   575      * deserialize it).

   576      * @param s the stream

   577      */

   578     private void readObject(java.io.ObjectInputStream s)

   579         throws java.io.IOException, ClassNotFoundException {

   580         // Read in capacity, and any hidden stuff

   581         s.defaultReadObject();

   582         head = new Node<E>(null, null);

   583         tail = head;

   584         // Read in all elements and place in queue

   585         for (;;) {

   586             E item = (E)s.readObject();

   587             if (item == null)

   588                 break;

   589             else

   590                 offer(item);

   591         }

   592     }

   593 

   594 }