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.45 by dl, Wed Oct 21 16:30:40 2009 UTC vs.
Revision 1.60 by dl, Fri Oct 30 12:06:31 2009 UTC

# Line 105 | Line 105 | public class LinkedTransferQueue<E> exte
105       * successful atomic operation per enq/deq pair. But it also
106       * enables lower cost variants of queue maintenance mechanics. (A
107       * variation of this idea applies even for non-dual queues that
108 <     * support deletion of embedded elements, such as
108 >     * support deletion of interior elements, such as
109       * j.u.c.ConcurrentLinkedQueue.)
110       *
111 <     * Once a node is matched, its item can never again change.  We
112 <     * may thus arrange that the linked list of them contains a prefix
113 <     * of zero or more matched nodes, followed by a suffix of zero or
114 <     * more unmatched nodes. (Note that we allow both the prefix and
115 <     * suffix to be zero length, which in turn means that we do not
116 <     * use a dummy header.)  If we were not concerned with either time
117 <     * or space efficiency, we could correctly perform enqueue and
118 <     * dequeue operations by traversing from a pointer to the initial
119 <     * node; CASing the item of the first unmatched node on match and
120 <     * CASing the next field of the trailing node on appends.  While
121 <     * this would be a terrible idea in itself, it does have the
122 <     * benefit of not requiring ANY atomic updates on head/tail
123 <     * fields.
111 >     * Once a node is matched, its match status can never again
112 >     * change.  We may thus arrange that the linked list of them
113 >     * contain a prefix of zero or more matched nodes, followed by a
114 >     * suffix of zero or more unmatched nodes. (Note that we allow
115 >     * both the prefix and suffix to be zero length, which in turn
116 >     * means that we do not use a dummy header.)  If we were not
117 >     * concerned with either time or space efficiency, we could
118 >     * correctly perform enqueue and dequeue operations by traversing
119 >     * from a pointer to the initial node; CASing the item of the
120 >     * first unmatched node on match and CASing the next field of the
121 >     * trailing node on appends. (Plus some special-casing when
122 >     * initially empty).  While this would be a terrible idea in
123 >     * itself, it does have the benefit of not requiring ANY atomic
124 >     * updates on head/tail fields.
125       *
126       * We introduce here an approach that lies between the extremes of
127 <     * never versus always updating queue (head and tail) pointers
128 <     * that reflects the tradeoff of sometimes require extra traversal
129 <     * steps to locate the first and/or last unmatched nodes, versus
130 <     * the reduced overhead and contention of fewer updates to queue
131 <     * pointers. For example, a possible snapshot of a queue is:
127 >     * never versus always updating queue (head and tail) pointers.
128 >     * This offers a tradeoff between sometimes requiring extra
129 >     * traversal steps to locate the first and/or last unmatched
130 >     * nodes, versus the reduced overhead and contention of fewer
131 >     * updates to queue pointers. For example, a possible snapshot of
132 >     * a queue is:
133       *
134       *  head           tail
135       *    |              |
# Line 139 | Line 141 | public class LinkedTransferQueue<E> exte
141       * similarly for "tail") is an empirical matter. We have found
142       * that using very small constants in the range of 1-3 work best
143       * over a range of platforms. Larger values introduce increasing
144 <     * costs of cache misses and risks of long traversal chains.
144 >     * costs of cache misses and risks of long traversal chains, while
145 >     * smaller values increase CAS contention and overhead.
146       *
147       * Dual queues with slack differ from plain M&S dual queues by
148       * virtue of only sometimes updating head or tail pointers when
# Line 158 | Line 161 | public class LinkedTransferQueue<E> exte
161       * targets.  Even when using very small slack values, this
162       * approach works well for dual queues because it allows all
163       * operations up to the point of matching or appending an item
164 <     * (hence potentially releasing another thread) to be read-only,
165 <     * thus not introducing any further contention. As described
166 <     * below, we implement this by performing slack maintenance
167 <     * retries only after these points.
164 >     * (hence potentially allowing progress by another thread) to be
165 >     * read-only, thus not introducing any further contention. As
166 >     * described below, we implement this by performing slack
167 >     * maintenance retries only after these points.
168       *
169       * As an accompaniment to such techniques, traversal overhead can
170       * be further reduced without increasing contention of head
171 <     * pointer updates.  During traversals, threads may sometimes
172 <     * shortcut the "next" link path from the current "head" node to
173 <     * be closer to the currently known first unmatched node. Again,
174 <     * this may be triggered with using thresholds or randomization.
171 >     * pointer updates: Threads may sometimes shortcut the "next" link
172 >     * path from the current "head" node to be closer to the currently
173 >     * known first unmatched node, and similarly for tail. Again, this
174 >     * may be triggered with using thresholds or randomization.
175       *
176       * These ideas must be further extended to avoid unbounded amounts
177       * of costly-to-reclaim garbage caused by the sequential "next"
# Line 180 | Line 183 | public class LinkedTransferQueue<E> exte
183       * (Similar issues arise in non-GC environments.)  To cope with
184       * this in our implementation, upon CASing to advance the head
185       * pointer, we set the "next" link of the previous head to point
186 <     * only to itself; thus limiting the length connected dead lists.
186 >     * only to itself; thus limiting the length of connected dead lists.
187       * (We also take similar care to wipe out possibly garbage
188       * retaining values held in other Node fields.)  However, doing so
189       * adds some further complexity to traversal: If any "next"
# Line 196 | Line 199 | public class LinkedTransferQueue<E> exte
199       * mechanics because an update may leave head at a detached node.
200       * And while direct writes are possible for tail updates, they
201       * increase the risk of long retraversals, and hence long garbage
202 <     * chains which can be much more costly than is worthwhile
202 >     * chains, which can be much more costly than is worthwhile
203       * considering that the cost difference of performing a CAS vs
204       * write is smaller when they are not triggered on each operation
205       * (especially considering that writes and CASes equally require
206       * additional GC bookkeeping ("write barriers") that are sometimes
207       * more costly than the writes themselves because of contention).
208       *
209 <     * Removal of internal nodes (due to timed out or interrupted
210 <     * waits, or calls to remove or Iterator.remove) uses a scheme
211 <     * roughly similar to that in Scherer, Lea, and Scott
212 <     * SynchronousQueue. Given a predecessor, we can unsplice any node
213 <     * except the (actual) tail of the queue. To avoid build-up of
214 <     * cancelled trailing nodes, upon a request to remove a trailing
215 <     * node, it is placed in field "cleanMe" to be unspliced later.
209 >     * Removal of interior nodes (due to timed out or interrupted
210 >     * waits, or calls to remove(x) or Iterator.remove) can use a
211 >     * scheme roughly similar to that described in Scherer, Lea, and
212 >     * Scott's SynchronousQueue. Given a predecessor, we can unsplice
213 >     * any node except the (actual) tail of the queue. To avoid
214 >     * build-up of cancelled trailing nodes, upon a request to remove
215 >     * a trailing node, it is placed in field "cleanMe" to be
216 >     * unspliced upon the next call to unsplice any other node.
217 >     * Situations needing such mechanics are not common but do occur
218 >     * in practice; for example when an unbounded series of short
219 >     * timed calls to poll repeatedly time out but never otherwise
220 >     * fall off the list because of an untimed call to take at the
221 >     * front of the queue. Note that maintaining field cleanMe does
222 >     * not otherwise much impact garbage retention even if never
223 >     * cleared by some other call because the held node will
224 >     * eventually either directly or indirectly lead to a self-link
225 >     * once off the list.
226       *
227       * *** Overview of implementation ***
228       *
229 <     * We use a threshold-based approach to updates, with a target
230 <     * slack of two.  The slack value is hard-wired: a path greater
229 >     * We use a threshold-based approach to updates, with a slack
230 >     * threshold of two -- that is, we update head/tail when the
231 >     * current pointer appears to be two or more steps away from the
232 >     * first/last node. The slack value is hard-wired: a path greater
233       * than one is naturally implemented by checking equality of
234       * traversal pointers except when the list has only one element,
235 <     * in which case we keep max slack at one. Avoiding tracking
236 <     * explicit counts across situations slightly simplifies an
235 >     * in which case we keep slack threshold at one. Avoiding tracking
236 >     * explicit counts across method calls slightly simplifies an
237       * already-messy implementation. Using randomization would
238       * probably work better if there were a low-quality dirt-cheap
239       * per-thread one available, but even ThreadLocalRandom is too
240       * heavy for these purposes.
241       *
242 <     * With such a small slack value, path short-circuiting is rarely
243 <     * worthwhile. However, it is used (in awaitMatch) immediately
244 <     * before a waiting thread starts to block, as a final bit of
245 <     * helping at a point when contention with others is extremely
246 <     * unlikely (since if other threads that could release it are
247 <     * operating, then the current thread wouldn't be blocking).
242 >     * With such a small slack threshold value, it is rarely
243 >     * worthwhile to augment this with path short-circuiting; i.e.,
244 >     * unsplicing nodes between head and the first unmatched node, or
245 >     * similarly for tail, rather than advancing head or tail
246 >     * proper. However, it is used (in awaitMatch) immediately before
247 >     * a waiting thread starts to block, as a final bit of helping at
248 >     * a point when contention with others is extremely unlikely
249 >     * (since if other threads that could release it are operating,
250 >     * then the current thread wouldn't be blocking).
251 >     *
252 >     * We allow both the head and tail fields to be null before any
253 >     * nodes are enqueued; initializing upon first append.  This
254 >     * simplifies some other logic, as well as providing more
255 >     * efficient explicit control paths instead of letting JVMs insert
256 >     * implicit NullPointerExceptions when they are null.  While not
257 >     * currently fully implemented, we also leave open the possibility
258 >     * of re-nulling these fields when empty (which is complicated to
259 >     * arrange, for little benefit.)
260       *
261       * All enqueue/dequeue operations are handled by the single method
262       * "xfer" with parameters indicating whether to act as some form
263       * of offer, put, poll, take, or transfer (each possibly with
264       * timeout). The relative complexity of using one monolithic
265       * method outweighs the code bulk and maintenance problems of
266 <     * using nine separate methods.
266 >     * using separate methods for each case.
267       *
268       * Operation consists of up to three phases. The first is
269       * implemented within method xfer, the second in tryAppend, and
# Line 249 | Line 276 | public class LinkedTransferQueue<E> exte
276       *    case matching it and returning, also if necessary updating
277       *    head to one past the matched node (or the node itself if the
278       *    list has no other unmatched nodes). If the CAS misses, then
279 <     *    a retry loops until the slack is at most two. Traversals
280 <     *    also check if the initial head is now off-list, in which
281 <     *    case they start at the new head.
279 >     *    a loop retries advancing head by two steps until either
280 >     *    success or the slack is at most two. By requiring that each
281 >     *    attempt advances head by two (if applicable), we ensure that
282 >     *    the slack does not grow without bound. Traversals also check
283 >     *    if the initial head is now off-list, in which case they
284 >     *    start at the new head.
285       *
286       *    If no candidates are found and the call was untimed
287       *    poll/offer, (argument "how" is NOW) return.
288       *
289       * 2. Try to append a new node (method tryAppend)
290       *
291 <     *    Starting at current tail pointer, try to append a new node
292 <     *    to the list (or if head was null, establish the first
293 <     *    node). Nodes can be appended only if their predecessors are
294 <     *    either already matched or are of the same mode. If we detect
295 <     *    otherwise, then a new node with opposite mode must have been
296 <     *    appended during traversal, so must restart at phase 1. The
297 <     *    traversal and update steps are otherwise similar to phase 1:
298 <     *    Retrying upon CAS misses and checking for staleness.  In
299 <     *    particular, if a self-link is encountered, then we can
300 <     *    safely jump to a node on the list by continuing the
301 <     *    traversal at current head.
291 >     *    Starting at current tail pointer, find the actual last node
292 >     *    and try to append a new node (or if head was null, establish
293 >     *    the first node). Nodes can be appended only if their
294 >     *    predecessors are either already matched or are of the same
295 >     *    mode. If we detect otherwise, then a new node with opposite
296 >     *    mode must have been appended during traversal, so we must
297 >     *    restart at phase 1. The traversal and update steps are
298 >     *    otherwise similar to phase 1: Retrying upon CAS misses and
299 >     *    checking for staleness.  In particular, if a self-link is
300 >     *    encountered, then we can safely jump to a node on the list
301 >     *    by continuing the traversal at current head.
302       *
303 <     *    On successful append, if the call was ASYNC, return
303 >     *    On successful append, if the call was ASYNC, return.
304       *
305       * 3. Await match or cancellation (method awaitMatch)
306       *
307       *    Wait for another thread to match node; instead cancelling if
308 <     *    current thread was interrupted or the wait timed out. On
308 >     *    the current thread was interrupted or the wait timed out. On
309       *    multiprocessors, we use front-of-queue spinning: If a node
310       *    appears to be the first unmatched node in the queue, it
311       *    spins a bit before blocking. In either case, before blocking
# Line 290 | Line 320 | public class LinkedTransferQueue<E> exte
320       *    to decide to occasionally perform a Thread.yield. While
321       *    yield has underdefined specs, we assume that might it help,
322       *    and will not hurt in limiting impact of spinning on busy
323 <     *    systems.  We also use much smaller (1/4) spins for nodes
324 <     *    that are not known to be front but whose predecessors have
325 <     *    not blocked -- these "chained" spins avoid artifacts of
323 >     *    systems.  We also use smaller (1/2) spins for nodes that are
324 >     *    not known to be front but whose predecessors have not
325 >     *    blocked -- these "chained" spins avoid artifacts of
326       *    front-of-queue rules which otherwise lead to alternating
327       *    nodes spinning vs blocking. Further, front threads that
328       *    represent phase changes (from data to request node or vice
329       *    versa) compared to their predecessors receive additional
330 <     *    spins, reflecting the longer code path lengths necessary to
331 <     *    release them under contention.
330 >     *    chained spins, reflecting longer paths typically required to
331 >     *    unblock threads during phase changes.
332       */
333  
334      /** True if on multiprocessor */
# Line 306 | Line 336 | public class LinkedTransferQueue<E> exte
336          Runtime.getRuntime().availableProcessors() > 1;
337  
338      /**
339 <     * The number of times to spin (with on average one randomly
340 <     * interspersed call to Thread.yield) on multiprocessor before
341 <     * blocking when a node is apparently the first waiter in the
342 <     * queue.  See above for explanation. Must be a power of two. The
343 <     * value is empirically derived -- it works pretty well across a
344 <     * variety of processors, numbers of CPUs, and OSes.
339 >     * The number of times to spin (with randomly interspersed calls
340 >     * to Thread.yield) on multiprocessor before blocking when a node
341 >     * is apparently the first waiter in the queue.  See above for
342 >     * explanation. Must be a power of two. The value is empirically
343 >     * derived -- it works pretty well across a variety of processors,
344 >     * numbers of CPUs, and OSes.
345       */
346      private static final int FRONT_SPINS   = 1 << 7;
347  
348      /**
349       * The number of times to spin before blocking when a node is
350 <     * preceded by another node that is apparently spinning.
350 >     * preceded by another node that is apparently spinning.  Also
351 >     * serves as an increment to FRONT_SPINS on phase changes, and as
352 >     * base average frequency for yielding during spins. Must be a
353 >     * power of two.
354       */
355 <    private static final int CHAINED_SPINS = FRONT_SPINS >>> 2;
355 >    private static final int CHAINED_SPINS = FRONT_SPINS >>> 1;
356  
357      /**
358 <     * Queue nodes. Uses Object, not E for items to allow forgetting
358 >     * Queue nodes. Uses Object, not E, for items to allow forgetting
359       * them after use.  Relies heavily on Unsafe mechanics to minimize
360 <     * unecessary ordering constraints: Writes that intrinsically
360 >     * unnecessary ordering constraints: Writes that intrinsically
361       * precede or follow CASes use simple relaxed forms.  Other
362       * cleanups use releasing/lazy writes.
363       */
364 <    static final class Node {
364 >    static final class Node<E> {
365          final boolean isData;   // false if this is a request node
366 <        volatile Object item;   // initially nonnull if isData; CASed to match
367 <        volatile Node next;
366 >        volatile Object item;   // initially non-null if isData; CASed to match
367 >        volatile Node<E> next;
368          volatile Thread waiter; // null until waiting
369  
370          // CAS methods for fields
371 <        final boolean casNext(Node cmp, Node val) {
371 >        final boolean casNext(Node<E> cmp, Node<E> val) {
372              return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
373          }
374  
375          final boolean casItem(Object cmp, Object val) {
376 +            assert cmp == null || cmp.getClass() != Node.class;
377              return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
378          }
379  
380          /**
381 <         * Create a new node. Uses relaxed write because item can only
382 <         * be seen if followed by CAS
381 >         * Creates a new node. Uses relaxed write because item can only
382 >         * be seen if followed by CAS.
383           */
384 <        Node(Object item, boolean isData) {
384 >        Node(E item, boolean isData) {
385              UNSAFE.putObject(this, itemOffset, item); // relaxed write
386              this.isData = isData;
387          }
# Line 376 | Line 410 | public class LinkedTransferQueue<E> exte
410           */
411          final boolean isMatched() {
412              Object x = item;
413 <            return x == this || (x != null) != isData;
413 >            return (x == this) || ((x == null) == isData);
414 >        }
415 >
416 >        /**
417 >         * Returns true if this is an unmatched request node.
418 >         */
419 >        final boolean isUnmatchedRequest() {
420 >            return !isData && item == null;
421          }
422  
423          /**
# Line 391 | Line 432 | public class LinkedTransferQueue<E> exte
432          }
433  
434          /**
435 <         * Tries to artifically match a data node -- used by remove.
435 >         * Tries to artificially match a data node -- used by remove.
436           */
437          final boolean tryMatchData() {
438 +            assert isData;
439              Object x = item;
440              if (x != null && x != this && casItem(x, null)) {
441                  LockSupport.unpark(waiter);
# Line 415 | Line 457 | public class LinkedTransferQueue<E> exte
457      }
458  
459      /** head of the queue; null until first enqueue */
460 <    private transient volatile Node head;
460 >    transient volatile Node<E> head;
461  
462      /** predecessor of dangling unspliceable node */
463 <    private transient volatile Node cleanMe; // decl here to reduce contention
463 >    private transient volatile Node<E> cleanMe; // decl here reduces contention
464  
465      /** tail of the queue; null until first append */
466 <    private transient volatile Node tail;
466 >    private transient volatile Node<E> tail;
467  
468      // CAS methods for fields
469 <    private boolean casTail(Node cmp, Node val) {
469 >    private boolean casTail(Node<E> cmp, Node<E> val) {
470          return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);
471      }
472  
473 <    private boolean casHead(Node cmp, Node val) {
473 >    private boolean casHead(Node<E> cmp, Node<E> val) {
474          return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);
475      }
476  
477 <    private boolean casCleanMe(Node cmp, Node val) {
477 >    private boolean casCleanMe(Node<E> cmp, Node<E> val) {
478          return UNSAFE.compareAndSwapObject(this, cleanMeOffset, cmp, val);
479      }
480  
# Line 445 | Line 487 | public class LinkedTransferQueue<E> exte
487      private static final int SYNC    = 2; // for transfer, take
488      private static final int TIMEOUT = 3; // for timed poll, tryTransfer
489  
490 +    @SuppressWarnings("unchecked")
491 +    static <E> E cast(Object item) {
492 +        assert item == null || item.getClass() != Node.class;
493 +        return (E) item;
494 +    }
495 +
496      /**
497       * Implements all queuing methods. See above for explanation.
498       *
499       * @param e the item or null for take
500 <     * @param haveData true if this is a put else a take
500 >     * @param haveData true if this is a put, else a take
501       * @param how NOW, ASYNC, SYNC, or TIMEOUT
502       * @param nanos timeout in nanosecs, used only if mode is TIMEOUT
503 <     * @return an item if matched, else e;
503 >     * @return an item if matched, else e
504       * @throws NullPointerException if haveData mode but e is null
505       */
506 <    private Object xfer(Object e, boolean haveData, int how, long nanos) {
506 >    private E xfer(E e, boolean haveData, int how, long nanos) {
507          if (haveData && (e == null))
508              throw new NullPointerException();
509 <        Node s = null;                        // the node to append, if needed
509 >        Node<E> s = null;                     // the node to append, if needed
510  
511          retry: for (;;) {                     // restart on append race
512  
513 <            for (Node h = head, p = h; p != null;) { // find & match first node
513 >            for (Node<E> h = head, p = h; p != null;) {
514 >                // find & match first node
515                  boolean isData = p.isData;
516                  Object item = p.item;
517                  if (item != p && (item != null) == isData) { // unmatched
518                      if (isData == haveData)   // can't match
519                          break;
520                      if (p.casItem(item, e)) { // match
521 <                        Thread w = p.waiter;
522 <                        while (p != h) {      // update head
523 <                            Node n = p.next;  // by 2 unless singleton
524 <                            if (n != null)
525 <                                p = n;
477 <                            if (head == h && casHead(h, p)) {
521 >                        for (Node<E> q = p; q != h;) {
522 >                            Node<E> n = q.next; // update head by 2
523 >                            if (n != null)    // unless singleton
524 >                                q = n;
525 >                            if (head == h && casHead(h, q)) {
526                                  h.forgetNext();
527                                  break;
528                              }                 // advance and retry
529                              if ((h = head)   == null ||
530 <                                (p = h.next) == null || !p.isMatched())
530 >                                (q = h.next) == null || !q.isMatched())
531                                  break;        // unless slack < 2
532                          }
533 <                        LockSupport.unpark(w);
534 <                        return item;
533 >                        LockSupport.unpark(p.waiter);
534 >                        return this.<E>cast(item);
535                      }
536                  }
537 <                Node n = p.next;
538 <                p = p != n ? n : (h = head);  // Use head if p offlist
537 >                Node<E> n = p.next;
538 >                p = (p != n) ? n : (h = head); // Use head if p offlist
539              }
540  
541              if (how >= ASYNC) {               // No matches available
542                  if (s == null)
543 <                    s = new Node(e, haveData);
544 <                Node pred = tryAppend(s, haveData);
543 >                    s = new Node<E>(e, haveData);
544 >                Node<E> pred = tryAppend(s, haveData);
545                  if (pred == null)
546                      continue retry;           // lost race vs opposite mode
547                  if (how >= SYNC)
548 <                    return awaitMatch(pred, s, e, how, nanos);
548 >                    return awaitMatch(s, pred, e, how, nanos);
549              }
550              return e; // not waiting
551          }
552      }
553  
554      /**
555 <     * Tries to append node s as tail
556 <     * @param haveData true if appending in data mode
555 >     * Tries to append node s as tail.
556 >     *
557       * @param s the node to append
558 +     * @param haveData true if appending in data mode
559       * @return null on failure due to losing race with append in
560       * different mode, else s's predecessor, or s itself if no
561       * predecessor
562       */
563 <    private Node tryAppend(Node s, boolean haveData) {
564 <        for (Node t = tail, p = t;;) { // move p to actual tail and append
565 <            Node n, u;                        // temps for reads of next & tail
563 >    private Node<E> tryAppend(Node<E> s, boolean haveData) {
564 >        for (Node<E> t = tail, p = t;;) { // move p to last node and append
565 >            Node<E> n, u;                     // temps for reads of next & tail
566              if (p == null && (p = head) == null) {
567                  if (casHead(null, s))
568                      return s;                 // initialize
569              }
570              else if (p.cannotPrecede(haveData))
571                  return null;                  // lost race vs opposite mode
572 <            else if ((n = p.next) != null)    // Not tail; keep traversing
572 >            else if ((n = p.next) != null)    // not last; keep traversing
573                  p = p != t && t != (u = tail) ? (t = u) : // stale tail
574 <                    p != n ? n : null;        // restart if off list
574 >                    (p != n) ? n : null;      // restart if off list
575              else if (!p.casNext(null, s))
576                  p = p.next;                   // re-read on CAS failure
577              else {
578 <                if (p != t) {                 // Update if slack now >= 2
578 >                if (p != t) {                 // update if slack now >= 2
579                      while ((tail != t || !casTail(t, s)) &&
580                             (t = tail)   != null &&
581                             (s = t.next) != null && // advance and retry
# Line 540 | Line 589 | public class LinkedTransferQueue<E> exte
589      /**
590       * Spins/yields/blocks until node s is matched or caller gives up.
591       *
543     * @param pred the predecessor of s or s or null if none
592       * @param s the waiting node
593 +     * @param pred the predecessor of s, or s itself if it has no
594 +     * predecessor, or null if unknown (the null case does not occur
595 +     * in any current calls but may in possible future extensions)
596       * @param e the comparison value for checking match
597       * @param how either SYNC or TIMEOUT
598       * @param nanos timeout value
599       * @return matched item, or e if unmatched on interrupt or timeout
600       */
601 <    private Object awaitMatch(Node pred, Node s, Object e,
551 <                              int how, long nanos) {
601 >    private E awaitMatch(Node<E> s, Node<E> pred, E e, int how, long nanos) {
602          long lastTime = (how == TIMEOUT) ? System.nanoTime() : 0L;
603          Thread w = Thread.currentThread();
604          int spins = -1; // initialized after first item and cancel checks
# Line 557 | Line 607 | public class LinkedTransferQueue<E> exte
607          for (;;) {
608              Object item = s.item;
609              if (item != e) {                  // matched
610 +                assert item != s;
611                  s.forgetContents();           // avoid garbage
612 <                return item;
612 >                return this.<E>cast(item);
613              }
614              if ((w.isInterrupted() || (how == TIMEOUT && nanos <= 0)) &&
615 <                     s.casItem(e, s)) {       // cancel
615 >                    s.casItem(e, s)) {       // cancel
616                  unsplice(pred, s);
617                  return e;
618              }
# Line 570 | Line 621 | public class LinkedTransferQueue<E> exte
621                  if ((spins = spinsFor(pred, s.isData)) > 0)
622                      randomYields = ThreadLocalRandom.current();
623              }
624 <            else if (spins > 0) {             // spin, occasionally yield
625 <                if (randomYields.nextInt(FRONT_SPINS) == 0)
626 <                    Thread.yield();
627 <                --spins;
624 >            else if (spins > 0) {             // spin
625 >                if (--spins == 0)
626 >                    shortenHeadPath();        // reduce slack before blocking
627 >                else if (randomYields.nextInt(CHAINED_SPINS) == 0)
628 >                    Thread.yield();           // occasionally yield
629              }
630              else if (s.waiter == null) {
631 <                shortenHeadPath();            // reduce slack before blocking
580 <                s.waiter = w;                 // request unpark
631 >                s.waiter = w;                 // request unpark then recheck
632              }
633              else if (how == TIMEOUT) {
634                  long now = System.nanoTime();
# Line 587 | Line 638 | public class LinkedTransferQueue<E> exte
638              }
639              else {
640                  LockSupport.park(this);
641 +                s.waiter = null;
642                  spins = -1;                   // spin if front upon wakeup
643              }
644          }
645      }
646  
647      /**
648 <     * Return spin/yield value for a node with given predecessor and
648 >     * Returns spin/yield value for a node with given predecessor and
649       * data mode. See above for explanation.
650       */
651 <    private static int spinsFor(Node pred, boolean haveData) {
651 >    private static int spinsFor(Node<?> pred, boolean haveData) {
652          if (MP && pred != null) {
653 <            boolean predData = pred.isData;
654 <            if (predData != haveData)         // front and phase change
655 <                return FRONT_SPINS + (FRONT_SPINS >>> 1);
604 <            if (predData != (pred.item != null)) // probably at front
653 >            if (pred.isData != haveData)      // phase change
654 >                return FRONT_SPINS + CHAINED_SPINS;
655 >            if (pred.isMatched())             // probably at front
656                  return FRONT_SPINS;
657              if (pred.waiter == null)          // pred apparently spinning
658                  return CHAINED_SPINS;
# Line 614 | Line 665 | public class LinkedTransferQueue<E> exte
665       * or trailing node; failing on contention.
666       */
667      private void shortenHeadPath() {
668 <        Node h, hn, p, q;
668 >        Node<E> h, hn, p, q;
669          if ((p = h = head) != null && h.isMatched() &&
670              (q = hn = h.next) != null) {
671 <            Node n;
671 >            Node<E> n;
672              while ((n = q.next) != q) {
673                  if (n == null || !q.isMatched()) {
674                      if (hn != q && h.next == hn)
# Line 633 | Line 684 | public class LinkedTransferQueue<E> exte
684      /* -------------- Traversal methods -------------- */
685  
686      /**
687 <     * Return the first unmatched node of the given mode, or null if
687 >     * Returns the first unmatched node of the given mode, or null if
688       * none.  Used by methods isEmpty, hasWaitingConsumer.
689       */
690 <    private Node firstOfMode(boolean data) {
691 <        for (Node p = head; p != null; ) {
690 >    private Node<E> firstOfMode(boolean data) {
691 >        for (Node<E> p = head; p != null; ) {
692              if (!p.isMatched())
693 <                return p.isData == data? p : null;
694 <            Node n = p.next;
695 <            p = n != p ? n : head;
693 >                return (p.isData == data) ? p : null;
694 >            Node<E> n = p.next;
695 >            p = (n != p) ? n : head;
696          }
697          return null;
698      }
699  
700      /**
701       * Returns the item in the first unmatched node with isData; or
702 <     * null if none. Used by peek.
702 >     * null if none.  Used by peek.
703       */
704 <    private Object firstDataItem() {
705 <        for (Node p = head; p != null; ) {
704 >    private E firstDataItem() {
705 >        for (Node<E> p = head; p != null; ) {
706              boolean isData = p.isData;
707              Object item = p.item;
708              if (item != p && (item != null) == isData)
709 <                return isData ? item : null;
710 <            Node n = p.next;
711 <            p = n != p ? n : head;
709 >                return isData ? this.<E>cast(item) : null;
710 >            Node<E> n = p.next;
711 >            p = (n != p) ? n : head;
712          }
713          return null;
714      }
715  
716      /**
717 <     * Traverse and count nodes of the given mode.
718 <     * Used by methds size and getWaitingConsumerCount.
717 >     * Traverses and counts unmatched nodes of the given mode.
718 >     * Used by methods size and getWaitingConsumerCount.
719       */
720      private int countOfMode(boolean data) {
721          int count = 0;
722 <        for (Node p = head; p != null; ) {
722 >        for (Node<E> p = head; p != null; ) {
723              if (!p.isMatched()) {
724                  if (p.isData != data)
725                      return 0;
726                  if (++count == Integer.MAX_VALUE) // saturated
727                      break;
728              }
729 <            Node n = p.next;
729 >            Node<E> n = p.next;
730              if (n != p)
731                  p = n;
732              else {
# Line 687 | Line 738 | public class LinkedTransferQueue<E> exte
738      }
739  
740      final class Itr implements Iterator<E> {
741 <        private Node nextNode;   // next node to return item for
742 <        private Object nextItem; // the corresponding item
743 <        private Node lastRet;    // last returned node, to support remove
741 >        private Node<E> nextNode;   // next node to return item for
742 >        private E nextItem;         // the corresponding item
743 >        private Node<E> lastRet;    // last returned node, to support remove
744 >        private Node<E> lastPred;   // predecessor to unlink lastRet
745  
746          /**
747           * Moves to next node after prev, or first node if prev null.
748           */
749 <        private void advance(Node prev) {
749 >        private void advance(Node<E> prev) {
750 >            lastPred = lastRet;
751              lastRet = prev;
752 <            Node p;
752 >            Node<E> p;
753              if (prev == null || (p = prev.next) == prev)
754                  p = head;
755              while (p != null) {
756                  Object item = p.item;
757                  if (p.isData) {
758                      if (item != null && item != p) {
759 <                        nextItem = item;
759 >                        nextItem = LinkedTransferQueue.this.<E>cast(item);
760                          nextNode = p;
761                          return;
762                      }
763                  }
764                  else if (item == null)
765                      break;
766 <                Node n = p.next;
767 <                p = n != p ? n : head;
766 >                Node<E> n = p.next;
767 >                p = (n != p) ? n : head;
768              }
769              nextNode = null;
770          }
# Line 725 | Line 778 | public class LinkedTransferQueue<E> exte
778          }
779  
780          public final E next() {
781 <            Node p = nextNode;
781 >            Node<E> p = nextNode;
782              if (p == null) throw new NoSuchElementException();
783 <            Object e = nextItem;
783 >            E e = nextItem;
784              advance(p);
785 <            return (E) e;
785 >            return e;
786          }
787  
788          public final void remove() {
789 <            Node p = lastRet;
789 >            Node<E> p = lastRet;
790              if (p == null) throw new IllegalStateException();
791 <            lastRet = null;
739 <            findAndRemoveNode(p);
791 >            findAndRemoveDataNode(lastPred, p);
792          }
793      }
794  
# Line 749 | Line 801 | public class LinkedTransferQueue<E> exte
801       * @param pred predecessor of node to be unspliced
802       * @param s the node to be unspliced
803       */
804 <    private void unsplice(Node pred, Node s) {
804 >    private void unsplice(Node<E> pred, Node<E> s) {
805          s.forgetContents(); // clear unneeded fields
806          /*
807           * At any given time, exactly one node on list cannot be
808 <         * deleted -- the last inserted node. To accommodate this, if
809 <         * we cannot delete s, we save its predecessor as "cleanMe",
808 >         * unlinked -- the last inserted node. To accommodate this, if
809 >         * we cannot unlink s, we save its predecessor as "cleanMe",
810           * processing the previously saved version first. Because only
811           * one node in the list can have a null next, at least one of
812           * node s or the node previously saved can always be
# Line 762 | Line 814 | public class LinkedTransferQueue<E> exte
814           */
815          if (pred != null && pred != s) {
816              while (pred.next == s) {
817 <                Node oldpred = cleanMe == null? null : reclean();
818 <                Node n = s.next;
817 >                Node<E> oldpred = (cleanMe == null) ? null : reclean();
818 >                Node<E> n = s.next;
819                  if (n != null) {
820                      if (n != s)
821                          pred.casNext(s, n);
822                      break;
823                  }
824                  if (oldpred == pred ||      // Already saved
825 <                    (oldpred == null && casCleanMe(null, pred)))
826 <                    break;                  // Postpone cleaning
825 >                    ((oldpred == null || oldpred.next == s) &&
826 >                     casCleanMe(oldpred, pred))) {
827 >                    break;
828 >                }
829              }
830          }
831      }
# Line 782 | Line 836 | public class LinkedTransferQueue<E> exte
836       *
837       * @return current cleanMe node (or null)
838       */
839 <    private Node reclean() {
839 >    private Node<E> reclean() {
840          /*
841           * cleanMe is, or at one time was, predecessor of a cancelled
842           * node s that was the tail so could not be unspliced.  If it
# Line 793 | Line 847 | public class LinkedTransferQueue<E> exte
847           * we can (must) clear cleanMe without unsplicing.  This can
848           * loop only due to contention.
849           */
850 <        Node pred;
850 >        Node<E> pred;
851          while ((pred = cleanMe) != null) {
852 <            Node s = pred.next;
853 <            Node n;
852 >            Node<E> s = pred.next;
853 >            Node<E> n;
854              if (s == null || s == pred || !s.isMatched())
855                  casCleanMe(pred, null); // already gone
856              else if ((n = s.next) != null) {
# Line 812 | Line 866 | public class LinkedTransferQueue<E> exte
866  
867      /**
868       * Main implementation of Iterator.remove(). Find
869 <     * and unsplice the given node.
869 >     * and unsplice the given data node.
870 >     * @param possiblePred possible predecessor of s
871 >     * @param s the node to remove
872       */
873 <    final void findAndRemoveNode(Node s) {
873 >    final void findAndRemoveDataNode(Node<E> possiblePred, Node<E> s) {
874 >        assert s.isData;
875          if (s.tryMatchData()) {
876 <            Node pred = null;
877 <            Node p = head;
878 <            while (p != null) {
879 <                if (p == s) {
880 <                    unsplice(pred, p);
881 <                    break;
882 <                }
883 <                if (!p.isData && !p.isMatched())
884 <                    break;
885 <                pred = p;
886 <                if ((p = p.next) == pred) { // stale
887 <                    pred = null;
888 <                    p = head;
876 >            if (possiblePred != null && possiblePred.next == s)
877 >                unsplice(possiblePred, s); // was actual predecessor
878 >            else {
879 >                for (Node<E> pred = null, p = head; p != null; ) {
880 >                    if (p == s) {
881 >                        unsplice(pred, p);
882 >                        break;
883 >                    }
884 >                    if (p.isUnmatchedRequest())
885 >                        break;
886 >                    pred = p;
887 >                    if ((p = p.next) == pred) { // stale
888 >                        pred = null;
889 >                        p = head;
890 >                    }
891                  }
892              }
893          }
# Line 839 | Line 898 | public class LinkedTransferQueue<E> exte
898       */
899      private boolean findAndRemove(Object e) {
900          if (e != null) {
901 <            Node pred = null;
843 <            Node p = head;
844 <            while (p != null) {
901 >            for (Node<E> pred = null, p = head; p != null; ) {
902                  Object item = p.item;
903                  if (p.isData) {
904                      if (item != null && item != p && e.equals(item) &&
# Line 853 | Line 910 | public class LinkedTransferQueue<E> exte
910                  else if (item == null)
911                      break;
912                  pred = p;
913 <                if ((p = p.next) == pred) {
913 >                if ((p = p.next) == pred) { // stale
914                      pred = null;
915                      p = head;
916                  }
# Line 989 | Line 1046 | public class LinkedTransferQueue<E> exte
1046      }
1047  
1048      public E take() throws InterruptedException {
1049 <        Object e = xfer(null, false, SYNC, 0);
1049 >        E e = xfer(null, false, SYNC, 0);
1050          if (e != null)
1051 <            return (E)e;
1051 >            return e;
1052          Thread.interrupted();
1053          throw new InterruptedException();
1054      }
1055  
1056      public E poll(long timeout, TimeUnit unit) throws InterruptedException {
1057 <        Object e = xfer(null, false, TIMEOUT, unit.toNanos(timeout));
1057 >        E e = xfer(null, false, TIMEOUT, unit.toNanos(timeout));
1058          if (e != null || !Thread.interrupted())
1059 <            return (E)e;
1059 >            return e;
1060          throw new InterruptedException();
1061      }
1062  
1063      public E poll() {
1064 <        return (E)xfer(null, false, NOW, 0);
1064 >        return xfer(null, false, NOW, 0);
1065      }
1066  
1067      /**
# Line 1061 | Line 1118 | public class LinkedTransferQueue<E> exte
1118      }
1119  
1120      public E peek() {
1121 <        return (E) firstDataItem();
1121 >        return firstDataItem();
1122      }
1123  
1124      /**
# Line 1124 | Line 1181 | public class LinkedTransferQueue<E> exte
1181      }
1182  
1183      /**
1184 <     * Save the state to a stream (that is, serialize it).
1184 >     * Saves the state to a stream (that is, serializes it).
1185       *
1186       * @serialData All of the elements (each an {@code E}) in
1187       * the proper order, followed by a null
# Line 1140 | Line 1197 | public class LinkedTransferQueue<E> exte
1197      }
1198  
1199      /**
1200 <     * Reconstitute the Queue instance from a stream (that is,
1201 <     * deserialize it).
1200 >     * Reconstitutes the Queue instance from a stream (that is,
1201 >     * deserializes it).
1202       *
1203       * @param s the stream
1204       */
# Line 1157 | Line 1214 | public class LinkedTransferQueue<E> exte
1214          }
1215      }
1216  
1160
1217      // Unsafe mechanics
1218  
1219      private static final sun.misc.Unsafe UNSAFE = getUnsafe();
# Line 1180 | Line 1236 | public class LinkedTransferQueue<E> exte
1236          }
1237      }
1238  
1239 <    private static sun.misc.Unsafe getUnsafe() {
1239 >    /**
1240 >     * Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
1241 >     * Replace with a simple call to Unsafe.getUnsafe when integrating
1242 >     * into a jdk.
1243 >     *
1244 >     * @return a sun.misc.Unsafe
1245 >     */
1246 >    static sun.misc.Unsafe getUnsafe() {
1247          try {
1248              return sun.misc.Unsafe.getUnsafe();
1249          } catch (SecurityException se) {

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines