util/concurrent/LinkedQueue.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain. Use, modify, and
 * redistribute this code in any way without acknowledgement.
 */

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


/**
 * An unbounded thread-safe queue based on linked nodes.  LinkedQueues
 * are an especially good choice when many threads will share access
 * to a common  queue. 
 *
 * <p> This implementation employs an efficient "wait-free" algorithm
 * based on one described in <a
 * href="http://www.cs.rochester.edu/u/michael/PODC96.html"> Simple,
 * Fast, and Practical Non-Blocking and Blocking Concurrent Queue
 * Algorithms</a> by Maged M. Michael and Michael L. Scott.)
 *
 * Beware that, unlike in most collections, the <tt>size</tt> method
 * is <em>NOT</em> a constant-time operation. Because of the
 * asynchronous nature of these queues, determining the current number
 * of elements requires an O(n) traversal.
 * 
 **/
public class LinkedQueue<E> extends AbstractQueue<E>
        implements Queue<E>, java.io.Serializable {

    /*
     * This is a straight adaptation of Michael & Scott algorithm.
     * For explanation, read the paper.  The only (minor) algorithmic
     * difference is that this version supports lazy deletion of
     * internal nodes (method remove(Object)) -- remove CAS'es item
     * fields to null. The normal queue operations unlink but then
     * pass over nodes with null item fields. Similarly, iteration
     * methods ignore those with nulls.
     */

    // Atomics support

    private final static AtomicReferenceFieldUpdater<LinkedQueue, AtomicLinkedNode> tailUpdater = new AtomicReferenceFieldUpdater<LinkedQueue, AtomicLinkedNode>(new LinkedQueue[0], new AtomicLinkedNode[0], "tail");
    private final static AtomicReferenceFieldUpdater<LinkedQueue, AtomicLinkedNode> headUpdater = new AtomicReferenceFieldUpdater<LinkedQueue, AtomicLinkedNode>(new LinkedQueue[0], new AtomicLinkedNode[0], "head");

    private boolean casTail(AtomicLinkedNode cmp, AtomicLinkedNode val) {
        return tailUpdater.compareAndSet(this, cmp, val);
    }

    private boolean casHead(AtomicLinkedNode cmp, AtomicLinkedNode val) {
        return headUpdater.compareAndSet(this, cmp, val);
    }


    /** 
     * Pointer to header node, initialized to a dummy node.  The first
     * actual node is at head.getNext().
     */
    private transient volatile AtomicLinkedNode head = new AtomicLinkedNode(null, null);

    /** Pointer to last node on list **/
    private transient volatile AtomicLinkedNode tail = head;


    /**
     * Creates an initially empty LinkedQueue.
     */
    public LinkedQueue() {}

    /**
     * Creates a LinkedQueue initially holding the elements
     * of the given collection. The elements are added in 
     * iterator traversal order.
     *
     * @param initialElements the collections whose elements are to be added.
     */
    public LinkedQueue(Collection<E> initialElements) {
        for (Iterator<E> it = initialElements.iterator(); it.hasNext();) 
            add(it.next());
    }

    public boolean offer(E x) {
        if (x == null) throw new IllegalArgumentException();
        AtomicLinkedNode n = new AtomicLinkedNode(x, null);
        for(;;) {
            AtomicLinkedNode t = tail;
            AtomicLinkedNode s = t.getNext();
            if (t == tail) {
                if (s == null) {
                    if (t.casNext(s, n)) {
                        casTail(t, n); 
                        return true;
                    }
                }
                else {
                    casTail(t, s);
                }
            }
        }
    }

    public E poll() {
        for (;;) {
            AtomicLinkedNode h = head;
            AtomicLinkedNode t = tail;
            AtomicLinkedNode first = h.getNext();
            if (h == head) {
                if (h == t) {
                    if (first == null)
                        return null;
                    else
                        casTail(t, first);
                }
                else if (casHead(h, first)) {
                    E item = (E)first.getItem();
                    if (item != null) {
                        first.setItem(null);
                        return item;
                    }
                    // else skip over deleted item, continue loop, 
                }
            }
        }
    }

    public E peek() { // same as poll except don't remove item
        for (;;) {
            AtomicLinkedNode h = head;
            AtomicLinkedNode t = tail;
            AtomicLinkedNode first = h.getNext();
            if (h == head) {
                if (h == t) {
                    if (first == null)
                        return null;
                    else
                        casTail(t, first);
                }
                else {
                    E item = (E)first.getItem();
                    if (item != null)
                        return item;
                    else // remove deleted node and continue
                        casHead(h, first);
                }
            }
        }
    }

    /**
     * Return the first actual (non-header) node on list.  This is yet
     * another variant of poll/peek; here returning out the first
     * node, not element (so we cannot collapse with peek() without
     * introducing race.)
     */
    AtomicLinkedNode first() { 
        for (;;) {
            AtomicLinkedNode h = head;
            AtomicLinkedNode t = tail;
            AtomicLinkedNode first = h.getNext();
            if (h == head) {
                if (h == t) {
                    if (first == null)
                        return null;
                    else
                        casTail(t, first);
                }
                else {
                    if (first.getItem() != null)
                        return first;
                    else // remove deleted node and continue
                        casHead(h, first);
                }
            }
        }
    }


    public boolean isEmpty() {
        return first() == null;
    }

    /**
     * Returns the number of elements in this collection. 
     * 
     * Beware that, unlike in most collection, this method> is
     * <em>NOT</em> a constant-time operation. Because of the
     * asynchronous nature of these queues, determining the current
     * number of elements requires an O(n) traversal.
     * @return the number of elements in this collection
     */ 
    public int size() {
        int count = 0;
        for (AtomicLinkedNode p = first(); p != null; p = p.getNext()) {
            if (p.getItem() != null)
                ++count;
        }
        return count;
    }

    public boolean contains(Object x) {
        if (x == null) return false;
        for (AtomicLinkedNode p = first(); p != null; p = p.getNext()) {
            Object item = p.getItem();
            if (item != null && 
                x.equals(item))
                return true;
        }
        return false;
    }

    public boolean remove(Object x) {
        if (x == null) return false;
        for (AtomicLinkedNode p = first(); p != null; p = p.getNext()) {
            Object item = p.getItem();
            if (item != null && 
                x.equals(item) &&
                p.casItem(item, null))
                return true;
        }
        return false;
    }
    
    public Object[] toArray() {
        // Use ArrayList to deal with resizing.
        ArrayList al = new ArrayList();
        for (AtomicLinkedNode p = first(); p != null; p = p.getNext()) {
            Object item = p.getItem();
            if (item != null)
                al.add(item);
        }
        return al.toArray();
    }

    public <T> T[] toArray(T[] a) {
        // try to use sent-in array
        int k = 0;
        AtomicLinkedNode p;
        for (p = first(); p != null && k < a.length; p = p.getNext()) {
            Object item = p.getItem();
            if (item != null)
                a[k++] = (T)item;
        }
        if (p == null) {
            if (k < a.length)
                a[k] = null;
            return a;
        }

        // If won't fit, use ArrayList version
        ArrayList al = new ArrayList();
        for (AtomicLinkedNode q = first(); q != null; q = q.getNext()) {
            Object item = q.getItem();
            if (item != null)
                al.add(item);
        }
        return (T[])al.toArray(a);
    }

    public Iterator<E> iterator() {
        return new Itr();
    }

    private class Itr implements Iterator<E> {
        /**
         * Next node to return item for.
         */
        private AtomicLinkedNode nextNode;

        /** 
         * nextItem holds on to item fields because once we claim
         * that an element exists in hasNext(), we must return it in
         * the following next() call even if it was in the process of
         * being removed when hasNext() was called.
         **/
        private E nextItem;

        /**
         * Node of the last returned item, to support remove.
         */
        private AtomicLinkedNode lastRet;

        Itr() { 
            advance();
        }
        
        /**
         * Move to next valid node. 
         * Return item to return for next(), or null if no such.
         */
        private E advance() { 
            lastRet = nextNode;
            E x = (E)nextItem;

            AtomicLinkedNode p = (nextNode == null)? first() : nextNode.getNext();
            for (;;) {
                if (p == null) {
                    nextNode = null;
                    nextItem = null;
                    return x;
                }
                E item = (E)p.getItem();
                if (item != null) {
                    nextNode = p;
                    nextItem = item;
                    return x;
                }
                else // skip over nulls 
                    p = p.getNext();
            }
        }
        
        public boolean hasNext() {
            return nextNode != null;
        }
        
        public E next() {
            if (nextNode == null) throw new NoSuchElementException();
            return advance();
        }
        
        public void remove() {
            AtomicLinkedNode l = lastRet;
            if (l == null) throw new IllegalStateException();
            // rely on a future traversal to relink.
            l.setItem(null);
            lastRet = null;
        }
    }

    /**
     * Save the state to a stream (that is, serialize it).
     *
     * @serialData All of the elements (each an <tt>E</tt>) in
     * the proper order, followed by a null
     */
    private void writeObject(java.io.ObjectOutputStream s)
        throws java.io.IOException {

        // Write out any hidden stuff
        s.defaultWriteObject();
        
        // Write out all elements in the proper order.
        for (AtomicLinkedNode p = first(); p != null; p = p.getNext()) {
            Object item = p.getItem();
            if (item != null)
                s.writeObject(item);
        }

        // Use trailing null as sentinel
        s.writeObject(null);
    }

    /**
     * Reconstitute the Queue instance from a stream (that is,
     * deserialize it).
     */
    private void readObject(java.io.ObjectInputStream s)
        throws java.io.IOException, ClassNotFoundException {
        // Read in capacity, and any hidden stuff
        s.defaultReadObject();

        // Read in all elements and place in queue
        for (;;) {
            E item = (E)s.readObject();
            if (item == null)
                break;
            add(item);
        }
    }

}
Revision:	1.5
Committed:	Sun Jun 22 14:27:18 2003 UTC (20 years, 11 months ago) by dl
Branch:	MAIN
Changes since 1.4:	+116 -93 lines
Log Message:	Added LinkedStack, plus package-private AtomicLinkedNode also used by LinkedQueue
#	Content
1	/*
2	* Written by Doug Lea with assistance from members of JCP JSR-166
3	* Expert Group and released to the public domain. Use, modify, and
4	* redistribute this code in any way without acknowledgement.
5	*/
6
7	package java.util.concurrent;
8	import java.util.*;
9	import java.util.concurrent.atomic.*;
10
11
12	/**
13	* An unbounded thread-safe queue based on linked nodes. LinkedQueues
14	* are an especially good choice when many threads will share access
15	* to a common queue.
16	*
17	* <p> This implementation employs an efficient "wait-free" algorithm
18	* based on one described in <a
19	* href="http://www.cs.rochester.edu/u/michael/PODC96.html"> Simple,
20	* Fast, and Practical Non-Blocking and Blocking Concurrent Queue
21	* Algorithms</a> by Maged M. Michael and Michael L. Scott.)
22	*
23	* Beware that, unlike in most collections, the <tt>size</tt> method
24	* is <em>NOT</em> a constant-time operation. Because of the
25	* asynchronous nature of these queues, determining the current number
26	* of elements requires an O(n) traversal.
27	*
28	**/
29	public class LinkedQueue<E> extends AbstractQueue<E>
30	implements Queue<E>, java.io.Serializable {
31
32	/*
33	* This is a straight adaptation of Michael & Scott algorithm.
34	* For explanation, read the paper. The only (minor) algorithmic
35	* difference is that this version supports lazy deletion of
36	* internal nodes (method remove(Object)) -- remove CAS'es item
37	* fields to null. The normal queue operations unlink but then
38	* pass over nodes with null item fields. Similarly, iteration
39	* methods ignore those with nulls.
40	*/
41
42	// Atomics support
43
44	private final static AtomicReferenceFieldUpdater<LinkedQueue, AtomicLinkedNode> tailUpdater = new AtomicReferenceFieldUpdater<LinkedQueue, AtomicLinkedNode>(new LinkedQueue[0], new AtomicLinkedNode[0], "tail");
45	private final static AtomicReferenceFieldUpdater<LinkedQueue, AtomicLinkedNode> headUpdater = new AtomicReferenceFieldUpdater<LinkedQueue, AtomicLinkedNode>(new LinkedQueue[0], new AtomicLinkedNode[0], "head");
46
47	private boolean casTail(AtomicLinkedNode cmp, AtomicLinkedNode val) {
48	return tailUpdater.compareAndSet(this, cmp, val);
49	}
50
51	private boolean casHead(AtomicLinkedNode cmp, AtomicLinkedNode val) {
52	return headUpdater.compareAndSet(this, cmp, val);
53	}
54
55
56	/**
57	* Pointer to header node, initialized to a dummy node. The first
58	* actual node is at head.getNext().
59	*/
60	private transient volatile AtomicLinkedNode head = new AtomicLinkedNode(null, null);
61
62	/ Pointer to last node on list /
63	private transient volatile AtomicLinkedNode tail = head;
64
65
66	/**
67	* Creates an initially empty LinkedQueue.
68	*/
69	public LinkedQueue() {}
70
71	/**
72	* Creates a LinkedQueue initially holding the elements
73	* of the given collection. The elements are added in
74	* iterator traversal order.
75	*
76	* @param initialElements the collections whose elements are to be added.
77	*/
78	public LinkedQueue(Collection<E> initialElements) {
79	for (Iterator<E> it = initialElements.iterator(); it.hasNext();)
80	add(it.next());
81	}
82
83	public boolean offer(E x) {
84	if (x == null) throw new IllegalArgumentException();
85	AtomicLinkedNode n = new AtomicLinkedNode(x, null);
86	for(;;) {
87	AtomicLinkedNode t = tail;
88	AtomicLinkedNode s = t.getNext();
89	if (t == tail) {
90	if (s == null) {
91	if (t.casNext(s, n)) {
92	casTail(t, n);
93	return true;
94	}
95	}
96	else {
97	casTail(t, s);
98	}
99	}
100	}
101	}
102
103	public E poll() {
104	for (;;) {
105	AtomicLinkedNode h = head;
106	AtomicLinkedNode t = tail;
107	AtomicLinkedNode first = h.getNext();
108	if (h == head) {
109	if (h == t) {
110	if (first == null)
111	return null;
112	else
113	casTail(t, first);
114	}
115	else if (casHead(h, first)) {
116	E item = (E)first.getItem();
117	if (item != null) {
118	first.setItem(null);
119	return item;
120	}
121	// else skip over deleted item, continue loop,
122	}
123	}
124	}
125	}
126
127	public E peek() { // same as poll except don't remove item
128	for (;;) {
129	AtomicLinkedNode h = head;
130	AtomicLinkedNode t = tail;
131	AtomicLinkedNode first = h.getNext();
132	if (h == head) {
133	if (h == t) {
134	if (first == null)
135	return null;
136	else
137	casTail(t, first);
138	}
139	else {
140	E item = (E)first.getItem();
141	if (item != null)
142	return item;
143	else // remove deleted node and continue
144	casHead(h, first);
145	}
146	}
147	}
148	}
149
150	/**
151	* Return the first actual (non-header) node on list. This is yet
152	* another variant of poll/peek; here returning out the first
153	* node, not element (so we cannot collapse with peek() without
154	* introducing race.)
155	*/
156	AtomicLinkedNode first() {
157	for (;;) {
158	AtomicLinkedNode h = head;
159	AtomicLinkedNode t = tail;
160	AtomicLinkedNode first = h.getNext();
161	if (h == head) {
162	if (h == t) {
163	if (first == null)
164	return null;
165	else
166	casTail(t, first);
167	}
168	else {
169	if (first.getItem() != null)
170	return first;
171	else // remove deleted node and continue
172	casHead(h, first);
173	}
174	}
175	}
176	}
177
178
179	public boolean isEmpty() {
180	return first() == null;
181	}
182
183	/**
184	* Returns the number of elements in this collection.
185	*
186	* Beware that, unlike in most collection, this method> is
187	* <em>NOT</em> a constant-time operation. Because of the
188	* asynchronous nature of these queues, determining the current
189	* number of elements requires an O(n) traversal.
190	* @return the number of elements in this collection
191	*/
192	public int size() {
193	int count = 0;
194	for (AtomicLinkedNode p = first(); p != null; p = p.getNext()) {
195	if (p.getItem() != null)
196	++count;
197	}
198	return count;
199	}
200
201	public boolean contains(Object x) {
202	if (x == null) return false;
203	for (AtomicLinkedNode p = first(); p != null; p = p.getNext()) {
204	Object item = p.getItem();
205	if (item != null &&
206	x.equals(item))
207	return true;
208	}
209	return false;
210	}
211
212	public boolean remove(Object x) {
213	if (x == null) return false;
214	for (AtomicLinkedNode p = first(); p != null; p = p.getNext()) {
215	Object item = p.getItem();
216	if (item != null &&
217	x.equals(item) &&
218	p.casItem(item, null))
219	return true;
220	}
221	return false;
222	}
223
224	public Object[] toArray() {
225	// Use ArrayList to deal with resizing.
226	ArrayList al = new ArrayList();
227	for (AtomicLinkedNode p = first(); p != null; p = p.getNext()) {
228	Object item = p.getItem();
229	if (item != null)
230	al.add(item);
231	}
232	return al.toArray();
233	}
234
235	public <T> T[] toArray(T[] a) {
236	// try to use sent-in array
237	int k = 0;
238	AtomicLinkedNode p;
239	for (p = first(); p != null && k < a.length; p = p.getNext()) {
240	Object item = p.getItem();
241	if (item != null)
242	a[k++] = (T)item;
243	}
244	if (p == null) {
245	if (k < a.length)
246	a[k] = null;
247	return a;
248	}
249
250	// If won't fit, use ArrayList version
251	ArrayList al = new ArrayList();
252	for (AtomicLinkedNode q = first(); q != null; q = q.getNext()) {
253	Object item = q.getItem();
254	if (item != null)
255	al.add(item);
256	}
257	return (T[])al.toArray(a);
258	}
259
260	public Iterator<E> iterator() {
261	return new Itr();
262	}
263
264	private class Itr implements Iterator<E> {
265	/**
266	* Next node to return item for.
267	*/
268	private AtomicLinkedNode nextNode;
269
270	/**
271	* nextItem holds on to item fields because once we claim
272	* that an element exists in hasNext(), we must return it in
273	* the following next() call even if it was in the process of
274	* being removed when hasNext() was called.
275	**/
276	private E nextItem;
277
278	/**
279	* Node of the last returned item, to support remove.
280	*/
281	private AtomicLinkedNode lastRet;
282
283	Itr() {
284	advance();
285	}
286
287	/**
288	* Move to next valid node.
289	* Return item to return for next(), or null if no such.
290	*/
291	private E advance() {
292	lastRet = nextNode;
293	E x = (E)nextItem;
294
295	AtomicLinkedNode p = (nextNode == null)? first() : nextNode.getNext();
296	for (;;) {
297	if (p == null) {
298	nextNode = null;
299	nextItem = null;
300	return x;
301	}
302	E item = (E)p.getItem();
303	if (item != null) {
304	nextNode = p;
305	nextItem = item;
306	return x;
307	}
308	else // skip over nulls
309	p = p.getNext();
310	}
311	}
312
313	public boolean hasNext() {
314	return nextNode != null;
315	}
316
317	public E next() {
318	if (nextNode == null) throw new NoSuchElementException();
319	return advance();
320	}
321
322	public void remove() {
323	AtomicLinkedNode l = lastRet;
324	if (l == null) throw new IllegalStateException();
325	// rely on a future traversal to relink.
326	l.setItem(null);
327	lastRet = null;
328	}
329	}
330
331	/**
332	* Save the state to a stream (that is, serialize it).
333	*
334	* @serialData All of the elements (each an <tt>E</tt>) in
335	* the proper order, followed by a null
336	*/
337	private void writeObject(java.io.ObjectOutputStream s)
338	throws java.io.IOException {
339
340	// Write out any hidden stuff
341	s.defaultWriteObject();
342
343	// Write out all elements in the proper order.
344	for (AtomicLinkedNode p = first(); p != null; p = p.getNext()) {
345	Object item = p.getItem();
346	if (item != null)
347	s.writeObject(item);
348	}
349
350	// Use trailing null as sentinel
351	s.writeObject(null);
352	}
353
354	/**
355	* Reconstitute the Queue instance from a stream (that is,
356	* deserialize it).
357	*/
358	private void readObject(java.io.ObjectInputStream s)
359	throws java.io.IOException, ClassNotFoundException {
360	// Read in capacity, and any hidden stuff
361	s.defaultReadObject();
362
363	// Read in all elements and place in queue
364	for (;;) {
365	E item = (E)s.readObject();
366	if (item == null)
367	break;
368	add(item);
369	}
370	}
371
372	}