1 /*
   2  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   3  *
   4  * This code is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU General Public License version 2 only, as
   6  * published by the Free Software Foundation.  Oracle designates this
   7  * particular file as subject to the "Classpath" exception as provided
   8  * by Oracle in the LICENSE file that accompanied this code.
   9  *
  10  * This code is distributed in the hope that it will be useful, but WITHOUT
  11  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  13  * version 2 for more details (a copy is included in the LICENSE file that
  14  * accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License version
  17  * 2 along with this work; if not, write to the Free Software Foundation,
  18  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  19  *
  20  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  21  * or visit www.oracle.com if you need additional information or have any
  22  * questions.
  23  */
  24 
  25 /*
  26  * This file is available under and governed by the GNU General Public
  27  * License version 2 only, as published by the Free Software Foundation.
  28  * However, the following notice accompanied the original version of this
  29  * file:
  30  *
  31  * Written by Doug Lea, Bill Scherer, and Michael Scott with
  32  * assistance from members of JCP JSR-166 Expert Group and released to
  33  * the public domain, as explained at
  34  * http://creativecommons.org/publicdomain/zero/1.0/
  35  */
  36 
  37 package java.util.concurrent;
  38 
  39 import java.lang.invoke.MethodHandles;
  40 import java.lang.invoke.VarHandle;
  41 import java.util.AbstractQueue;
  42 import java.util.Collection;
  43 import java.util.Collections;
  44 import java.util.Iterator;
  45 import java.util.Objects;
  46 import java.util.Spliterator;
  47 import java.util.Spliterators;
  48 import java.util.concurrent.locks.LockSupport;
  49 import java.util.concurrent.locks.ReentrantLock;
  50 
  51 /**
  52  * A {@linkplain BlockingQueue blocking queue} in which each insert
  53  * operation must wait for a corresponding remove operation by another
  54  * thread, and vice versa.  A synchronous queue does not have any
  55  * internal capacity, not even a capacity of one.  You cannot
  56  * {@code peek} at a synchronous queue because an element is only
  57  * present when you try to remove it; you cannot insert an element
  58  * (using any method) unless another thread is trying to remove it;
  59  * you cannot iterate as there is nothing to iterate.  The
  60  * <em>head</em> of the queue is the element that the first queued
  61  * inserting thread is trying to add to the queue; if there is no such
  62  * queued thread then no element is available for removal and
  63  * {@code poll()} will return {@code null}.  For purposes of other
  64  * {@code Collection} methods (for example {@code contains}), a
  65  * {@code SynchronousQueue} acts as an empty collection.  This queue
  66  * does not permit {@code null} elements.
  67  *
  68  * <p>Synchronous queues are similar to rendezvous channels used in
  69  * CSP and Ada. They are well suited for handoff designs, in which an
  70  * object running in one thread must sync up with an object running
  71  * in another thread in order to hand it some information, event, or
  72  * task.
  73  *
  74  * <p>This class supports an optional fairness policy for ordering
  75  * waiting producer and consumer threads.  By default, this ordering
  76  * is not guaranteed. However, a queue constructed with fairness set
  77  * to {@code true} grants threads access in FIFO order.
  78  *
  79  * <p>This class and its iterator implement all of the <em>optional</em>
  80  * methods of the {@link Collection} and {@link Iterator} interfaces.
  81  *
  82  * <p>This class is a member of the
  83  * <a href="{@docRoot}/../technotes/guides/collections/index.html">
  84  * Java Collections Framework</a>.
  85  *
  86  * @since 1.5
  87  * @author Doug Lea and Bill Scherer and Michael Scott
  88  * @param <E> the type of elements held in this queue
  89  */
  90 public class SynchronousQueue<E> extends AbstractQueue<E>
  91     implements BlockingQueue<E>, java.io.Serializable {
  92     private static final long serialVersionUID = -3223113410248163686L;
  93 
  94     /*
  95      * This class implements extensions of the dual stack and dual
  96      * queue algorithms described in "Nonblocking Concurrent Objects
  97      * with Condition Synchronization", by W. N. Scherer III and
  98      * M. L. Scott.  18th Annual Conf. on Distributed Computing,
  99      * Oct. 2004 (see also
 100      * http://www.cs.rochester.edu/u/scott/synchronization/pseudocode/duals.html).
 101      * The (Lifo) stack is used for non-fair mode, and the (Fifo)
 102      * queue for fair mode. The performance of the two is generally
 103      * similar. Fifo usually supports higher throughput under
 104      * contention but Lifo maintains higher thread locality in common
 105      * applications.
 106      *
 107      * A dual queue (and similarly stack) is one that at any given
 108      * time either holds "data" -- items provided by put operations,
 109      * or "requests" -- slots representing take operations, or is
 110      * empty. A call to "fulfill" (i.e., a call requesting an item
 111      * from a queue holding data or vice versa) dequeues a
 112      * complementary node.  The most interesting feature of these
 113      * queues is that any operation can figure out which mode the
 114      * queue is in, and act accordingly without needing locks.
 115      *
 116      * Both the queue and stack extend abstract class Transferer
 117      * defining the single method transfer that does a put or a
 118      * take. These are unified into a single method because in dual
 119      * data structures, the put and take operations are symmetrical,
 120      * so nearly all code can be combined. The resulting transfer
 121      * methods are on the long side, but are easier to follow than
 122      * they would be if broken up into nearly-duplicated parts.
 123      *
 124      * The queue and stack data structures share many conceptual
 125      * similarities but very few concrete details. For simplicity,
 126      * they are kept distinct so that they can later evolve
 127      * separately.
 128      *
 129      * The algorithms here differ from the versions in the above paper
 130      * in extending them for use in synchronous queues, as well as
 131      * dealing with cancellation. The main differences include:
 132      *
 133      *  1. The original algorithms used bit-marked pointers, but
 134      *     the ones here use mode bits in nodes, leading to a number
 135      *     of further adaptations.
 136      *  2. SynchronousQueues must block threads waiting to become
 137      *     fulfilled.
 138      *  3. Support for cancellation via timeout and interrupts,
 139      *     including cleaning out cancelled nodes/threads
 140      *     from lists to avoid garbage retention and memory depletion.
 141      *
 142      * Blocking is mainly accomplished using LockSupport park/unpark,
 143      * except that nodes that appear to be the next ones to become
 144      * fulfilled first spin a bit (on multiprocessors only). On very
 145      * busy synchronous queues, spinning can dramatically improve
 146      * throughput. And on less busy ones, the amount of spinning is
 147      * small enough not to be noticeable.
 148      *
 149      * Cleaning is done in different ways in queues vs stacks.  For
 150      * queues, we can almost always remove a node immediately in O(1)
 151      * time (modulo retries for consistency checks) when it is
 152      * cancelled. But if it may be pinned as the current tail, it must
 153      * wait until some subsequent cancellation. For stacks, we need a
 154      * potentially O(n) traversal to be sure that we can remove the
 155      * node, but this can run concurrently with other threads
 156      * accessing the stack.
 157      *
 158      * While garbage collection takes care of most node reclamation
 159      * issues that otherwise complicate nonblocking algorithms, care
 160      * is taken to "forget" references to data, other nodes, and
 161      * threads that might be held on to long-term by blocked
 162      * threads. In cases where setting to null would otherwise
 163      * conflict with main algorithms, this is done by changing a
 164      * node's link to now point to the node itself. This doesn't arise
 165      * much for Stack nodes (because blocked threads do not hang on to
 166      * old head pointers), but references in Queue nodes must be
 167      * aggressively forgotten to avoid reachability of everything any
 168      * node has ever referred to since arrival.
 169      */
 170 
 171     /**
 172      * Shared internal API for dual stacks and queues.
 173      */
 174     abstract static class Transferer<E> {
 175         /**
 176          * Performs a put or take.
 177          *
 178          * @param e if non-null, the item to be handed to a consumer;
 179          *          if null, requests that transfer return an item
 180          *          offered by producer.
 181          * @param timed if this operation should timeout
 182          * @param nanos the timeout, in nanoseconds
 183          * @return if non-null, the item provided or received; if null,
 184          *         the operation failed due to timeout or interrupt --
 185          *         the caller can distinguish which of these occurred
 186          *         by checking Thread.interrupted.
 187          */
 188         abstract E transfer(E e, boolean timed, long nanos);
 189     }
 190 
 191     /**
 192      * The number of times to spin before blocking in timed waits.
 193      * The value is empirically derived -- it works well across a
 194      * variety of processors and OSes. Empirically, the best value
 195      * seems not to vary with number of CPUs (beyond 2) so is just
 196      * a constant.
 197      */
 198     static final int MAX_TIMED_SPINS =
 199         (Runtime.getRuntime().availableProcessors() < 2) ? 0 : 32;
 200 
 201     /**
 202      * The number of times to spin before blocking in untimed waits.
 203      * This is greater than timed value because untimed waits spin
 204      * faster since they don't need to check times on each spin.
 205      */
 206     static final int MAX_UNTIMED_SPINS = MAX_TIMED_SPINS * 16;
 207 
 208     /**
 209      * The number of nanoseconds for which it is faster to spin
 210      * rather than to use timed park. A rough estimate suffices.
 211      */
 212     static final long SPIN_FOR_TIMEOUT_THRESHOLD = 1000L;
 213 
 214     /** Dual stack */
 215     static final class TransferStack<E> extends Transferer<E> {
 216         /*
 217          * This extends Scherer-Scott dual stack algorithm, differing,
 218          * among other ways, by using "covering" nodes rather than
 219          * bit-marked pointers: Fulfilling operations push on marker
 220          * nodes (with FULFILLING bit set in mode) to reserve a spot
 221          * to match a waiting node.
 222          */
 223 
 224         /* Modes for SNodes, ORed together in node fields */
 225         /** Node represents an unfulfilled consumer */
 226         static final int REQUEST    = 0;
 227         /** Node represents an unfulfilled producer */
 228         static final int DATA       = 1;
 229         /** Node is fulfilling another unfulfilled DATA or REQUEST */
 230         static final int FULFILLING = 2;
 231 
 232         /** Returns true if m has fulfilling bit set. */
 233         static boolean isFulfilling(int m) { return (m & FULFILLING) != 0; }
 234 
 235         /** Node class for TransferStacks. */
 236         static final class SNode {
 237             volatile SNode next;        // next node in stack
 238             volatile SNode match;       // the node matched to this
 239             volatile Thread waiter;     // to control park/unpark
 240             Object item;                // data; or null for REQUESTs
 241             int mode;
 242             // Note: item and mode fields don't need to be volatile
 243             // since they are always written before, and read after,
 244             // other volatile/atomic operations.
 245 
 246             SNode(Object item) {
 247                 this.item = item;
 248             }
 249 
 250             boolean casNext(SNode cmp, SNode val) {
 251                 return cmp == next &&
 252                     SNEXT.compareAndSet(this, cmp, val);
 253             }
 254 
 255             /**
 256              * Tries to match node s to this node, if so, waking up thread.
 257              * Fulfillers call tryMatch to identify their waiters.
 258              * Waiters block until they have been matched.
 259              *
 260              * @param s the node to match
 261              * @return true if successfully matched to s
 262              */
 263             boolean tryMatch(SNode s) {
 264                 if (match == null &&
 265                     SMATCH.compareAndSet(this, null, s)) {
 266                     Thread w = waiter;
 267                     if (w != null) {    // waiters need at most one unpark
 268                         waiter = null;
 269                         LockSupport.unpark(w);
 270                     }
 271                     return true;
 272                 }
 273                 return match == s;
 274             }
 275 
 276             /**
 277              * Tries to cancel a wait by matching node to itself.
 278              */
 279             void tryCancel() {
 280                 SMATCH.compareAndSet(this, null, this);
 281             }
 282 
 283             boolean isCancelled() {
 284                 return match == this;
 285             }
 286 
 287             // VarHandle mechanics
 288             private static final VarHandle SMATCH;
 289             private static final VarHandle SNEXT;
 290             static {
 291                 try {
 292                     MethodHandles.Lookup l = MethodHandles.lookup();
 293                     SMATCH = l.findVarHandle(SNode.class, "match", SNode.class);
 294                     SNEXT = l.findVarHandle(SNode.class, "next", SNode.class);
 295                 } catch (ReflectiveOperationException e) {
 296                     throw new Error(e);
 297                 }
 298             }
 299         }
 300 
 301         /** The head (top) of the stack */
 302         volatile SNode head;
 303 
 304         boolean casHead(SNode h, SNode nh) {
 305             return h == head &&
 306                 SHEAD.compareAndSet(this, h, nh);
 307         }
 308 
 309         /**
 310          * Creates or resets fields of a node. Called only from transfer
 311          * where the node to push on stack is lazily created and
 312          * reused when possible to help reduce intervals between reads
 313          * and CASes of head and to avoid surges of garbage when CASes
 314          * to push nodes fail due to contention.
 315          */
 316         static SNode snode(SNode s, Object e, SNode next, int mode) {
 317             if (s == null) s = new SNode(e);
 318             s.mode = mode;
 319             s.next = next;
 320             return s;
 321         }
 322 
 323         /**
 324          * Puts or takes an item.
 325          */
 326         @SuppressWarnings("unchecked")
 327         E transfer(E e, boolean timed, long nanos) {
 328             /*
 329              * Basic algorithm is to loop trying one of three actions:
 330              *
 331              * 1. If apparently empty or already containing nodes of same
 332              *    mode, try to push node on stack and wait for a match,
 333              *    returning it, or null if cancelled.
 334              *
 335              * 2. If apparently containing node of complementary mode,
 336              *    try to push a fulfilling node on to stack, match
 337              *    with corresponding waiting node, pop both from
 338              *    stack, and return matched item. The matching or
 339              *    unlinking might not actually be necessary because of
 340              *    other threads performing action 3:
 341              *
 342              * 3. If top of stack already holds another fulfilling node,
 343              *    help it out by doing its match and/or pop
 344              *    operations, and then continue. The code for helping
 345              *    is essentially the same as for fulfilling, except
 346              *    that it doesn't return the item.
 347              */
 348 
 349             SNode s = null; // constructed/reused as needed
 350             int mode = (e == null) ? REQUEST : DATA;
 351 
 352             for (;;) {
 353                 SNode h = head;
 354                 if (h == null || h.mode == mode) {  // empty or same-mode
 355                     if (timed && nanos <= 0L) {     // can't wait
 356                         if (h != null && h.isCancelled())
 357                             casHead(h, h.next);     // pop cancelled node
 358                         else
 359                             return null;
 360                     } else if (casHead(h, s = snode(s, e, h, mode))) {
 361                         SNode m = awaitFulfill(s, timed, nanos);
 362                         if (m == s) {               // wait was cancelled
 363                             clean(s);
 364                             return null;
 365                         }
 366                         if ((h = head) != null && h.next == s)
 367                             casHead(h, s.next);     // help s's fulfiller
 368                         return (E) ((mode == REQUEST) ? m.item : s.item);
 369                     }
 370                 } else if (!isFulfilling(h.mode)) { // try to fulfill
 371                     if (h.isCancelled())            // already cancelled
 372                         casHead(h, h.next);         // pop and retry
 373                     else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
 374                         for (;;) { // loop until matched or waiters disappear
 375                             SNode m = s.next;       // m is s's match
 376                             if (m == null) {        // all waiters are gone
 377                                 casHead(s, null);   // pop fulfill node
 378                                 s = null;           // use new node next time
 379                                 break;              // restart main loop
 380                             }
 381                             SNode mn = m.next;
 382                             if (m.tryMatch(s)) {
 383                                 casHead(s, mn);     // pop both s and m
 384                                 return (E) ((mode == REQUEST) ? m.item : s.item);
 385                             } else                  // lost match
 386                                 s.casNext(m, mn);   // help unlink
 387                         }
 388                     }
 389                 } else {                            // help a fulfiller
 390                     SNode m = h.next;               // m is h's match
 391                     if (m == null)                  // waiter is gone
 392                         casHead(h, null);           // pop fulfilling node
 393                     else {
 394                         SNode mn = m.next;
 395                         if (m.tryMatch(h))          // help match
 396                             casHead(h, mn);         // pop both h and m
 397                         else                        // lost match
 398                             h.casNext(m, mn);       // help unlink
 399                     }
 400                 }
 401             }
 402         }
 403 
 404         /**
 405          * Spins/blocks until node s is matched by a fulfill operation.
 406          *
 407          * @param s the waiting node
 408          * @param timed true if timed wait
 409          * @param nanos timeout value
 410          * @return matched node, or s if cancelled
 411          */
 412         SNode awaitFulfill(SNode s, boolean timed, long nanos) {
 413             /*
 414              * When a node/thread is about to block, it sets its waiter
 415              * field and then rechecks state at least one more time
 416              * before actually parking, thus covering race vs
 417              * fulfiller noticing that waiter is non-null so should be
 418              * woken.
 419              *
 420              * When invoked by nodes that appear at the point of call
 421              * to be at the head of the stack, calls to park are
 422              * preceded by spins to avoid blocking when producers and
 423              * consumers are arriving very close in time.  This can
 424              * happen enough to bother only on multiprocessors.
 425              *
 426              * The order of checks for returning out of main loop
 427              * reflects fact that interrupts have precedence over
 428              * normal returns, which have precedence over
 429              * timeouts. (So, on timeout, one last check for match is
 430              * done before giving up.) Except that calls from untimed
 431              * SynchronousQueue.{poll/offer} don't check interrupts
 432              * and don't wait at all, so are trapped in transfer
 433              * method rather than calling awaitFulfill.
 434              */
 435             final long deadline = timed ? System.nanoTime() + nanos : 0L;
 436             Thread w = Thread.currentThread();
 437             int spins = shouldSpin(s)
 438                 ? (timed ? MAX_TIMED_SPINS : MAX_UNTIMED_SPINS)
 439                 : 0;
 440             for (;;) {
 441                 if (w.isInterrupted())
 442                     s.tryCancel();
 443                 SNode m = s.match;
 444                 if (m != null)
 445                     return m;
 446                 if (timed) {
 447                     nanos = deadline - System.nanoTime();
 448                     if (nanos <= 0L) {
 449                         s.tryCancel();
 450                         continue;
 451                     }
 452                 }
 453                 if (spins > 0) {
 454                     Thread.onSpinWait();
 455                     spins = shouldSpin(s) ? (spins - 1) : 0;
 456                 }
 457                 else if (s.waiter == null)
 458                     s.waiter = w; // establish waiter so can park next iter
 459                 else if (!timed)
 460                     LockSupport.park(this);
 461                 else if (nanos > SPIN_FOR_TIMEOUT_THRESHOLD)
 462                     LockSupport.parkNanos(this, nanos);
 463             }
 464         }
 465 
 466         /**
 467          * Returns true if node s is at head or there is an active
 468          * fulfiller.
 469          */
 470         boolean shouldSpin(SNode s) {
 471             SNode h = head;
 472             return (h == s || h == null || isFulfilling(h.mode));
 473         }
 474 
 475         /**
 476          * Unlinks s from the stack.
 477          */
 478         void clean(SNode s) {
 479             s.item = null;   // forget item
 480             s.waiter = null; // forget thread
 481 
 482             /*
 483              * At worst we may need to traverse entire stack to unlink
 484              * s. If there are multiple concurrent calls to clean, we
 485              * might not see s if another thread has already removed
 486              * it. But we can stop when we see any node known to
 487              * follow s. We use s.next unless it too is cancelled, in
 488              * which case we try the node one past. We don't check any
 489              * further because we don't want to doubly traverse just to
 490              * find sentinel.
 491              */
 492 
 493             SNode past = s.next;
 494             if (past != null && past.isCancelled())
 495                 past = past.next;
 496 
 497             // Absorb cancelled nodes at head
 498             SNode p;
 499             while ((p = head) != null && p != past && p.isCancelled())
 500                 casHead(p, p.next);
 501 
 502             // Unsplice embedded nodes
 503             while (p != null && p != past) {
 504                 SNode n = p.next;
 505                 if (n != null && n.isCancelled())
 506                     p.casNext(n, n.next);
 507                 else
 508                     p = n;
 509             }
 510         }
 511 
 512         // VarHandle mechanics
 513         private static final VarHandle SHEAD;
 514         static {
 515             try {
 516                 MethodHandles.Lookup l = MethodHandles.lookup();
 517                 SHEAD = l.findVarHandle(TransferStack.class, "head", SNode.class);
 518             } catch (ReflectiveOperationException e) {
 519                 throw new Error(e);
 520             }
 521         }
 522     }
 523 
 524     /** Dual Queue */
 525     static final class TransferQueue<E> extends Transferer<E> {
 526         /*
 527          * This extends Scherer-Scott dual queue algorithm, differing,
 528          * among other ways, by using modes within nodes rather than
 529          * marked pointers. The algorithm is a little simpler than
 530          * that for stacks because fulfillers do not need explicit
 531          * nodes, and matching is done by CAS'ing QNode.item field
 532          * from non-null to null (for put) or vice versa (for take).
 533          */
 534 
 535         /** Node class for TransferQueue. */
 536         static final class QNode {
 537             volatile QNode next;          // next node in queue
 538             volatile Object item;         // CAS'ed to or from null
 539             volatile Thread waiter;       // to control park/unpark
 540             final boolean isData;
 541 
 542             QNode(Object item, boolean isData) {
 543                 this.item = item;
 544                 this.isData = isData;
 545             }
 546 
 547             boolean casNext(QNode cmp, QNode val) {
 548                 return next == cmp &&
 549                     QNEXT.compareAndSet(this, cmp, val);
 550             }
 551 
 552             boolean casItem(Object cmp, Object val) {
 553                 return item == cmp &&
 554                     QITEM.compareAndSet(this, cmp, val);
 555             }
 556 
 557             /**
 558              * Tries to cancel by CAS'ing ref to this as item.
 559              */
 560             void tryCancel(Object cmp) {
 561                 QITEM.compareAndSet(this, cmp, this);
 562             }
 563 
 564             boolean isCancelled() {
 565                 return item == this;
 566             }
 567 
 568             /**
 569              * Returns true if this node is known to be off the queue
 570              * because its next pointer has been forgotten due to
 571              * an advanceHead operation.
 572              */
 573             boolean isOffList() {
 574                 return next == this;
 575             }
 576 
 577             // VarHandle mechanics
 578             private static final VarHandle QITEM;
 579             private static final VarHandle QNEXT;
 580             static {
 581                 try {
 582                     MethodHandles.Lookup l = MethodHandles.lookup();
 583                     QITEM = l.findVarHandle(QNode.class, "item", Object.class);
 584                     QNEXT = l.findVarHandle(QNode.class, "next", QNode.class);
 585                 } catch (ReflectiveOperationException e) {
 586                     throw new Error(e);
 587                 }
 588             }
 589         }
 590 
 591         /** Head of queue */
 592         transient volatile QNode head;
 593         /** Tail of queue */
 594         transient volatile QNode tail;
 595         /**
 596          * Reference to a cancelled node that might not yet have been
 597          * unlinked from queue because it was the last inserted node
 598          * when it was cancelled.
 599          */
 600         transient volatile QNode cleanMe;
 601 
 602         TransferQueue() {
 603             QNode h = new QNode(null, false); // initialize to dummy node.
 604             head = h;
 605             tail = h;
 606         }
 607 
 608         /**
 609          * Tries to cas nh as new head; if successful, unlink
 610          * old head's next node to avoid garbage retention.
 611          */
 612         void advanceHead(QNode h, QNode nh) {
 613             if (h == head &&
 614                 QHEAD.compareAndSet(this, h, nh))
 615                 h.next = h; // forget old next
 616         }
 617 
 618         /**
 619          * Tries to cas nt as new tail.
 620          */
 621         void advanceTail(QNode t, QNode nt) {
 622             if (tail == t)
 623                 QTAIL.compareAndSet(this, t, nt);
 624         }
 625 
 626         /**
 627          * Tries to CAS cleanMe slot.
 628          */
 629         boolean casCleanMe(QNode cmp, QNode val) {
 630             return cleanMe == cmp &&
 631                 QCLEANME.compareAndSet(this, cmp, val);
 632         }
 633 
 634         /**
 635          * Puts or takes an item.
 636          */
 637         @SuppressWarnings("unchecked")
 638         E transfer(E e, boolean timed, long nanos) {
 639             /* Basic algorithm is to loop trying to take either of
 640              * two actions:
 641              *
 642              * 1. If queue apparently empty or holding same-mode nodes,
 643              *    try to add node to queue of waiters, wait to be
 644              *    fulfilled (or cancelled) and return matching item.
 645              *
 646              * 2. If queue apparently contains waiting items, and this
 647              *    call is of complementary mode, try to fulfill by CAS'ing
 648              *    item field of waiting node and dequeuing it, and then
 649              *    returning matching item.
 650              *
 651              * In each case, along the way, check for and try to help
 652              * advance head and tail on behalf of other stalled/slow
 653              * threads.
 654              *
 655              * The loop starts off with a null check guarding against
 656              * seeing uninitialized head or tail values. This never
 657              * happens in current SynchronousQueue, but could if
 658              * callers held non-volatile/final ref to the
 659              * transferer. The check is here anyway because it places
 660              * null checks at top of loop, which is usually faster
 661              * than having them implicitly interspersed.
 662              */
 663 
 664             QNode s = null; // constructed/reused as needed
 665             boolean isData = (e != null);
 666 
 667             for (;;) {
 668                 QNode t = tail;
 669                 QNode h = head;
 670                 if (t == null || h == null)         // saw uninitialized value
 671                     continue;                       // spin
 672 
 673                 if (h == t || t.isData == isData) { // empty or same-mode
 674                     QNode tn = t.next;
 675                     if (t != tail)                  // inconsistent read
 676                         continue;
 677                     if (tn != null) {               // lagging tail
 678                         advanceTail(t, tn);
 679                         continue;
 680                     }
 681                     if (timed && nanos <= 0L)       // can't wait
 682                         return null;
 683                     if (s == null)
 684                         s = new QNode(e, isData);
 685                     if (!t.casNext(null, s))        // failed to link in
 686                         continue;
 687 
 688                     advanceTail(t, s);              // swing tail and wait
 689                     Object x = awaitFulfill(s, e, timed, nanos);
 690                     if (x == s) {                   // wait was cancelled
 691                         clean(t, s);
 692                         return null;
 693                     }
 694 
 695                     if (!s.isOffList()) {           // not already unlinked
 696                         advanceHead(t, s);          // unlink if head
 697                         if (x != null)              // and forget fields
 698                             s.item = s;
 699                         s.waiter = null;
 700                     }
 701                     return (x != null) ? (E)x : e;
 702 
 703                 } else {                            // complementary-mode
 704                     QNode m = h.next;               // node to fulfill
 705                     if (t != tail || m == null || h != head)
 706                         continue;                   // inconsistent read
 707 
 708                     Object x = m.item;
 709                     if (isData == (x != null) ||    // m already fulfilled
 710                         x == m ||                   // m cancelled
 711                         !m.casItem(x, e)) {         // lost CAS
 712                         advanceHead(h, m);          // dequeue and retry
 713                         continue;
 714                     }
 715 
 716                     advanceHead(h, m);              // successfully fulfilled
 717                     LockSupport.unpark(m.waiter);
 718                     return (x != null) ? (E)x : e;
 719                 }
 720             }
 721         }
 722 
 723         /**
 724          * Spins/blocks until node s is fulfilled.
 725          *
 726          * @param s the waiting node
 727          * @param e the comparison value for checking match
 728          * @param timed true if timed wait
 729          * @param nanos timeout value
 730          * @return matched item, or s if cancelled
 731          */
 732         Object awaitFulfill(QNode s, E e, boolean timed, long nanos) {
 733             /* Same idea as TransferStack.awaitFulfill */
 734             final long deadline = timed ? System.nanoTime() + nanos : 0L;
 735             Thread w = Thread.currentThread();
 736             int spins = (head.next == s)
 737                 ? (timed ? MAX_TIMED_SPINS : MAX_UNTIMED_SPINS)
 738                 : 0;
 739             for (;;) {
 740                 if (w.isInterrupted())
 741                     s.tryCancel(e);
 742                 Object x = s.item;
 743                 if (x != e)
 744                     return x;
 745                 if (timed) {
 746                     nanos = deadline - System.nanoTime();
 747                     if (nanos <= 0L) {
 748                         s.tryCancel(e);
 749                         continue;
 750                     }
 751                 }
 752                 if (spins > 0) {
 753                     --spins;
 754                     Thread.onSpinWait();
 755                 }
 756                 else if (s.waiter == null)
 757                     s.waiter = w;
 758                 else if (!timed)
 759                     LockSupport.park(this);
 760                 else if (nanos > SPIN_FOR_TIMEOUT_THRESHOLD)
 761                     LockSupport.parkNanos(this, nanos);
 762             }
 763         }
 764 
 765         /**
 766          * Gets rid of cancelled node s with original predecessor pred.
 767          */
 768         void clean(QNode pred, QNode s) {
 769             s.waiter = null; // forget thread
 770             /*
 771              * At any given time, exactly one node on list cannot be
 772              * deleted -- the last inserted node. To accommodate this,
 773              * if we cannot delete s, we save its predecessor as
 774              * "cleanMe", deleting the previously saved version
 775              * first. At least one of node s or the node previously
 776              * saved can always be deleted, so this always terminates.
 777              */
 778             while (pred.next == s) { // Return early if already unlinked
 779                 QNode h = head;
 780                 QNode hn = h.next;   // Absorb cancelled first node as head
 781                 if (hn != null && hn.isCancelled()) {
 782                     advanceHead(h, hn);
 783                     continue;
 784                 }
 785                 QNode t = tail;      // Ensure consistent read for tail
 786                 if (t == h)
 787                     return;
 788                 QNode tn = t.next;
 789                 if (t != tail)
 790                     continue;
 791                 if (tn != null) {
 792                     advanceTail(t, tn);
 793                     continue;
 794                 }
 795                 if (s != t) {        // If not tail, try to unsplice
 796                     QNode sn = s.next;
 797                     if (sn == s || pred.casNext(s, sn))
 798                         return;
 799                 }
 800                 QNode dp = cleanMe;
 801                 if (dp != null) {    // Try unlinking previous cancelled node
 802                     QNode d = dp.next;
 803                     QNode dn;
 804                     if (d == null ||               // d is gone or
 805                         d == dp ||                 // d is off list or
 806                         !d.isCancelled() ||        // d not cancelled or
 807                         (d != t &&                 // d not tail and
 808                          (dn = d.next) != null &&  //   has successor
 809                          dn != d &&                //   that is on list
 810                          dp.casNext(d, dn)))       // d unspliced
 811                         casCleanMe(dp, null);
 812                     if (dp == pred)
 813                         return;      // s is already saved node
 814                 } else if (casCleanMe(null, pred))
 815                     return;          // Postpone cleaning s
 816             }
 817         }
 818 
 819         // VarHandle mechanics
 820         private static final VarHandle QHEAD;
 821         private static final VarHandle QTAIL;
 822         private static final VarHandle QCLEANME;
 823         static {
 824             try {
 825                 MethodHandles.Lookup l = MethodHandles.lookup();
 826                 QHEAD = l.findVarHandle(TransferQueue.class, "head",
 827                                         QNode.class);
 828                 QTAIL = l.findVarHandle(TransferQueue.class, "tail",
 829                                         QNode.class);
 830                 QCLEANME = l.findVarHandle(TransferQueue.class, "cleanMe",
 831                                            QNode.class);
 832             } catch (ReflectiveOperationException e) {
 833                 throw new Error(e);
 834             }
 835         }
 836     }
 837 
 838     /**
 839      * The transferer. Set only in constructor, but cannot be declared
 840      * as final without further complicating serialization.  Since
 841      * this is accessed only at most once per public method, there
 842      * isn't a noticeable performance penalty for using volatile
 843      * instead of final here.
 844      */
 845     private transient volatile Transferer<E> transferer;
 846 
 847     /**
 848      * Creates a {@code SynchronousQueue} with nonfair access policy.
 849      */
 850     public SynchronousQueue() {
 851         this(false);
 852     }
 853 
 854     /**
 855      * Creates a {@code SynchronousQueue} with the specified fairness policy.
 856      *
 857      * @param fair if true, waiting threads contend in FIFO order for
 858      *        access; otherwise the order is unspecified.
 859      */
 860     public SynchronousQueue(boolean fair) {
 861         transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
 862     }
 863 
 864     /**
 865      * Adds the specified element to this queue, waiting if necessary for
 866      * another thread to receive it.
 867      *
 868      * @throws InterruptedException {@inheritDoc}
 869      * @throws NullPointerException {@inheritDoc}
 870      */
 871     public void put(E e) throws InterruptedException {
 872         if (e == null) throw new NullPointerException();
 873         if (transferer.transfer(e, false, 0) == null) {
 874             Thread.interrupted();
 875             throw new InterruptedException();
 876         }
 877     }
 878 
 879     /**
 880      * Inserts the specified element into this queue, waiting if necessary
 881      * up to the specified wait time for another thread to receive it.
 882      *
 883      * @return {@code true} if successful, or {@code false} if the
 884      *         specified waiting time elapses before a consumer appears
 885      * @throws InterruptedException {@inheritDoc}
 886      * @throws NullPointerException {@inheritDoc}
 887      */
 888     public boolean offer(E e, long timeout, TimeUnit unit)
 889         throws InterruptedException {
 890         if (e == null) throw new NullPointerException();
 891         if (transferer.transfer(e, true, unit.toNanos(timeout)) != null)
 892             return true;
 893         if (!Thread.interrupted())
 894             return false;
 895         throw new InterruptedException();
 896     }
 897 
 898     /**
 899      * Inserts the specified element into this queue, if another thread is
 900      * waiting to receive it.
 901      *
 902      * @param e the element to add
 903      * @return {@code true} if the element was added to this queue, else
 904      *         {@code false}
 905      * @throws NullPointerException if the specified element is null
 906      */
 907     public boolean offer(E e) {
 908         if (e == null) throw new NullPointerException();
 909         return transferer.transfer(e, true, 0) != null;
 910     }
 911 
 912     /**
 913      * Retrieves and removes the head of this queue, waiting if necessary
 914      * for another thread to insert it.
 915      *
 916      * @return the head of this queue
 917      * @throws InterruptedException {@inheritDoc}
 918      */
 919     public E take() throws InterruptedException {
 920         E e = transferer.transfer(null, false, 0);
 921         if (e != null)
 922             return e;
 923         Thread.interrupted();
 924         throw new InterruptedException();
 925     }
 926 
 927     /**
 928      * Retrieves and removes the head of this queue, waiting
 929      * if necessary up to the specified wait time, for another thread
 930      * to insert it.
 931      *
 932      * @return the head of this queue, or {@code null} if the
 933      *         specified waiting time elapses before an element is present
 934      * @throws InterruptedException {@inheritDoc}
 935      */
 936     public E poll(long timeout, TimeUnit unit) throws InterruptedException {
 937         E e = transferer.transfer(null, true, unit.toNanos(timeout));
 938         if (e != null || !Thread.interrupted())
 939             return e;
 940         throw new InterruptedException();
 941     }
 942 
 943     /**
 944      * Retrieves and removes the head of this queue, if another thread
 945      * is currently making an element available.
 946      *
 947      * @return the head of this queue, or {@code null} if no
 948      *         element is available
 949      */
 950     public E poll() {
 951         return transferer.transfer(null, true, 0);
 952     }
 953 
 954     /**
 955      * Always returns {@code true}.
 956      * A {@code SynchronousQueue} has no internal capacity.
 957      *
 958      * @return {@code true}
 959      */
 960     public boolean isEmpty() {
 961         return true;
 962     }
 963 
 964     /**
 965      * Always returns zero.
 966      * A {@code SynchronousQueue} has no internal capacity.
 967      *
 968      * @return zero
 969      */
 970     public int size() {
 971         return 0;
 972     }
 973 
 974     /**
 975      * Always returns zero.
 976      * A {@code SynchronousQueue} has no internal capacity.
 977      *
 978      * @return zero
 979      */
 980     public int remainingCapacity() {
 981         return 0;
 982     }
 983 
 984     /**
 985      * Does nothing.
 986      * A {@code SynchronousQueue} has no internal capacity.
 987      */
 988     public void clear() {
 989     }
 990 
 991     /**
 992      * Always returns {@code false}.
 993      * A {@code SynchronousQueue} has no internal capacity.
 994      *
 995      * @param o the element
 996      * @return {@code false}
 997      */
 998     public boolean contains(Object o) {
 999         return false;
1000     }
1001 
1002     /**
1003      * Always returns {@code false}.
1004      * A {@code SynchronousQueue} has no internal capacity.
1005      *
1006      * @param o the element to remove
1007      * @return {@code false}
1008      */
1009     public boolean remove(Object o) {
1010         return false;
1011     }
1012 
1013     /**
1014      * Returns {@code false} unless the given collection is empty.
1015      * A {@code SynchronousQueue} has no internal capacity.
1016      *
1017      * @param c the collection
1018      * @return {@code false} unless given collection is empty
1019      */
1020     public boolean containsAll(Collection<?> c) {
1021         return c.isEmpty();
1022     }
1023 
1024     /**
1025      * Always returns {@code false}.
1026      * A {@code SynchronousQueue} has no internal capacity.
1027      *
1028      * @param c the collection
1029      * @return {@code false}
1030      */
1031     public boolean removeAll(Collection<?> c) {
1032         return false;
1033     }
1034 
1035     /**
1036      * Always returns {@code false}.
1037      * A {@code SynchronousQueue} has no internal capacity.
1038      *
1039      * @param c the collection
1040      * @return {@code false}
1041      */
1042     public boolean retainAll(Collection<?> c) {
1043         return false;
1044     }
1045 
1046     /**
1047      * Always returns {@code null}.
1048      * A {@code SynchronousQueue} does not return elements
1049      * unless actively waited on.
1050      *
1051      * @return {@code null}
1052      */
1053     public E peek() {
1054         return null;
1055     }
1056 
1057     /**
1058      * Returns an empty iterator in which {@code hasNext} always returns
1059      * {@code false}.
1060      *
1061      * @return an empty iterator
1062      */
1063     public Iterator<E> iterator() {
1064         return Collections.emptyIterator();
1065     }
1066 
1067     /**
1068      * Returns an empty spliterator in which calls to
1069      * {@link java.util.Spliterator#trySplit()} always return {@code null}.
1070      *
1071      * @return an empty spliterator
1072      * @since 1.8
1073      */
1074     public Spliterator<E> spliterator() {
1075         return Spliterators.emptySpliterator();
1076     }
1077 
1078     /**
1079      * Returns a zero-length array.
1080      * @return a zero-length array
1081      */
1082     public Object[] toArray() {
1083         return new Object[0];
1084     }
1085 
1086     /**
1087      * Sets the zeroth element of the specified array to {@code null}
1088      * (if the array has non-zero length) and returns it.
1089      *
1090      * @param a the array
1091      * @return the specified array
1092      * @throws NullPointerException if the specified array is null
1093      */
1094     public <T> T[] toArray(T[] a) {
1095         if (a.length > 0)
1096             a[0] = null;
1097         return a;
1098     }
1099 
1100     /**
1101      * Always returns {@code "[]"}.
1102      * @return {@code "[]"}
1103      * @since 9
1104      */
1105     public String toString() {
1106         return "[]";
1107     }
1108 
1109     /**
1110      * @throws UnsupportedOperationException {@inheritDoc}
1111      * @throws ClassCastException            {@inheritDoc}
1112      * @throws NullPointerException          {@inheritDoc}
1113      * @throws IllegalArgumentException      {@inheritDoc}
1114      */
1115     public int drainTo(Collection<? super E> c) {
1116         Objects.requireNonNull(c);
1117         if (c == this)
1118             throw new IllegalArgumentException();
1119         int n = 0;
1120         for (E e; (e = poll()) != null; n++)
1121             c.add(e);
1122         return n;
1123     }
1124 
1125     /**
1126      * @throws UnsupportedOperationException {@inheritDoc}
1127      * @throws ClassCastException            {@inheritDoc}
1128      * @throws NullPointerException          {@inheritDoc}
1129      * @throws IllegalArgumentException      {@inheritDoc}
1130      */
1131     public int drainTo(Collection<? super E> c, int maxElements) {
1132         Objects.requireNonNull(c);
1133         if (c == this)
1134             throw new IllegalArgumentException();
1135         int n = 0;
1136         for (E e; n < maxElements && (e = poll()) != null; n++)
1137             c.add(e);
1138         return n;
1139     }
1140 
1141     /*
1142      * To cope with serialization strategy in the 1.5 version of
1143      * SynchronousQueue, we declare some unused classes and fields
1144      * that exist solely to enable serializability across versions.
1145      * These fields are never used, so are initialized only if this
1146      * object is ever serialized or deserialized.
1147      */
1148 
1149     @SuppressWarnings("serial")
1150     static class WaitQueue implements java.io.Serializable { }
1151     static class LifoWaitQueue extends WaitQueue {
1152         private static final long serialVersionUID = -3633113410248163686L;
1153     }
1154     static class FifoWaitQueue extends WaitQueue {
1155         private static final long serialVersionUID = -3623113410248163686L;
1156     }
1157     private ReentrantLock qlock;
1158     private WaitQueue waitingProducers;
1159     private WaitQueue waitingConsumers;
1160 
1161     /**
1162      * Saves this queue to a stream (that is, serializes it).
1163      * @param s the stream
1164      * @throws java.io.IOException if an I/O error occurs
1165      */
1166     private void writeObject(java.io.ObjectOutputStream s)
1167         throws java.io.IOException {
1168         boolean fair = transferer instanceof TransferQueue;
1169         if (fair) {
1170             qlock = new ReentrantLock(true);
1171             waitingProducers = new FifoWaitQueue();
1172             waitingConsumers = new FifoWaitQueue();
1173         }
1174         else {
1175             qlock = new ReentrantLock();
1176             waitingProducers = new LifoWaitQueue();
1177             waitingConsumers = new LifoWaitQueue();
1178         }
1179         s.defaultWriteObject();
1180     }
1181 
1182     /**
1183      * Reconstitutes this queue from a stream (that is, deserializes it).
1184      * @param s the stream
1185      * @throws ClassNotFoundException if the class of a serialized object
1186      *         could not be found
1187      * @throws java.io.IOException if an I/O error occurs
1188      */
1189     private void readObject(java.io.ObjectInputStream s)
1190         throws java.io.IOException, ClassNotFoundException {
1191         s.defaultReadObject();
1192         if (waitingProducers instanceof FifoWaitQueue)
1193             transferer = new TransferQueue<E>();
1194         else
1195             transferer = new TransferStack<E>();
1196     }
1197 
1198     static {
1199         // Reduce the risk of rare disastrous classloading in first call to
1200         // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
1201         Class<?> ensureLoaded = LockSupport.class;
1202     }
1203 }