ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jsr166y/LinkedTransferQueue.java
Revision: 1.30
Committed: Mon Jul 27 03:22:39 2009 UTC (14 years, 9 months ago) by jsr166
Branch: MAIN
Changes since 1.29: +4 -4 lines
Log Message:
@throws NullPointerException {@inheritDoc}

File Contents

# User Rev Content
1 dl 1.1 /*
2     * Written by Doug Lea with assistance from members of JCP JSR-166
3     * Expert Group and released to the public domain, as explained at
4     * http://creativecommons.org/licenses/publicdomain
5     */
6    
7     package jsr166y;
8 jsr166 1.26
9 dl 1.1 import java.util.concurrent.*;
10 jsr166 1.26
11     import java.util.AbstractQueue;
12     import java.util.Collection;
13     import java.util.Iterator;
14     import java.util.NoSuchElementException;
15     import java.util.concurrent.locks.LockSupport;
16     import java.util.concurrent.atomic.AtomicReference;
17     import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
18 dl 1.1
19     /**
20     * An unbounded {@linkplain TransferQueue} based on linked nodes.
21     * This queue orders elements FIFO (first-in-first-out) with respect
22     * to any given producer. The <em>head</em> of the queue is that
23     * element that has been on the queue the longest time for some
24     * producer. The <em>tail</em> of the queue is that element that has
25     * been on the queue the shortest time for some producer.
26     *
27 jsr166 1.11 * <p>Beware that, unlike in most collections, the {@code size}
28 dl 1.1 * method is <em>NOT</em> a constant-time operation. Because of the
29     * asynchronous nature of these queues, determining the current number
30     * of elements requires a traversal of the elements.
31     *
32     * <p>This class and its iterator implement all of the
33     * <em>optional</em> methods of the {@link Collection} and {@link
34     * Iterator} interfaces.
35     *
36     * <p>Memory consistency effects: As with other concurrent
37     * collections, actions in a thread prior to placing an object into a
38     * {@code LinkedTransferQueue}
39     * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
40     * actions subsequent to the access or removal of that element from
41     * the {@code LinkedTransferQueue} in another thread.
42     *
43     * <p>This class is a member of the
44     * <a href="{@docRoot}/../technotes/guides/collections/index.html">
45     * Java Collections Framework</a>.
46     *
47 dl 1.3 * @since 1.7
48 dl 1.1 * @author Doug Lea
49     * @param <E> the type of elements held in this collection
50     */
51     public class LinkedTransferQueue<E> extends AbstractQueue<E>
52     implements TransferQueue<E>, java.io.Serializable {
53     private static final long serialVersionUID = -3223113410248163686L;
54    
55     /*
56     * This class extends the approach used in FIFO-mode
57     * SynchronousQueues. See the internal documentation, as well as
58     * the PPoPP 2006 paper "Scalable Synchronous Queues" by Scherer,
59     * Lea & Scott
60     * (http://www.cs.rice.edu/~wns1/papers/2006-PPoPP-SQ.pdf)
61     *
62 dl 1.9 * The main extension is to provide different Wait modes for the
63     * main "xfer" method that puts or takes items. These don't
64     * impact the basic dual-queue logic, but instead control whether
65     * or how threads block upon insertion of request or data nodes
66     * into the dual queue. It also uses slightly different
67     * conventions for tracking whether nodes are off-list or
68     * cancelled.
69 dl 1.1 */
70    
71     // Wait modes for xfer method
72     static final int NOWAIT = 0;
73     static final int TIMEOUT = 1;
74     static final int WAIT = 2;
75    
76     /** The number of CPUs, for spin control */
77     static final int NCPUS = Runtime.getRuntime().availableProcessors();
78    
79     /**
80     * The number of times to spin before blocking in timed waits.
81     * The value is empirically derived -- it works well across a
82     * variety of processors and OSes. Empirically, the best value
83     * seems not to vary with number of CPUs (beyond 2) so is just
84     * a constant.
85     */
86 jsr166 1.22 static final int maxTimedSpins = (NCPUS < 2) ? 0 : 32;
87 dl 1.1
88     /**
89     * The number of times to spin before blocking in untimed waits.
90     * This is greater than timed value because untimed waits spin
91     * faster since they don't need to check times on each spin.
92     */
93     static final int maxUntimedSpins = maxTimedSpins * 16;
94    
95     /**
96     * The number of nanoseconds for which it is faster to spin
97     * rather than to use timed park. A rough estimate suffices.
98     */
99     static final long spinForTimeoutThreshold = 1000L;
100    
101 jsr166 1.5 /**
102 dl 1.9 * Node class for LinkedTransferQueue. Opportunistically
103     * subclasses from AtomicReference to represent item. Uses Object,
104     * not E, to allow setting item to "this" after use, to avoid
105     * garbage retention. Similarly, setting the next field to this is
106     * used as sentinel that node is off list.
107 dl 1.1 */
108 jsr166 1.25 static final class Node<E> extends AtomicReference<Object> {
109     volatile Node<E> next;
110 dl 1.1 volatile Thread waiter; // to control park/unpark
111     final boolean isData;
112 jsr166 1.25
113     Node(E item, boolean isData) {
114 dl 1.1 super(item);
115     this.isData = isData;
116     }
117    
118 jsr166 1.25 @SuppressWarnings("rawtypes")
119     static final AtomicReferenceFieldUpdater<Node, Node>
120 dl 1.1 nextUpdater = AtomicReferenceFieldUpdater.newUpdater
121 jsr166 1.25 (Node.class, Node.class, "next");
122 dl 1.1
123 jsr166 1.25 final boolean casNext(Node<E> cmp, Node<E> val) {
124 dl 1.1 return nextUpdater.compareAndSet(this, cmp, val);
125     }
126 dl 1.15
127     final void clearNext() {
128     nextUpdater.lazySet(this, this);
129     }
130    
131 jsr166 1.24 private static final long serialVersionUID = -3375979862319811754L;
132 dl 1.1 }
133    
134     /**
135     * Padded version of AtomicReference used for head, tail and
136     * cleanMe, to alleviate contention across threads CASing one vs
137     * the other.
138     */
139     static final class PaddedAtomicReference<T> extends AtomicReference<T> {
140     // enough padding for 64bytes with 4byte refs
141     Object p0, p1, p2, p3, p4, p5, p6, p7, p8, p9, pa, pb, pc, pd, pe;
142     PaddedAtomicReference(T r) { super(r); }
143 jsr166 1.24 private static final long serialVersionUID = 8170090609809740854L;
144 dl 1.1 }
145    
146    
147 dl 1.7 /** head of the queue */
148 jsr166 1.25 private transient final PaddedAtomicReference<Node<E>> head;
149 jsr166 1.23
150 dl 1.7 /** tail of the queue */
151 jsr166 1.25 private transient final PaddedAtomicReference<Node<E>> tail;
152 dl 1.1
153     /**
154     * Reference to a cancelled node that might not yet have been
155     * unlinked from queue because it was the last inserted node
156     * when it cancelled.
157     */
158 jsr166 1.25 private transient final PaddedAtomicReference<Node<E>> cleanMe;
159 dl 1.1
160     /**
161     * Tries to cas nh as new head; if successful, unlink
162     * old head's next node to avoid garbage retention.
163     */
164 jsr166 1.25 private boolean advanceHead(Node<E> h, Node<E> nh) {
165 dl 1.1 if (h == head.get() && head.compareAndSet(h, nh)) {
166 dl 1.15 h.clearNext(); // forget old next
167 dl 1.1 return true;
168     }
169     return false;
170     }
171 jsr166 1.5
172 dl 1.1 /**
173     * Puts or takes an item. Used for most queue operations (except
174 dl 1.9 * poll() and tryTransfer()). See the similar code in
175     * SynchronousQueue for detailed explanation.
176 jsr166 1.17 *
177 jsr166 1.4 * @param e the item or if null, signifies that this is a take
178 dl 1.1 * @param mode the wait mode: NOWAIT, TIMEOUT, WAIT
179     * @param nanos timeout in nanosecs, used only if mode is TIMEOUT
180     * @return an item, or null on failure
181     */
182 jsr166 1.25 private E xfer(E e, int mode, long nanos) {
183 dl 1.1 boolean isData = (e != null);
184 jsr166 1.25 Node<E> s = null;
185     final PaddedAtomicReference<Node<E>> head = this.head;
186     final PaddedAtomicReference<Node<E>> tail = this.tail;
187 dl 1.1
188     for (;;) {
189 jsr166 1.25 Node<E> t = tail.get();
190     Node<E> h = head.get();
191 dl 1.1
192     if (t != null && (t == h || t.isData == isData)) {
193     if (s == null)
194 jsr166 1.25 s = new Node<E>(e, isData);
195     Node<E> last = t.next;
196 dl 1.1 if (last != null) {
197     if (t == tail.get())
198     tail.compareAndSet(t, last);
199     }
200     else if (t.casNext(null, s)) {
201     tail.compareAndSet(t, s);
202     return awaitFulfill(t, s, e, mode, nanos);
203     }
204     }
205 jsr166 1.5
206 dl 1.1 else if (h != null) {
207 jsr166 1.25 Node<E> first = h.next;
208 jsr166 1.5 if (t == tail.get() && first != null &&
209 dl 1.1 advanceHead(h, first)) {
210     Object x = first.get();
211     if (x != first && first.compareAndSet(x, e)) {
212     LockSupport.unpark(first.waiter);
213 jsr166 1.25 return isData ? e : (E) x;
214 dl 1.1 }
215     }
216     }
217     }
218     }
219    
220    
221     /**
222     * Version of xfer for poll() and tryTransfer, which
223 jsr166 1.17 * simplifies control paths both here and in xfer.
224 dl 1.1 */
225 jsr166 1.25 private E fulfill(E e) {
226 dl 1.1 boolean isData = (e != null);
227 jsr166 1.25 final PaddedAtomicReference<Node<E>> head = this.head;
228     final PaddedAtomicReference<Node<E>> tail = this.tail;
229 dl 1.1
230     for (;;) {
231 jsr166 1.25 Node<E> t = tail.get();
232     Node<E> h = head.get();
233 dl 1.1
234     if (t != null && (t == h || t.isData == isData)) {
235 jsr166 1.25 Node<E> last = t.next;
236 dl 1.1 if (t == tail.get()) {
237     if (last != null)
238     tail.compareAndSet(t, last);
239     else
240     return null;
241     }
242     }
243     else if (h != null) {
244 jsr166 1.25 Node<E> first = h.next;
245 jsr166 1.5 if (t == tail.get() &&
246 dl 1.1 first != null &&
247     advanceHead(h, first)) {
248     Object x = first.get();
249     if (x != first && first.compareAndSet(x, e)) {
250     LockSupport.unpark(first.waiter);
251 jsr166 1.25 return isData ? e : (E) x;
252 dl 1.1 }
253     }
254     }
255     }
256     }
257    
258     /**
259     * Spins/blocks until node s is fulfilled or caller gives up,
260     * depending on wait mode.
261     *
262     * @param pred the predecessor of waiting node
263     * @param s the waiting node
264     * @param e the comparison value for checking match
265     * @param mode mode
266     * @param nanos timeout value
267     * @return matched item, or s if cancelled
268     */
269 jsr166 1.25 private E awaitFulfill(Node<E> pred, Node<E> s, E e,
270     int mode, long nanos) {
271 dl 1.1 if (mode == NOWAIT)
272     return null;
273    
274 jsr166 1.22 long lastTime = (mode == TIMEOUT) ? System.nanoTime() : 0;
275 dl 1.1 Thread w = Thread.currentThread();
276     int spins = -1; // set to desired spin count below
277     for (;;) {
278     if (w.isInterrupted())
279     s.compareAndSet(e, s);
280     Object x = s.get();
281     if (x != e) { // Node was matched or cancelled
282     advanceHead(pred, s); // unlink if head
283 jsr166 1.17 if (x == s) { // was cancelled
284 dl 1.9 clean(pred, s);
285     return null;
286     }
287 jsr166 1.5 else if (x != null) {
288 dl 1.1 s.set(s); // avoid garbage retention
289 jsr166 1.25 return (E) x;
290 dl 1.1 }
291     else
292     return e;
293     }
294     if (mode == TIMEOUT) {
295     long now = System.nanoTime();
296     nanos -= now - lastTime;
297     lastTime = now;
298     if (nanos <= 0) {
299     s.compareAndSet(e, s); // try to cancel
300     continue;
301     }
302     }
303     if (spins < 0) {
304 jsr166 1.25 Node<E> h = head.get(); // only spin if at head
305 dl 1.1 spins = ((h != null && h.next == s) ?
306 jsr166 1.22 ((mode == TIMEOUT) ?
307 dl 1.1 maxTimedSpins : maxUntimedSpins) : 0);
308     }
309     if (spins > 0)
310     --spins;
311     else if (s.waiter == null)
312     s.waiter = w;
313     else if (mode != TIMEOUT) {
314 dl 1.12 LockSupport.park(this);
315 dl 1.1 s.waiter = null;
316     spins = -1;
317     }
318     else if (nanos > spinForTimeoutThreshold) {
319 dl 1.12 LockSupport.parkNanos(this, nanos);
320 dl 1.1 s.waiter = null;
321     spins = -1;
322     }
323     }
324     }
325    
326     /**
327 jsr166 1.17 * Returns validated tail for use in cleaning methods.
328 dl 1.9 */
329 jsr166 1.25 private Node<E> getValidatedTail() {
330 dl 1.9 for (;;) {
331 jsr166 1.25 Node<E> h = head.get();
332     Node<E> first = h.next;
333 dl 1.9 if (first != null && first.next == first) { // help advance
334     advanceHead(h, first);
335     continue;
336     }
337 jsr166 1.25 Node<E> t = tail.get();
338     Node<E> last = t.next;
339 dl 1.9 if (t == tail.get()) {
340     if (last != null)
341     tail.compareAndSet(t, last); // help advance
342     else
343     return t;
344     }
345     }
346 jsr166 1.10 }
347 dl 1.9
348     /**
349 dl 1.1 * Gets rid of cancelled node s with original predecessor pred.
350 jsr166 1.17 *
351 dl 1.9 * @param pred predecessor of cancelled node
352     * @param s the cancelled node
353 dl 1.1 */
354 jsr166 1.25 private void clean(Node<E> pred, Node<E> s) {
355 dl 1.1 Thread w = s.waiter;
356     if (w != null) { // Wake up thread
357     s.waiter = null;
358     if (w != Thread.currentThread())
359     LockSupport.unpark(w);
360     }
361 dl 1.15
362     if (pred == null)
363     return;
364    
365 dl 1.9 /*
366     * At any given time, exactly one node on list cannot be
367     * deleted -- the last inserted node. To accommodate this, if
368     * we cannot delete s, we save its predecessor as "cleanMe",
369     * processing the previously saved version first. At least one
370     * of node s or the node previously saved can always be
371     * processed, so this always terminates.
372     */
373     while (pred.next == s) {
374 jsr166 1.25 Node<E> oldpred = reclean(); // First, help get rid of cleanMe
375     Node<E> t = getValidatedTail();
376 dl 1.9 if (s != t) { // If not tail, try to unsplice
377 jsr166 1.25 Node<E> sn = s.next; // s.next == s means s already off list
378 jsr166 1.10 if (sn == s || pred.casNext(s, sn))
379 dl 1.9 break;
380     }
381     else if (oldpred == pred || // Already saved
382     (oldpred == null && cleanMe.compareAndSet(null, pred)))
383     break; // Postpone cleaning
384     }
385     }
386 jsr166 1.5
387 dl 1.9 /**
388     * Tries to unsplice the cancelled node held in cleanMe that was
389     * previously uncleanable because it was at tail.
390 jsr166 1.17 *
391 dl 1.9 * @return current cleanMe node (or null)
392     */
393 jsr166 1.25 private Node<E> reclean() {
394 jsr166 1.10 /*
395 dl 1.9 * cleanMe is, or at one time was, predecessor of cancelled
396     * node s that was the tail so could not be unspliced. If s
397     * is no longer the tail, try to unsplice if necessary and
398     * make cleanMe slot available. This differs from similar
399     * code in clean() because we must check that pred still
400     * points to a cancelled node that must be unspliced -- if
401     * not, we can (must) clear cleanMe without unsplicing.
402     * This can loop only due to contention on casNext or
403 jsr166 1.10 * clearing cleanMe.
404 dl 1.9 */
405 jsr166 1.25 Node<E> pred;
406 dl 1.9 while ((pred = cleanMe.get()) != null) {
407 jsr166 1.25 Node<E> t = getValidatedTail();
408     Node<E> s = pred.next;
409 jsr166 1.10 if (s != t) {
410 jsr166 1.25 Node<E> sn;
411 dl 1.9 if (s == null || s == pred || s.get() != s ||
412     (sn = s.next) == s || pred.casNext(s, sn))
413     cleanMe.compareAndSet(pred, null);
414 dl 1.1 }
415 dl 1.9 else // s is still tail; cannot clean
416     break;
417 dl 1.1 }
418 dl 1.9 return pred;
419 dl 1.1 }
420 jsr166 1.5
421 dl 1.1 /**
422 jsr166 1.11 * Creates an initially empty {@code LinkedTransferQueue}.
423 dl 1.1 */
424     public LinkedTransferQueue() {
425 jsr166 1.25 Node<E> dummy = new Node<E>(null, false);
426     head = new PaddedAtomicReference<Node<E>>(dummy);
427     tail = new PaddedAtomicReference<Node<E>>(dummy);
428     cleanMe = new PaddedAtomicReference<Node<E>>(null);
429 dl 1.1 }
430    
431     /**
432 jsr166 1.11 * Creates a {@code LinkedTransferQueue}
433 dl 1.1 * initially containing the elements of the given collection,
434     * added in traversal order of the collection's iterator.
435 jsr166 1.17 *
436 dl 1.1 * @param c the collection of elements to initially contain
437     * @throws NullPointerException if the specified collection or any
438     * of its elements are null
439     */
440     public LinkedTransferQueue(Collection<? extends E> c) {
441 dl 1.7 this();
442 dl 1.1 addAll(c);
443     }
444    
445 jsr166 1.29 /**
446     * @throws InterruptedException {@inheritDoc}
447     * @throws NullPointerException {@inheritDoc}
448     */
449 dl 1.1 public void put(E e) throws InterruptedException {
450     if (e == null) throw new NullPointerException();
451     if (Thread.interrupted()) throw new InterruptedException();
452     xfer(e, NOWAIT, 0);
453     }
454    
455 jsr166 1.29 /**
456     * @throws InterruptedException {@inheritDoc}
457     * @throws NullPointerException {@inheritDoc}
458     */
459 jsr166 1.5 public boolean offer(E e, long timeout, TimeUnit unit)
460 dl 1.1 throws InterruptedException {
461     if (e == null) throw new NullPointerException();
462     if (Thread.interrupted()) throw new InterruptedException();
463     xfer(e, NOWAIT, 0);
464     return true;
465     }
466    
467 jsr166 1.29 /**
468     * @throws NullPointerException {@inheritDoc}
469     */
470 dl 1.1 public boolean offer(E e) {
471     if (e == null) throw new NullPointerException();
472     xfer(e, NOWAIT, 0);
473     return true;
474     }
475    
476 jsr166 1.29 /**
477     * @throws NullPointerException {@inheritDoc}
478     */
479 dl 1.15 public boolean add(E e) {
480     if (e == null) throw new NullPointerException();
481     xfer(e, NOWAIT, 0);
482     return true;
483     }
484    
485 jsr166 1.29 /**
486     * @throws InterruptedException {@inheritDoc}
487     * @throws NullPointerException {@inheritDoc}
488     */
489 dl 1.1 public void transfer(E e) throws InterruptedException {
490     if (e == null) throw new NullPointerException();
491     if (xfer(e, WAIT, 0) == null) {
492 jsr166 1.6 Thread.interrupted();
493 dl 1.1 throw new InterruptedException();
494 jsr166 1.6 }
495 dl 1.1 }
496    
497 jsr166 1.29 /**
498     * @throws InterruptedException {@inheritDoc}
499     * @throws NullPointerException {@inheritDoc}
500     */
501 dl 1.1 public boolean tryTransfer(E e, long timeout, TimeUnit unit)
502     throws InterruptedException {
503     if (e == null) throw new NullPointerException();
504     if (xfer(e, TIMEOUT, unit.toNanos(timeout)) != null)
505     return true;
506     if (!Thread.interrupted())
507     return false;
508     throw new InterruptedException();
509     }
510    
511 jsr166 1.29 /**
512     * @throws NullPointerException {@inheritDoc}
513     */
514 dl 1.1 public boolean tryTransfer(E e) {
515     if (e == null) throw new NullPointerException();
516     return fulfill(e) != null;
517     }
518    
519 jsr166 1.29 /**
520     * @throws InterruptedException {@inheritDoc}
521     */
522 dl 1.1 public E take() throws InterruptedException {
523     Object e = xfer(null, WAIT, 0);
524     if (e != null)
525 jsr166 1.22 return (E) e;
526 jsr166 1.6 Thread.interrupted();
527 dl 1.1 throw new InterruptedException();
528     }
529    
530 jsr166 1.29 /**
531     * @throws InterruptedException {@inheritDoc}
532     */
533 dl 1.1 public E poll(long timeout, TimeUnit unit) throws InterruptedException {
534     Object e = xfer(null, TIMEOUT, unit.toNanos(timeout));
535     if (e != null || !Thread.interrupted())
536 jsr166 1.22 return (E) e;
537 dl 1.1 throw new InterruptedException();
538     }
539    
540     public E poll() {
541 jsr166 1.25 return fulfill(null);
542 dl 1.1 }
543    
544 jsr166 1.29 /**
545 jsr166 1.30 * @throws NullPointerException {@inheritDoc}
546     * @throws IllegalArgumentException {@inheritDoc}
547 jsr166 1.29 */
548 dl 1.1 public int drainTo(Collection<? super E> c) {
549     if (c == null)
550     throw new NullPointerException();
551     if (c == this)
552     throw new IllegalArgumentException();
553     int n = 0;
554     E e;
555     while ( (e = poll()) != null) {
556     c.add(e);
557     ++n;
558     }
559     return n;
560     }
561    
562 jsr166 1.29 /**
563 jsr166 1.30 * @throws NullPointerException {@inheritDoc}
564     * @throws IllegalArgumentException {@inheritDoc}
565 jsr166 1.29 */
566 dl 1.1 public int drainTo(Collection<? super E> c, int maxElements) {
567     if (c == null)
568     throw new NullPointerException();
569     if (c == this)
570     throw new IllegalArgumentException();
571     int n = 0;
572     E e;
573     while (n < maxElements && (e = poll()) != null) {
574     c.add(e);
575     ++n;
576     }
577     return n;
578     }
579    
580     // Traversal-based methods
581    
582     /**
583 jsr166 1.17 * Returns head after performing any outstanding helping steps.
584 dl 1.1 */
585 jsr166 1.25 private Node<E> traversalHead() {
586 dl 1.1 for (;;) {
587 jsr166 1.25 Node<E> t = tail.get();
588     Node<E> h = head.get();
589 dl 1.1 if (h != null && t != null) {
590 jsr166 1.25 Node<E> last = t.next;
591     Node<E> first = h.next;
592 dl 1.1 if (t == tail.get()) {
593 jsr166 1.5 if (last != null)
594 dl 1.1 tail.compareAndSet(t, last);
595     else if (first != null) {
596     Object x = first.get();
597 jsr166 1.5 if (x == first)
598     advanceHead(h, first);
599 dl 1.1 else
600     return h;
601     }
602     else
603     return h;
604     }
605     }
606 dl 1.15 reclean();
607 dl 1.1 }
608     }
609    
610    
611     public Iterator<E> iterator() {
612     return new Itr();
613     }
614    
615     /**
616 jsr166 1.4 * Iterators. Basic strategy is to traverse list, treating
617 dl 1.1 * non-data (i.e., request) nodes as terminating list.
618     * Once a valid data node is found, the item is cached
619     * so that the next call to next() will return it even
620     * if subsequently removed.
621     */
622     class Itr implements Iterator<E> {
623 jsr166 1.25 Node<E> next; // node to return next
624     Node<E> pnext; // predecessor of next
625     Node<E> snext; // successor of next
626     Node<E> curr; // last returned node, for remove()
627     Node<E> pcurr; // predecessor of curr, for remove()
628 jsr166 1.18 E nextItem; // Cache of next item, once committed to in next
629 jsr166 1.5
630 dl 1.1 Itr() {
631 dl 1.15 findNext();
632 dl 1.1 }
633 jsr166 1.5
634 dl 1.15 /**
635 jsr166 1.17 * Ensures next points to next valid node, or null if none.
636 dl 1.15 */
637     void findNext() {
638 dl 1.1 for (;;) {
639 jsr166 1.25 Node<E> pred = pnext;
640     Node<E> q = next;
641 dl 1.15 if (pred == null || pred == q) {
642     pred = traversalHead();
643     q = pred.next;
644     }
645     if (q == null || !q.isData) {
646     next = null;
647     return;
648     }
649     Object x = q.get();
650 jsr166 1.25 Node<E> s = q.next;
651 dl 1.15 if (x != null && q != x && q != s) {
652 jsr166 1.22 nextItem = (E) x;
653 dl 1.15 snext = s;
654     pnext = pred;
655     next = q;
656     return;
657 dl 1.1 }
658 dl 1.15 pnext = q;
659     next = s;
660 dl 1.1 }
661     }
662 jsr166 1.5
663 dl 1.1 public boolean hasNext() {
664 dl 1.15 return next != null;
665 dl 1.1 }
666 jsr166 1.5
667 dl 1.1 public E next() {
668 dl 1.15 if (next == null) throw new NoSuchElementException();
669     pcurr = pnext;
670     curr = next;
671     pnext = next;
672     next = snext;
673     E x = nextItem;
674     findNext();
675     return x;
676 dl 1.1 }
677 jsr166 1.5
678 dl 1.1 public void remove() {
679 jsr166 1.25 Node<E> p = curr;
680 dl 1.15 if (p == null)
681 dl 1.1 throw new IllegalStateException();
682     Object x = p.get();
683     if (x != null && x != p && p.compareAndSet(x, p))
684 dl 1.15 clean(pcurr, p);
685 dl 1.1 }
686     }
687    
688     public E peek() {
689     for (;;) {
690 jsr166 1.25 Node<E> h = traversalHead();
691     Node<E> p = h.next;
692 dl 1.1 if (p == null)
693     return null;
694     Object x = p.get();
695     if (p != x) {
696     if (!p.isData)
697     return null;
698     if (x != null)
699 jsr166 1.22 return (E) x;
700 dl 1.1 }
701     }
702     }
703    
704 dl 1.2 public boolean isEmpty() {
705     for (;;) {
706 jsr166 1.25 Node<E> h = traversalHead();
707     Node<E> p = h.next;
708 dl 1.2 if (p == null)
709     return true;
710     Object x = p.get();
711     if (p != x) {
712     if (!p.isData)
713     return true;
714     if (x != null)
715     return false;
716     }
717     }
718     }
719    
720 dl 1.1 public boolean hasWaitingConsumer() {
721     for (;;) {
722 jsr166 1.25 Node<E> h = traversalHead();
723     Node<E> p = h.next;
724 dl 1.1 if (p == null)
725     return false;
726     Object x = p.get();
727 jsr166 1.5 if (p != x)
728 dl 1.1 return !p.isData;
729     }
730     }
731 jsr166 1.5
732 dl 1.1 /**
733     * Returns the number of elements in this queue. If this queue
734 jsr166 1.11 * contains more than {@code Integer.MAX_VALUE} elements, returns
735     * {@code Integer.MAX_VALUE}.
736 dl 1.1 *
737     * <p>Beware that, unlike in most collections, this method is
738     * <em>NOT</em> a constant-time operation. Because of the
739     * asynchronous nature of these queues, determining the current
740     * number of elements requires an O(n) traversal.
741     *
742     * @return the number of elements in this queue
743     */
744     public int size() {
745     int count = 0;
746 jsr166 1.25 Node<E> h = traversalHead();
747     for (Node<E> p = h.next; p != null && p.isData; p = p.next) {
748 dl 1.1 Object x = p.get();
749 jsr166 1.5 if (x != null && x != p) {
750 dl 1.1 if (++count == Integer.MAX_VALUE) // saturated
751     break;
752     }
753     }
754     return count;
755     }
756    
757     public int getWaitingConsumerCount() {
758     int count = 0;
759 jsr166 1.25 Node<E> h = traversalHead();
760     for (Node<E> p = h.next; p != null && !p.isData; p = p.next) {
761 dl 1.1 if (p.get() == null) {
762     if (++count == Integer.MAX_VALUE)
763     break;
764     }
765     }
766     return count;
767     }
768    
769     public int remainingCapacity() {
770     return Integer.MAX_VALUE;
771     }
772    
773 dl 1.15 public boolean remove(Object o) {
774     if (o == null)
775     return false;
776     for (;;) {
777 jsr166 1.25 Node<E> pred = traversalHead();
778 dl 1.15 for (;;) {
779 jsr166 1.25 Node<E> q = pred.next;
780 dl 1.15 if (q == null || !q.isData)
781     return false;
782     if (q == pred) // restart
783     break;
784     Object x = q.get();
785 jsr166 1.17 if (x != null && x != q && o.equals(x) &&
786 dl 1.15 q.compareAndSet(x, q)) {
787     clean(pred, q);
788     return true;
789     }
790     pred = q;
791     }
792     }
793     }
794    
795 dl 1.1 /**
796     * Save the state to a stream (that is, serialize it).
797     *
798 jsr166 1.11 * @serialData All of the elements (each an {@code E}) in
799 dl 1.1 * the proper order, followed by a null
800     * @param s the stream
801     */
802     private void writeObject(java.io.ObjectOutputStream s)
803     throws java.io.IOException {
804     s.defaultWriteObject();
805 jsr166 1.16 for (E e : this)
806     s.writeObject(e);
807 dl 1.1 // Use trailing null as sentinel
808     s.writeObject(null);
809     }
810    
811     /**
812     * Reconstitute the Queue instance from a stream (that is,
813     * deserialize it).
814 jsr166 1.19 *
815 dl 1.1 * @param s the stream
816     */
817     private void readObject(java.io.ObjectInputStream s)
818     throws java.io.IOException, ClassNotFoundException {
819     s.defaultReadObject();
820 dl 1.7 resetHeadAndTail();
821 dl 1.1 for (;;) {
822 jsr166 1.25 @SuppressWarnings("unchecked") E item = (E) s.readObject();
823 dl 1.1 if (item == null)
824     break;
825     else
826     offer(item);
827     }
828     }
829 dl 1.7
830     // Support for resetting head/tail while deserializing
831 dl 1.12 private void resetHeadAndTail() {
832 jsr166 1.25 Node<E> dummy = new Node<E>(null, false);
833 jsr166 1.20 UNSAFE.putObjectVolatile(this, headOffset,
834 jsr166 1.25 new PaddedAtomicReference<Node<E>>(dummy));
835 jsr166 1.20 UNSAFE.putObjectVolatile(this, tailOffset,
836 jsr166 1.25 new PaddedAtomicReference<Node<E>>(dummy));
837 jsr166 1.20 UNSAFE.putObjectVolatile(this, cleanMeOffset,
838 jsr166 1.25 new PaddedAtomicReference<Node<E>>(null));
839 dl 1.12 }
840 dl 1.7
841 jsr166 1.28 // Unsafe mechanics
842    
843     private static final sun.misc.Unsafe UNSAFE = getUnsafe();
844     private static final long headOffset =
845     objectFieldOffset("head", LinkedTransferQueue.class);
846     private static final long tailOffset =
847     objectFieldOffset("tail", LinkedTransferQueue.class);
848     private static final long cleanMeOffset =
849     objectFieldOffset("cleanMe", LinkedTransferQueue.class);
850    
851     private static long objectFieldOffset(String field, Class<?> klazz) {
852     try {
853     return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
854     } catch (NoSuchFieldException e) {
855     // Convert Exception to corresponding Error
856     NoSuchFieldError error = new NoSuchFieldError(field);
857     error.initCause(e);
858     throw error;
859     }
860     }
861    
862     /**
863     * Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package.
864     * Replace with a simple call to Unsafe.getUnsafe when integrating
865     * into a jdk.
866     *
867     * @return a sun.misc.Unsafe
868     */
869 jsr166 1.25 private static sun.misc.Unsafe getUnsafe() {
870 jsr166 1.13 try {
871 jsr166 1.25 return sun.misc.Unsafe.getUnsafe();
872 jsr166 1.13 } catch (SecurityException se) {
873     try {
874     return java.security.AccessController.doPrivileged
875 jsr166 1.28 (new java.security
876     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
877 jsr166 1.25 public sun.misc.Unsafe run() throws Exception {
878 jsr166 1.28 java.lang.reflect.Field f = sun.misc
879     .Unsafe.class.getDeclaredField("theUnsafe");
880     f.setAccessible(true);
881     return (sun.misc.Unsafe) f.get(null);
882 jsr166 1.13 }});
883     } catch (java.security.PrivilegedActionException e) {
884 jsr166 1.25 throw new RuntimeException("Could not initialize intrinsics",
885     e.getCause());
886 jsr166 1.13 }
887     }
888     }
889 dl 1.1 }