ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/ConcurrentLinkedQueue.java
Revision: 1.13
Committed: Wed Aug 20 14:49:25 2003 UTC (20 years, 9 months ago) by dl
Branch: MAIN
Changes since 1.12: +44 -1 lines
Log Message:
AtomicLinkedNode now an inner class of Concurrentlinkedqueue

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