[cvs] / jsr166 / src / main / java / util / concurrent / LinkedTransferQueue.java Repository:
ViewVC logotype

Diff of /jsr166/src/main/java/util/concurrent/LinkedTransferQueue.java

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 1.7, Tue Aug 4 20:41:40 2009 UTC revision 1.8, Tue Oct 27 23:21:32 2009 UTC
# Line 13  Line 13 
13  import java.util.NoSuchElementException;  import java.util.NoSuchElementException;
14  import java.util.Queue;  import java.util.Queue;
15  import java.util.concurrent.locks.LockSupport;  import java.util.concurrent.locks.LockSupport;
 import java.util.concurrent.atomic.AtomicReference;  
   
16  /**  /**
17   * An unbounded {@link TransferQueue} based on linked nodes.   * An unbounded {@link TransferQueue} based on linked nodes.
18   * This queue orders elements FIFO (first-in-first-out) with respect   * This queue orders elements FIFO (first-in-first-out) with respect
# Line 52  Line 50 
50      private static final long serialVersionUID = -3223113410248163686L;      private static final long serialVersionUID = -3223113410248163686L;
51    
52      /*      /*
53       * This class extends the approach used in FIFO-mode       * *** Overview of Dual Queues with Slack ***
      * SynchronousQueues. See the internal documentation, as well as  
      * the PPoPP 2006 paper "Scalable Synchronous Queues" by Scherer,  
      * Lea & Scott  
      * (http://www.cs.rice.edu/~wns1/papers/2006-PPoPP-SQ.pdf)  
54       *       *
55       * The main extension is to provide different Wait modes for the       * Dual Queues, introduced by Scherer and Scott
56       * main "xfer" method that puts or takes items.  These don't       * (http://www.cs.rice.edu/~wns1/papers/2004-DISC-DDS.pdf) are
57       * impact the basic dual-queue logic, but instead control whether       * (linked) queues in which nodes may represent either data or
58       * or how threads block upon insertion of request or data nodes       * requests.  When a thread tries to enqueue a data node, but
59       * into the dual queue. It also uses slightly different       * encounters a request node, it instead "matches" and removes it;
60       * conventions for tracking whether nodes are off-list or       * and vice versa for enqueuing requests. Blocking Dual Queues
61       * cancelled.       * arrange that threads enqueuing unmatched requests block until
62       */       * other threads provide the match. Dual Synchronous Queues (see
63         * Scherer, Lea, & Scott
64         * http://www.cs.rochester.edu/u/scott/papers/2009_Scherer_CACM_SSQ.pdf)
65         * additionally arrange that threads enqueuing unmatched data also
66         * block.  Dual Transfer Queues support all of these modes, as
67         * dictated by callers.
68         *
69         * A FIFO dual queue may be implemented using a variation of the
70         * Michael & Scott (M&S) lock-free queue algorithm
71         * (http://www.cs.rochester.edu/u/scott/papers/1996_PODC_queues.pdf).
72         * It maintains two pointer fields, "head", pointing to a
73         * (matched) node that in turn points to the first actual
74         * (unmatched) queue node (or null if empty); and "tail" that
75         * points to the last node on the queue (or again null if
76         * empty). For example, here is a possible queue with four data
77         * elements:
78         *
79         *  head                tail
80         *    |                   |
81         *    v                   v
82         *    M -> U -> U -> U -> U
83         *
84         * The M&S queue algorithm is known to be prone to scalability and
85         * overhead limitations when maintaining (via CAS) these head and
86         * tail pointers. This has led to the development of
87         * contention-reducing variants such as elimination arrays (see
88         * Moir et al http://portal.acm.org/citation.cfm?id=1074013) and
89         * optimistic back pointers (see Ladan-Mozes & Shavit
90         * http://people.csail.mit.edu/edya/publications/OptimisticFIFOQueue-journal.pdf).
91         * However, the nature of dual queues enables a simpler tactic for
92         * improving M&S-style implementations when dual-ness is needed.
93         *
94         * In a dual queue, each node must atomically maintain its match
95         * status. While there are other possible variants, we implement
96         * this here as: for a data-mode node, matching entails CASing an
97         * "item" field from a non-null data value to null upon match, and
98         * vice-versa for request nodes, CASing from null to a data
99         * value. (Note that the linearization properties of this style of
100         * queue are easy to verify -- elements are made available by
101         * linking, and unavailable by matching.) Compared to plain M&S
102         * queues, this property of dual queues requires one additional
103         * successful atomic operation per enq/deq pair. But it also
104         * enables lower cost variants of queue maintenance mechanics. (A
105         * variation of this idea applies even for non-dual queues that
106         * support deletion of interior elements, such as
107         * j.u.c.ConcurrentLinkedQueue.)
108         *
109         * Once a node is matched, its match status can never again
110         * change.  We may thus arrange that the linked list of them
111         * contain a prefix of zero or more matched nodes, followed by a
112         * suffix of zero or more unmatched nodes. (Note that we allow
113         * both the prefix and suffix to be zero length, which in turn
114         * means that we do not use a dummy header.)  If we were not
115         * concerned with either time or space efficiency, we could
116         * correctly perform enqueue and dequeue operations by traversing
117         * from a pointer to the initial node; CASing the item of the
118         * first unmatched node on match and CASing the next field of the
119         * trailing node on appends. (Plus some special-casing when
120         * initially empty).  While this would be a terrible idea in
121         * itself, it does have the benefit of not requiring ANY atomic
122         * updates on head/tail fields.
123         *
124         * We introduce here an approach that lies between the extremes of
125         * never versus always updating queue (head and tail) pointers.
126         * This offers a tradeoff between sometimes requiring extra
127         * traversal steps to locate the first and/or last unmatched
128         * nodes, versus the reduced overhead and contention of fewer
129         * updates to queue pointers. For example, a possible snapshot of
130         * a queue is:
131         *
132         *  head           tail
133         *    |              |
134         *    v              v
135         *    M -> M -> U -> U -> U -> U
136         *
137         * The best value for this "slack" (the targeted maximum distance
138         * between the value of "head" and the first unmatched node, and
139         * similarly for "tail") is an empirical matter. We have found
140         * that using very small constants in the range of 1-3 work best
141         * over a range of platforms. Larger values introduce increasing
142         * costs of cache misses and risks of long traversal chains, while
143         * smaller values increase CAS contention and overhead.
144         *
145         * Dual queues with slack differ from plain M&S dual queues by
146         * virtue of only sometimes updating head or tail pointers when
147         * matching, appending, or even traversing nodes; in order to
148         * maintain a targeted slack.  The idea of "sometimes" may be
149         * operationalized in several ways. The simplest is to use a
150         * per-operation counter incremented on each traversal step, and
151         * to try (via CAS) to update the associated queue pointer
152         * whenever the count exceeds a threshold. Another, that requires
153         * more overhead, is to use random number generators to update
154         * with a given probability per traversal step.
155         *
156         * In any strategy along these lines, because CASes updating
157         * fields may fail, the actual slack may exceed targeted
158         * slack. However, they may be retried at any time to maintain
159         * targets.  Even when using very small slack values, this
160         * approach works well for dual queues because it allows all
161         * operations up to the point of matching or appending an item
162         * (hence potentially allowing progress by another thread) to be
163         * read-only, thus not introducing any further contention. As
164         * described below, we implement this by performing slack
165         * maintenance retries only after these points.
166         *
167         * As an accompaniment to such techniques, traversal overhead can
168         * be further reduced without increasing contention of head
169         * pointer updates: Threads may sometimes shortcut the "next" link
170         * path from the current "head" node to be closer to the currently
171         * known first unmatched node, and similarly for tail. Again, this
172         * may be triggered with using thresholds or randomization.
173         *
174         * These ideas must be further extended to avoid unbounded amounts
175         * of costly-to-reclaim garbage caused by the sequential "next"
176         * links of nodes starting at old forgotten head nodes: As first
177         * described in detail by Boehm
178         * (http://portal.acm.org/citation.cfm?doid=503272.503282) if a GC
179         * delays noticing that any arbitrarily old node has become
180         * garbage, all newer dead nodes will also be unreclaimed.
181         * (Similar issues arise in non-GC environments.)  To cope with
182         * this in our implementation, upon CASing to advance the head
183         * pointer, we set the "next" link of the previous head to point
184         * only to itself; thus limiting the length of connected dead lists.
185         * (We also take similar care to wipe out possibly garbage
186         * retaining values held in other Node fields.)  However, doing so
187         * adds some further complexity to traversal: If any "next"
188         * pointer links to itself, it indicates that the current thread
189         * has lagged behind a head-update, and so the traversal must
190         * continue from the "head".  Traversals trying to find the
191         * current tail starting from "tail" may also encounter
192         * self-links, in which case they also continue at "head".
193         *
194         * It is tempting in slack-based scheme to not even use CAS for
195         * updates (similarly to Ladan-Mozes & Shavit). However, this
196         * cannot be done for head updates under the above link-forgetting
197         * mechanics because an update may leave head at a detached node.
198         * And while direct writes are possible for tail updates, they
199         * increase the risk of long retraversals, and hence long garbage
200         * chains, which can be much more costly than is worthwhile
201         * considering that the cost difference of performing a CAS vs
202         * write is smaller when they are not triggered on each operation
203         * (especially considering that writes and CASes equally require
204         * additional GC bookkeeping ("write barriers") that are sometimes
205         * more costly than the writes themselves because of contention).
206         *
207         * Removal of interior nodes (due to timed out or interrupted
208         * waits, or calls to remove(x) or Iterator.remove) can use a
209         * scheme roughly similar to that described in Scherer, Lea, and
210         * Scott's SynchronousQueue. Given a predecessor, we can unsplice
211         * any node except the (actual) tail of the queue. To avoid
212         * build-up of cancelled trailing nodes, upon a request to remove
213         * a trailing node, it is placed in field "cleanMe" to be
214         * unspliced upon the next call to unsplice any other node.
215         * Situations needing such mechanics are not common but do occur
216         * in practice; for example when an unbounded series of short
217         * timed calls to poll repeatedly time out but never otherwise
218         * fall off the list because of an untimed call to take at the
219         * front of the queue. Note that maintaining field cleanMe does
220         * not otherwise much impact garbage retention even if never
221         * cleared by some other call because the held node will
222         * eventually either directly or indirectly lead to a self-link
223         * once off the list.
224         *
225         * *** Overview of implementation ***
226         *
227         * We use a threshold-based approach to updates, with a slack
228         * threshold of two -- that is, we update head/tail when the
229         * current pointer appears to be two or more steps away from the
230         * first/last node. The slack value is hard-wired: a path greater
231         * than one is naturally implemented by checking equality of
232         * traversal pointers except when the list has only one element,
233         * in which case we keep slack threshold at one. Avoiding tracking
234         * explicit counts across method calls slightly simplifies an
235         * already-messy implementation. Using randomization would
236         * probably work better if there were a low-quality dirt-cheap
237         * per-thread one available, but even ThreadLocalRandom is too
238         * heavy for these purposes.
239         *
240         * With such a small slack threshold value, it is rarely
241         * worthwhile to augment this with path short-circuiting; i.e.,
242         * unsplicing nodes between head and the first unmatched node, or
243         * similarly for tail, rather than advancing head or tail
244         * proper. However, it is used (in awaitMatch) immediately before
245         * a waiting thread starts to block, as a final bit of helping at
246         * a point when contention with others is extremely unlikely
247         * (since if other threads that could release it are operating,
248         * then the current thread wouldn't be blocking).
249         *
250         * We allow both the head and tail fields to be null before any
251         * nodes are enqueued; initializing upon first append.  This
252         * simplifies some other logic, as well as providing more
253         * efficient explicit control paths instead of letting JVMs insert
254         * implicit NullPointerExceptions when they are null.  While not
255         * currently fully implemented, we also leave open the possibility
256         * of re-nulling these fields when empty (which is complicated to
257         * arrange, for little benefit.)
258         *
259         * All enqueue/dequeue operations are handled by the single method
260         * "xfer" with parameters indicating whether to act as some form
261         * of offer, put, poll, take, or transfer (each possibly with
262         * timeout). The relative complexity of using one monolithic
263         * method outweighs the code bulk and maintenance problems of
264         * using separate methods for each case.
265         *
266         * Operation consists of up to three phases. The first is
267         * implemented within method xfer, the second in tryAppend, and
268         * the third in method awaitMatch.
269         *
270         * 1. Try to match an existing node
271         *
272         *    Starting at head, skip already-matched nodes until finding
273         *    an unmatched node of opposite mode, if one exists, in which
274         *    case matching it and returning, also if necessary updating
275         *    head to one past the matched node (or the node itself if the
276         *    list has no other unmatched nodes). If the CAS misses, then
277         *    a loop retries advancing head by two steps until either
278         *    success or the slack is at most two. By requiring that each
279         *    attempt advances head by two (if applicable), we ensure that
280         *    the slack does not grow without bound. Traversals also check
281         *    if the initial head is now off-list, in which case they
282         *    start at the new head.
283         *
284         *    If no candidates are found and the call was untimed
285         *    poll/offer, (argument "how" is NOW) return.
286         *
287         * 2. Try to append a new node (method tryAppend)
288         *
289         *    Starting at current tail pointer, find the actual last node
290         *    and try to append a new node (or if head was null, establish
291         *    the first node). Nodes can be appended only if their
292         *    predecessors are either already matched or are of the same
293         *    mode. If we detect otherwise, then a new node with opposite
294         *    mode must have been appended during traversal, so we must
295         *    restart at phase 1. The traversal and update steps are
296         *    otherwise similar to phase 1: Retrying upon CAS misses and
297         *    checking for staleness.  In particular, if a self-link is
298         *    encountered, then we can safely jump to a node on the list
299         *    by continuing the traversal at current head.
300         *
301         *    On successful append, if the call was ASYNC, return.
302         *
303         * 3. Await match or cancellation (method awaitMatch)
304         *
305         *    Wait for another thread to match node; instead cancelling if
306         *    the current thread was interrupted or the wait timed out. On
307         *    multiprocessors, we use front-of-queue spinning: If a node
308         *    appears to be the first unmatched node in the queue, it
309         *    spins a bit before blocking. In either case, before blocking
310         *    it tries to unsplice any nodes between the current "head"
311         *    and the first unmatched node.
312         *
313         *    Front-of-queue spinning vastly improves performance of
314         *    heavily contended queues. And so long as it is relatively
315         *    brief and "quiet", spinning does not much impact performance
316         *    of less-contended queues.  During spins threads check their
317         *    interrupt status and generate a thread-local random number
318         *    to decide to occasionally perform a Thread.yield. While
319         *    yield has underdefined specs, we assume that might it help,
320         *    and will not hurt in limiting impact of spinning on busy
321         *    systems.  We also use smaller (1/2) spins for nodes that are
322         *    not known to be front but whose predecessors have not
323         *    blocked -- these "chained" spins avoid artifacts of
324         *    front-of-queue rules which otherwise lead to alternating
325         *    nodes spinning vs blocking. Further, front threads that
326         *    represent phase changes (from data to request node or vice
327         *    versa) compared to their predecessors receive additional
328         *    chained spins, reflecting longer paths typically required to
329         *    unblock threads during phase changes.
330         */
331    
332        /** True if on multiprocessor */
333        private static final boolean MP =
334            Runtime.getRuntime().availableProcessors() > 1;
335    
336        /**
337         * The number of times to spin (with randomly interspersed calls
338         * to Thread.yield) on multiprocessor before blocking when a node
339         * is apparently the first waiter in the queue.  See above for
340         * explanation. Must be a power of two. The value is empirically
341         * derived -- it works pretty well across a variety of processors,
342         * numbers of CPUs, and OSes.
343         */
344        private static final int FRONT_SPINS   = 1 << 7;
345    
346        /**
347         * The number of times to spin before blocking when a node is
348         * preceded by another node that is apparently spinning.  Also
349         * serves as an increment to FRONT_SPINS on phase changes, and as
350         * base average frequency for yielding during spins. Must be a
351         * power of two.
352         */
353        private static final int CHAINED_SPINS = FRONT_SPINS >>> 1;
354    
355        /**
356         * Queue nodes. Uses Object, not E, for items to allow forgetting
357         * them after use.  Relies heavily on Unsafe mechanics to minimize
358         * unnecessary ordering constraints: Writes that intrinsically
359         * precede or follow CASes use simple relaxed forms.  Other
360         * cleanups use releasing/lazy writes.
361         */
362        static final class Node<E> {
363            final boolean isData;   // false if this is a request node
364            volatile Object item;   // initially non-null if isData; CASed to match
365            volatile Node<E> next;
366            volatile Thread waiter; // null until waiting
367    
368      // Wait modes for xfer method          // CAS methods for fields
369      static final int NOWAIT  = 0;          final boolean casNext(Node<E> cmp, Node<E> val) {
370      static final int TIMEOUT = 1;              return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
371      static final int WAIT    = 2;          }
372    
373      /** The number of CPUs, for spin control */          final boolean casItem(Object cmp, Object val) {
374      static final int NCPUS = Runtime.getRuntime().availableProcessors();              assert cmp.getClass() != Node.class;
375                return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
376            }
377    
378      /**      /**
379       * The number of times to spin before blocking in timed waits.           * Creates a new node. Uses relaxed write because item can only
380       * The value is empirically derived -- it works well across a           * be seen if followed by CAS.
      * variety of processors and OSes. Empirically, the best value  
      * seems not to vary with number of CPUs (beyond 2) so is just  
      * a constant.  
381       */       */
382      static final int maxTimedSpins = (NCPUS < 2) ? 0 : 32;          Node(E item, boolean isData) {
383                UNSAFE.putObject(this, itemOffset, item); // relaxed write
384                this.isData = isData;
385            }
386    
387      /**      /**
388       * The number of times to spin before blocking in untimed waits.           * Links node to itself to avoid garbage retention.  Called
389       * This is greater than timed value because untimed waits spin           * only after CASing head field, so uses relaxed write.
      * faster since they don't need to check times on each spin.  
390       */       */
391      static final int maxUntimedSpins = maxTimedSpins * 16;          final void forgetNext() {
392                UNSAFE.putObject(this, nextOffset, this);
393            }
394    
395      /**      /**
396       * The number of nanoseconds for which it is faster to spin           * Sets item to self (using a releasing/lazy write) and waiter
397       * rather than to use timed park. A rough estimate suffices.           * to null, to avoid garbage retention after extracting or
398             * cancelling.
399       */       */
400      static final long spinForTimeoutThreshold = 1000L;          final void forgetContents() {
401                UNSAFE.putOrderedObject(this, itemOffset, this);
402                UNSAFE.putOrderedObject(this, waiterOffset, null);
403            }
404    
405      /**      /**
406       * Node class for LinkedTransferQueue. Opportunistically           * Returns true if this node has been matched, including the
407       * subclasses from AtomicReference to represent item. Uses Object,           * case of artificial matches due to cancellation.
      * not E, to allow setting item to "this" after use, to avoid  
      * garbage retention. Similarly, setting the next field to this is  
      * used as sentinel that node is off list.  
408       */       */
409      static final class Node<E> extends AtomicReference<Object> {          final boolean isMatched() {
410          volatile Node<E> next;              Object x = item;
411          volatile Thread waiter;       // to control park/unpark              return x == this || (x != null) != isData;
412          final boolean isData;          }
413    
414          Node(E item, boolean isData) {          /**
415              super(item);           * Returns true if a node with the given mode cannot be
416              this.isData = isData;           * appended to this node because this node is unmatched and
417             * has opposite data mode.
418             */
419            final boolean cannotPrecede(boolean haveData) {
420                boolean d = isData;
421                Object x;
422                return d != haveData && (x = item) != this && (x != null) == d;
423          }          }
424    
425          // Unsafe mechanics          /**
426             * Tries to artificially match a data node -- used by remove.
427             */
428            final boolean tryMatchData() {
429                Object x = item;
430                if (x != null && x != this && casItem(x, null)) {
431                    LockSupport.unpark(waiter);
432                    return true;
433                }
434                return false;
435            }
436    
437            // Unsafe mechanics
438          private static final sun.misc.Unsafe UNSAFE = sun.misc.Unsafe.getUnsafe();          private static final sun.misc.Unsafe UNSAFE = sun.misc.Unsafe.getUnsafe();
439          private static final long nextOffset =          private static final long nextOffset =
440              objectFieldOffset(UNSAFE, "next", Node.class);              objectFieldOffset(UNSAFE, "next", Node.class);
441            private static final long itemOffset =
442          final boolean casNext(Node<E> cmp, Node<E> val) {              objectFieldOffset(UNSAFE, "item", Node.class);
443              return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);          private static final long waiterOffset =
444          }              objectFieldOffset(UNSAFE, "waiter", Node.class);
   
         final void clearNext() {  
             UNSAFE.putOrderedObject(this, nextOffset, this);  
         }  
445    
446          private static final long serialVersionUID = -3375979862319811754L;          private static final long serialVersionUID = -3375979862319811754L;
447      }      }
448    
449      /**      /** head of the queue; null until first enqueue */
450       * Padded version of AtomicReference used for head, tail and      transient volatile Node<E> head;
      * cleanMe, to alleviate contention across threads CASing one vs  
      * the other.  
      */  
     static final class PaddedAtomicReference<T> extends AtomicReference<T> {  
         // enough padding for 64bytes with 4byte refs  
         Object p0, p1, p2, p3, p4, p5, p6, p7, p8, p9, pa, pb, pc, pd, pe;  
         PaddedAtomicReference(T r) { super(r); }  
         private static final long serialVersionUID = 8170090609809740854L;  
     }  
   
451    
452      /** head of the queue */      /** predecessor of dangling unspliceable node */
453      private transient final PaddedAtomicReference<Node<E>> head;      private transient volatile Node<E> cleanMe; // decl here reduces contention
454    
455      /** tail of the queue */      /** tail of the queue; null until first append */
456      private transient final PaddedAtomicReference<Node<E>> tail;      private transient volatile Node<E> tail;
457    
458      /**      // CAS methods for fields
459       * Reference to a cancelled node that might not yet have been      private boolean casTail(Node<E> cmp, Node<E> val) {
460       * unlinked from queue because it was the last inserted node          return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);
461       * when it cancelled.      }
      */  
     private transient final PaddedAtomicReference<Node<E>> cleanMe;  
462    
463      /**      private boolean casHead(Node<E> cmp, Node<E> val) {
464       * Tries to cas nh as new head; if successful, unlink          return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);
      * old head's next node to avoid garbage retention.  
      */  
     private boolean advanceHead(Node<E> h, Node<E> nh) {  
         if (h == head.get() && head.compareAndSet(h, nh)) {  
             h.clearNext(); // forget old next  
             return true;  
465          }          }
466          return false;  
467        private boolean casCleanMe(Node<E> cmp, Node<E> val) {
468            return UNSAFE.compareAndSwapObject(this, cleanMeOffset, cmp, val);
469      }      }
470    
471        /*
472         * Possible values for "how" argument in xfer method. Beware that
473         * the order of assigned numerical values matters.
474         */
475        private static final int NOW     = 0; // for untimed poll, tryTransfer
476        private static final int ASYNC   = 1; // for offer, put, add
477        private static final int SYNC    = 2; // for transfer, take
478        private static final int TIMEOUT = 3; // for timed poll, tryTransfer
479    
480      /**      /**
481       * Puts or takes an item. Used for most queue operations (except       * Implements all queuing methods. See above for explanation.
      * poll() and tryTransfer()). See the similar code in  
      * SynchronousQueue for detailed explanation.  
482       *       *
483       * @param e the item or if null, signifies that this is a take       * @param e the item or null for take
484       * @param mode the wait mode: NOWAIT, TIMEOUT, WAIT       * @param haveData true if this is a put, else a take
485         * @param how NOW, ASYNC, SYNC, or TIMEOUT
486       * @param nanos timeout in nanosecs, used only if mode is TIMEOUT       * @param nanos timeout in nanosecs, used only if mode is TIMEOUT
487       * @return an item, or null on failure       * @return an item if matched, else e
488         * @throws NullPointerException if haveData mode but e is null
489       */       */
490      private E xfer(E e, int mode, long nanos) {      private E xfer(E e, boolean haveData, int how, long nanos) {
491          boolean isData = (e != null);          if (haveData && (e == null))
492          Node<E> s = null;              throw new NullPointerException();
493          final PaddedAtomicReference<Node<E>> head = this.head;          Node<E> s = null;                     // the node to append, if needed
         final PaddedAtomicReference<Node<E>> tail = this.tail;  
494    
495          for (;;) {          retry: for (;;) {                     // restart on append race
             Node<E> t = tail.get();  
             Node<E> h = head.get();  
496    
497              if (t == h || t.isData == isData) {              for (Node<E> h = head, p = h; p != null;) {
498                  if (s == null)                  // find & match first node
499                      s = new Node<E>(e, isData);                  boolean isData = p.isData;
500                  Node<E> last = t.next;                  Object item = p.item;
501                  if (last != null) {                  if (item != p && (item != null) == isData) { // unmatched
502                      if (t == tail.get())                      if (isData == haveData)   // can't match
503                          tail.compareAndSet(t, last);                          break;
504                        if (p.casItem(item, e)) { // match
505                            for (Node<E> q = p; q != h;) {
506                                Node<E> n = q.next; // update head by 2
507                                if (n != null)    // unless singleton
508                                    q = n;
509                                if (head == h && casHead(h, q)) {
510                                    h.forgetNext();
511                                    break;
512                                }                 // advance and retry
513                                if ((h = head)   == null ||
514                                    (q = h.next) == null || !q.isMatched())
515                                    break;        // unless slack < 2
516                  }                  }
517                  else if (t.casNext(null, s)) {                          LockSupport.unpark(p.waiter);
518                      tail.compareAndSet(t, s);                          return this.<E>cast(item);
                     return awaitFulfill(t, s, e, mode, nanos);  
519                  }                  }
             } else {  
                 Node<E> first = h.next;  
                 if (t == tail.get() && first != null &&  
                     advanceHead(h, first)) {  
                     Object x = first.get();  
                     if (x != first && first.compareAndSet(x, e)) {  
                         LockSupport.unpark(first.waiter);  
                         return isData ? e : (E) x;  
520                      }                      }
521                    Node<E> n = p.next;
522                    p = (p != n) ? n : (h = head); // Use head if p offlist
523                  }                  }
524    
525                if (how >= ASYNC) {               // No matches available
526                    if (s == null)
527                        s = new Node<E>(e, haveData);
528                    Node<E> pred = tryAppend(s, haveData);
529                    if (pred == null)
530                        continue retry;           // lost race vs opposite mode
531                    if (how >= SYNC)
532                        return awaitMatch(s, pred, e, how, nanos);
533              }              }
534                return e; // not waiting
535          }          }
536      }      }
537    
   
538      /**      /**
539       * Version of xfer for poll() and tryTransfer, which       * Tries to append node s as tail.
540       * simplifies control paths both here and in xfer.       *
541         * @param s the node to append
542         * @param haveData true if appending in data mode
543         * @return null on failure due to losing race with append in
544         * different mode, else s's predecessor, or s itself if no
545         * predecessor
546       */       */
547      private E fulfill(E e) {      private Node<E> tryAppend(Node<E> s, boolean haveData) {
548          boolean isData = (e != null);          for (Node<E> t = tail, p = t;;) { // move p to last node and append
549          final PaddedAtomicReference<Node<E>> head = this.head;              Node<E> n, u;                     // temps for reads of next & tail
550          final PaddedAtomicReference<Node<E>> tail = this.tail;              if (p == null && (p = head) == null) {
551                    if (casHead(null, s))
552          for (;;) {                      return s;                 // initialize
             Node<E> t = tail.get();  
             Node<E> h = head.get();  
   
             if (t == h || t.isData == isData) {  
                 Node<E> last = t.next;  
                 if (t == tail.get()) {  
                     if (last != null)  
                         tail.compareAndSet(t, last);  
                     else  
                         return null;  
                 }  
             } else {  
                 Node<E> first = h.next;  
                 if (t == tail.get() &&  
                     first != null &&  
                     advanceHead(h, first)) {  
                     Object x = first.get();  
                     if (x != first && first.compareAndSet(x, e)) {  
                         LockSupport.unpark(first.waiter);  
                         return isData ? e : (E) x;  
553                      }                      }
554                else if (p.cannotPrecede(haveData))
555                    return null;                  // lost race vs opposite mode
556                else if ((n = p.next) != null)    // not last; keep traversing
557                    p = p != t && t != (u = tail) ? (t = u) : // stale tail
558                        (p != n) ? n : null;      // restart if off list
559                else if (!p.casNext(null, s))
560                    p = p.next;                   // re-read on CAS failure
561                else {
562                    if (p != t) {                 // update if slack now >= 2
563                        while ((tail != t || !casTail(t, s)) &&
564                               (t = tail)   != null &&
565                               (s = t.next) != null && // advance and retry
566                               (s = s.next) != null && s != t);
567                  }                  }
568                    return p;
569              }              }
570          }          }
571      }      }
572    
573      /**      /**
574       * Spins/blocks until node s is fulfilled or caller gives up,       * Spins/yields/blocks until node s is matched or caller gives up.
      * depending on wait mode.  
575       *       *
      * @param pred the predecessor of waiting node  
576       * @param s the waiting node       * @param s the waiting node
577         * @param pred the predecessor of s, or s itself if it has no
578         * predecessor, or null if unknown (the null case does not occur
579         * in any current calls but may in possible future extensions)
580       * @param e the comparison value for checking match       * @param e the comparison value for checking match
581       * @param mode mode       * @param how either SYNC or TIMEOUT
582       * @param nanos timeout value       * @param nanos timeout value
583       * @return matched item, or null if cancelled       * @return matched item, or e if unmatched on interrupt or timeout
584       */       */
585      private E awaitFulfill(Node<E> pred, Node<E> s, E e,      private E awaitMatch(Node<E> s, Node<E> pred, E e, int how, long nanos) {
586                             int mode, long nanos) {          long lastTime = (how == TIMEOUT) ? System.nanoTime() : 0L;
         if (mode == NOWAIT)  
             return null;  
   
         long lastTime = (mode == TIMEOUT) ? System.nanoTime() : 0;  
587          Thread w = Thread.currentThread();          Thread w = Thread.currentThread();
588          int spins = -1; // set to desired spin count below          int spins = -1; // initialized after first item and cancel checks
589            ThreadLocalRandom randomYields = null; // bound if needed
590    
591          for (;;) {          for (;;) {
592              if (w.isInterrupted())              Object item = s.item;
593                  s.compareAndSet(e, s);              if (item != e) {                  // matched
594              Object x = s.get();                  assert item != s;
595              if (x != e) {                 // Node was matched or cancelled                  s.forgetContents();           // avoid garbage
596                  advanceHead(pred, s);     // unlink if head                  return this.<E>cast(item);
597                  if (x == s) {             // was cancelled              }
598                      clean(pred, s);              if ((w.isInterrupted() || (how == TIMEOUT && nanos <= 0)) &&
599                      return null;                      s.casItem(e, s)) {       // cancel
600                    unsplice(pred, s);
601                    return e;
602                  }                  }
603                  else if (x != null) {  
604                      s.set(s);             // avoid garbage retention              if (spins < 0) {                  // establish spins at/near front
605                      return (E) x;                  if ((spins = spinsFor(pred, s.isData)) > 0)
606                        randomYields = ThreadLocalRandom.current();
607                }
608                else if (spins > 0) {             // spin
609                    if (--spins == 0)
610                        shortenHeadPath();        // reduce slack before blocking
611                    else if (randomYields.nextInt(CHAINED_SPINS) == 0)
612                        Thread.yield();           // occasionally yield
613                  }                  }
614                  else              else if (s.waiter == null) {
615                      return e;                  s.waiter = w;                 // request unpark then recheck
616              }              }
617              if (mode == TIMEOUT) {              else if (how == TIMEOUT) {
618                  long now = System.nanoTime();                  long now = System.nanoTime();
619                  nanos -= now - lastTime;                  if ((nanos -= now - lastTime) > 0)
620                        LockSupport.parkNanos(this, nanos);
621                  lastTime = now;                  lastTime = now;
622                  if (nanos <= 0) {              }
623                      s.compareAndSet(e, s); // try to cancel              else {
                     continue;  
                 }  
             }  
             if (spins < 0) {  
                 Node<E> h = head.get(); // only spin if at head  
                 spins = ((h.next != s) ? 0 :  
                          (mode == TIMEOUT) ? maxTimedSpins :  
                          maxUntimedSpins);  
             }  
             if (spins > 0)  
                 --spins;  
             else if (s.waiter == null)  
                 s.waiter = w;  
             else if (mode != TIMEOUT) {  
624                  LockSupport.park(this);                  LockSupport.park(this);
625                  s.waiter = null;                  s.waiter = null;
626                  spins = -1;                  spins = -1;                   // spin if front upon wakeup
627                }
628              }              }
             else if (nanos > spinForTimeoutThreshold) {  
                 LockSupport.parkNanos(this, nanos);  
                 s.waiter = null;  
                 spins = -1;  
629              }              }
630    
631        /**
632         * Returns spin/yield value for a node with given predecessor and
633         * data mode. See above for explanation.
634         */
635        private static int spinsFor(Node<?> pred, boolean haveData) {
636            if (MP && pred != null) {
637                if (pred.isData != haveData)      // phase change
638                    return FRONT_SPINS + CHAINED_SPINS;
639                if (pred.isMatched())             // probably at front
640                    return FRONT_SPINS;
641                if (pred.waiter == null)          // pred apparently spinning
642                    return CHAINED_SPINS;
643          }          }
644            return 0;
645      }      }
646    
647      /**      /**
648       * Returns validated tail for use in cleaning methods.       * Tries (once) to unsplice nodes between head and first unmatched
649         * or trailing node; failing on contention.
650       */       */
651      private Node<E> getValidatedTail() {      private void shortenHeadPath() {
652          for (;;) {          Node<E> h, hn, p, q;
653              Node<E> h = head.get();          if ((p = h = head) != null && h.isMatched() &&
654              Node<E> first = h.next;              (q = hn = h.next) != null) {
655              if (first != null && first.get() == first) { // help advance              Node<E> n;
656                  advanceHead(h, first);              while ((n = q.next) != q) {
657                  continue;                  if (n == null || !q.isMatched()) {
658              }                      if (hn != q && h.next == hn)
659              Node<E> t = tail.get();                          h.casNext(hn, q);
660              Node<E> last = t.next;                      break;
             if (t == tail.get()) {  
                 if (last != null)  
                     tail.compareAndSet(t, last); // help advance  
                 else  
                     return t;  
661              }              }
662                    p = q;
663                    q = n;
664          }          }
665      }      }
666        }
667    
668        /* -------------- Traversal methods -------------- */
669    
670      /**      /**
671       * Gets rid of cancelled node s with original predecessor pred.       * Returns the first unmatched node of the given mode, or null if
672       *       * none.  Used by methods isEmpty, hasWaitingConsumer.
      * @param pred predecessor of cancelled node  
      * @param s the cancelled node  
673       */       */
674      private void clean(Node<E> pred, Node<E> s) {      private Node<E> firstOfMode(boolean data) {
675          Thread w = s.waiter;          for (Node<E> p = head; p != null; ) {
676          if (w != null) {             // Wake up thread              if (!p.isMatched())
677              s.waiter = null;                  return (p.isData == data) ? p : null;
678              if (w != Thread.currentThread())              Node<E> n = p.next;
679                  LockSupport.unpark(w);              p = (n != p) ? n : head;
680            }
681            return null;
682          }          }
683    
684          if (pred == null)      @SuppressWarnings("unchecked")
685        static <E> E cast(Object item) {
686            assert item.getClass() != Node.class;
687            return (E) item;
688        }
689    
690        /**
691         * Returns the item in the first unmatched node with isData; or
692         * null if none.  Used by peek.
693         */
694        private E firstDataItem() {
695            for (Node<E> p = head; p != null; ) {
696                boolean isData = p.isData;
697                Object item = p.item;
698                if (item != p && (item != null) == isData)
699                    return isData ? this.<E>cast(item) : null;
700                Node<E> n = p.next;
701                p = (n != p) ? n : head;
702            }
703            return null;
704        }
705    
706        /**
707         * Traverses and counts unmatched nodes of the given mode.
708         * Used by methods size and getWaitingConsumerCount.
709         */
710        private int countOfMode(boolean data) {
711            int count = 0;
712            for (Node<E> p = head; p != null; ) {
713                if (!p.isMatched()) {
714                    if (p.isData != data)
715                        return 0;
716                    if (++count == Integer.MAX_VALUE) // saturated
717                        break;
718                }
719                Node<E> n = p.next;
720                if (n != p)
721                    p = n;
722                else {
723                    count = 0;
724                    p = head;
725                }
726            }
727            return count;
728        }
729    
730        final class Itr implements Iterator<E> {
731            private Node<E> nextNode;   // next node to return item for
732            private E nextItem;         // the corresponding item
733            private Node<E> lastRet;    // last returned node, to support remove
734    
735            /**
736             * Moves to next node after prev, or first node if prev null.
737             */
738            private void advance(Node<E> prev) {
739                lastRet = prev;
740                Node<E> p;
741                if (prev == null || (p = prev.next) == prev)
742                    p = head;
743                while (p != null) {
744                    Object item = p.item;
745                    if (p.isData) {
746                        if (item != null && item != p) {
747                            nextItem = LinkedTransferQueue.this.<E>cast(item);
748                            nextNode = p;
749              return;              return;
750                        }
751                    }
752                    else if (item == null)
753                        break;
754                    Node<E> n = p.next;
755                    p = (n != p) ? n : head;
756                }
757                nextNode = null;
758            }
759    
760            Itr() {
761                advance(null);
762            }
763    
764            public final boolean hasNext() {
765                return nextNode != null;
766            }
767    
768            public final E next() {
769                Node<E> p = nextNode;
770                if (p == null) throw new NoSuchElementException();
771                E e = nextItem;
772                advance(p);
773                return e;
774            }
775    
776            public final void remove() {
777                Node<E> p = lastRet;
778                if (p == null) throw new IllegalStateException();
779                lastRet = null;
780                findAndRemoveNode(p);
781            }
782        }
783    
784        /* -------------- Removal methods -------------- */
785    
786        /**
787         * Unsplices (now or later) the given deleted/cancelled node with
788         * the given predecessor.
789         *
790         * @param pred predecessor of node to be unspliced
791         * @param s the node to be unspliced
792         */
793        private void unsplice(Node<E> pred, Node<E> s) {
794            s.forgetContents(); // clear unneeded fields
795          /*          /*
796           * At any given time, exactly one node on list cannot be           * At any given time, exactly one node on list cannot be
797           * deleted -- the last inserted node. To accommodate this, if           * unlinked -- the last inserted node. To accommodate this, if
798           * we cannot delete s, we save its predecessor as "cleanMe",           * we cannot unlink s, we save its predecessor as "cleanMe",
799           * processing the previously saved version first. At least one           * processing the previously saved version first. Because only
800           * of node s or the node previously saved can always be           * one node in the list can have a null next, at least one of
801             * node s or the node previously saved can always be
802           * processed, so this always terminates.           * processed, so this always terminates.
803           */           */
804            if (pred != null && pred != s) {
805          while (pred.next == s) {          while (pred.next == s) {
806              Node<E> oldpred = reclean();  // First, help get rid of cleanMe                  Node<E> oldpred = (cleanMe == null) ? null : reclean();
807              Node<E> t = getValidatedTail();                  Node<E> n = s.next;
808              if (s != t) {               // If not tail, try to unsplice                  if (n != null) {
809                  Node<E> sn = s.next;      // s.next == s means s already off list                      if (n != s)
810                  if (sn == s || pred.casNext(s, sn))                          pred.casNext(s, n);
811                      break;                      break;
812              }              }
813              else if (oldpred == pred || // Already saved                  if (oldpred == pred ||      // Already saved
814                       (oldpred == null && cleanMe.compareAndSet(null, pred)))                      (oldpred == null && casCleanMe(null, pred)))
815                  break;                  // Postpone cleaning                  break;                  // Postpone cleaning
816          }          }
817      }      }
818        }
819    
820      /**      /**
821       * Tries to unsplice the cancelled node held in cleanMe that was       * Tries to unsplice the deleted/cancelled node held in cleanMe
822       * previously uncleanable because it was at tail.       * that was previously uncleanable because it was at tail.
823       *       *
824       * @return current cleanMe node (or null)       * @return current cleanMe node (or null)
825       */       */
826      private Node<E> reclean() {      private Node<E> reclean() {
827          /*          /*
828           * cleanMe is, or at one time was, predecessor of cancelled           * cleanMe is, or at one time was, predecessor of a cancelled
829           * node s that was the tail so could not be unspliced.  If s           * node s that was the tail so could not be unspliced.  If it
830           * is no longer the tail, try to unsplice if necessary and           * is no longer the tail, try to unsplice if necessary and
831           * make cleanMe slot available.  This differs from similar           * make cleanMe slot available.  This differs from similar
832           * code in clean() because we must check that pred still           * code in unsplice() because we must check that pred still
833           * points to a cancelled node that must be unspliced -- if           * points to a matched node that can be unspliced -- if not,
834           * not, we can (must) clear cleanMe without unsplicing.           * we can (must) clear cleanMe without unsplicing.  This can
835           * This can loop only due to contention on casNext or           * loop only due to contention.
          * clearing cleanMe.  
836           */           */
837          Node<E> pred;          Node<E> pred;
838          while ((pred = cleanMe.get()) != null) {          while ((pred = cleanMe) != null) {
             Node<E> t = getValidatedTail();  
839              Node<E> s = pred.next;              Node<E> s = pred.next;
840              if (s != t) {              Node<E> n;
841                  Node<E> sn;              if (s == null || s == pred || !s.isMatched())
842                  if (s == null || s == pred || s.get() != s ||                  casCleanMe(pred, null); // already gone
843                      (sn = s.next) == s || pred.casNext(s, sn))              else if ((n = s.next) != null) {
844                      cleanMe.compareAndSet(pred, null);                  if (n != s)
845                        pred.casNext(s, n);
846                    casCleanMe(pred, null);
847              }              }
848              else // s is still tail; cannot clean              else
849                  break;                  break;
850          }          }
851          return pred;          return pred;
852      }      }
853    
854      /**      /**
855         * Main implementation of Iterator.remove(). Find
856         * and unsplice the given node.
857         */
858        final void findAndRemoveNode(Node<E> s) {
859            if (s.tryMatchData()) {
860                Node<E> pred = null;
861                Node<E> p = head;
862                while (p != null) {
863                    if (p == s) {
864                        unsplice(pred, p);
865                        break;
866                    }
867                    if (!p.isData && !p.isMatched())
868                        break;
869                    pred = p;
870                    if ((p = p.next) == pred) { // stale
871                        pred = null;
872                        p = head;
873                    }
874                }
875            }
876        }
877    
878        /**
879         * Main implementation of remove(Object)
880         */
881        private boolean findAndRemove(Object e) {
882            if (e != null) {
883                Node<E> pred = null;
884                Node<E> p = head;
885                while (p != null) {
886                    Object item = p.item;
887                    if (p.isData) {
888                        if (item != null && item != p && e.equals(item) &&
889                            p.tryMatchData()) {
890                            unsplice(pred, p);
891                            return true;
892                        }
893                    }
894                    else if (item == null)
895                        break;
896                    pred = p;
897                    if ((p = p.next) == pred) {
898                        pred = null;
899                        p = head;
900                    }
901                }
902            }
903            return false;
904        }
905    
906    
907        /**
908       * Creates an initially empty {@code LinkedTransferQueue}.       * Creates an initially empty {@code LinkedTransferQueue}.
909       */       */
910      public LinkedTransferQueue() {      public LinkedTransferQueue() {
         Node<E> dummy = new Node<E>(null, false);  
         head = new PaddedAtomicReference<Node<E>>(dummy);  
         tail = new PaddedAtomicReference<Node<E>>(dummy);  
         cleanMe = new PaddedAtomicReference<Node<E>>(null);  
911      }      }
912    
913      /**      /**
# Line 446  Line 931 
931       * @throws NullPointerException if the specified element is null       * @throws NullPointerException if the specified element is null
932       */       */
933      public void put(E e) {      public void put(E e) {
934          offer(e);          xfer(e, true, ASYNC, 0);
935      }      }
936    
937      /**      /**
# Line 459  Line 944 
944       * @throws NullPointerException if the specified element is null       * @throws NullPointerException if the specified element is null
945       */       */
946      public boolean offer(E e, long timeout, TimeUnit unit) {      public boolean offer(E e, long timeout, TimeUnit unit) {
947          return offer(e);          xfer(e, true, ASYNC, 0);
948            return true;
949      }      }
950    
951      /**      /**
# Line 471  Line 957 
957       * @throws NullPointerException if the specified element is null       * @throws NullPointerException if the specified element is null
958       */       */
959      public boolean offer(E e) {      public boolean offer(E e) {
960          if (e == null) throw new NullPointerException();          xfer(e, true, ASYNC, 0);
         xfer(e, NOWAIT, 0);  
961          return true;          return true;
962      }      }
963    
# Line 485  Line 970 
970       * @throws NullPointerException if the specified element is null       * @throws NullPointerException if the specified element is null
971       */       */
972      public boolean add(E e) {      public boolean add(E e) {
973          return offer(e);          xfer(e, true, ASYNC, 0);
974            return true;
975      }      }
976    
977      /**      /**
# Line 499  Line 985 
985       * @throws NullPointerException if the specified element is null       * @throws NullPointerException if the specified element is null
986       */       */
987      public boolean tryTransfer(E e) {      public boolean tryTransfer(E e) {
988          if (e == null) throw new NullPointerException();          return xfer(e, true, NOW, 0) == null;
         return fulfill(e) != null;  
989      }      }
990    
991      /**      /**
# Line 515  Line 1000 
1000       * @throws NullPointerException if the specified element is null       * @throws NullPointerException if the specified element is null
1001       */       */
1002      public void transfer(E e) throws InterruptedException {      public void transfer(E e) throws InterruptedException {
1003          if (e == null) throw new NullPointerException();          if (xfer(e, true, SYNC, 0) != null) {
1004          if (xfer(e, WAIT, 0) == null) {              Thread.interrupted(); // failure possible only due to interrupt
             Thread.interrupted();  
1005              throw new InterruptedException();              throw new InterruptedException();
1006          }          }
1007      }      }
# Line 538  Line 1022 
1022       */       */
1023      public boolean tryTransfer(E e, long timeout, TimeUnit unit)      public boolean tryTransfer(E e, long timeout, TimeUnit unit)
1024          throws InterruptedException {          throws InterruptedException {
1025          if (e == null) throw new NullPointerException();          if (xfer(e, true, TIMEOUT, unit.toNanos(timeout)) == null)
         if (xfer(e, TIMEOUT, unit.toNanos(timeout)) != null)  
1026              return true;              return true;
1027          if (!Thread.interrupted())          if (!Thread.interrupted())
1028              return false;              return false;
# Line 547  Line 1030 
1030      }      }
1031    
1032      public E take() throws InterruptedException {      public E take() throws InterruptedException {
1033          E e = xfer(null, WAIT, 0);          E e = xfer(null, false, SYNC, 0);
1034          if (e != null)          if (e != null)
1035              return e;              return e;
1036          Thread.interrupted();          Thread.interrupted();
# Line 555  Line 1038 
1038      }      }
1039    
1040      public E poll(long timeout, TimeUnit unit) throws InterruptedException {      public E poll(long timeout, TimeUnit unit) throws InterruptedException {
1041          E e = xfer(null, TIMEOUT, unit.toNanos(timeout));          E e = xfer(null, false, TIMEOUT, unit.toNanos(timeout));
1042          if (e != null || !Thread.interrupted())          if (e != null || !Thread.interrupted())
1043              return e;              return e;
1044          throw new InterruptedException();          throw new InterruptedException();
1045      }      }
1046    
1047      public E poll() {      public E poll() {
1048          return fulfill(null);          return xfer(null, false, NOW, 0);
1049      }      }
1050    
1051      /**      /**
# Line 601  Line 1084 
1084          return n;          return n;
1085      }      }
1086    
     // Traversal-based methods  
   
     /**  
      * Returns head after performing any outstanding helping steps.  
      */  
     private Node<E> traversalHead() {  
         for (;;) {  
             Node<E> t = tail.get();  
             Node<E> h = head.get();  
             Node<E> last = t.next;  
             Node<E> first = h.next;  
             if (t == tail.get()) {  
                 if (last != null)  
                     tail.compareAndSet(t, last);  
                 else if (first != null) {  
                     Object x = first.get();  
                     if (x == first)  
                         advanceHead(h, first);  
                     else  
                         return h;  
                 }  
                 else  
                     return h;  
             }  
             reclean();  
         }  
     }  
   
1087      /**      /**
1088       * Returns an iterator over the elements in this queue in proper       * Returns an iterator over the elements in this queue in proper
1089       * sequence, from head to tail.       * sequence, from head to tail.
# Line 646  Line 1101 
1101          return new Itr();          return new Itr();
1102      }      }
1103    
     /**  
      * Iterators. Basic strategy is to traverse list, treating  
      * non-data (i.e., request) nodes as terminating list.  
      * Once a valid data node is found, the item is cached  
      * so that the next call to next() will return it even  
      * if subsequently removed.  
      */  
     class Itr implements Iterator<E> {  
         Node<E> next;        // node to return next  
         Node<E> pnext;       // predecessor of next  
         Node<E> curr;        // last returned node, for remove()  
         Node<E> pcurr;       // predecessor of curr, for remove()  
         E nextItem;          // Cache of next item, once committed to in next  
   
         Itr() {  
             advance();  
         }  
   
         /**  
          * Moves to next valid node and returns item to return for  
          * next(), or null if no such.  
          */  
         private E advance() {  
             pcurr = pnext;  
             curr = next;  
             E item = nextItem;  
   
             for (;;) {  
                 pnext = (next == null) ? traversalHead() : next;  
                 next = pnext.next;  
                 if (next == pnext) {  
                     next = null;  
                     continue;  // restart  
                 }  
                 if (next == null)  
                     break;  
                 Object x = next.get();  
                 if (x != null && x != next) {  
                     nextItem = (E) x;  
                     break;  
                 }  
             }  
             return item;  
         }  
   
         public boolean hasNext() {  
             return next != null;  
         }  
   
         public E next() {  
             if (next == null)  
                 throw new NoSuchElementException();  
             return advance();  
         }  
   
         public void remove() {  
             Node<E> p = curr;  
             if (p == null)  
                 throw new IllegalStateException();  
             Object x = p.get();  
             if (x != null && x != p && p.compareAndSet(x, p))  
                 clean(pcurr, p);  
         }  
     }  
   
1104      public E peek() {      public E peek() {
1105          for (;;) {          return firstDataItem();
             Node<E> h = traversalHead();  
             Node<E> p = h.next;  
             if (p == null)  
                 return null;  
             Object x = p.get();  
             if (p != x) {  
                 if (!p.isData)  
                     return null;  
                 if (x != null)  
                     return (E) x;  
             }  
         }  
1106      }      }
1107    
1108      /**      /**
# Line 733  Line 1111 
1111       * @return {@code true} if this queue contains no elements       * @return {@code true} if this queue contains no elements
1112       */       */
1113      public boolean isEmpty() {      public boolean isEmpty() {
1114          for (;;) {          return firstOfMode(true) == null;
             Node<E> h = traversalHead();  
             Node<E> p = h.next;  
             if (p == null)  
                 return true;  
             Object x = p.get();  
             if (p != x) {  
                 if (!p.isData)  
                     return true;  
                 if (x != null)  
                     return false;  
             }  
         }  
1115      }      }
1116    
1117      public boolean hasWaitingConsumer() {      public boolean hasWaitingConsumer() {
1118          for (;;) {          return firstOfMode(false) != null;
             Node<E> h = traversalHead();  
             Node<E> p = h.next;  
             if (p == null)  
                 return false;  
             Object x = p.get();  
             if (p != x)  
                 return !p.isData;  
         }  
1119      }      }
1120    
1121      /**      /**
# Line 773  Line 1131 
1131       * @return the number of elements in this queue       * @return the number of elements in this queue
1132       */       */
1133      public int size() {      public int size() {
1134          for (;;) {          return countOfMode(true);
             int count = 0;  
             Node<E> pred = traversalHead();  
             for (;;) {  
                 Node<E> q = pred.next;  
                 if (q == pred) // restart  
                     break;  
                 if (q == null || !q.isData)  
                     return count;  
                 Object x = q.get();  
                 if (x != null && x != q) {  
                     if (++count == Integer.MAX_VALUE) // saturated  
                         return count;  
                 }  
                 pred = q;  
             }  
         }  
1135      }      }
1136    
1137      public int getWaitingConsumerCount() {      public int getWaitingConsumerCount() {
1138          // converse of size -- count valid non-data nodes          return countOfMode(false);
         for (;;) {  
             int count = 0;  
             Node<E> pred = traversalHead();  
             for (;;) {  
                 Node<E> q = pred.next;  
                 if (q == pred) // restart  
                     break;  
                 if (q == null || q.isData)  
                     return count;  
                 Object x = q.get();  
                 if (x == null) {  
                     if (++count == Integer.MAX_VALUE) // saturated  
                         return count;  
                 }  
                 pred = q;  
             }  
         }  
1139      }      }
1140    
1141      /**      /**
# Line 825  Line 1150 
1150       * @return {@code true} if this queue changed as a result of the call       * @return {@code true} if this queue changed as a result of the call
1151       */       */
1152      public boolean remove(Object o) {      public boolean remove(Object o) {
1153          if (o == null)          return findAndRemove(o);
             return false;  
         for (;;) {  
             Node<E> pred = traversalHead();  
             for (;;) {  
                 Node<E> q = pred.next;  
                 if (q == pred) // restart  
                     break;  
                 if (q == null || !q.isData)  
                     return false;  
                 Object x = q.get();  
                 if (x != null && x != q && o.equals(x) &&  
                     q.compareAndSet(x, q)) {  
                     clean(pred, q);  
                     return true;  
                 }  
                 pred = q;  
             }  
         }  
1154      }      }
1155    
1156      /**      /**
# Line 858  Line 1165 
1165      }      }
1166    
1167      /**      /**
1168       * Save the state to a stream (that is, serialize it).       * Saves the state to a stream (that is, serializes it).
1169       *       *
1170       * @serialData All of the elements (each an {@code E}) in       * @serialData All of the elements (each an {@code E}) in
1171       * the proper order, followed by a null       * the proper order, followed by a null
# Line 874  Line 1181 
1181      }      }
1182    
1183      /**      /**
1184       * Reconstitute the Queue instance from a stream (that is,       * Reconstitutes the Queue instance from a stream (that is,
1185       * deserialize it).       * deserializes it).
1186       *       *
1187       * @param s the stream       * @param s the stream
1188       */       */
1189      private void readObject(java.io.ObjectInputStream s)      private void readObject(java.io.ObjectInputStream s)
1190          throws java.io.IOException, ClassNotFoundException {          throws java.io.IOException, ClassNotFoundException {
1191          s.defaultReadObject();          s.defaultReadObject();
         resetHeadAndTail();  
1192          for (;;) {          for (;;) {
1193              @SuppressWarnings("unchecked") E item = (E) s.readObject();              @SuppressWarnings("unchecked") E item = (E) s.readObject();
1194              if (item == null)              if (item == null)
# Line 892  Line 1198 
1198          }          }
1199      }      }
1200    
     // Support for resetting head/tail while deserializing  
     private void resetHeadAndTail() {  
         Node<E> dummy = new Node<E>(null, false);  
         UNSAFE.putObjectVolatile(this, headOffset,  
                                  new PaddedAtomicReference<Node<E>>(dummy));  
         UNSAFE.putObjectVolatile(this, tailOffset,  
                                  new PaddedAtomicReference<Node<E>>(dummy));  
         UNSAFE.putObjectVolatile(this, cleanMeOffset,  
                                  new PaddedAtomicReference<Node<E>>(null));  
     }  
   
1201      // Unsafe mechanics      // Unsafe mechanics
1202    
1203      private static final sun.misc.Unsafe UNSAFE = sun.misc.Unsafe.getUnsafe();      private static final sun.misc.Unsafe UNSAFE = sun.misc.Unsafe.getUnsafe();
# Line 913  Line 1208 
1208      private static final long cleanMeOffset =      private static final long cleanMeOffset =
1209          objectFieldOffset(UNSAFE, "cleanMe", LinkedTransferQueue.class);          objectFieldOffset(UNSAFE, "cleanMe", LinkedTransferQueue.class);
1210    
   
1211      static long objectFieldOffset(sun.misc.Unsafe UNSAFE,      static long objectFieldOffset(sun.misc.Unsafe UNSAFE,
1212                                    String field, Class<?> klazz) {                                    String field, Class<?> klazz) {
1213          try {          try {
# Line 925  Line 1219 
1219              throw error;              throw error;
1220          }          }
1221      }      }
1222    
1223  }  }

Legend:
Removed from v.1.7  
changed lines
  Added in v.1.8

Doug Lea
ViewVC Help
Powered by ViewVC 1.0.8