ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jsr166y/LinkedTransferQueue.java
(Generate patch)

Comparing jsr166/src/jsr166y/LinkedTransferQueue.java (file contents):
Revision 1.12 by dl, Mon Jan 12 17:16:18 2009 UTC vs.
Revision 1.21 by jsr166, Wed Jul 22 01:36:51 2009 UTC

# Line 44 | Line 44 | import java.lang.reflect.*;
44   * @since 1.7
45   * @author Doug Lea
46   * @param <E> the type of elements held in this collection
47 – *
47   */
48   public class LinkedTransferQueue<E> extends AbstractQueue<E>
49      implements TransferQueue<E>, java.io.Serializable {
# Line 116 | Line 115 | public class LinkedTransferQueue<E> exte
115              nextUpdater = AtomicReferenceFieldUpdater.newUpdater
116              (QNode.class, QNode.class, "next");
117  
118 <        boolean casNext(QNode cmp, QNode val) {
118 >        final boolean casNext(QNode cmp, QNode val) {
119              return nextUpdater.compareAndSet(this, cmp, val);
120          }
121 +
122 +        final void clearNext() {
123 +            nextUpdater.lazySet(this, this);
124 +        }
125 +
126      }
127  
128      /**
# Line 151 | Line 155 | public class LinkedTransferQueue<E> exte
155       */
156      private boolean advanceHead(QNode h, QNode nh) {
157          if (h == head.get() && head.compareAndSet(h, nh)) {
158 <            h.next = h; // forget old next
158 >            h.clearNext(); // forget old next
159              return true;
160          }
161          return false;
# Line 161 | Line 165 | public class LinkedTransferQueue<E> exte
165       * Puts or takes an item. Used for most queue operations (except
166       * poll() and tryTransfer()). See the similar code in
167       * SynchronousQueue for detailed explanation.
168 +     *
169       * @param e the item or if null, signifies that this is a take
170       * @param mode the wait mode: NOWAIT, TIMEOUT, WAIT
171       * @param nanos timeout in nanosecs, used only if mode is TIMEOUT
# Line 207 | Line 212 | public class LinkedTransferQueue<E> exte
212  
213      /**
214       * Version of xfer for poll() and tryTransfer, which
215 <     * simplifies control paths both here and in xfer
215 >     * simplifies control paths both here and in xfer.
216       */
217      private Object fulfill(Object e) {
218          boolean isData = (e != null);
# Line 267 | Line 272 | public class LinkedTransferQueue<E> exte
272              Object x = s.get();
273              if (x != e) {                 // Node was matched or cancelled
274                  advanceHead(pred, s);     // unlink if head
275 <                if (x == s) {              // was cancelled
275 >                if (x == s) {             // was cancelled
276                      clean(pred, s);
277                      return null;
278                  }
# Line 311 | Line 316 | public class LinkedTransferQueue<E> exte
316      }
317  
318      /**
319 <     * Returns validated tail for use in cleaning methods
319 >     * Returns validated tail for use in cleaning methods.
320       */
321      private QNode getValidatedTail() {
322          for (;;) {
# Line 334 | Line 339 | public class LinkedTransferQueue<E> exte
339  
340      /**
341       * Gets rid of cancelled node s with original predecessor pred.
342 +     *
343       * @param pred predecessor of cancelled node
344       * @param s the cancelled node
345       */
# Line 344 | Line 350 | public class LinkedTransferQueue<E> exte
350              if (w != Thread.currentThread())
351                  LockSupport.unpark(w);
352          }
353 +
354 +        if (pred == null)
355 +            return;
356 +
357          /*
358           * At any given time, exactly one node on list cannot be
359           * deleted -- the last inserted node. To accommodate this, if
# Line 369 | Line 379 | public class LinkedTransferQueue<E> exte
379      /**
380       * Tries to unsplice the cancelled node held in cleanMe that was
381       * previously uncleanable because it was at tail.
382 +     *
383       * @return current cleanMe node (or null)
384       */
385      private QNode reclean() {
# Line 413 | Line 424 | public class LinkedTransferQueue<E> exte
424       * Creates a {@code LinkedTransferQueue}
425       * initially containing the elements of the given collection,
426       * added in traversal order of the collection's iterator.
427 +     *
428       * @param c the collection of elements to initially contain
429       * @throws NullPointerException if the specified collection or any
430       *         of its elements are null
# Line 442 | Line 454 | public class LinkedTransferQueue<E> exte
454          return true;
455      }
456  
457 +    public boolean add(E e) {
458 +        if (e == null) throw new NullPointerException();
459 +        xfer(e, NOWAIT, 0);
460 +        return true;
461 +    }
462 +
463      public void transfer(E e) throws InterruptedException {
464          if (e == null) throw new NullPointerException();
465          if (xfer(e, WAIT, 0) == null) {
# Line 515 | Line 533 | public class LinkedTransferQueue<E> exte
533      // Traversal-based methods
534  
535      /**
536 <     * Return head after performing any outstanding helping steps
536 >     * Returns head after performing any outstanding helping steps.
537       */
538      private QNode traversalHead() {
539          for (;;) {
# Line 538 | Line 556 | public class LinkedTransferQueue<E> exte
556                          return h;
557                  }
558              }
559 +            reclean();
560          }
561      }
562  
# Line 554 | Line 573 | public class LinkedTransferQueue<E> exte
573       * if subsequently removed.
574       */
575      class Itr implements Iterator<E> {
576 <        QNode nextNode;    // Next node to return next
577 <        QNode currentNode; // last returned node, for remove()
578 <        QNode prevNode;    // predecessor of last returned node
579 <        E nextItem;        // Cache of next item, once commited to in next
576 >        QNode next;        // node to return next
577 >        QNode pnext;       // predecessor of next
578 >        QNode snext;       // successor of next
579 >        QNode curr;        // last returned node, for remove()
580 >        QNode pcurr;       // predecessor of curr, for remove()
581 >        E nextItem;        // Cache of next item, once committed to in next
582  
583          Itr() {
584 <            nextNode = traversalHead();
564 <            advance();
584 >            findNext();
585          }
586  
587 <        E advance() {
588 <            prevNode = currentNode;
589 <            currentNode = nextNode;
590 <            E x = nextItem;
571 <
572 <            QNode p = nextNode.next;
587 >        /**
588 >         * Ensures next points to next valid node, or null if none.
589 >         */
590 >        void findNext() {
591              for (;;) {
592 <                if (p == null || !p.isData) {
593 <                    nextNode = null;
594 <                    nextItem = null;
595 <                    return x;
592 >                QNode pred = pnext;
593 >                QNode q = next;
594 >                if (pred == null || pred == q) {
595 >                    pred = traversalHead();
596 >                    q = pred.next;
597 >                }
598 >                if (q == null || !q.isData) {
599 >                    next = null;
600 >                    return;
601 >                }
602 >                Object x = q.get();
603 >                QNode s = q.next;
604 >                if (x != null && q != x && q != s) {
605 >                    nextItem = (E)x;
606 >                    snext = s;
607 >                    pnext = pred;
608 >                    next = q;
609 >                    return;
610                  }
611 <                Object item = p.get();
612 <                if (item != p && item != null) {
581 <                    nextNode = p;
582 <                    nextItem = (E)item;
583 <                    return x;
584 <                }
585 <                prevNode = p;
586 <                p = p.next;
611 >                pnext = q;
612 >                next = s;
613              }
614          }
615  
616          public boolean hasNext() {
617 <            return nextNode != null;
617 >            return next != null;
618          }
619  
620          public E next() {
621 <            if (nextNode == null) throw new NoSuchElementException();
622 <            return advance();
621 >            if (next == null) throw new NoSuchElementException();
622 >            pcurr = pnext;
623 >            curr = next;
624 >            pnext = next;
625 >            next = snext;
626 >            E x = nextItem;
627 >            findNext();
628 >            return x;
629          }
630  
631          public void remove() {
632 <            QNode p = currentNode;
633 <            QNode prev = prevNode;
602 <            if (prev == null || p == null)
632 >            QNode p = curr;
633 >            if (p == null)
634                  throw new IllegalStateException();
635              Object x = p.get();
636              if (x != null && x != p && p.compareAndSet(x, p))
637 <                clean(prev, p);
637 >                clean(pcurr, p);
638          }
639      }
640  
# Line 692 | Line 723 | public class LinkedTransferQueue<E> exte
723          return Integer.MAX_VALUE;
724      }
725  
726 +    public boolean remove(Object o) {
727 +        if (o == null)
728 +            return false;
729 +        for (;;) {
730 +            QNode pred = traversalHead();
731 +            for (;;) {
732 +                QNode q = pred.next;
733 +                if (q == null || !q.isData)
734 +                    return false;
735 +                if (q == pred) // restart
736 +                    break;
737 +                Object x = q.get();
738 +                if (x != null && x != q && o.equals(x) &&
739 +                    q.compareAndSet(x, q)) {
740 +                    clean(pred, q);
741 +                    return true;
742 +                }
743 +                pred = q;
744 +            }
745 +        }
746 +    }
747 +
748      /**
749       * Save the state to a stream (that is, serialize it).
750       *
# Line 702 | Line 755 | public class LinkedTransferQueue<E> exte
755      private void writeObject(java.io.ObjectOutputStream s)
756          throws java.io.IOException {
757          s.defaultWriteObject();
758 <        for (Iterator<E> it = iterator(); it.hasNext(); )
759 <            s.writeObject(it.next());
758 >        for (E e : this)
759 >            s.writeObject(e);
760          // Use trailing null as sentinel
761          s.writeObject(null);
762      }
# Line 711 | Line 764 | public class LinkedTransferQueue<E> exte
764      /**
765       * Reconstitute the Queue instance from a stream (that is,
766       * deserialize it).
767 +     *
768       * @param s the stream
769       */
770      private void readObject(java.io.ObjectInputStream s)
# Line 730 | Line 784 | public class LinkedTransferQueue<E> exte
784      // Support for resetting head/tail while deserializing
785      private void resetHeadAndTail() {
786          QNode dummy = new QNode(null, false);
787 <        _unsafe.putObjectVolatile(this, headOffset,
787 >        UNSAFE.putObjectVolatile(this, headOffset,
788                                    new PaddedAtomicReference<QNode>(dummy));
789 <        _unsafe.putObjectVolatile(this, tailOffset,
789 >        UNSAFE.putObjectVolatile(this, tailOffset,
790                                    new PaddedAtomicReference<QNode>(dummy));
791 <        _unsafe.putObjectVolatile(this, cleanMeOffset,
791 >        UNSAFE.putObjectVolatile(this, cleanMeOffset,
792                                    new PaddedAtomicReference<QNode>(null));
793      }
794  
795      // Temporary Unsafe mechanics for preliminary release
796 <    private static final Unsafe _unsafe;
796 >    private static Unsafe getUnsafe() throws Throwable {
797 >        try {
798 >            return Unsafe.getUnsafe();
799 >        } catch (SecurityException se) {
800 >            try {
801 >                return java.security.AccessController.doPrivileged
802 >                    (new java.security.PrivilegedExceptionAction<Unsafe>() {
803 >                        public Unsafe run() throws Exception {
804 >                            return getUnsafePrivileged();
805 >                        }});
806 >            } catch (java.security.PrivilegedActionException e) {
807 >                throw e.getCause();
808 >            }
809 >        }
810 >    }
811 >
812 >    private static Unsafe getUnsafePrivileged()
813 >            throws NoSuchFieldException, IllegalAccessException {
814 >        Field f = Unsafe.class.getDeclaredField("theUnsafe");
815 >        f.setAccessible(true);
816 >        return (Unsafe) f.get(null);
817 >    }
818 >
819 >    private static long fieldOffset(String fieldName)
820 >            throws NoSuchFieldException {
821 >        return UNSAFE.objectFieldOffset
822 >            (LinkedTransferQueue.class.getDeclaredField(fieldName));
823 >    }
824 >
825 >    private static final Unsafe UNSAFE;
826      private static final long headOffset;
827      private static final long tailOffset;
828      private static final long cleanMeOffset;
829      static {
830          try {
831 <            if (LinkedTransferQueue.class.getClassLoader() != null) {
832 <                Field f = Unsafe.class.getDeclaredField("theUnsafe");
833 <                f.setAccessible(true);
834 <                _unsafe = (Unsafe)f.get(null);
835 <            }
753 <            else
754 <                _unsafe = Unsafe.getUnsafe();
755 <            headOffset = _unsafe.objectFieldOffset
756 <                (LinkedTransferQueue.class.getDeclaredField("head"));
757 <            tailOffset = _unsafe.objectFieldOffset
758 <                (LinkedTransferQueue.class.getDeclaredField("tail"));
759 <            cleanMeOffset = _unsafe.objectFieldOffset
760 <                (LinkedTransferQueue.class.getDeclaredField("cleanMe"));
761 <        } catch (Exception e) {
831 >            UNSAFE = getUnsafe();
832 >            headOffset = fieldOffset("head");
833 >            tailOffset = fieldOffset("tail");
834 >            cleanMeOffset = fieldOffset("cleanMe");
835 >        } catch (Throwable e) {
836              throw new RuntimeException("Could not initialize intrinsics", e);
837          }
838      }

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines