ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/ConcurrentLinkedQueue.java
Revision: 1.23
Committed: Tue Dec 23 19:38:09 2003 UTC (20 years, 5 months ago) by dl
Branch: MAIN
Changes since 1.22: +44 -44 lines
Log Message:
cache finals across volatiles; avoid readResolve; doc improvments; timed invokeAll interleaves

File Contents

# 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 {@linkplain Queue queue} based on linked nodes.
14 * This queue orders elements FIFO (first-in-first-out).
15 * The <em>head</em> of the queue is that element that has been on the
16 * queue the longest time.
17 * The <em>tail</em> of the queue is that element that has been on the
18 * queue the shortest time. New elements
19 * are inserted at the tail of the queue, and the queue retrieval
20 * operations obtain elements at the head of the queue.
21 * A <tt>ConcurrentLinkedQueue</tt> is an appropriate choice when
22 * many threads will share access to a common collection.
23 * This queue does not permit <tt>null</tt> elements.
24 *
25 * <p>This implementation employs an efficient &quot;wait-free&quot;
26 * algorithm based on one described in <a
27 * href="http://www.cs.rochester.edu/u/michael/PODC96.html"> Simple,
28 * Fast, and Practical Non-Blocking and Blocking Concurrent Queue
29 * Algorithms</a> by Maged M. Michael and Michael L. Scott.
30 *
31 * <p>Beware that, unlike in most collections, the <tt>size</tt> method
32 * is <em>NOT</em> a constant-time operation. Because of the
33 * asynchronous nature of these queues, determining the current number
34 * of elements requires a traversal of the elements.
35 *
36 * <p>This class implements all of the <em>optional</em> methods
37 * of the {@link Collection} and {@link Iterator} interfaces.
38 *
39 * @since 1.5
40 * @author Doug Lea
41 * @param <E> the type of elements held in this collection
42 *
43 **/
44 public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
45 implements Queue<E>, java.io.Serializable {
46 private static final long serialVersionUID = 196745693267521676L;
47
48 /*
49 * This is a straight adaptation of Michael & Scott algorithm.
50 * For explanation, read the paper. The only (minor) algorithmic
51 * difference is that this version supports lazy deletion of
52 * internal nodes (method remove(Object)) -- remove CAS'es item
53 * fields to null. The normal queue operations unlink but then
54 * pass over nodes with null item fields. Similarly, iteration
55 * methods ignore those with nulls.
56 */
57
58 private static class Node<E> {
59 private volatile E item;
60 private volatile Node<E> next;
61
62 private static final
63 AtomicReferenceFieldUpdater<Node, Node>
64 nextUpdater =
65 AtomicReferenceFieldUpdater.newUpdater
66 (Node.class, Node.class, "next");
67 private static final
68 AtomicReferenceFieldUpdater<Node, Object>
69 itemUpdater =
70 AtomicReferenceFieldUpdater.newUpdater
71 (Node.class, Object.class, "item");
72
73 Node(E x) { item = x; }
74
75 Node(E x, Node<E> n) { item = x; next = n; }
76
77 E getItem() {
78 return item;
79 }
80
81 boolean casItem(E cmp, E val) {
82 return itemUpdater.compareAndSet(this, cmp, val);
83 }
84
85 void setItem(E val) {
86 itemUpdater.set(this, val);
87 }
88
89 Node<E> getNext() {
90 return next;
91 }
92
93 boolean casNext(Node<E> cmp, Node<E> val) {
94 return nextUpdater.compareAndSet(this, cmp, val);
95 }
96
97 void setNext(Node<E> val) {
98 nextUpdater.set(this, val);
99 }
100
101 }
102
103 private static final
104 AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>
105 tailUpdater =
106 AtomicReferenceFieldUpdater.newUpdater
107 (ConcurrentLinkedQueue.class, Node.class, "tail");
108 private static final
109 AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>
110 headUpdater =
111 AtomicReferenceFieldUpdater.newUpdater
112 (ConcurrentLinkedQueue.class, Node.class, "head");
113
114 private boolean casTail(Node<E> cmp, Node<E> val) {
115 return tailUpdater.compareAndSet(this, cmp, val);
116 }
117
118 private boolean casHead(Node<E> cmp, Node<E> val) {
119 return headUpdater.compareAndSet(this, cmp, val);
120 }
121
122
123 /**
124 * Pointer to header node, initialized to a dummy node. The first
125 * actual node is at head.getNext().
126 */
127 private transient volatile Node<E> head = new Node<E>(null, null);
128
129 /** Pointer to last node on list **/
130 private transient volatile Node<E> tail = head;
131
132
133 /**
134 * Creates a <tt>ConcurrentLinkedQueue</tt> that is initially empty.
135 */
136 public ConcurrentLinkedQueue() {}
137
138 /**
139 * Creates a <tt>ConcurrentLinkedQueue</tt>
140 * initially containing the elements of the given collection,
141 * added in traversal order of the collection's iterator.
142 * @param c the collection of elements to initially contain
143 * @throws NullPointerException if <tt>c</tt> or any element within it
144 * is <tt>null</tt>
145 */
146 public ConcurrentLinkedQueue(Collection<? extends E> c) {
147 for (Iterator<? extends E> it = c.iterator(); it.hasNext();)
148 add(it.next());
149 }
150
151 // Have to override just to update the javadoc
152
153 /**
154 * Adds the specified element to the tail of this queue.
155 * @param o the element to add.
156 * @return <tt>true</tt> (as per the general contract of
157 * <tt>Collection.add</tt>).
158 *
159 * @throws NullPointerException if the specified element is <tt>null</tt>
160 */
161 public boolean add(E o) {
162 return offer(o);
163 }
164
165 /**
166 * Inserts the specified element to the tail of this queue.
167 *
168 * @param o the element to add.
169 * @return <tt>true</tt> (as per the general contract of
170 * <tt>Queue.offer</tt>).
171 * @throws NullPointerException if the specified element is <tt>null</tt>
172 */
173 public boolean offer(E o) {
174 if (o == null) throw new NullPointerException();
175 Node<E> n = new Node<E>(o, null);
176 for(;;) {
177 Node<E> t = tail;
178 Node<E> s = t.getNext();
179 if (t == tail) {
180 if (s == null) {
181 if (t.casNext(s, n)) {
182 casTail(t, n);
183 return true;
184 }
185 } else {
186 casTail(t, s);
187 }
188 }
189 }
190 }
191
192 public E poll() {
193 for (;;) {
194 Node<E> h = head;
195 Node<E> t = tail;
196 Node<E> first = h.getNext();
197 if (h == head) {
198 if (h == t) {
199 if (first == null)
200 return null;
201 else
202 casTail(t, first);
203 } else if (casHead(h, first)) {
204 E item = first.getItem();
205 if (item != null) {
206 first.setItem(null);
207 return item;
208 }
209 // else skip over deleted item, continue loop,
210 }
211 }
212 }
213 }
214
215 public E peek() { // same as poll except don't remove item
216 for (;;) {
217 Node<E> h = head;
218 Node<E> t = tail;
219 Node<E> first = h.getNext();
220 if (h == head) {
221 if (h == t) {
222 if (first == null)
223 return null;
224 else
225 casTail(t, first);
226 } else {
227 E item = first.getItem();
228 if (item != null)
229 return item;
230 else // remove deleted node and continue
231 casHead(h, first);
232 }
233 }
234 }
235 }
236
237 /**
238 * Returns the first actual (non-header) node on list. This is yet
239 * another variant of poll/peek; here returning out the first
240 * node, not element (so we cannot collapse with peek() without
241 * introducing race.)
242 */
243 Node<E> first() {
244 for (;;) {
245 Node<E> h = head;
246 Node<E> t = tail;
247 Node<E> first = h.getNext();
248 if (h == head) {
249 if (h == t) {
250 if (first == null)
251 return null;
252 else
253 casTail(t, first);
254 } else {
255 if (first.getItem() != null)
256 return first;
257 else // remove deleted node and continue
258 casHead(h, first);
259 }
260 }
261 }
262 }
263
264
265 public boolean isEmpty() {
266 return first() == null;
267 }
268
269 /**
270 * Returns the number of elements in this queue. If this queue
271 * contains more than <tt>Integer.MAX_VALUE</tt> elements, returns
272 * <tt>Integer.MAX_VALUE</tt>.
273 *
274 * <p>Beware that, unlike in most collections, this method is
275 * <em>NOT</em> a constant-time operation. Because of the
276 * asynchronous nature of these queues, determining the current
277 * number of elements requires an O(n) traversal.
278 *
279 * @return the number of elements in this queue.
280 */
281 public int size() {
282 int count = 0;
283 for (Node<E> p = first(); p != null; p = p.getNext()) {
284 if (p.getItem() != null) {
285 // Collections.size() spec says to max out
286 if (++count == Integer.MAX_VALUE)
287 break;
288 }
289 }
290 return count;
291 }
292
293 public boolean contains(Object o) {
294 if (o == null) return false;
295 for (Node<E> p = first(); p != null; p = p.getNext()) {
296 E item = p.getItem();
297 if (item != null &&
298 o.equals(item))
299 return true;
300 }
301 return false;
302 }
303
304 public boolean remove(Object o) {
305 if (o == null) return false;
306 for (Node<E> p = first(); p != null; p = p.getNext()) {
307 E item = p.getItem();
308 if (item != null &&
309 o.equals(item) &&
310 p.casItem(item, null))
311 return true;
312 }
313 return false;
314 }
315
316 public Object[] toArray() {
317 // Use ArrayList to deal with resizing.
318 ArrayList<E> al = new ArrayList<E>();
319 for (Node<E> p = first(); p != null; p = p.getNext()) {
320 E item = p.getItem();
321 if (item != null)
322 al.add(item);
323 }
324 return al.toArray();
325 }
326
327 public <T> T[] toArray(T[] a) {
328 // try to use sent-in array
329 int k = 0;
330 Node<E> p;
331 for (p = first(); p != null && k < a.length; p = p.getNext()) {
332 E item = p.getItem();
333 if (item != null)
334 a[k++] = (T)item;
335 }
336 if (p == null) {
337 if (k < a.length)
338 a[k] = null;
339 return a;
340 }
341
342 // If won't fit, use ArrayList version
343 ArrayList<E> al = new ArrayList<E>();
344 for (Node<E> q = first(); q != null; q = q.getNext()) {
345 E item = q.getItem();
346 if (item != null)
347 al.add(item);
348 }
349 return (T[])al.toArray(a);
350 }
351
352 /**
353 * Returns an iterator over the elements in this queue in proper sequence.
354 * The returned iterator is a "weakly consistent" iterator that
355 * will never throw {@link java.util.ConcurrentModificationException},
356 * and guarantees to traverse elements as they existed upon
357 * construction of the iterator, and may (but is not guaranteed to)
358 * reflect any modifications subsequent to construction.
359 *
360 * @return an iterator over the elements in this queue in proper sequence.
361 */
362 public Iterator<E> iterator() {
363 return new Itr();
364 }
365
366 private class Itr implements Iterator<E> {
367 /**
368 * Next node to return item for.
369 */
370 private Node<E> nextNode;
371
372 /**
373 * nextItem holds on to item fields because once we claim
374 * that an element exists in hasNext(), we must return it in
375 * the following next() call even if it was in the process of
376 * being removed when hasNext() was called.
377 **/
378 private E nextItem;
379
380 /**
381 * Node of the last returned item, to support remove.
382 */
383 private Node<E> lastRet;
384
385 Itr() {
386 advance();
387 }
388
389 /**
390 * Move to next valid node.
391 * Return item to return for next(), or null if no such.
392 */
393 private E advance() {
394 lastRet = nextNode;
395 E x = nextItem;
396
397 Node<E> p = (nextNode == null)? first() : nextNode.getNext();
398 for (;;) {
399 if (p == null) {
400 nextNode = null;
401 nextItem = null;
402 return x;
403 }
404 E item = p.getItem();
405 if (item != null) {
406 nextNode = p;
407 nextItem = item;
408 return x;
409 } else // skip over nulls
410 p = p.getNext();
411 }
412 }
413
414 public boolean hasNext() {
415 return nextNode != null;
416 }
417
418 public E next() {
419 if (nextNode == null) throw new NoSuchElementException();
420 return advance();
421 }
422
423 public void remove() {
424 Node<E> l = lastRet;
425 if (l == null) throw new IllegalStateException();
426 // rely on a future traversal to relink.
427 l.setItem(null);
428 lastRet = null;
429 }
430 }
431
432 /**
433 * Save the state to a stream (that is, serialize it).
434 *
435 * @serialData All of the elements (each an <tt>E</tt>) in
436 * the proper order, followed by a null
437 * @param s the stream
438 */
439 private void writeObject(java.io.ObjectOutputStream s)
440 throws java.io.IOException {
441
442 // Write out any hidden stuff
443 s.defaultWriteObject();
444
445 // Write out all elements in the proper order.
446 for (Node<E> p = first(); p != null; p = p.getNext()) {
447 Object item = p.getItem();
448 if (item != null)
449 s.writeObject(item);
450 }
451
452 // Use trailing null as sentinel
453 s.writeObject(null);
454 }
455
456 /**
457 * Reconstitute the Queue instance from a stream (that is,
458 * deserialize it).
459 * @param s the stream
460 */
461 private void readObject(java.io.ObjectInputStream s)
462 throws java.io.IOException, ClassNotFoundException {
463 // Read in capacity, and any hidden stuff
464 s.defaultReadObject();
465 head = new Node<E>(null, null);
466 tail = head;
467 // Read in all elements and place in queue
468 for (;;) {
469 E item = (E)s.readObject();
470 if (item == null)
471 break;
472 else
473 offer(item);
474 }
475 }
476
477 }