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.  Sun designates this

     7  * particular file as subject to the "Classpath" exception as provided

     8  * by Sun 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 Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,

    21  * CA 95054 USA or visit www.sun.com if you need additional information or

    22  * have any 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 with assistance from members of JCP JSR-166

    32  * Expert Group and released to the public domain, as explained at

    33  * http://creativecommons.org/licenses/publicdomain

    34  */

    35 

    36 package java.util.concurrent.locks;

    37 import java.util.*;

    38 import java.util.concurrent.*;

    39 import java.util.concurrent.atomic.*;

    40 import sun.misc.Unsafe;

    41 

    42 /**

    43  * A version of {@link AbstractQueuedSynchronizer} in

    44  * which synchronization state is maintained as a <tt>long</tt>.

    45  * This class has exactly the same structure, properties, and methods

    46  * as <tt>AbstractQueuedSynchronizer</tt> with the exception

    47  * that all state-related parameters and results are defined

    48  * as <tt>long</tt> rather than <tt>int</tt>. This class

    49  * may be useful when creating synchronizers such as

    50  * multilevel locks and barriers that require

    51  * 64 bits of state.

    52  *

    53  * <p>See {@link AbstractQueuedSynchronizer} for usage

    54  * notes and examples.

    55  *

    56  * @since 1.6

    57  * @author Doug Lea

    58  */

    59 public abstract class AbstractQueuedLongSynchronizer

    60     extends AbstractOwnableSynchronizer

    61     implements java.io.Serializable {

    62 

    63     private static final long serialVersionUID = 7373984972572414692L;

    64 

    65     /*

    66       To keep sources in sync, the remainder of this source file is

    67       exactly cloned from AbstractQueuedSynchronizer, replacing class

    68       name and changing ints related with sync state to longs. Please

    69       keep it that way.

    70     */

    71 

    72     /**

    73      * Creates a new <tt>AbstractQueuedLongSynchronizer</tt> instance

    74      * with initial synchronization state of zero.

    75      */

    76     protected AbstractQueuedLongSynchronizer() { }

    77 

    78     /**

    79      * Wait queue node class.

    80      *

    81      * <p>The wait queue is a variant of a "CLH" (Craig, Landin, and

    82      * Hagersten) lock queue. CLH locks are normally used for

    83      * spinlocks.  We instead use them for blocking synchronizers, but

    84      * use the same basic tactic of holding some of the control

    85      * information about a thread in the predecessor of its node.  A

    86      * "status" field in each node keeps track of whether a thread

    87      * should block.  A node is signalled when its predecessor

    88      * releases.  Each node of the queue otherwise serves as a

    89      * specific-notification-style monitor holding a single waiting

    90      * thread. The status field does NOT control whether threads are

    91      * granted locks etc though.  A thread may try to acquire if it is

    92      * first in the queue. But being first does not guarantee success;

    93      * it only gives the right to contend.  So the currently released

    94      * contender thread may need to rewait.

    95      *

    96      * <p>To enqueue into a CLH lock, you atomically splice it in as new

    97      * tail. To dequeue, you just set the head field.

    98      * <pre>

    99      *      +------+  prev +-----+       +-----+

   100      * head |      | <---- |     | <---- |     |  tail

   101      *      +------+       +-----+       +-----+

   102      * </pre>

   103      *

   104      * <p>Insertion into a CLH queue requires only a single atomic

   105      * operation on "tail", so there is a simple atomic point of

   106      * demarcation from unqueued to queued. Similarly, dequeing

   107      * involves only updating the "head". However, it takes a bit

   108      * more work for nodes to determine who their successors are,

   109      * in part to deal with possible cancellation due to timeouts

   110      * and interrupts.

   111      *

   112      * <p>The "prev" links (not used in original CLH locks), are mainly

   113      * needed to handle cancellation. If a node is cancelled, its

   114      * successor is (normally) relinked to a non-cancelled

   115      * predecessor. For explanation of similar mechanics in the case

   116      * of spin locks, see the papers by Scott and Scherer at

   117      * http://www.cs.rochester.edu/u/scott/synchronization/

   118      *

   119      * <p>We also use "next" links to implement blocking mechanics.

   120      * The thread id for each node is kept in its own node, so a

   121      * predecessor signals the next node to wake up by traversing

   122      * next link to determine which thread it is.  Determination of

   123      * successor must avoid races with newly queued nodes to set

   124      * the "next" fields of their predecessors.  This is solved

   125      * when necessary by checking backwards from the atomically

   126      * updated "tail" when a node's successor appears to be null.

   127      * (Or, said differently, the next-links are an optimization

   128      * so that we don't usually need a backward scan.)

   129      *

   130      * <p>Cancellation introduces some conservatism to the basic

   131      * algorithms.  Since we must poll for cancellation of other

   132      * nodes, we can miss noticing whether a cancelled node is

   133      * ahead or behind us. This is dealt with by always unparking

   134      * successors upon cancellation, allowing them to stabilize on

   135      * a new predecessor, unless we can identify an uncancelled

   136      * predecessor who will carry this responsibility.

   137      *

   138      * <p>CLH queues need a dummy header node to get started. But

   139      * we don't create them on construction, because it would be wasted

   140      * effort if there is never contention. Instead, the node

   141      * is constructed and head and tail pointers are set upon first

   142      * contention.

   143      *

   144      * <p>Threads waiting on Conditions use the same nodes, but

   145      * use an additional link. Conditions only need to link nodes

   146      * in simple (non-concurrent) linked queues because they are

   147      * only accessed when exclusively held.  Upon await, a node is

   148      * inserted into a condition queue.  Upon signal, the node is

   149      * transferred to the main queue.  A special value of status

   150      * field is used to mark which queue a node is on.

   151      *

   152      * <p>Thanks go to Dave Dice, Mark Moir, Victor Luchangco, Bill

   153      * Scherer and Michael Scott, along with members of JSR-166

   154      * expert group, for helpful ideas, discussions, and critiques

   155      * on the design of this class.

   156      */

   157     static final class Node {

   158         /** Marker to indicate a node is waiting in shared mode */

   159         static final Node SHARED = new Node();

   160         /** Marker to indicate a node is waiting in exclusive mode */

   161         static final Node EXCLUSIVE = null;

   162 

   163         /** waitStatus value to indicate thread has cancelled */

   164         static final int CANCELLED =  1;

   165         /** waitStatus value to indicate successor's thread needs unparking */

   166         static final int SIGNAL    = -1;

   167         /** waitStatus value to indicate thread is waiting on condition */

   168         static final int CONDITION = -2;

   169 

   170         /**

   171          * Status field, taking on only the values:

   172          *   SIGNAL:     The successor of this node is (or will soon be)

   173          *               blocked (via park), so the current node must

   174          *               unpark its successor when it releases or

   175          *               cancels. To avoid races, acquire methods must

   176          *               first indicate they need a signal,

   177          *               then retry the atomic acquire, and then,

   178          *               on failure, block.

   179          *   CANCELLED:  This node is cancelled due to timeout or interrupt.

   180          *               Nodes never leave this state. In particular,

   181          *               a thread with cancelled node never again blocks.

   182          *   CONDITION:  This node is currently on a condition queue.

   183          *               It will not be used as a sync queue node until

   184          *               transferred. (Use of this value here

   185          *               has nothing to do with the other uses

   186          *               of the field, but simplifies mechanics.)

   187          *   0:          None of the above

   188          *

   189          * The values are arranged numerically to simplify use.

   190          * Non-negative values mean that a node doesn't need to

   191          * signal. So, most code doesn't need to check for particular

   192          * values, just for sign.

   193          *

   194          * The field is initialized to 0 for normal sync nodes, and

   195          * CONDITION for condition nodes.  It is modified using CAS

   196          * (or when possible, unconditional volatile writes).

   197          */

   198         volatile int waitStatus;

   199 

   200         /**

   201          * Link to predecessor node that current node/thread relies on

   202          * for checking waitStatus. Assigned during enqueing, and nulled

   203          * out (for sake of GC) only upon dequeuing.  Also, upon

   204          * cancellation of a predecessor, we short-circuit while

   205          * finding a non-cancelled one, which will always exist

   206          * because the head node is never cancelled: A node becomes

   207          * head only as a result of successful acquire. A

   208          * cancelled thread never succeeds in acquiring, and a thread only

   209          * cancels itself, not any other node.

   210          */

   211         volatile Node prev;

   212 

   213         /**

   214          * Link to the successor node that the current node/thread

   215          * unparks upon release. Assigned during enqueuing, adjusted

   216          * when bypassing cancelled predecessors, and nulled out (for

   217          * sake of GC) when dequeued.  The enq operation does not

   218          * assign next field of a predecessor until after attachment,

   219          * so seeing a null next field does not necessarily mean that

   220          * node is at end of queue. However, if a next field appears

   221          * to be null, we can scan prev's from the tail to

   222          * double-check.  The next field of cancelled nodes is set to

   223          * point to the node itself instead of null, to make life

   224          * easier for isOnSyncQueue.

   225          */

   226         volatile Node next;

   227 

   228         /**

   229          * The thread that enqueued this node.  Initialized on

   230          * construction and nulled out after use.

   231          */

   232         volatile Thread thread;

   233 

   234         /**

   235          * Link to next node waiting on condition, or the special

   236          * value SHARED.  Because condition queues are accessed only

   237          * when holding in exclusive mode, we just need a simple

   238          * linked queue to hold nodes while they are waiting on

   239          * conditions. They are then transferred to the queue to

   240          * re-acquire. And because conditions can only be exclusive,

   241          * we save a field by using special value to indicate shared

   242          * mode.

   243          */

   244         Node nextWaiter;

   245 

   246         /**

   247          * Returns true if node is waiting in shared mode

   248          */

   249         final boolean isShared() {

   250             return nextWaiter == SHARED;

   251         }

   252 

   253         /**

   254          * Returns previous node, or throws NullPointerException if null.

   255          * Use when predecessor cannot be null.  The null check could

   256          * be elided, but is present to help the VM.

   257          *

   258          * @return the predecessor of this node

   259          */

   260         final Node predecessor() throws NullPointerException {

   261             Node p = prev;

   262             if (p == null)

   263                 throw new NullPointerException();

   264             else

   265                 return p;

   266         }

   267 

   268         Node() {    // Used to establish initial head or SHARED marker

   269         }

   270 

   271         Node(Thread thread, Node mode) {     // Used by addWaiter

   272             this.nextWaiter = mode;

   273             this.thread = thread;

   274         }

   275 

   276         Node(Thread thread, int waitStatus) { // Used by Condition

   277             this.waitStatus = waitStatus;

   278             this.thread = thread;

   279         }

   280     }

   281 

   282     /**

   283      * Head of the wait queue, lazily initialized.  Except for

   284      * initialization, it is modified only via method setHead.  Note:

   285      * If head exists, its waitStatus is guaranteed not to be

   286      * CANCELLED.

   287      */

   288     private transient volatile Node head;

   289 

   290     /**

   291      * Tail of the wait queue, lazily initialized.  Modified only via

   292      * method enq to add new wait node.

   293      */

   294     private transient volatile Node tail;

   295 

   296     /**

   297      * The synchronization state.

   298      */

   299     private volatile long state;

   300 

   301     /**

   302      * Returns the current value of synchronization state.

   303      * This operation has memory semantics of a <tt>volatile</tt> read.

   304      * @return current state value

   305      */

   306     protected final long getState() {

   307         return state;

   308     }

   309 

   310     /**

   311      * Sets the value of synchronization state.

   312      * This operation has memory semantics of a <tt>volatile</tt> write.

   313      * @param newState the new state value

   314      */

   315     protected final void setState(long newState) {

   316         state = newState;

   317     }

   318 

   319     /**

   320      * Atomically sets synchronization state to the given updated

   321      * value if the current state value equals the expected value.

   322      * This operation has memory semantics of a <tt>volatile</tt> read

   323      * and write.

   324      *

   325      * @param expect the expected value

   326      * @param update the new value

   327      * @return true if successful. False return indicates that the actual

   328      *         value was not equal to the expected value.

   329      */

   330     protected final boolean compareAndSetState(long expect, long update) {

   331         // See below for intrinsics setup to support this

   332         return unsafe.compareAndSwapLong(this, stateOffset, expect, update);

   333     }

   334 

   335     // Queuing utilities

   336 

   337     /**

   338      * The number of nanoseconds for which it is faster to spin

   339      * rather than to use timed park. A rough estimate suffices

   340      * to improve responsiveness with very short timeouts.

   341      */

   342     static final long spinForTimeoutThreshold = 1000L;

   343 

   344     /**

   345      * Inserts node into queue, initializing if necessary. See picture above.

   346      * @param node the node to insert

   347      * @return node's predecessor

   348      */

   349     private Node enq(final Node node) {

   350         for (;;) {

   351             Node t = tail;

   352             if (t == null) { // Must initialize

   353                 if (compareAndSetHead(new Node()))

   354                     tail = head;

   355             } else {

   356                 node.prev = t;

   357                 if (compareAndSetTail(t, node)) {

   358                     t.next = node;

   359                     return t;

   360                 }

   361             }

   362         }

   363     }

   364 

   365     /**

   366      * Creates and enqueues node for current thread and given mode.

   367      *

   368      * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared

   369      * @return the new node

   370      */

   371     private Node addWaiter(Node mode) {

   372         Node node = new Node(Thread.currentThread(), mode);

   373         // Try the fast path of enq; backup to full enq on failure

   374         Node pred = tail;

   375         if (pred != null) {

   376             node.prev = pred;

   377             if (compareAndSetTail(pred, node)) {

   378                 pred.next = node;

   379                 return node;

   380             }

   381         }

   382         enq(node);

   383         return node;

   384     }

   385 

   386     /**

   387      * Sets head of queue to be node, thus dequeuing. Called only by

   388      * acquire methods.  Also nulls out unused fields for sake of GC

   389      * and to suppress unnecessary signals and traversals.

   390      *

   391      * @param node the node

   392      */

   393     private void setHead(Node node) {

   394         head = node;

   395         node.thread = null;

   396         node.prev = null;

   397     }

   398 

   399     /**

   400      * Wakes up node's successor, if one exists.

   401      *

   402      * @param node the node

   403      */

   404     private void unparkSuccessor(Node node) {

   405         /*

   406          * Try to clear status in anticipation of signalling.  It is

   407          * OK if this fails or if status is changed by waiting thread.

   408          */

   409         compareAndSetWaitStatus(node, Node.SIGNAL, 0);

   410 

   411         /*

   412          * Thread to unpark is held in successor, which is normally

   413          * just the next node.  But if cancelled or apparently null,

   414          * traverse backwards from tail to find the actual

   415          * non-cancelled successor.

   416          */

   417         Node s = node.next;

   418         if (s == null || s.waitStatus > 0) {

   419             s = null;

   420             for (Node t = tail; t != null && t != node; t = t.prev)

   421                 if (t.waitStatus <= 0)

   422                     s = t;

   423         }

   424         if (s != null)

   425             LockSupport.unpark(s.thread);

   426     }

   427 

   428     /**

   429      * Sets head of queue, and checks if successor may be waiting

   430      * in shared mode, if so propagating if propagate > 0.

   431      *

   432      * @param pred the node holding waitStatus for node

   433      * @param node the node

   434      * @param propagate the return value from a tryAcquireShared

   435      */

   436     private void setHeadAndPropagate(Node node, long propagate) {

   437         setHead(node);

   438         if (propagate > 0 && node.waitStatus != 0) {

   439             /*

   440              * Don't bother fully figuring out successor.  If it

   441              * looks null, call unparkSuccessor anyway to be safe.

   442              */

   443             Node s = node.next;

   444             if (s == null || s.isShared())

   445                 unparkSuccessor(node);

   446         }

   447     }

   448 

   449     // Utilities for various versions of acquire

   450 

   451     /**

   452      * Cancels an ongoing attempt to acquire.

   453      *

   454      * @param node the node

   455      */

   456     private void cancelAcquire(Node node) {

   457         // Ignore if node doesn't exist

   458         if (node == null)

   459             return;

   460 

   461         node.thread = null;

   462 

   463         // Skip cancelled predecessors

   464         Node pred = node.prev;

   465         while (pred.waitStatus > 0)

   466             node.prev = pred = pred.prev;

   467 

   468         // Getting this before setting waitStatus ensures staleness

   469         Node predNext = pred.next;

   470 

   471         // Can use unconditional write instead of CAS here

   472         node.waitStatus = Node.CANCELLED;

   473 

   474         // If we are the tail, remove ourselves

   475         if (node == tail && compareAndSetTail(node, pred)) {

   476             compareAndSetNext(pred, predNext, null);

   477         } else {

   478             // If "active" predecessor found...

   479             if (pred != head

   480                 && (pred.waitStatus == Node.SIGNAL

   481                     || compareAndSetWaitStatus(pred, 0, Node.SIGNAL))

   482                 && pred.thread != null) {

   483 

   484                 // If successor is active, set predecessor's next link

   485                 Node next = node.next;

   486                 if (next != null && next.waitStatus <= 0)

   487                     compareAndSetNext(pred, predNext, next);

   488             } else {

   489                 unparkSuccessor(node);

   490             }

   491 

   492             node.next = node; // help GC

   493         }

   494     }

   495 

   496     /**

   497      * Checks and updates status for a node that failed to acquire.

   498      * Returns true if thread should block. This is the main signal

   499      * control in all acquire loops.  Requires that pred == node.prev

   500      *

   501      * @param pred node's predecessor holding status

   502      * @param node the node

   503      * @return {@code true} if thread should block

   504      */

   505     private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {

   506         int s = pred.waitStatus;

   507         if (s < 0)

   508             /*

   509              * This node has already set status asking a release

   510              * to signal it, so it can safely park.

   511              */

   512             return true;

   513         if (s > 0) {

   514             /*

   515              * Predecessor was cancelled. Skip over predecessors and

   516              * indicate retry.

   517              */

   518             do {

   519                 node.prev = pred = pred.prev;

   520             } while (pred.waitStatus > 0);

   521             pred.next = node;

   522         }

   523         else

   524             /*

   525              * Indicate that we need a signal, but don't park yet. Caller

   526              * will need to retry to make sure it cannot acquire before

   527              * parking.

   528              */

   529             compareAndSetWaitStatus(pred, 0, Node.SIGNAL);

   530         return false;

   531     }

   532 

   533     /**

   534      * Convenience method to interrupt current thread.

   535      */

   536     private static void selfInterrupt() {

   537         Thread.currentThread().interrupt();

   538     }

   539 

   540     /**

   541      * Convenience method to park and then check if interrupted

   542      *

   543      * @return {@code true} if interrupted

   544      */

   545     private final boolean parkAndCheckInterrupt() {

   546         LockSupport.park(this);

   547         return Thread.interrupted();

   548     }

   549 

   550     /*

   551      * Various flavors of acquire, varying in exclusive/shared and

   552      * control modes.  Each is mostly the same, but annoyingly

   553      * different.  Only a little bit of factoring is possible due to

   554      * interactions of exception mechanics (including ensuring that we

   555      * cancel if tryAcquire throws exception) and other control, at

   556      * least not without hurting performance too much.

   557      */

   558 

   559     /**

   560      * Acquires in exclusive uninterruptible mode for thread already in

   561      * queue. Used by condition wait methods as well as acquire.

   562      *

   563      * @param node the node

   564      * @param arg the acquire argument

   565      * @return {@code true} if interrupted while waiting

   566      */

   567     final boolean acquireQueued(final Node node, long arg) {

   568         boolean failed = true;

   569         try {

   570             boolean interrupted = false;

   571             for (;;) {

   572                 final Node p = node.predecessor();

   573                 if (p == head && tryAcquire(arg)) {

   574                     setHead(node);

   575                     p.next = null; // help GC

   576                     failed = false;

   577                     return interrupted;

   578                 }

   579                 if (shouldParkAfterFailedAcquire(p, node) &&

   580                     parkAndCheckInterrupt())

   581                     interrupted = true;

   582             }

   583         } finally {

   584             if (failed)

   585                 cancelAcquire(node);

   586         }

   587     }

   588 

   589     /**

   590      * Acquires in exclusive interruptible mode.

   591      * @param arg the acquire argument

   592      */

   593     private void doAcquireInterruptibly(long arg)

   594         throws InterruptedException {

   595         final Node node = addWaiter(Node.EXCLUSIVE);

   596         boolean failed = true;

   597         try {

   598             for (;;) {

   599                 final Node p = node.predecessor();

   600                 if (p == head && tryAcquire(arg)) {

   601                     setHead(node);

   602                     p.next = null; // help GC

   603                     failed = false;

   604                     return;

   605                 }

   606                 if (shouldParkAfterFailedAcquire(p, node) &&

   607                     parkAndCheckInterrupt())

   608                     throw new InterruptedException();

   609             }

   610         } finally {

   611             if (failed)

   612                 cancelAcquire(node);

   613         }

   614     }

   615 

   616     /**

   617      * Acquires in exclusive timed mode.

   618      *

   619      * @param arg the acquire argument

   620      * @param nanosTimeout max wait time

   621      * @return {@code true} if acquired

   622      */

   623     private boolean doAcquireNanos(long arg, long nanosTimeout)

   624         throws InterruptedException {

   625         long lastTime = System.nanoTime();

   626         final Node node = addWaiter(Node.EXCLUSIVE);

   627         boolean failed = true;

   628         try {

   629             for (;;) {

   630                 final Node p = node.predecessor();

   631                 if (p == head && tryAcquire(arg)) {

   632                     setHead(node);

   633                     p.next = null; // help GC

   634                     failed = false;

   635                     return true;

   636                 }

   637                 if (nanosTimeout <= 0)

   638                     return false;

   639                 if (shouldParkAfterFailedAcquire(p, node) &&

   640                     nanosTimeout > spinForTimeoutThreshold)

   641                     LockSupport.parkNanos(this, nanosTimeout);

   642                 long now = System.nanoTime();

   643                 nanosTimeout -= now - lastTime;

   644                 lastTime = now;

   645                 if (Thread.interrupted())

   646                     throw new InterruptedException();

   647             }

   648         } finally {

   649             if (failed)

   650                 cancelAcquire(node);

   651         }

   652     }

   653 

   654     /**

   655      * Acquires in shared uninterruptible mode.

   656      * @param arg the acquire argument

   657      */

   658     private void doAcquireShared(long arg) {

   659         final Node node = addWaiter(Node.SHARED);

   660         boolean failed = true;

   661         try {

   662             boolean interrupted = false;

   663             for (;;) {

   664                 final Node p = node.predecessor();

   665                 if (p == head) {

   666                     long r = tryAcquireShared(arg);

   667                     if (r >= 0) {

   668                         setHeadAndPropagate(node, r);

   669                         p.next = null; // help GC

   670                         if (interrupted)

   671                             selfInterrupt();

   672                         failed = false;

   673                         return;

   674                     }

   675                 }

   676                 if (shouldParkAfterFailedAcquire(p, node) &&

   677                     parkAndCheckInterrupt())

   678                     interrupted = true;

   679             }

   680         } finally {

   681             if (failed)

   682                 cancelAcquire(node);

   683         }

   684     }

   685 

   686     /**

   687      * Acquires in shared interruptible mode.

   688      * @param arg the acquire argument

   689      */

   690     private void doAcquireSharedInterruptibly(long arg)

   691         throws InterruptedException {

   692         final Node node = addWaiter(Node.SHARED);

   693         boolean failed = true;

   694         try {

   695             for (;;) {

   696                 final Node p = node.predecessor();

   697                 if (p == head) {

   698                     long r = tryAcquireShared(arg);

   699                     if (r >= 0) {

   700                         setHeadAndPropagate(node, r);

   701                         p.next = null; // help GC

   702                         failed = false;

   703                         return;

   704                     }

   705                 }

   706                 if (shouldParkAfterFailedAcquire(p, node) &&

   707                     parkAndCheckInterrupt())

   708                     throw new InterruptedException();

   709             }

   710         } finally {

   711             if (failed)

   712                 cancelAcquire(node);

   713         }

   714     }

   715 

   716     /**

   717      * Acquires in shared timed mode.

   718      *

   719      * @param arg the acquire argument

   720      * @param nanosTimeout max wait time

   721      * @return {@code true} if acquired

   722      */

   723     private boolean doAcquireSharedNanos(long arg, long nanosTimeout)

   724         throws InterruptedException {

   725 

   726         long lastTime = System.nanoTime();

   727         final Node node = addWaiter(Node.SHARED);

   728         boolean failed = true;

   729         try {

   730             for (;;) {

   731                 final Node p = node.predecessor();

   732                 if (p == head) {

   733                     long r = tryAcquireShared(arg);

   734                     if (r >= 0) {

   735                         setHeadAndPropagate(node, r);

   736                         p.next = null; // help GC

   737                         failed = false;

   738                         return true;

   739                     }

   740                 }

   741                 if (nanosTimeout <= 0)

   742                     return false;

   743                 if (shouldParkAfterFailedAcquire(p, node) &&

   744                     nanosTimeout > spinForTimeoutThreshold)

   745                     LockSupport.parkNanos(this, nanosTimeout);

   746                 long now = System.nanoTime();

   747                 nanosTimeout -= now - lastTime;

   748                 lastTime = now;

   749                 if (Thread.interrupted())

   750                     throw new InterruptedException();

   751             }

   752         } finally {

   753             if (failed)

   754                 cancelAcquire(node);

   755         }

   756     }

   757 

   758     // Main exported methods

   759 

   760     /**

   761      * Attempts to acquire in exclusive mode. This method should query

   762      * if the state of the object permits it to be acquired in the

   763      * exclusive mode, and if so to acquire it.

   764      *

   765      * <p>This method is always invoked by the thread performing

   766      * acquire.  If this method reports failure, the acquire method

   767      * may queue the thread, if it is not already queued, until it is

   768      * signalled by a release from some other thread. This can be used

   769      * to implement method {@link Lock#tryLock()}.

   770      *

   771      * <p>The default

   772      * implementation throws {@link UnsupportedOperationException}.

   773      *

   774      * @param arg the acquire argument. This value is always the one

   775      *        passed to an acquire method, or is the value saved on entry

   776      *        to a condition wait.  The value is otherwise uninterpreted

   777      *        and can represent anything you like.

   778      * @return {@code true} if successful. Upon success, this object has

   779      *         been acquired.

   780      * @throws IllegalMonitorStateException if acquiring would place this

   781      *         synchronizer in an illegal state. This exception must be

   782      *         thrown in a consistent fashion for synchronization to work

   783      *         correctly.

   784      * @throws UnsupportedOperationException if exclusive mode is not supported

   785      */

   786     protected boolean tryAcquire(long arg) {

   787         throw new UnsupportedOperationException();

   788     }

   789 

   790     /**

   791      * Attempts to set the state to reflect a release in exclusive

   792      * mode.

   793      *

   794      * <p>This method is always invoked by the thread performing release.

   795      *

   796      * <p>The default implementation throws

   797      * {@link UnsupportedOperationException}.

   798      *

   799      * @param arg the release argument. This value is always the one

   800      *        passed to a release method, or the current state value upon

   801      *        entry to a condition wait.  The value is otherwise

   802      *        uninterpreted and can represent anything you like.

   803      * @return {@code true} if this object is now in a fully released

   804      *         state, so that any waiting threads may attempt to acquire;

   805      *         and {@code false} otherwise.

   806      * @throws IllegalMonitorStateException if releasing would place this

   807      *         synchronizer in an illegal state. This exception must be

   808      *         thrown in a consistent fashion for synchronization to work

   809      *         correctly.

   810      * @throws UnsupportedOperationException if exclusive mode is not supported

   811      */

   812     protected boolean tryRelease(long arg) {

   813         throw new UnsupportedOperationException();

   814     }

   815 

   816     /**

   817      * Attempts to acquire in shared mode. This method should query if

   818      * the state of the object permits it to be acquired in the shared

   819      * mode, and if so to acquire it.

   820      *

   821      * <p>This method is always invoked by the thread performing

   822      * acquire.  If this method reports failure, the acquire method

   823      * may queue the thread, if it is not already queued, until it is

   824      * signalled by a release from some other thread.

   825      *

   826      * <p>The default implementation throws {@link

   827      * UnsupportedOperationException}.

   828      *

   829      * @param arg the acquire argument. This value is always the one

   830      *        passed to an acquire method, or is the value saved on entry

   831      *        to a condition wait.  The value is otherwise uninterpreted

   832      *        and can represent anything you like.

   833      * @return a negative value on failure; zero if acquisition in shared

   834      *         mode succeeded but no subsequent shared-mode acquire can

   835      *         succeed; and a positive value if acquisition in shared

   836      *         mode succeeded and subsequent shared-mode acquires might

   837      *         also succeed, in which case a subsequent waiting thread

   838      *         must check availability. (Support for three different

   839      *         return values enables this method to be used in contexts

   840      *         where acquires only sometimes act exclusively.)  Upon

   841      *         success, this object has been acquired.

   842      * @throws IllegalMonitorStateException if acquiring would place this

   843      *         synchronizer in an illegal state. This exception must be

   844      *         thrown in a consistent fashion for synchronization to work

   845      *         correctly.

   846      * @throws UnsupportedOperationException if shared mode is not supported

   847      */

   848     protected long tryAcquireShared(long arg) {

   849         throw new UnsupportedOperationException();

   850     }

   851 

   852     /**

   853      * Attempts to set the state to reflect a release in shared mode.

   854      *

   855      * <p>This method is always invoked by the thread performing release.

   856      *

   857      * <p>The default implementation throws

   858      * {@link UnsupportedOperationException}.

   859      *

   860      * @param arg the release argument. This value is always the one

   861      *        passed to a release method, or the current state value upon

   862      *        entry to a condition wait.  The value is otherwise

   863      *        uninterpreted and can represent anything you like.

   864      * @return {@code true} if this release of shared mode may permit a

   865      *         waiting acquire (shared or exclusive) to succeed; and

   866      *         {@code false} otherwise

   867      * @throws IllegalMonitorStateException if releasing would place this

   868      *         synchronizer in an illegal state. This exception must be

   869      *         thrown in a consistent fashion for synchronization to work

   870      *         correctly.

   871      * @throws UnsupportedOperationException if shared mode is not supported

   872      */

   873     protected boolean tryReleaseShared(long arg) {

   874         throw new UnsupportedOperationException();

   875     }

   876 

   877     /**

   878      * Returns {@code true} if synchronization is held exclusively with

   879      * respect to the current (calling) thread.  This method is invoked

   880      * upon each call to a non-waiting {@link ConditionObject} method.

   881      * (Waiting methods instead invoke {@link #release}.)

   882      *

   883      * <p>The default implementation throws {@link

   884      * UnsupportedOperationException}. This method is invoked

   885      * internally only within {@link ConditionObject} methods, so need

   886      * not be defined if conditions are not used.

   887      *

   888      * @return {@code true} if synchronization is held exclusively;

   889      *         {@code false} otherwise

   890      * @throws UnsupportedOperationException if conditions are not supported

   891      */

   892     protected boolean isHeldExclusively() {

   893         throw new UnsupportedOperationException();

   894     }

   895 

   896     /**

   897      * Acquires in exclusive mode, ignoring interrupts.  Implemented

   898      * by invoking at least once {@link #tryAcquire},

   899      * returning on success.  Otherwise the thread is queued, possibly

   900      * repeatedly blocking and unblocking, invoking {@link

   901      * #tryAcquire} until success.  This method can be used

   902      * to implement method {@link Lock#lock}.

   903      *

   904      * @param arg the acquire argument.  This value is conveyed to

   905      *        {@link #tryAcquire} but is otherwise uninterpreted and

   906      *        can represent anything you like.

   907      */

   908     public final void acquire(long arg) {

   909         if (!tryAcquire(arg) &&

   910             acquireQueued(addWaiter(Node.EXCLUSIVE), arg))

   911             selfInterrupt();

   912     }

   913 

   914     /**

   915      * Acquires in exclusive mode, aborting if interrupted.

   916      * Implemented by first checking interrupt status, then invoking

   917      * at least once {@link #tryAcquire}, returning on

   918      * success.  Otherwise the thread is queued, possibly repeatedly

   919      * blocking and unblocking, invoking {@link #tryAcquire}

   920      * until success or the thread is interrupted.  This method can be

   921      * used to implement method {@link Lock#lockInterruptibly}.

   922      *

   923      * @param arg the acquire argument.  This value is conveyed to

   924      *        {@link #tryAcquire} but is otherwise uninterpreted and

   925      *        can represent anything you like.

   926      * @throws InterruptedException if the current thread is interrupted

   927      */

   928     public final void acquireInterruptibly(long arg) throws InterruptedException {

   929         if (Thread.interrupted())

   930             throw new InterruptedException();

   931         if (!tryAcquire(arg))

   932             doAcquireInterruptibly(arg);

   933     }

   934 

   935     /**

   936      * Attempts to acquire in exclusive mode, aborting if interrupted,

   937      * and failing if the given timeout elapses.  Implemented by first

   938      * checking interrupt status, then invoking at least once {@link

   939      * #tryAcquire}, returning on success.  Otherwise, the thread is

   940      * queued, possibly repeatedly blocking and unblocking, invoking

   941      * {@link #tryAcquire} until success or the thread is interrupted

   942      * or the timeout elapses.  This method can be used to implement

   943      * method {@link Lock#tryLock(long, TimeUnit)}.

   944      *

   945      * @param arg the acquire argument.  This value is conveyed to

   946      *        {@link #tryAcquire} but is otherwise uninterpreted and

   947      *        can represent anything you like.

   948      * @param nanosTimeout the maximum number of nanoseconds to wait

   949      * @return {@code true} if acquired; {@code false} if timed out

   950      * @throws InterruptedException if the current thread is interrupted

   951      */

   952     public final boolean tryAcquireNanos(long arg, long nanosTimeout) throws InterruptedException {

   953         if (Thread.interrupted())

   954             throw new InterruptedException();

   955         return tryAcquire(arg) ||

   956             doAcquireNanos(arg, nanosTimeout);

   957     }

   958 

   959     /**

   960      * Releases in exclusive mode.  Implemented by unblocking one or

   961      * more threads if {@link #tryRelease} returns true.

   962      * This method can be used to implement method {@link Lock#unlock}.

   963      *

   964      * @param arg the release argument.  This value is conveyed to

   965      *        {@link #tryRelease} but is otherwise uninterpreted and

   966      *        can represent anything you like.

   967      * @return the value returned from {@link #tryRelease}

   968      */

   969     public final boolean release(long arg) {

   970         if (tryRelease(arg)) {

   971             Node h = head;

   972             if (h != null && h.waitStatus != 0)

   973                 unparkSuccessor(h);

   974             return true;

   975         }

   976         return false;

   977     }

   978 

   979     /**

   980      * Acquires in shared mode, ignoring interrupts.  Implemented by

   981      * first invoking at least once {@link #tryAcquireShared},

   982      * returning on success.  Otherwise the thread is queued, possibly

   983      * repeatedly blocking and unblocking, invoking {@link

   984      * #tryAcquireShared} until success.

   985      *

   986      * @param arg the acquire argument.  This value is conveyed to

   987      *        {@link #tryAcquireShared} but is otherwise uninterpreted

   988      *        and can represent anything you like.

   989      */

   990     public final void acquireShared(long arg) {

   991         if (tryAcquireShared(arg) < 0)

   992             doAcquireShared(arg);

   993     }

   994 

   995     /**

   996      * Acquires in shared mode, aborting if interrupted.  Implemented

   997      * by first checking interrupt status, then invoking at least once

   998      * {@link #tryAcquireShared}, returning on success.  Otherwise the

   999      * thread is queued, possibly repeatedly blocking and unblocking,

  1000      * invoking {@link #tryAcquireShared} until success or the thread

  1001      * is interrupted.

  1002      * @param arg the acquire argument

  1003      * This value is conveyed to {@link #tryAcquireShared} but is

  1004      * otherwise uninterpreted and can represent anything

  1005      * you like.

  1006      * @throws InterruptedException if the current thread is interrupted

  1007      */

  1008     public final void acquireSharedInterruptibly(long arg) throws InterruptedException {

  1009         if (Thread.interrupted())

  1010             throw new InterruptedException();

  1011         if (tryAcquireShared(arg) < 0)

  1012             doAcquireSharedInterruptibly(arg);

  1013     }

  1014 

  1015     /**

  1016      * Attempts to acquire in shared mode, aborting if interrupted, and

  1017      * failing if the given timeout elapses.  Implemented by first

  1018      * checking interrupt status, then invoking at least once {@link

  1019      * #tryAcquireShared}, returning on success.  Otherwise, the

  1020      * thread is queued, possibly repeatedly blocking and unblocking,

  1021      * invoking {@link #tryAcquireShared} until success or the thread

  1022      * is interrupted or the timeout elapses.

  1023      *

  1024      * @param arg the acquire argument.  This value is conveyed to

  1025      *        {@link #tryAcquireShared} but is otherwise uninterpreted

  1026      *        and can represent anything you like.

  1027      * @param nanosTimeout the maximum number of nanoseconds to wait

  1028      * @return {@code true} if acquired; {@code false} if timed out

  1029      * @throws InterruptedException if the current thread is interrupted

  1030      */

  1031     public final boolean tryAcquireSharedNanos(long arg, long nanosTimeout) throws InterruptedException {

  1032         if (Thread.interrupted())

  1033             throw new InterruptedException();

  1034         return tryAcquireShared(arg) >= 0 ||

  1035             doAcquireSharedNanos(arg, nanosTimeout);

  1036     }

  1037 

  1038     /**

  1039      * Releases in shared mode.  Implemented by unblocking one or more

  1040      * threads if {@link #tryReleaseShared} returns true.

  1041      *

  1042      * @param arg the release argument.  This value is conveyed to

  1043      *        {@link #tryReleaseShared} but is otherwise uninterpreted

  1044      *        and can represent anything you like.

  1045      * @return the value returned from {@link #tryReleaseShared}

  1046      */

  1047     public final boolean releaseShared(long arg) {

  1048         if (tryReleaseShared(arg)) {

  1049             Node h = head;

  1050             if (h != null && h.waitStatus != 0)

  1051                 unparkSuccessor(h);

  1052             return true;

  1053         }

  1054         return false;

  1055     }

  1056 

  1057     // Queue inspection methods

  1058 

  1059     /**

  1060      * Queries whether any threads are waiting to acquire. Note that

  1061      * because cancellations due to interrupts and timeouts may occur

  1062      * at any time, a {@code true} return does not guarantee that any

  1063      * other thread will ever acquire.

  1064      *

  1065      * <p>In this implementation, this operation returns in

  1066      * constant time.

  1067      *

  1068      * @return {@code true} if there may be other threads waiting to acquire

  1069      */

  1070     public final boolean hasQueuedThreads() {

  1071         return head != tail;

  1072     }

  1073 

  1074     /**

  1075      * Queries whether any threads have ever contended to acquire this

  1076      * synchronizer; that is if an acquire method has ever blocked.

  1077      *

  1078      * <p>In this implementation, this operation returns in

  1079      * constant time.

  1080      *

  1081      * @return {@code true} if there has ever been contention

  1082      */

  1083     public final boolean hasContended() {

  1084         return head != null;

  1085     }

  1086 

  1087     /**

  1088      * Returns the first (longest-waiting) thread in the queue, or

  1089      * {@code null} if no threads are currently queued.

  1090      *

  1091      * <p>In this implementation, this operation normally returns in

  1092      * constant time, but may iterate upon contention if other threads are

  1093      * concurrently modifying the queue.

  1094      *

  1095      * @return the first (longest-waiting) thread in the queue, or

  1096      *         {@code null} if no threads are currently queued

  1097      */

  1098     public final Thread getFirstQueuedThread() {

  1099         // handle only fast path, else relay

  1100         return (head == tail) ? null : fullGetFirstQueuedThread();

  1101     }

  1102 

  1103     /**

  1104      * Version of getFirstQueuedThread called when fastpath fails

  1105      */

  1106     private Thread fullGetFirstQueuedThread() {

  1107         /*

  1108          * The first node is normally head.next. Try to get its

  1109          * thread field, ensuring consistent reads: If thread

  1110          * field is nulled out or s.prev is no longer head, then

  1111          * some other thread(s) concurrently performed setHead in

  1112          * between some of our reads. We try this twice before

  1113          * resorting to traversal.

  1114          */

  1115         Node h, s;

  1116         Thread st;

  1117         if (((h = head) != null && (s = h.next) != null &&

  1118              s.prev == head && (st = s.thread) != null) ||

  1119             ((h = head) != null && (s = h.next) != null &&

  1120              s.prev == head && (st = s.thread) != null))

  1121             return st;

  1122 

  1123         /*

  1124          * Head's next field might not have been set yet, or may have

  1125          * been unset after setHead. So we must check to see if tail

  1126          * is actually first node. If not, we continue on, safely

  1127          * traversing from tail back to head to find first,

  1128          * guaranteeing termination.

  1129          */

  1130 

  1131         Node t = tail;

  1132         Thread firstThread = null;

  1133         while (t != null && t != head) {

  1134             Thread tt = t.thread;

  1135             if (tt != null)