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