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 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.
     */

    static class Node {
        private volatile Object item;
        private volatile Node next;
        Node(Object x, Node n) { item = x; next = n; }
    }

    // Atomics support

    private final static AtomicReferenceFieldUpdater<LinkedQueue, Node> tailUpdater = new AtomicReferenceFieldUpdater<LinkedQueue, Node>(new LinkedQueue[0], new Node[0], "tail");
    private final static AtomicReferenceFieldUpdater<LinkedQueue, Node> headUpdater = new AtomicReferenceFieldUpdater<LinkedQueue, Node>(new LinkedQueue[0], new Node[0], "head");
    private final static AtomicReferenceFieldUpdater<Node, Node> nextUpdater =
    new AtomicReferenceFieldUpdater<Node, Node>(new Node[0], new Node[0], "next");
    private final static AtomicReferenceFieldUpdater<Node, Object> itemUpdater
     = new AtomicReferenceFieldUpdater<Node, Object>(new Node[0], new Object[0], "item");

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

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

    private boolean casNext(Node node, Node cmp, Node val) {
        return nextUpdater.compareAndSet(node, cmp, val);
    }

    private boolean casItem(Node node, Object cmp, Object val) {
        return itemUpdater.compareAndSet(node, cmp, val);
    }


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

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

    /**
     * Return the first actual (non-header) node on list.
     */
    Node first() { return head.next; }

    public LinkedQueue() {}

    public LinkedQueue(Collection<E> initialElements) {
        for (Iterator<E> it = initialElements.iterator(); it.hasNext();) 
            add(it.next());
    }


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

    public boolean offer(E x) {
        return add(x);
    }

    public E poll() {
        for (;;) {
            Node h = head;
            Node t = tail;
            Node first = h.next;
            if (h == head) {
                if (h == t) {
                    if (first == null)
                        return null;
                    else
                        casTail(t, first);
                }
                else if (casHead(h, first)) {
                    E item = (E)first.item;
                    if (item != null) {
                        itemUpdater.set(first, null);
                        return item;
                    }
                    // else skip over deleted item, continue loop, 
                }
            }
        }
    }

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

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

    /**
     * Returns the number of elements in this collection. 
     * 
     * Beware that, unlike 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 (Node p = first(); p != null; p = p.next) {
            if (p.item != null)
                ++count;
        }
        return count;
    }

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

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

    public <T> T[] toArray(T[] a) {
        // try to use sent-in array
        int k = 0;
        Node p;
        for (p = first(); p != null && k < a.length; p = p.next) {
            Object item = p.item;
            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 (Node q = first(); q != null; q = q.next) {
            Object item = q.item;
            if (item != null)
                al.add(item);
        }
        return (T[])al.toArray(a);
    }

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

    private class Itr implements Iterator<E> {
        private Node current;
        /** 
         * currentItem 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 currentItem;

        Itr() { 
            for (current = first(); current != null; current = current.next) {
                E item = (E)current.item;
                if (item != null) {
                    currentItem = item;
                    return;
                }
            }
        }
        
        /**
         * Move to next valid node. 
         * Return previous item, or null if no such.
         */
        private E advance() { 
            E x = (E)currentItem;
            for (;;) {
                current = current.next;
                if (current == null) {
                    currentItem = null;
                    return x;
                }
                E item = (E)current.item;
                if (item != null) {
                    currentItem = item;
                    return x;
                }
            }
        }
        
        public boolean hasNext() {
            return current != null;
        }
        
        public E next() {
            if (current == null) throw new NoSuchElementException();
            return advance();
        }
        
        public void remove() {
            if (current == null) throw new NoSuchElementException();
            // java.util.Iterator contract requires throw if already removed
            if (currentItem == null) throw new IllegalStateException();
            // rely on a future traversal to relink.
            currentItem = null;
            itemUpdater.set(current, 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 (Node p = first(); p != null; p = p.next) 
            s.writeObject(p.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.4
Committed:	Wed Jun 18 11:41:04 2003 UTC (20 years, 11 months ago) by dl
Branch:	MAIN
Changes since 1.3:	+6 -1 lines
Log Message:	Added some internal documentation.
#	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.2	* Beware that, unlike most collections, the <tt>size</tt> method is
24			* <em>NOT</em> a constant-time operation. Because of the asynchronous
25			* nature of these queues, determining the current number of elements
26			* requires an O(n) traversal.
27			*
28	tim	1.1	**/
29	dl	1.2	public class LinkedQueue<E> extends AbstractQueue<E>
30	tim	1.1	implements Queue<E>, java.io.Serializable {
31
32	dl	1.2	/*
33			* This is a straight adaptation of Michael & Scott algorithm.
34	dl	1.4	* 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	dl	1.2	*/
41
42			static class Node {
43			private volatile Object item;
44			private volatile Node next;
45			Node(Object x, Node n) { item = x; next = n; }
46			}
47
48			// Atomics support
49
50			private final static AtomicReferenceFieldUpdater<LinkedQueue, Node> tailUpdater = new AtomicReferenceFieldUpdater<LinkedQueue, Node>(new LinkedQueue[0], new Node[0], "tail");
51			private final static AtomicReferenceFieldUpdater<LinkedQueue, Node> headUpdater = new AtomicReferenceFieldUpdater<LinkedQueue, Node>(new LinkedQueue[0], new Node[0], "head");
52			private final static AtomicReferenceFieldUpdater<Node, Node> nextUpdater =
53			new AtomicReferenceFieldUpdater<Node, Node>(new Node[0], new Node[0], "next");
54			private final static AtomicReferenceFieldUpdater<Node, Object> itemUpdater
55			= new AtomicReferenceFieldUpdater<Node, Object>(new Node[0], new Object[0], "item");
56
57			private boolean casTail(Node cmp, Node val) {
58			return tailUpdater.compareAndSet(this, cmp, val);
59			}
60
61			private boolean casHead(Node cmp, Node val) {
62			return headUpdater.compareAndSet(this, cmp, val);
63			}
64
65			private boolean casNext(Node node, Node cmp, Node val) {
66			return nextUpdater.compareAndSet(node, cmp, val);
67			}
68
69			private boolean casItem(Node node, Object cmp, Object val) {
70			return itemUpdater.compareAndSet(node, cmp, val);
71			}
72
73
74			/**
75			* Pointer to header node, initialized to a dummy node. The first
76			* actual node is at head.next.
77			*/
78			private transient volatile Node head = new Node(null, null);
79
80			/ Pointer to last node on list /
81			private transient volatile Node tail = head;
82
83			/**
84			* Return the first actual (non-header) node on list.
85			*/
86			Node first() { return head.next; }
87
88	tim	1.1	public LinkedQueue() {}
89	dl	1.2
90			public LinkedQueue(Collection<E> initialElements) {
91			for (Iterator<E> it = initialElements.iterator(); it.hasNext();)
92			add(it.next());
93			}
94
95
96	tim	1.1	public boolean add(E x) {
97	dl	1.2	if (x == null) throw new IllegalArgumentException();
98			Node n = new Node(x, null);
99			for(;;) {
100			Node t = tail;
101			Node s = t.next;
102			if (t == tail) {
103			if (s == null) {
104			if (casNext(t, s, n)) {
105			casTail(t, n);
106			return true;
107			}
108			}
109			else {
110			casTail(t, s);
111			}
112			}
113			}
114	tim	1.1	}
115	dl	1.2
116	tim	1.1	public boolean offer(E x) {
117	dl	1.2	return add(x);
118			}
119
120			public E poll() {
121			for (;;) {
122			Node h = head;
123			Node t = tail;
124			Node first = h.next;
125			if (h == head) {
126			if (h == t) {
127			if (first == null)
128			return null;
129			else
130			casTail(t, first);
131			}
132			else if (casHead(h, first)) {
133			E item = (E)first.item;
134			if (item != null) {
135			itemUpdater.set(first, null);
136			return item;
137			}
138			// else skip over deleted item, continue loop,
139			}
140			}
141			}
142			}
143
144			public E peek() { // same as poll except don't remove item
145			for (;;) {
146			Node h = head;
147			Node t = tail;
148			Node first = h.next;
149			if (h == head) {
150			if (h == t) {
151			if (first == null)
152			return null;
153			else
154			casTail(t, first);
155			}
156			else {
157			E item = (E)first.item;
158			if (item != null)
159			return item;
160			else // remove deleted node and continue
161			casHead(h, first);
162			}
163			}
164			}
165			}
166
167			public boolean isEmpty() {
168			return peek() == null;
169	tim	1.1	}
170	dl	1.2
171			/**
172			* Returns the number of elements in this collection.
173			*
174			* Beware that, unlike most collection, this method> is
175			* <em>NOT</em> a constant-time operation. Because of the
176			* asynchronous nature of these queues, determining the current
177			* number of elements requires an O(n) traversal.
178			* @return the number of elements in this collection
179			*/
180			public int size() {
181			int count = 0;
182			for (Node p = first(); p != null; p = p.next) {
183			if (p.item != null)
184			++count;
185			}
186			return count;
187	tim	1.1	}
188	dl	1.2
189			public boolean contains(Object x) {
190			if (x == null) return false;
191			for (Node p = first(); p != null; p = p.next) {
192			Object item = p.item;
193			if (item != null &&
194			x.equals(item))
195			return true;
196			}
197			return false;
198	tim	1.1	}
199
200			public boolean remove(Object x) {
201	dl	1.2	if (x == null) return false;
202			for (Node p = first(); p != null; p = p.next) {
203			Object item = p.item;
204			if (item != null &&
205			x.equals(item) &&
206			casItem(p, item, null))
207			return true;
208			}
209	tim	1.1	return false;
210			}
211	dl	1.2
212			public Object[] toArray() {
213			// Use ArrayList to deal with resizing.
214			ArrayList al = new ArrayList();
215			for (Node p = first(); p != null; p = p.next) {
216			Object item = p.item;
217			if (item != null)
218			al.add(item);
219			}
220			return al.toArray();
221	tim	1.1	}
222	dl	1.2
223			public <T> T[] toArray(T[] a) {
224			// try to use sent-in array
225			int k = 0;
226			Node p;
227			for (p = first(); p != null && k < a.length; p = p.next) {
228			Object item = p.item;
229			if (item != null)
230			a[k++] = (T)item;
231			}
232			if (p == null) {
233			if (k < a.length)
234			a[k] = null;
235			return a;
236			}
237
238			// If won't fit, use ArrayList version
239			ArrayList al = new ArrayList();
240			for (Node q = first(); q != null; q = q.next) {
241			Object item = q.item;
242			if (item != null)
243			al.add(item);
244			}
245			return (T[])al.toArray(a);
246	tim	1.1	}
247	dl	1.2
248			public Iterator<E> iterator() {
249			return new Itr();
250	tim	1.1	}
251	dl	1.2
252			private class Itr implements Iterator<E> {
253			private Node current;
254			/**
255			* currentItem holds on to item fields because once we claim
256			* that an element exists in hasNext(), we must return it in
257			* the following next() call even if it was in the process of
258			* being removed when hasNext() was called.
259			**/
260			private E currentItem;
261
262			Itr() {
263			for (current = first(); current != null; current = current.next) {
264			E item = (E)current.item;
265			if (item != null) {
266			currentItem = item;
267			return;
268			}
269			}
270			}
271
272			/**
273			* Move to next valid node.
274			* Return previous item, or null if no such.
275			*/
276			private E advance() {
277			E x = (E)currentItem;
278			for (;;) {
279			current = current.next;
280			if (current == null) {
281			currentItem = null;
282			return x;
283			}
284			E item = (E)current.item;
285			if (item != null) {
286			currentItem = item;
287			return x;
288			}
289			}
290			}
291
292			public boolean hasNext() {
293			return current != null;
294			}
295
296			public E next() {
297			if (current == null) throw new NoSuchElementException();
298			return advance();
299			}
300
301			public void remove() {
302			if (current == null) throw new NoSuchElementException();
303			// java.util.Iterator contract requires throw if already removed
304			if (currentItem == null) throw new IllegalStateException();
305			// rely on a future traversal to relink.
306			currentItem = null;
307			itemUpdater.set(current, null);
308			}
309	tim	1.1	}
310	dl	1.2
311			/**
312			* Save the state to a stream (that is, serialize it).
313			*
314			* @serialData All of the elements (each an <tt>E</tt>) in
315			* the proper order, followed by a null
316			*/
317			private void writeObject(java.io.ObjectOutputStream s)
318			throws java.io.IOException {
319
320			// Write out any hidden stuff
321			s.defaultWriteObject();
322
323			// Write out all elements in the proper order.
324			for (Node p = first(); p != null; p = p.next)
325			s.writeObject(p.item);
326
327			// Use trailing null as sentinel
328			s.writeObject(null);
329	tim	1.1	}
330
331	dl	1.2	/**
332			* Reconstitute the Queue instance from a stream (that is,
333			* deserialize it).
334			*/
335			private void readObject(java.io.ObjectInputStream s)
336			throws java.io.IOException, ClassNotFoundException {
337			// Read in capacity, and any hidden stuff
338			s.defaultReadObject();
339
340			// Read in all elements and place in queue
341			for (;;) {
342			E item = (E)s.readObject();
343			if (item == null)
344			break;
345			add(item);
346			}
347	tim	1.1	}
348
349			}