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

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, as explained at
4 * http://creativecommons.org/licenses/publicdomain
5 */
6
7 package jsr166y;
8
9 import java.util.concurrent.*;
10
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
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 * <p>Beware that, unlike in most collections, the {@code size}
28 * 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 * @since 1.7
48 * @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 * 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 */
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 static final int maxTimedSpins = (NCPUS < 2) ? 0 : 32;
87
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 /**
102 * 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 */
108 static final class Node<E> extends AtomicReference<Object> {
109 volatile Node<E> next;
110 volatile Thread waiter; // to control park/unpark
111 final boolean isData;
112
113 Node(E item, boolean isData) {
114 super(item);
115 this.isData = isData;
116 }
117
118 @SuppressWarnings("rawtypes")
119 static final AtomicReferenceFieldUpdater<Node, Node>
120 nextUpdater = AtomicReferenceFieldUpdater.newUpdater
121 (Node.class, Node.class, "next");
122
123 final boolean casNext(Node<E> cmp, Node<E> val) {
124 return nextUpdater.compareAndSet(this, cmp, val);
125 }
126
127 final void clearNext() {
128 nextUpdater.lazySet(this, this);
129 }
130
131 private static final long serialVersionUID = -3375979862319811754L;
132 }
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 private static final long serialVersionUID = 8170090609809740854L;
144 }
145
146
147 /** head of the queue */
148 private transient final PaddedAtomicReference<Node<E>> head;
149
150 /** tail of the queue */
151 private transient final PaddedAtomicReference<Node<E>> tail;
152
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 private transient final PaddedAtomicReference<Node<E>> cleanMe;
159
160 /**
161 * Tries to cas nh as new head; if successful, unlink
162 * old head's next node to avoid garbage retention.
163 */
164 private boolean advanceHead(Node<E> h, Node<E> nh) {
165 if (h == head.get() && head.compareAndSet(h, nh)) {
166 h.clearNext(); // forget old next
167 return true;
168 }
169 return false;
170 }
171
172 /**
173 * Puts or takes an item. Used for most queue operations (except
174 * poll() and tryTransfer()). See the similar code in
175 * SynchronousQueue for detailed explanation.
176 *
177 * @param e the item or if null, signifies that this is a take
178 * @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 private E xfer(E e, int mode, long nanos) {
183 boolean isData = (e != null);
184 Node<E> s = null;
185 final PaddedAtomicReference<Node<E>> head = this.head;
186 final PaddedAtomicReference<Node<E>> tail = this.tail;
187
188 for (;;) {
189 Node<E> t = tail.get();
190 Node<E> h = head.get();
191
192 if (t != null && (t == h || t.isData == isData)) {
193 if (s == null)
194 s = new Node<E>(e, isData);
195 Node<E> last = t.next;
196 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
206 else if (h != null) {
207 Node<E> first = h.next;
208 if (t == tail.get() && first != null &&
209 advanceHead(h, first)) {
210 Object x = first.get();
211 if (x != first && first.compareAndSet(x, e)) {
212 LockSupport.unpark(first.waiter);
213 return isData ? e : (E) x;
214 }
215 }
216 }
217 }
218 }
219
220
221 /**
222 * Version of xfer for poll() and tryTransfer, which
223 * simplifies control paths both here and in xfer.
224 */
225 private E fulfill(E e) {
226 boolean isData = (e != null);
227 final PaddedAtomicReference<Node<E>> head = this.head;
228 final PaddedAtomicReference<Node<E>> tail = this.tail;
229
230 for (;;) {
231 Node<E> t = tail.get();
232 Node<E> h = head.get();
233
234 if (t != null && (t == h || t.isData == isData)) {
235 Node<E> last = t.next;
236 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 Node<E> first = h.next;
245 if (t == tail.get() &&
246 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 return isData ? e : (E) x;
252 }
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 private E awaitFulfill(Node<E> pred, Node<E> s, E e,
270 int mode, long nanos) {
271 if (mode == NOWAIT)
272 return null;
273
274 long lastTime = (mode == TIMEOUT) ? System.nanoTime() : 0;
275 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 if (x == s) { // was cancelled
284 clean(pred, s);
285 return null;
286 }
287 else if (x != null) {
288 s.set(s); // avoid garbage retention
289 return (E) x;
290 }
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 Node<E> h = head.get(); // only spin if at head
305 spins = ((h != null && h.next == s) ?
306 ((mode == TIMEOUT) ?
307 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 LockSupport.park(this);
315 s.waiter = null;
316 spins = -1;
317 }
318 else if (nanos > spinForTimeoutThreshold) {
319 LockSupport.parkNanos(this, nanos);
320 s.waiter = null;
321 spins = -1;
322 }
323 }
324 }
325
326 /**
327 * Returns validated tail for use in cleaning methods.
328 */
329 private Node<E> getValidatedTail() {
330 for (;;) {
331 Node<E> h = head.get();
332 Node<E> first = h.next;
333 if (first != null && first.next == first) { // help advance
334 advanceHead(h, first);
335 continue;
336 }
337 Node<E> t = tail.get();
338 Node<E> last = t.next;
339 if (t == tail.get()) {
340 if (last != null)
341 tail.compareAndSet(t, last); // help advance
342 else
343 return t;
344 }
345 }
346 }
347
348 /**
349 * Gets rid of cancelled node s with original predecessor pred.
350 *
351 * @param pred predecessor of cancelled node
352 * @param s the cancelled node
353 */
354 private void clean(Node<E> pred, Node<E> s) {
355 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
362 if (pred == null)
363 return;
364
365 /*
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 Node<E> oldpred = reclean(); // First, help get rid of cleanMe
375 Node<E> t = getValidatedTail();
376 if (s != t) { // If not tail, try to unsplice
377 Node<E> sn = s.next; // s.next == s means s already off list
378 if (sn == s || pred.casNext(s, sn))
379 break;
380 }
381 else if (oldpred == pred || // Already saved
382 (oldpred == null && cleanMe.compareAndSet(null, pred)))
383 break; // Postpone cleaning
384 }
385 }
386
387 /**
388 * Tries to unsplice the cancelled node held in cleanMe that was
389 * previously uncleanable because it was at tail.
390 *
391 * @return current cleanMe node (or null)
392 */
393 private Node<E> reclean() {
394 /*
395 * 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 * clearing cleanMe.
404 */
405 Node<E> pred;
406 while ((pred = cleanMe.get()) != null) {
407 Node<E> t = getValidatedTail();
408 Node<E> s = pred.next;
409 if (s != t) {
410 Node<E> sn;
411 if (s == null || s == pred || s.get() != s ||
412 (sn = s.next) == s || pred.casNext(s, sn))
413 cleanMe.compareAndSet(pred, null);
414 }
415 else // s is still tail; cannot clean
416 break;
417 }
418 return pred;
419 }
420
421 /**
422 * Creates an initially empty {@code LinkedTransferQueue}.
423 */
424 public LinkedTransferQueue() {
425 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 }
430
431 /**
432 * Creates a {@code LinkedTransferQueue}
433 * initially containing the elements of the given collection,
434 * added in traversal order of the collection's iterator.
435 *
436 * @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 this();
442 addAll(c);
443 }
444
445 /**
446 * @throws InterruptedException {@inheritDoc}
447 * @throws NullPointerException {@inheritDoc}
448 */
449 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 /**
456 * @throws InterruptedException {@inheritDoc}
457 * @throws NullPointerException {@inheritDoc}
458 */
459 public boolean offer(E e, long timeout, TimeUnit unit)
460 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 /**
468 * @throws NullPointerException {@inheritDoc}
469 */
470 public boolean offer(E e) {
471 if (e == null) throw new NullPointerException();
472 xfer(e, NOWAIT, 0);
473 return true;
474 }
475
476 /**
477 * @throws NullPointerException {@inheritDoc}
478 */
479 public boolean add(E e) {
480 if (e == null) throw new NullPointerException();
481 xfer(e, NOWAIT, 0);
482 return true;
483 }
484
485 /**
486 * @throws InterruptedException {@inheritDoc}
487 * @throws NullPointerException {@inheritDoc}
488 */
489 public void transfer(E e) throws InterruptedException {
490 if (e == null) throw new NullPointerException();
491 if (xfer(e, WAIT, 0) == null) {
492 Thread.interrupted();
493 throw new InterruptedException();
494 }
495 }
496
497 /**
498 * @throws InterruptedException {@inheritDoc}
499 * @throws NullPointerException {@inheritDoc}
500 */
501 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 /**
512 * @throws NullPointerException {@inheritDoc}
513 */
514 public boolean tryTransfer(E e) {
515 if (e == null) throw new NullPointerException();
516 return fulfill(e) != null;
517 }
518
519 /**
520 * @throws InterruptedException {@inheritDoc}
521 */
522 public E take() throws InterruptedException {
523 Object e = xfer(null, WAIT, 0);
524 if (e != null)
525 return (E) e;
526 Thread.interrupted();
527 throw new InterruptedException();
528 }
529
530 /**
531 * @throws InterruptedException {@inheritDoc}
532 */
533 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 return (E) e;
537 throw new InterruptedException();
538 }
539
540 public E poll() {
541 return fulfill(null);
542 }
543
544 /**
545 * @throws NullPointerException {@inheritDoc}
546 * @throwsIllegalArgumentException {@inheritDoc}
547 */
548 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 /**
563 * @throws NullPointerException {@inheritDoc}
564 * @throwsIllegalArgumentException {@inheritDoc}
565 */
566 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 * Returns head after performing any outstanding helping steps.
584 */
585 private Node<E> traversalHead() {
586 for (;;) {
587 Node<E> t = tail.get();
588 Node<E> h = head.get();
589 if (h != null && t != null) {
590 Node<E> last = t.next;
591 Node<E> first = h.next;
592 if (t == tail.get()) {
593 if (last != null)
594 tail.compareAndSet(t, last);
595 else if (first != null) {
596 Object x = first.get();
597 if (x == first)
598 advanceHead(h, first);
599 else
600 return h;
601 }
602 else
603 return h;
604 }
605 }
606 reclean();
607 }
608 }
609
610
611 public Iterator<E> iterator() {
612 return new Itr();
613 }
614
615 /**
616 * Iterators. Basic strategy is to traverse list, treating
617 * 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 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 E nextItem; // Cache of next item, once committed to in next
629
630 Itr() {
631 findNext();
632 }
633
634 /**
635 * Ensures next points to next valid node, or null if none.
636 */
637 void findNext() {
638 for (;;) {
639 Node<E> pred = pnext;
640 Node<E> q = next;
641 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 Node<E> s = q.next;
651 if (x != null && q != x && q != s) {
652 nextItem = (E) x;
653 snext = s;
654 pnext = pred;
655 next = q;
656 return;
657 }
658 pnext = q;
659 next = s;
660 }
661 }
662
663 public boolean hasNext() {
664 return next != null;
665 }
666
667 public E next() {
668 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 }
677
678 public void remove() {
679 Node<E> p = curr;
680 if (p == null)
681 throw new IllegalStateException();
682 Object x = p.get();
683 if (x != null && x != p && p.compareAndSet(x, p))
684 clean(pcurr, p);
685 }
686 }
687
688 public E peek() {
689 for (;;) {
690 Node<E> h = traversalHead();
691 Node<E> p = h.next;
692 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 return (E) x;
700 }
701 }
702 }
703
704 public boolean isEmpty() {
705 for (;;) {
706 Node<E> h = traversalHead();
707 Node<E> p = h.next;
708 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 public boolean hasWaitingConsumer() {
721 for (;;) {
722 Node<E> h = traversalHead();
723 Node<E> p = h.next;
724 if (p == null)
725 return false;
726 Object x = p.get();
727 if (p != x)
728 return !p.isData;
729 }
730 }
731
732 /**
733 * Returns the number of elements in this queue. If this queue
734 * contains more than {@code Integer.MAX_VALUE} elements, returns
735 * {@code Integer.MAX_VALUE}.
736 *
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 Node<E> h = traversalHead();
747 for (Node<E> p = h.next; p != null && p.isData; p = p.next) {
748 Object x = p.get();
749 if (x != null && x != p) {
750 if (++count == Integer.MAX_VALUE) // saturated
751 break;
752 }
753 }
754 return count;
755 }
756
757 public int getWaitingConsumerCount() {
758 int count = 0;
759 Node<E> h = traversalHead();
760 for (Node<E> p = h.next; p != null && !p.isData; p = p.next) {
761 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 public boolean remove(Object o) {
774 if (o == null)
775 return false;
776 for (;;) {
777 Node<E> pred = traversalHead();
778 for (;;) {
779 Node<E> q = pred.next;
780 if (q == null || !q.isData)
781 return false;
782 if (q == pred) // restart
783 break;
784 Object x = q.get();
785 if (x != null && x != q && o.equals(x) &&
786 q.compareAndSet(x, q)) {
787 clean(pred, q);
788 return true;
789 }
790 pred = q;
791 }
792 }
793 }
794
795 /**
796 * Save the state to a stream (that is, serialize it).
797 *
798 * @serialData All of the elements (each an {@code E}) in
799 * 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 for (E e : this)
806 s.writeObject(e);
807 // 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 *
815 * @param s the stream
816 */
817 private void readObject(java.io.ObjectInputStream s)
818 throws java.io.IOException, ClassNotFoundException {
819 s.defaultReadObject();
820 resetHeadAndTail();
821 for (;;) {
822 @SuppressWarnings("unchecked") E item = (E) s.readObject();
823 if (item == null)
824 break;
825 else
826 offer(item);
827 }
828 }
829
830 // Support for resetting head/tail while deserializing
831 private void resetHeadAndTail() {
832 Node<E> dummy = new Node<E>(null, false);
833 UNSAFE.putObjectVolatile(this, headOffset,
834 new PaddedAtomicReference<Node<E>>(dummy));
835 UNSAFE.putObjectVolatile(this, tailOffset,
836 new PaddedAtomicReference<Node<E>>(dummy));
837 UNSAFE.putObjectVolatile(this, cleanMeOffset,
838 new PaddedAtomicReference<Node<E>>(null));
839 }
840
841 // 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 private static sun.misc.Unsafe getUnsafe() {
870 try {
871 return sun.misc.Unsafe.getUnsafe();
872 } catch (SecurityException se) {
873 try {
874 return java.security.AccessController.doPrivileged
875 (new java.security
876 .PrivilegedExceptionAction<sun.misc.Unsafe>() {
877 public sun.misc.Unsafe run() throws Exception {
878 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 }});
883 } catch (java.security.PrivilegedActionException e) {
884 throw new RuntimeException("Could not initialize intrinsics",
885 e.getCause());
886 }
887 }
888 }
889 }