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 with assistance from members of JCP JSR-166 32 * Expert Group and released to the public domain, as explained at 33 * http://creativecommons.org/publicdomain/zero/1.0/ 34 */ 35 36 package java.util.concurrent.locks; 37 38 import java.util.ArrayList; 39 import java.util.Collection; 40 import java.util.Date; 41 import java.util.concurrent.TimeUnit; 42 43 /** 44 * Provides a framework for implementing blocking locks and related 45 * synchronizers (semaphores, events, etc) that rely on 46 * first-in-first-out (FIFO) wait queues. This class is designed to 47 * be a useful basis for most kinds of synchronizers that rely on a 48 * single atomic {@code int} value to represent state. Subclasses 49 * must define the protected methods that change this state, and which 50 * define what that state means in terms of this object being acquired 51 * or released. Given these, the other methods in this class carry 52 * out all queuing and blocking mechanics. Subclasses can maintain 53 * other state fields, but only the atomically updated {@code int} 54 * value manipulated using methods {@link #getState}, {@link 55 * #setState} and {@link #compareAndSetState} is tracked with respect 56 * to synchronization. 57 * 58 * <p>Subclasses should be defined as non-public internal helper 59 * classes that are used to implement the synchronization properties 60 * of their enclosing class. Class 61 * {@code AbstractQueuedSynchronizer} does not implement any 62 * synchronization interface. Instead it defines methods such as 63 * {@link #acquireInterruptibly} that can be invoked as 64 * appropriate by concrete locks and related synchronizers to 65 * implement their public methods. 66 * 67 * <p>This class supports either or both a default <em>exclusive</em> 68 * mode and a <em>shared</em> mode. When acquired in exclusive mode, 69 * attempted acquires by other threads cannot succeed. Shared mode 70 * acquires by multiple threads may (but need not) succeed. This class 71 * does not "understand" these differences except in the 72 * mechanical sense that when a shared mode acquire succeeds, the next 73 * waiting thread (if one exists) must also determine whether it can 74 * acquire as well. Threads waiting in the different modes share the 75 * same FIFO queue. Usually, implementation subclasses support only 76 * one of these modes, but both can come into play for example in a 77 * {@link ReadWriteLock}. Subclasses that support only exclusive or 78 * only shared modes need not define the methods supporting the unused mode. 79 * 80 * <p>This class defines a nested {@link ConditionObject} class that 81 * can be used as a {@link Condition} implementation by subclasses 82 * supporting exclusive mode for which method {@link 83 * #isHeldExclusively} reports whether synchronization is exclusively 84 * held with respect to the current thread, method {@link #release} 85 * invoked with the current {@link #getState} value fully releases 86 * this object, and {@link #acquire}, given this saved state value, 87 * eventually restores this object to its previous acquired state. No 88 * {@code AbstractQueuedSynchronizer} method otherwise creates such a 89 * condition, so if this constraint cannot be met, do not use it. The 90 * behavior of {@link ConditionObject} depends of course on the 91 * semantics of its synchronizer implementation. 92 * 93 * <p>This class provides inspection, instrumentation, and monitoring 94 * methods for the internal queue, as well as similar methods for 95 * condition objects. These can be exported as desired into classes 96 * using an {@code AbstractQueuedSynchronizer} for their 97 * synchronization mechanics. 98 * 99 * <p>Serialization of this class stores only the underlying atomic 100 * integer maintaining state, so deserialized objects have empty 101 * thread queues. Typical subclasses requiring serializability will 102 * define a {@code readObject} method that restores this to a known 103 * initial state upon deserialization. 104 * 105 * <h3>Usage</h3> 106 * 107 * <p>To use this class as the basis of a synchronizer, redefine the 108 * following methods, as applicable, by inspecting and/or modifying 109 * the synchronization state using {@link #getState}, {@link 110 * #setState} and/or {@link #compareAndSetState}: 111 * 112 * <ul> 113 * <li>{@link #tryAcquire} 114 * <li>{@link #tryRelease} 115 * <li>{@link #tryAcquireShared} 116 * <li>{@link #tryReleaseShared} 117 * <li>{@link #isHeldExclusively} 118 * </ul> 119 * 120 * Each of these methods by default throws {@link 121 * UnsupportedOperationException}. Implementations of these methods 122 * must be internally thread-safe, and should in general be short and 123 * not block. Defining these methods is the <em>only</em> supported 124 * means of using this class. All other methods are declared 125 * {@code final} because they cannot be independently varied. 126 * 127 * <p>You may also find the inherited methods from {@link 128 * AbstractOwnableSynchronizer} useful to keep track of the thread 129 * owning an exclusive synchronizer. You are encouraged to use them 130 * -- this enables monitoring and diagnostic tools to assist users in 131 * determining which threads hold locks. 132 * 133 * <p>Even though this class is based on an internal FIFO queue, it 134 * does not automatically enforce FIFO acquisition policies. The core 135 * of exclusive synchronization takes the form: 136 * 137 * <pre> 138 * Acquire: 139 * while (!tryAcquire(arg)) { 140 * <em>enqueue thread if it is not already queued</em>; 141 * <em>possibly block current thread</em>; 142 * } 143 * 144 * Release: 145 * if (tryRelease(arg)) 146 * <em>unblock the first queued thread</em>; 147 * </pre> 148 * 149 * (Shared mode is similar but may involve cascading signals.) 150 * 151 * <p id="barging">Because checks in acquire are invoked before 152 * enqueuing, a newly acquiring thread may <em>barge</em> ahead of 153 * others that are blocked and queued. However, you can, if desired, 154 * define {@code tryAcquire} and/or {@code tryAcquireShared} to 155 * disable barging by internally invoking one or more of the inspection 156 * methods, thereby providing a <em>fair</em> FIFO acquisition order. 157 * In particular, most fair synchronizers can define {@code tryAcquire} 158 * to return {@code false} if {@link #hasQueuedPredecessors} (a method 159 * specifically designed to be used by fair synchronizers) returns 160 * {@code true}. Other variations are possible. 161 * 162 * <p>Throughput and scalability are generally highest for the 163 * default barging (also known as <em>greedy</em>, 164 * <em>renouncement</em>, and <em>convoy-avoidance</em>) strategy. 165 * While this is not guaranteed to be fair or starvation-free, earlier 166 * queued threads are allowed to recontend before later queued 167 * threads, and each recontention has an unbiased chance to succeed 168 * against incoming threads. Also, while acquires do not 169 * "spin" in the usual sense, they may perform multiple 170 * invocations of {@code tryAcquire} interspersed with other 171 * computations before blocking. This gives most of the benefits of 172 * spins when exclusive synchronization is only briefly held, without 173 * most of the liabilities when it isn't. If so desired, you can 174 * augment this by preceding calls to acquire methods with 175 * "fast-path" checks, possibly prechecking {@link #hasContended} 176 * and/or {@link #hasQueuedThreads} to only do so if the synchronizer 177 * is likely not to be contended. 178 * 179 * <p>This class provides an efficient and scalable basis for 180 * synchronization in part by specializing its range of use to 181 * synchronizers that can rely on {@code int} state, acquire, and 182 * release parameters, and an internal FIFO wait queue. When this does 183 * not suffice, you can build synchronizers from a lower level using 184 * {@link java.util.concurrent.atomic atomic} classes, your own custom 185 * {@link java.util.Queue} classes, and {@link LockSupport} blocking 186 * support. 187 * 188 * <h3>Usage Examples</h3> 189 * 190 * <p>Here is a non-reentrant mutual exclusion lock class that uses 191 * the value zero to represent the unlocked state, and one to 192 * represent the locked state. While a non-reentrant lock 193 * does not strictly require recording of the current owner 194 * thread, this class does so anyway to make usage easier to monitor. 195 * It also supports conditions and exposes 196 * one of the instrumentation methods: 197 * 198 * <pre> {@code 199 * class Mutex implements Lock, java.io.Serializable { 200 * 201 * // Our internal helper class 202 * private static class Sync extends AbstractQueuedSynchronizer { 203 * // Reports whether in locked state 204 * protected boolean isHeldExclusively() { 205 * return getState() == 1; 206 * } 207 * 208 * // Acquires the lock if state is zero 209 * public boolean tryAcquire(int acquires) { 210 * assert acquires == 1; // Otherwise unused 211 * if (compareAndSetState(0, 1)) { 212 * setExclusiveOwnerThread(Thread.currentThread()); 213 * return true; 214 * } 215 * return false; 216 * } 217 * 218 * // Releases the lock by setting state to zero 219 * protected boolean tryRelease(int releases) { 220 * assert releases == 1; // Otherwise unused 221 * if (getState() == 0) throw new IllegalMonitorStateException(); 222 * setExclusiveOwnerThread(null); 223 * setState(0); 224 * return true; 225 * } 226 * 227 * // Provides a Condition 228 * Condition newCondition() { return new ConditionObject(); } 229 * 230 * // Deserializes properly 231 * private void readObject(ObjectInputStream s) 232 * throws IOException, ClassNotFoundException { 233 * s.defaultReadObject(); 234 * setState(0); // reset to unlocked state 235 * } 236 * } 237 * 238 * // The sync object does all the hard work. We just forward to it. 239 * private final Sync sync = new Sync(); 240 * 241 * public void lock() { sync.acquire(1); } 242 * public boolean tryLock() { return sync.tryAcquire(1); } 243 * public void unlock() { sync.release(1); } 244 * public Condition newCondition() { return sync.newCondition(); } 245 * public boolean isLocked() { return sync.isHeldExclusively(); } 246 * public boolean hasQueuedThreads() { return sync.hasQueuedThreads(); } 247 * public void lockInterruptibly() throws InterruptedException { 248 * sync.acquireInterruptibly(1); 249 * } 250 * public boolean tryLock(long timeout, TimeUnit unit) 251 * throws InterruptedException { 252 * return sync.tryAcquireNanos(1, unit.toNanos(timeout)); 253 * } 254 * }}</pre> 255 * 256 * <p>Here is a latch class that is like a 257 * {@link java.util.concurrent.CountDownLatch CountDownLatch} 258 * except that it only requires a single {@code signal} to 259 * fire. Because a latch is non-exclusive, it uses the {@code shared} 260 * acquire and release methods. 261 * 262 * <pre> {@code 263 * class BooleanLatch { 264 * 265 * private static class Sync extends AbstractQueuedSynchronizer { 266 * boolean isSignalled() { return getState() != 0; } 267 * 268 * protected int tryAcquireShared(int ignore) { 269 * return isSignalled() ? 1 : -1; 270 * } 271 * 272 * protected boolean tryReleaseShared(int ignore) { 273 * setState(1); 274 * return true; 275 * } 276 * } 277 * 278 * private final Sync sync = new Sync(); 279 * public boolean isSignalled() { return sync.isSignalled(); } 280 * public void signal() { sync.releaseShared(1); } 281 * public void await() throws InterruptedException { 282 * sync.acquireSharedInterruptibly(1); 283 * } 284 * }}</pre> 285 * 286 * @since 1.5 287 * @author Doug Lea 288 */ 289 public abstract class AbstractQueuedSynchronizer 290 extends AbstractOwnableSynchronizer 291 implements java.io.Serializable { 292 293 private static final long serialVersionUID = 7373984972572414691L; 294 295 /** 296 * Creates a new {@code AbstractQueuedSynchronizer} instance 297 * with initial synchronization state of zero. 298 */ AbstractQueuedSynchronizer()299 protected AbstractQueuedSynchronizer() { } 300 301 /** 302 * Wait queue node class. 303 * 304 * <p>The wait queue is a variant of a "CLH" (Craig, Landin, and 305 * Hagersten) lock queue. CLH locks are normally used for 306 * spinlocks. We instead use them for blocking synchronizers, but 307 * use the same basic tactic of holding some of the control 308 * information about a thread in the predecessor of its node. A 309 * "status" field in each node keeps track of whether a thread 310 * should block. A node is signalled when its predecessor 311 * releases. Each node of the queue otherwise serves as a 312 * specific-notification-style monitor holding a single waiting 313 * thread. The status field does NOT control whether threads are 314 * granted locks etc though. A thread may try to acquire if it is 315 * first in the queue. But being first does not guarantee success; 316 * it only gives the right to contend. So the currently released 317 * contender thread may need to rewait. 318 * 319 * <p>To enqueue into a CLH lock, you atomically splice it in as new 320 * tail. To dequeue, you just set the head field. 321 * <pre> 322 * +------+ prev +-----+ +-----+ 323 * head | | <---- | | <---- | | tail 324 * +------+ +-----+ +-----+ 325 * </pre> 326 * 327 * <p>Insertion into a CLH queue requires only a single atomic 328 * operation on "tail", so there is a simple atomic point of 329 * demarcation from unqueued to queued. Similarly, dequeuing 330 * involves only updating the "head". However, it takes a bit 331 * more work for nodes to determine who their successors are, 332 * in part to deal with possible cancellation due to timeouts 333 * and interrupts. 334 * 335 * <p>The "prev" links (not used in original CLH locks), are mainly 336 * needed to handle cancellation. If a node is cancelled, its 337 * successor is (normally) relinked to a non-cancelled 338 * predecessor. For explanation of similar mechanics in the case 339 * of spin locks, see the papers by Scott and Scherer at 340 * http://www.cs.rochester.edu/u/scott/synchronization/ 341 * 342 * <p>We also use "next" links to implement blocking mechanics. 343 * The thread id for each node is kept in its own node, so a 344 * predecessor signals the next node to wake up by traversing 345 * next link to determine which thread it is. Determination of 346 * successor must avoid races with newly queued nodes to set 347 * the "next" fields of their predecessors. This is solved 348 * when necessary by checking backwards from the atomically 349 * updated "tail" when a node's successor appears to be null. 350 * (Or, said differently, the next-links are an optimization 351 * so that we don't usually need a backward scan.) 352 * 353 * <p>Cancellation introduces some conservatism to the basic 354 * algorithms. Since we must poll for cancellation of other 355 * nodes, we can miss noticing whether a cancelled node is 356 * ahead or behind us. This is dealt with by always unparking 357 * successors upon cancellation, allowing them to stabilize on 358 * a new predecessor, unless we can identify an uncancelled 359 * predecessor who will carry this responsibility. 360 * 361 * <p>CLH queues need a dummy header node to get started. But 362 * we don't create them on construction, because it would be wasted 363 * effort if there is never contention. Instead, the node 364 * is constructed and head and tail pointers are set upon first 365 * contention. 366 * 367 * <p>Threads waiting on Conditions use the same nodes, but 368 * use an additional link. Conditions only need to link nodes 369 * in simple (non-concurrent) linked queues because they are 370 * only accessed when exclusively held. Upon await, a node is 371 * inserted into a condition queue. Upon signal, the node is 372 * transferred to the main queue. A special value of status 373 * field is used to mark which queue a node is on. 374 * 375 * <p>Thanks go to Dave Dice, Mark Moir, Victor Luchangco, Bill 376 * Scherer and Michael Scott, along with members of JSR-166 377 * expert group, for helpful ideas, discussions, and critiques 378 * on the design of this class. 379 */ 380 static final class Node { 381 /** Marker to indicate a node is waiting in shared mode */ 382 static final Node SHARED = new Node(); 383 /** Marker to indicate a node is waiting in exclusive mode */ 384 static final Node EXCLUSIVE = null; 385 386 /** waitStatus value to indicate thread has cancelled. */ 387 static final int CANCELLED = 1; 388 /** waitStatus value to indicate successor's thread needs unparking. */ 389 static final int SIGNAL = -1; 390 /** waitStatus value to indicate thread is waiting on condition. */ 391 static final int CONDITION = -2; 392 /** 393 * waitStatus value to indicate the next acquireShared should 394 * unconditionally propagate. 395 */ 396 static final int PROPAGATE = -3; 397 398 /** 399 * Status field, taking on only the values: 400 * SIGNAL: The successor of this node is (or will soon be) 401 * blocked (via park), so the current node must 402 * unpark its successor when it releases or 403 * cancels. To avoid races, acquire methods must 404 * first indicate they need a signal, 405 * then retry the atomic acquire, and then, 406 * on failure, block. 407 * CANCELLED: This node is cancelled due to timeout or interrupt. 408 * Nodes never leave this state. In particular, 409 * a thread with cancelled node never again blocks. 410 * CONDITION: This node is currently on a condition queue. 411 * It will not be used as a sync queue node 412 * until transferred, at which time the status 413 * will be set to 0. (Use of this value here has 414 * nothing to do with the other uses of the 415 * field, but simplifies mechanics.) 416 * PROPAGATE: A releaseShared should be propagated to other 417 * nodes. This is set (for head node only) in 418 * doReleaseShared to ensure propagation 419 * continues, even if other operations have 420 * since intervened. 421 * 0: None of the above 422 * 423 * The values are arranged numerically to simplify use. 424 * Non-negative values mean that a node doesn't need to 425 * signal. So, most code doesn't need to check for particular 426 * values, just for sign. 427 * 428 * The field is initialized to 0 for normal sync nodes, and 429 * CONDITION for condition nodes. It is modified using CAS 430 * (or when possible, unconditional volatile writes). 431 */ 432 volatile int waitStatus; 433 434 /** 435 * Link to predecessor node that current node/thread relies on 436 * for checking waitStatus. Assigned during enqueuing, and nulled 437 * out (for sake of GC) only upon dequeuing. Also, upon 438 * cancellation of a predecessor, we short-circuit while 439 * finding a non-cancelled one, which will always exist 440 * because the head node is never cancelled: A node becomes 441 * head only as a result of successful acquire. A 442 * cancelled thread never succeeds in acquiring, and a thread only 443 * cancels itself, not any other node. 444 */ 445 volatile Node prev; 446 447 /** 448 * Link to the successor node that the current node/thread 449 * unparks upon release. Assigned during enqueuing, adjusted 450 * when bypassing cancelled predecessors, and nulled out (for 451 * sake of GC) when dequeued. The enq operation does not 452 * assign next field of a predecessor until after attachment, 453 * so seeing a null next field does not necessarily mean that 454 * node is at end of queue. However, if a next field appears 455 * to be null, we can scan prev's from the tail to 456 * double-check. The next field of cancelled nodes is set to 457 * point to the node itself instead of null, to make life 458 * easier for isOnSyncQueue. 459 */ 460 volatile Node next; 461 462 /** 463 * The thread that enqueued this node. Initialized on 464 * construction and nulled out after use. 465 */ 466 volatile Thread thread; 467 468 /** 469 * Link to next node waiting on condition, or the special 470 * value SHARED. Because condition queues are accessed only 471 * when holding in exclusive mode, we just need a simple 472 * linked queue to hold nodes while they are waiting on 473 * conditions. They are then transferred to the queue to 474 * re-acquire. And because conditions can only be exclusive, 475 * we save a field by using special value to indicate shared 476 * mode. 477 */ 478 Node nextWaiter; 479 480 /** 481 * Returns true if node is waiting in shared mode. 482 */ isShared()483 final boolean isShared() { 484 return nextWaiter == SHARED; 485 } 486 487 /** 488 * Returns previous node, or throws NullPointerException if null. 489 * Use when predecessor cannot be null. The null check could 490 * be elided, but is present to help the VM. 491 * 492 * @return the predecessor of this node 493 */ predecessor()494 final Node predecessor() throws NullPointerException { 495 Node p = prev; 496 if (p == null) 497 throw new NullPointerException(); 498 else 499 return p; 500 } 501 502 /** Establishes initial head or SHARED marker. */ Node()503 Node() {} 504 505 /** Constructor used by addWaiter. */ Node(Node nextWaiter)506 Node(Node nextWaiter) { 507 this.nextWaiter = nextWaiter; 508 U.putObject(this, THREAD, Thread.currentThread()); 509 } 510 511 /** Constructor used by addConditionWaiter. */ Node(int waitStatus)512 Node(int waitStatus) { 513 U.putInt(this, WAITSTATUS, waitStatus); 514 U.putObject(this, THREAD, Thread.currentThread()); 515 } 516 517 /** CASes waitStatus field. */ compareAndSetWaitStatus(int expect, int update)518 final boolean compareAndSetWaitStatus(int expect, int update) { 519 return U.compareAndSwapInt(this, WAITSTATUS, expect, update); 520 } 521 522 /** CASes next field. */ compareAndSetNext(Node expect, Node update)523 final boolean compareAndSetNext(Node expect, Node update) { 524 return U.compareAndSwapObject(this, NEXT, expect, update); 525 } 526 527 private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe(); 528 private static final long NEXT; 529 static final long PREV; 530 private static final long THREAD; 531 private static final long WAITSTATUS; 532 static { 533 try { 534 NEXT = U.objectFieldOffset 535 (Node.class.getDeclaredField("next")); 536 PREV = U.objectFieldOffset 537 (Node.class.getDeclaredField("prev")); 538 THREAD = U.objectFieldOffset 539 (Node.class.getDeclaredField("thread")); 540 WAITSTATUS = U.objectFieldOffset 541 (Node.class.getDeclaredField("waitStatus")); 542 } catch (ReflectiveOperationException e) { 543 throw new Error(e); 544 } 545 } 546 } 547 548 /** 549 * Head of the wait queue, lazily initialized. Except for 550 * initialization, it is modified only via method setHead. Note: 551 * If head exists, its waitStatus is guaranteed not to be 552 * CANCELLED. 553 */ 554 private transient volatile Node head; 555 556 /** 557 * Tail of the wait queue, lazily initialized. Modified only via 558 * method enq to add new wait node. 559 */ 560 private transient volatile Node tail; 561 562 /** 563 * The synchronization state. 564 */ 565 private volatile int state; 566 567 /** 568 * Returns the current value of synchronization state. 569 * This operation has memory semantics of a {@code volatile} read. 570 * @return current state value 571 */ getState()572 protected final int getState() { 573 return state; 574 } 575 576 /** 577 * Sets the value of synchronization state. 578 * This operation has memory semantics of a {@code volatile} write. 579 * @param newState the new state value 580 */ setState(int newState)581 protected final void setState(int newState) { 582 state = newState; 583 } 584 585 /** 586 * Atomically sets synchronization state to the given updated 587 * value if the current state value equals the expected value. 588 * This operation has memory semantics of a {@code volatile} read 589 * and write. 590 * 591 * @param expect the expected value 592 * @param update the new value 593 * @return {@code true} if successful. False return indicates that the actual 594 * value was not equal to the expected value. 595 */ compareAndSetState(int expect, int update)596 protected final boolean compareAndSetState(int expect, int update) { 597 return U.compareAndSwapInt(this, STATE, expect, update); 598 } 599 600 // Queuing utilities 601 602 /** 603 * The number of nanoseconds for which it is faster to spin 604 * rather than to use timed park. A rough estimate suffices 605 * to improve responsiveness with very short timeouts. 606 */ 607 static final long SPIN_FOR_TIMEOUT_THRESHOLD = 1000L; 608 609 /** 610 * Inserts node into queue, initializing if necessary. See picture above. 611 * @param node the node to insert 612 * @return node's predecessor 613 */ enq(Node node)614 private Node enq(Node node) { 615 for (;;) { 616 Node oldTail = tail; 617 if (oldTail != null) { 618 U.putObject(node, Node.PREV, oldTail); 619 if (compareAndSetTail(oldTail, node)) { 620 oldTail.next = node; 621 return oldTail; 622 } 623 } else { 624 initializeSyncQueue(); 625 } 626 } 627 } 628 629 /** 630 * Creates and enqueues node for current thread and given mode. 631 * 632 * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared 633 * @return the new node 634 */ addWaiter(Node mode)635 private Node addWaiter(Node mode) { 636 Node node = new Node(mode); 637 638 for (;;) { 639 Node oldTail = tail; 640 if (oldTail != null) { 641 U.putObject(node, Node.PREV, oldTail); 642 if (compareAndSetTail(oldTail, node)) { 643 oldTail.next = node; 644 return node; 645 } 646 } else { 647 initializeSyncQueue(); 648 } 649 } 650 } 651 652 /** 653 * Sets head of queue to be node, thus dequeuing. Called only by 654 * acquire methods. Also nulls out unused fields for sake of GC 655 * and to suppress unnecessary signals and traversals. 656 * 657 * @param node the node 658 */ setHead(Node node)659 private void setHead(Node node) { 660 head = node; 661 node.thread = null; 662 node.prev = null; 663 } 664 665 /** 666 * Wakes up node's successor, if one exists. 667 * 668 * @param node the node 669 */ unparkSuccessor(Node node)670 private void unparkSuccessor(Node node) { 671 /* 672 * If status is negative (i.e., possibly needing signal) try 673 * to clear in anticipation of signalling. It is OK if this 674 * fails or if status is changed by waiting thread. 675 */ 676 int ws = node.waitStatus; 677 if (ws < 0) 678 node.compareAndSetWaitStatus(ws, 0); 679 680 /* 681 * Thread to unpark is held in successor, which is normally 682 * just the next node. But if cancelled or apparently null, 683 * traverse backwards from tail to find the actual 684 * non-cancelled successor. 685 */ 686 Node s = node.next; 687 if (s == null || s.waitStatus > 0) { 688 s = null; 689 for (Node p = tail; p != node && p != null; p = p.prev) 690 if (p.waitStatus <= 0) 691 s = p; 692 } 693 if (s != null) 694 LockSupport.unpark(s.thread); 695 } 696 697 /** 698 * Release action for shared mode -- signals successor and ensures 699 * propagation. (Note: For exclusive mode, release just amounts 700 * to calling unparkSuccessor of head if it needs signal.) 701 */ doReleaseShared()702 private void doReleaseShared() { 703 /* 704 * Ensure that a release propagates, even if there are other 705 * in-progress acquires/releases. This proceeds in the usual 706 * way of trying to unparkSuccessor of head if it needs 707 * signal. But if it does not, status is set to PROPAGATE to 708 * ensure that upon release, propagation continues. 709 * Additionally, we must loop in case a new node is added 710 * while we are doing this. Also, unlike other uses of 711 * unparkSuccessor, we need to know if CAS to reset status 712 * fails, if so rechecking. 713 */ 714 for (;;) { 715 Node h = head; 716 if (h != null && h != tail) { 717 int ws = h.waitStatus; 718 if (ws == Node.SIGNAL) { 719 if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0)) 720 continue; // loop to recheck cases 721 unparkSuccessor(h); 722 } 723 else if (ws == 0 && 724 !h.compareAndSetWaitStatus(0, Node.PROPAGATE)) 725 continue; // loop on failed CAS 726 } 727 if (h == head) // loop if head changed 728 break; 729 } 730 } 731 732 /** 733 * Sets head of queue, and checks if successor may be waiting 734 * in shared mode, if so propagating if either propagate > 0 or 735 * PROPAGATE status was set. 736 * 737 * @param node the node 738 * @param propagate the return value from a tryAcquireShared 739 */ setHeadAndPropagate(Node node, int propagate)740 private void setHeadAndPropagate(Node node, int propagate) { 741 Node h = head; // Record old head for check below 742 setHead(node); 743 /* 744 * Try to signal next queued node if: 745 * Propagation was indicated by caller, 746 * or was recorded (as h.waitStatus either before 747 * or after setHead) by a previous operation 748 * (note: this uses sign-check of waitStatus because 749 * PROPAGATE status may transition to SIGNAL.) 750 * and 751 * The next node is waiting in shared mode, 752 * or we don't know, because it appears null 753 * 754 * The conservatism in both of these checks may cause 755 * unnecessary wake-ups, but only when there are multiple 756 * racing acquires/releases, so most need signals now or soon 757 * anyway. 758 */ 759 if (propagate > 0 || h == null || h.waitStatus < 0 || 760 (h = head) == null || h.waitStatus < 0) { 761 Node s = node.next; 762 if (s == null || s.isShared()) 763 doReleaseShared(); 764 } 765 } 766 767 // Utilities for various versions of acquire 768 769 /** 770 * Cancels an ongoing attempt to acquire. 771 * 772 * @param node the node 773 */ cancelAcquire(Node node)774 private void cancelAcquire(Node node) { 775 // Ignore if node doesn't exist 776 if (node == null) 777 return; 778 779 node.thread = null; 780 781 // Skip cancelled predecessors 782 Node pred = node.prev; 783 while (pred.waitStatus > 0) 784 node.prev = pred = pred.prev; 785 786 // predNext is the apparent node to unsplice. CASes below will 787 // fail if not, in which case, we lost race vs another cancel 788 // or signal, so no further action is necessary. 789 Node predNext = pred.next; 790 791 // Can use unconditional write instead of CAS here. 792 // After this atomic step, other Nodes can skip past us. 793 // Before, we are free of interference from other threads. 794 node.waitStatus = Node.CANCELLED; 795 796 // If we are the tail, remove ourselves. 797 if (node == tail && compareAndSetTail(node, pred)) { 798 pred.compareAndSetNext(predNext, null); 799 } else { 800 // If successor needs signal, try to set pred's next-link 801 // so it will get one. Otherwise wake it up to propagate. 802 int ws; 803 if (pred != head && 804 ((ws = pred.waitStatus) == Node.SIGNAL || 805 (ws <= 0 && pred.compareAndSetWaitStatus(ws, Node.SIGNAL))) && 806 pred.thread != null) { 807 Node next = node.next; 808 if (next != null && next.waitStatus <= 0) 809 pred.compareAndSetNext(predNext, next); 810 } else { 811 unparkSuccessor(node); 812 } 813 814 node.next = node; // help GC 815 } 816 } 817 818 /** 819 * Checks and updates status for a node that failed to acquire. 820 * Returns true if thread should block. This is the main signal 821 * control in all acquire loops. Requires that pred == node.prev. 822 * 823 * @param pred node's predecessor holding status 824 * @param node the node 825 * @return {@code true} if thread should block 826 */ shouldParkAfterFailedAcquire(Node pred, Node node)827 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 828 int ws = pred.waitStatus; 829 if (ws == Node.SIGNAL) 830 /* 831 * This node has already set status asking a release 832 * to signal it, so it can safely park. 833 */ 834 return true; 835 if (ws > 0) { 836 /* 837 * Predecessor was cancelled. Skip over predecessors and 838 * indicate retry. 839 */ 840 do { 841 node.prev = pred = pred.prev; 842 } while (pred.waitStatus > 0); 843 pred.next = node; 844 } else { 845 /* 846 * waitStatus must be 0 or PROPAGATE. Indicate that we 847 * need a signal, but don't park yet. Caller will need to 848 * retry to make sure it cannot acquire before parking. 849 */ 850 pred.compareAndSetWaitStatus(ws, Node.SIGNAL); 851 } 852 return false; 853 } 854 855 /** 856 * Convenience method to interrupt current thread. 857 */ selfInterrupt()858 static void selfInterrupt() { 859 Thread.currentThread().interrupt(); 860 } 861 862 /** 863 * Convenience method to park and then check if interrupted. 864 * 865 * @return {@code true} if interrupted 866 */ parkAndCheckInterrupt()867 private final boolean parkAndCheckInterrupt() { 868 LockSupport.park(this); 869 return Thread.interrupted(); 870 } 871 872 /* 873 * Various flavors of acquire, varying in exclusive/shared and 874 * control modes. Each is mostly the same, but annoyingly 875 * different. Only a little bit of factoring is possible due to 876 * interactions of exception mechanics (including ensuring that we 877 * cancel if tryAcquire throws exception) and other control, at 878 * least not without hurting performance too much. 879 */ 880 881 /** 882 * Acquires in exclusive uninterruptible mode for thread already in 883 * queue. Used by condition wait methods as well as acquire. 884 * 885 * @param node the node 886 * @param arg the acquire argument 887 * @return {@code true} if interrupted while waiting 888 */ acquireQueued(final Node node, int arg)889 final boolean acquireQueued(final Node node, int arg) { 890 try { 891 boolean interrupted = false; 892 for (;;) { 893 final Node p = node.predecessor(); 894 if (p == head && tryAcquire(arg)) { 895 setHead(node); 896 p.next = null; // help GC 897 return interrupted; 898 } 899 if (shouldParkAfterFailedAcquire(p, node) && 900 parkAndCheckInterrupt()) 901 interrupted = true; 902 } 903 } catch (Throwable t) { 904 cancelAcquire(node); 905 throw t; 906 } 907 } 908 909 /** 910 * Acquires in exclusive interruptible mode. 911 * @param arg the acquire argument 912 */ doAcquireInterruptibly(int arg)913 private void doAcquireInterruptibly(int arg) 914 throws InterruptedException { 915 final Node node = addWaiter(Node.EXCLUSIVE); 916 try { 917 for (;;) { 918 final Node p = node.predecessor(); 919 if (p == head && tryAcquire(arg)) { 920 setHead(node); 921 p.next = null; // help GC 922 return; 923 } 924 if (shouldParkAfterFailedAcquire(p, node) && 925 parkAndCheckInterrupt()) 926 throw new InterruptedException(); 927 } 928 } catch (Throwable t) { 929 cancelAcquire(node); 930 throw t; 931 } 932 } 933 934 /** 935 * Acquires in exclusive timed mode. 936 * 937 * @param arg the acquire argument 938 * @param nanosTimeout max wait time 939 * @return {@code true} if acquired 940 */ doAcquireNanos(int arg, long nanosTimeout)941 private boolean doAcquireNanos(int arg, long nanosTimeout) 942 throws InterruptedException { 943 if (nanosTimeout <= 0L) 944 return false; 945 final long deadline = System.nanoTime() + nanosTimeout; 946 final Node node = addWaiter(Node.EXCLUSIVE); 947 try { 948 for (;;) { 949 final Node p = node.predecessor(); 950 if (p == head && tryAcquire(arg)) { 951 setHead(node); 952 p.next = null; // help GC 953 return true; 954 } 955 nanosTimeout = deadline - System.nanoTime(); 956 if (nanosTimeout <= 0L) { 957 cancelAcquire(node); 958 return false; 959 } 960 if (shouldParkAfterFailedAcquire(p, node) && 961 nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD) 962 LockSupport.parkNanos(this, nanosTimeout); 963 if (Thread.interrupted()) 964 throw new InterruptedException(); 965 } 966 } catch (Throwable t) { 967 cancelAcquire(node); 968 throw t; 969 } 970 } 971 972 /** 973 * Acquires in shared uninterruptible mode. 974 * @param arg the acquire argument 975 */ doAcquireShared(int arg)976 private void doAcquireShared(int arg) { 977 final Node node = addWaiter(Node.SHARED); 978 try { 979 boolean interrupted = false; 980 for (;;) { 981 final Node p = node.predecessor(); 982 if (p == head) { 983 int r = tryAcquireShared(arg); 984 if (r >= 0) { 985 setHeadAndPropagate(node, r); 986 p.next = null; // help GC 987 if (interrupted) 988 selfInterrupt(); 989 return; 990 } 991 } 992 if (shouldParkAfterFailedAcquire(p, node) && 993 parkAndCheckInterrupt()) 994 interrupted = true; 995 } 996 } catch (Throwable t) { 997 cancelAcquire(node); 998 throw t; 999 } 1000 } 1001 1002 /** 1003 * Acquires in shared interruptible mode. 1004 * @param arg the acquire argument 1005 */ doAcquireSharedInterruptibly(int arg)1006 private void doAcquireSharedInterruptibly(int arg) 1007 throws InterruptedException { 1008 final Node node = addWaiter(Node.SHARED); 1009 try { 1010 for (;;) { 1011 final Node p = node.predecessor(); 1012 if (p == head) { 1013 int r = tryAcquireShared(arg); 1014 if (r >= 0) { 1015 setHeadAndPropagate(node, r); 1016 p.next = null; // help GC 1017 return; 1018 } 1019 } 1020 if (shouldParkAfterFailedAcquire(p, node) && 1021 parkAndCheckInterrupt()) 1022 throw new InterruptedException(); 1023 } 1024 } catch (Throwable t) { 1025 cancelAcquire(node); 1026 throw t; 1027 } 1028 } 1029 1030 /** 1031 * Acquires in shared timed mode. 1032 * 1033 * @param arg the acquire argument 1034 * @param nanosTimeout max wait time 1035 * @return {@code true} if acquired 1036 */ doAcquireSharedNanos(int arg, long nanosTimeout)1037 private boolean doAcquireSharedNanos(int arg, long nanosTimeout) 1038 throws InterruptedException { 1039 if (nanosTimeout <= 0L) 1040 return false; 1041 final long deadline = System.nanoTime() + nanosTimeout; 1042 final Node node = addWaiter(Node.SHARED); 1043 try { 1044 for (;;) { 1045 final Node p = node.predecessor(); 1046 if (p == head) { 1047 int r = tryAcquireShared(arg); 1048 if (r >= 0) { 1049 setHeadAndPropagate(node, r); 1050 p.next = null; // help GC 1051 return true; 1052 } 1053 } 1054 nanosTimeout = deadline - System.nanoTime(); 1055 if (nanosTimeout <= 0L) { 1056 cancelAcquire(node); 1057 return false; 1058 } 1059 if (shouldParkAfterFailedAcquire(p, node) && 1060 nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD) 1061 LockSupport.parkNanos(this, nanosTimeout); 1062 if (Thread.interrupted()) 1063 throw new InterruptedException(); 1064 } 1065 } catch (Throwable t) { 1066 cancelAcquire(node); 1067 throw t; 1068 } 1069 } 1070 1071 // Main exported methods 1072 1073 /** 1074 * Attempts to acquire in exclusive mode. This method should query 1075 * if the state of the object permits it to be acquired in the 1076 * exclusive mode, and if so to acquire it. 1077 * 1078 * <p>This method is always invoked by the thread performing 1079 * acquire. If this method reports failure, the acquire method 1080 * may queue the thread, if it is not already queued, until it is 1081 * signalled by a release from some other thread. This can be used 1082 * to implement method {@link Lock#tryLock()}. 1083 * 1084 * <p>The default 1085 * implementation throws {@link UnsupportedOperationException}. 1086 * 1087 * @param arg the acquire argument. This value is always the one 1088 * passed to an acquire method, or is the value saved on entry 1089 * to a condition wait. The value is otherwise uninterpreted 1090 * and can represent anything you like. 1091 * @return {@code true} if successful. Upon success, this object has 1092 * been acquired. 1093 * @throws IllegalMonitorStateException if acquiring would place this 1094 * synchronizer in an illegal state. This exception must be 1095 * thrown in a consistent fashion for synchronization to work 1096 * correctly. 1097 * @throws UnsupportedOperationException if exclusive mode is not supported 1098 */ tryAcquire(int arg)1099 protected boolean tryAcquire(int arg) { 1100 throw new UnsupportedOperationException(); 1101 } 1102 1103 /** 1104 * Attempts to set the state to reflect a release in exclusive 1105 * mode. 1106 * 1107 * <p>This method is always invoked by the thread performing release. 1108 * 1109 * <p>The default implementation throws 1110 * {@link UnsupportedOperationException}. 1111 * 1112 * @param arg the release argument. This value is always the one 1113 * passed to a release method, or the current state value upon 1114 * entry to a condition wait. The value is otherwise 1115 * uninterpreted and can represent anything you like. 1116 * @return {@code true} if this object is now in a fully released 1117 * state, so that any waiting threads may attempt to acquire; 1118 * and {@code false} otherwise. 1119 * @throws IllegalMonitorStateException if releasing would place this 1120 * synchronizer in an illegal state. This exception must be 1121 * thrown in a consistent fashion for synchronization to work 1122 * correctly. 1123 * @throws UnsupportedOperationException if exclusive mode is not supported 1124 */ tryRelease(int arg)1125 protected boolean tryRelease(int arg) { 1126 throw new UnsupportedOperationException(); 1127 } 1128 1129 /** 1130 * Attempts to acquire in shared mode. This method should query if 1131 * the state of the object permits it to be acquired in the shared 1132 * mode, and if so to acquire it. 1133 * 1134 * <p>This method is always invoked by the thread performing 1135 * acquire. If this method reports failure, the acquire method 1136 * may queue the thread, if it is not already queued, until it is 1137 * signalled by a release from some other thread. 1138 * 1139 * <p>The default implementation throws {@link 1140 * UnsupportedOperationException}. 1141 * 1142 * @param arg the acquire argument. This value is always the one 1143 * passed to an acquire method, or is the value saved on entry 1144 * to a condition wait. The value is otherwise uninterpreted 1145 * and can represent anything you like. 1146 * @return a negative value on failure; zero if acquisition in shared 1147 * mode succeeded but no subsequent shared-mode acquire can 1148 * succeed; and a positive value if acquisition in shared 1149 * mode succeeded and subsequent shared-mode acquires might 1150 * also succeed, in which case a subsequent waiting thread 1151 * must check availability. (Support for three different 1152 * return values enables this method to be used in contexts 1153 * where acquires only sometimes act exclusively.) Upon 1154 * success, this object has been acquired. 1155 * @throws IllegalMonitorStateException if acquiring would place this 1156 * synchronizer in an illegal state. This exception must be 1157 * thrown in a consistent fashion for synchronization to work 1158 * correctly. 1159 * @throws UnsupportedOperationException if shared mode is not supported 1160 */ tryAcquireShared(int arg)1161 protected int tryAcquireShared(int arg) { 1162 throw new UnsupportedOperationException(); 1163 } 1164 1165 /** 1166 * Attempts to set the state to reflect a release in shared mode. 1167 * 1168 * <p>This method is always invoked by the thread performing release. 1169 * 1170 * <p>The default implementation throws 1171 * {@link UnsupportedOperationException}. 1172 * 1173 * @param arg the release argument. This value is always the one 1174 * passed to a release method, or the current state value upon 1175 * entry to a condition wait. The value is otherwise 1176 * uninterpreted and can represent anything you like. 1177 * @return {@code true} if this release of shared mode may permit a 1178 * waiting acquire (shared or exclusive) to succeed; and 1179 * {@code false} otherwise 1180 * @throws IllegalMonitorStateException if releasing would place this 1181 * synchronizer in an illegal state. This exception must be 1182 * thrown in a consistent fashion for synchronization to work 1183 * correctly. 1184 * @throws UnsupportedOperationException if shared mode is not supported 1185 */ tryReleaseShared(int arg)1186 protected boolean tryReleaseShared(int arg) { 1187 throw new UnsupportedOperationException(); 1188 } 1189 1190 /** 1191 * Returns {@code true} if synchronization is held exclusively with 1192 * respect to the current (calling) thread. This method is invoked 1193 * upon each call to a non-waiting {@link ConditionObject} method. 1194 * (Waiting methods instead invoke {@link #release}.) 1195 * 1196 * <p>The default implementation throws {@link 1197 * UnsupportedOperationException}. This method is invoked 1198 * internally only within {@link ConditionObject} methods, so need 1199 * not be defined if conditions are not used. 1200 * 1201 * @return {@code true} if synchronization is held exclusively; 1202 * {@code false} otherwise 1203 * @throws UnsupportedOperationException if conditions are not supported 1204 */ isHeldExclusively()1205 protected boolean isHeldExclusively() { 1206 throw new UnsupportedOperationException(); 1207 } 1208 1209 /** 1210 * Acquires in exclusive mode, ignoring interrupts. Implemented 1211 * by invoking at least once {@link #tryAcquire}, 1212 * returning on success. Otherwise the thread is queued, possibly 1213 * repeatedly blocking and unblocking, invoking {@link 1214 * #tryAcquire} until success. This method can be used 1215 * to implement method {@link Lock#lock}. 1216 * 1217 * @param arg the acquire argument. This value is conveyed to 1218 * {@link #tryAcquire} but is otherwise uninterpreted and 1219 * can represent anything you like. 1220 */ acquire(int arg)1221 public final void acquire(int arg) { 1222 if (!tryAcquire(arg) && 1223 acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) 1224 selfInterrupt(); 1225 } 1226 1227 /** 1228 * Acquires in exclusive mode, aborting if interrupted. 1229 * Implemented by first checking interrupt status, then invoking 1230 * at least once {@link #tryAcquire}, returning on 1231 * success. Otherwise the thread is queued, possibly repeatedly 1232 * blocking and unblocking, invoking {@link #tryAcquire} 1233 * until success or the thread is interrupted. This method can be 1234 * used to implement method {@link Lock#lockInterruptibly}. 1235 * 1236 * @param arg the acquire argument. This value is conveyed to 1237 * {@link #tryAcquire} but is otherwise uninterpreted and 1238 * can represent anything you like. 1239 * @throws InterruptedException if the current thread is interrupted 1240 */ acquireInterruptibly(int arg)1241 public final void acquireInterruptibly(int arg) 1242 throws InterruptedException { 1243 if (Thread.interrupted()) 1244 throw new InterruptedException(); 1245 if (!tryAcquire(arg)) 1246 doAcquireInterruptibly(arg); 1247 } 1248 1249 /** 1250 * Attempts to acquire in exclusive mode, aborting if interrupted, 1251 * and failing if the given timeout elapses. Implemented by first 1252 * checking interrupt status, then invoking at least once {@link 1253 * #tryAcquire}, returning on success. Otherwise, the thread is 1254 * queued, possibly repeatedly blocking and unblocking, invoking 1255 * {@link #tryAcquire} until success or the thread is interrupted 1256 * or the timeout elapses. This method can be used to implement 1257 * method {@link Lock#tryLock(long, TimeUnit)}. 1258 * 1259 * @param arg the acquire argument. This value is conveyed to 1260 * {@link #tryAcquire} but is otherwise uninterpreted and 1261 * can represent anything you like. 1262 * @param nanosTimeout the maximum number of nanoseconds to wait 1263 * @return {@code true} if acquired; {@code false} if timed out 1264 * @throws InterruptedException if the current thread is interrupted 1265 */ tryAcquireNanos(int arg, long nanosTimeout)1266 public final boolean tryAcquireNanos(int arg, long nanosTimeout) 1267 throws InterruptedException { 1268 if (Thread.interrupted()) 1269 throw new InterruptedException(); 1270 return tryAcquire(arg) || 1271 doAcquireNanos(arg, nanosTimeout); 1272 } 1273 1274 /** 1275 * Releases in exclusive mode. Implemented by unblocking one or 1276 * more threads if {@link #tryRelease} returns true. 1277 * This method can be used to implement method {@link Lock#unlock}. 1278 * 1279 * @param arg the release argument. This value is conveyed to 1280 * {@link #tryRelease} but is otherwise uninterpreted and 1281 * can represent anything you like. 1282 * @return the value returned from {@link #tryRelease} 1283 */ release(int arg)1284 public final boolean release(int arg) { 1285 if (tryRelease(arg)) { 1286 Node h = head; 1287 if (h != null && h.waitStatus != 0) 1288 unparkSuccessor(h); 1289 return true; 1290 } 1291 return false; 1292 } 1293 1294 /** 1295 * Acquires in shared mode, ignoring interrupts. Implemented by 1296 * first invoking at least once {@link #tryAcquireShared}, 1297 * returning on success. Otherwise the thread is queued, possibly 1298 * repeatedly blocking and unblocking, invoking {@link 1299 * #tryAcquireShared} until success. 1300 * 1301 * @param arg the acquire argument. This value is conveyed to 1302 * {@link #tryAcquireShared} but is otherwise uninterpreted 1303 * and can represent anything you like. 1304 */ acquireShared(int arg)1305 public final void acquireShared(int arg) { 1306 if (tryAcquireShared(arg) < 0) 1307 doAcquireShared(arg); 1308 } 1309 1310 /** 1311 * Acquires in shared mode, aborting if interrupted. Implemented 1312 * by first checking interrupt status, then invoking at least once 1313 * {@link #tryAcquireShared}, returning on success. Otherwise the 1314 * thread is queued, possibly repeatedly blocking and unblocking, 1315 * invoking {@link #tryAcquireShared} until success or the thread 1316 * is interrupted. 1317 * @param arg the acquire argument. 1318 * This value is conveyed to {@link #tryAcquireShared} but is 1319 * otherwise uninterpreted and can represent anything 1320 * you like. 1321 * @throws InterruptedException if the current thread is interrupted 1322 */ acquireSharedInterruptibly(int arg)1323 public final void acquireSharedInterruptibly(int arg) 1324 throws InterruptedException { 1325 if (Thread.interrupted()) 1326 throw new InterruptedException(); 1327 if (tryAcquireShared(arg) < 0) 1328 doAcquireSharedInterruptibly(arg); 1329 } 1330 1331 /** 1332 * Attempts to acquire in shared mode, aborting if interrupted, and 1333 * failing if the given timeout elapses. Implemented by first 1334 * checking interrupt status, then invoking at least once {@link 1335 * #tryAcquireShared}, returning on success. Otherwise, the 1336 * thread is queued, possibly repeatedly blocking and unblocking, 1337 * invoking {@link #tryAcquireShared} until success or the thread 1338 * is interrupted or the timeout elapses. 1339 * 1340 * @param arg the acquire argument. This value is conveyed to 1341 * {@link #tryAcquireShared} but is otherwise uninterpreted 1342 * and can represent anything you like. 1343 * @param nanosTimeout the maximum number of nanoseconds to wait 1344 * @return {@code true} if acquired; {@code false} if timed out 1345 * @throws InterruptedException if the current thread is interrupted 1346 */ tryAcquireSharedNanos(int arg, long nanosTimeout)1347 public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) 1348 throws InterruptedException { 1349 if (Thread.interrupted()) 1350 throw new InterruptedException(); 1351 return tryAcquireShared(arg) >= 0 || 1352 doAcquireSharedNanos(arg, nanosTimeout); 1353 } 1354 1355 /** 1356 * Releases in shared mode. Implemented by unblocking one or more 1357 * threads if {@link #tryReleaseShared} returns true. 1358 * 1359 * @param arg the release argument. This value is conveyed to 1360 * {@link #tryReleaseShared} but is otherwise uninterpreted 1361 * and can represent anything you like. 1362 * @return the value returned from {@link #tryReleaseShared} 1363 */ releaseShared(int arg)1364 public final boolean releaseShared(int arg) { 1365 if (tryReleaseShared(arg)) { 1366 doReleaseShared(); 1367 return true; 1368 } 1369 return false; 1370 } 1371 1372 // Queue inspection methods 1373 1374 /** 1375 * Queries whether any threads are waiting to acquire. Note that 1376 * because cancellations due to interrupts and timeouts may occur 1377 * at any time, a {@code true} return does not guarantee that any 1378 * other thread will ever acquire. 1379 * 1380 * <p>In this implementation, this operation returns in 1381 * constant time. 1382 * 1383 * @return {@code true} if there may be other threads waiting to acquire 1384 */ hasQueuedThreads()1385 public final boolean hasQueuedThreads() { 1386 return head != tail; 1387 } 1388 1389 /** 1390 * Queries whether any threads have ever contended to acquire this 1391 * synchronizer; that is, if an acquire method has ever blocked. 1392 * 1393 * <p>In this implementation, this operation returns in 1394 * constant time. 1395 * 1396 * @return {@code true} if there has ever been contention 1397 */ hasContended()1398 public final boolean hasContended() { 1399 return head != null; 1400 } 1401 1402 /** 1403 * Returns the first (longest-waiting) thread in the queue, or 1404 * {@code null} if no threads are currently queued. 1405 * 1406 * <p>In this implementation, this operation normally returns in 1407 * constant time, but may iterate upon contention if other threads are 1408 * concurrently modifying the queue. 1409 * 1410 * @return the first (longest-waiting) thread in the queue, or 1411 * {@code null} if no threads are currently queued 1412 */ getFirstQueuedThread()1413 public final Thread getFirstQueuedThread() { 1414 // handle only fast path, else relay 1415 return (head == tail) ? null : fullGetFirstQueuedThread(); 1416 } 1417 1418 /** 1419 * Version of getFirstQueuedThread called when fastpath fails. 1420 */ fullGetFirstQueuedThread()1421 private Thread fullGetFirstQueuedThread() { 1422 /* 1423 * The first node is normally head.next. Try to get its 1424 * thread field, ensuring consistent reads: If thread 1425 * field is nulled out or s.prev is no longer head, then 1426 * some other thread(s) concurrently performed setHead in 1427 * between some of our reads. We try this twice before 1428 * resorting to traversal. 1429 */ 1430 Node h, s; 1431 Thread st; 1432 if (((h = head) != null && (s = h.next) != null && 1433 s.prev == head && (st = s.thread) != null) || 1434 ((h = head) != null && (s = h.next) != null && 1435 s.prev == head && (st = s.thread) != null)) 1436 return st; 1437 1438 /* 1439 * Head's next field might not have been set yet, or may have 1440 * been unset after setHead. So we must check to see if tail 1441 * is actually first node. If not, we continue on, safely 1442 * traversing from tail back to head to find first, 1443 * guaranteeing termination. 1444 */ 1445 1446 Thread firstThread = null; 1447 for (Node p = tail; p != null && p != head; p = p.prev) { 1448 Thread t = p.thread; 1449 if (t != null) 1450 firstThread = t; 1451 } 1452 return firstThread; 1453 } 1454 1455 /** 1456 * Returns true if the given thread is currently queued. 1457 * 1458 * <p>This implementation traverses the queue to determine 1459 * presence of the given thread. 1460 * 1461 * @param thread the thread 1462 * @return {@code true} if the given thread is on the queue 1463 * @throws NullPointerException if the thread is null 1464 */ isQueued(Thread thread)1465 public final boolean isQueued(Thread thread) { 1466 if (thread == null) 1467 throw new NullPointerException(); 1468 for (Node p = tail; p != null; p = p.prev) 1469 if (p.thread == thread) 1470 return true; 1471 return false; 1472 } 1473 1474 /** 1475 * Returns {@code true} if the apparent first queued thread, if one 1476 * exists, is waiting in exclusive mode. If this method returns 1477 * {@code true}, and the current thread is attempting to acquire in 1478 * shared mode (that is, this method is invoked from {@link 1479 * #tryAcquireShared}) then it is guaranteed that the current thread 1480 * is not the first queued thread. Used only as a heuristic in 1481 * ReentrantReadWriteLock. 1482 */ apparentlyFirstQueuedIsExclusive()1483 final boolean apparentlyFirstQueuedIsExclusive() { 1484 Node h, s; 1485 return (h = head) != null && 1486 (s = h.next) != null && 1487 !s.isShared() && 1488 s.thread != null; 1489 } 1490 1491 /** 1492 * Queries whether any threads have been waiting to acquire longer 1493 * than the current thread. 1494 * 1495 * <p>An invocation of this method is equivalent to (but may be 1496 * more efficient than): 1497 * <pre> {@code 1498 * getFirstQueuedThread() != Thread.currentThread() 1499 * && hasQueuedThreads()}</pre> 1500 * 1501 * <p>Note that because cancellations due to interrupts and 1502 * timeouts may occur at any time, a {@code true} return does not 1503 * guarantee that some other thread will acquire before the current 1504 * thread. Likewise, it is possible for another thread to win a 1505 * race to enqueue after this method has returned {@code false}, 1506 * due to the queue being empty. 1507 * 1508 * <p>This method is designed to be used by a fair synchronizer to 1509 * avoid <a href="AbstractQueuedSynchronizer.html#barging">barging</a>. 1510 * Such a synchronizer's {@link #tryAcquire} method should return 1511 * {@code false}, and its {@link #tryAcquireShared} method should 1512 * return a negative value, if this method returns {@code true} 1513 * (unless this is a reentrant acquire). For example, the {@code 1514 * tryAcquire} method for a fair, reentrant, exclusive mode 1515 * synchronizer might look like this: 1516 * 1517 * <pre> {@code 1518 * protected boolean tryAcquire(int arg) { 1519 * if (isHeldExclusively()) { 1520 * // A reentrant acquire; increment hold count 1521 * return true; 1522 * } else if (hasQueuedPredecessors()) { 1523 * return false; 1524 * } else { 1525 * // try to acquire normally 1526 * } 1527 * }}</pre> 1528 * 1529 * @return {@code true} if there is a queued thread preceding the 1530 * current thread, and {@code false} if the current thread 1531 * is at the head of the queue or the queue is empty 1532 * @since 1.7 1533 */ hasQueuedPredecessors()1534 public final boolean hasQueuedPredecessors() { 1535 // The correctness of this depends on head being initialized 1536 // before tail and on head.next being accurate if the current 1537 // thread is first in queue. 1538 Node t = tail; // Read fields in reverse initialization order 1539 Node h = head; 1540 Node s; 1541 return h != t && 1542 ((s = h.next) == null || s.thread != Thread.currentThread()); 1543 } 1544 1545 1546 // Instrumentation and monitoring methods 1547 1548 /** 1549 * Returns an estimate of the number of threads waiting to 1550 * acquire. The value is only an estimate because the number of 1551 * threads may change dynamically while this method traverses 1552 * internal data structures. This method is designed for use in 1553 * monitoring system state, not for synchronization control. 1554 * 1555 * @return the estimated number of threads waiting to acquire 1556 */ getQueueLength()1557 public final int getQueueLength() { 1558 int n = 0; 1559 for (Node p = tail; p != null; p = p.prev) { 1560 if (p.thread != null) 1561 ++n; 1562 } 1563 return n; 1564 } 1565 1566 /** 1567 * Returns a collection containing threads that may be waiting to 1568 * acquire. Because the actual set of threads may change 1569 * dynamically while constructing this result, the returned 1570 * collection is only a best-effort estimate. The elements of the 1571 * returned collection are in no particular order. This method is 1572 * designed to facilitate construction of subclasses that provide 1573 * more extensive monitoring facilities. 1574 * 1575 * @return the collection of threads 1576 */ getQueuedThreads()1577 public final Collection<Thread> getQueuedThreads() { 1578 ArrayList<Thread> list = new ArrayList<>(); 1579 for (Node p = tail; p != null; p = p.prev) { 1580 Thread t = p.thread; 1581 if (t != null) 1582 list.add(t); 1583 } 1584 return list; 1585 } 1586 1587 /** 1588 * Returns a collection containing threads that may be waiting to 1589 * acquire in exclusive mode. This has the same properties 1590 * as {@link #getQueuedThreads} except that it only returns 1591 * those threads waiting due to an exclusive acquire. 1592 * 1593 * @return the collection of threads 1594 */ getExclusiveQueuedThreads()1595 public final Collection<Thread> getExclusiveQueuedThreads() { 1596 ArrayList<Thread> list = new ArrayList<>(); 1597 for (Node p = tail; p != null; p = p.prev) { 1598 if (!p.isShared()) { 1599 Thread t = p.thread; 1600 if (t != null) 1601 list.add(t); 1602 } 1603 } 1604 return list; 1605 } 1606 1607 /** 1608 * Returns a collection containing threads that may be waiting to 1609 * acquire in shared mode. This has the same properties 1610 * as {@link #getQueuedThreads} except that it only returns 1611 * those threads waiting due to a shared acquire. 1612 * 1613 * @return the collection of threads 1614 */ getSharedQueuedThreads()1615 public final Collection<Thread> getSharedQueuedThreads() { 1616 ArrayList<Thread> list = new ArrayList<>(); 1617 for (Node p = tail; p != null; p = p.prev) { 1618 if (p.isShared()) { 1619 Thread t = p.thread; 1620 if (t != null) 1621 list.add(t); 1622 } 1623 } 1624 return list; 1625 } 1626 1627 /** 1628 * Returns a string identifying this synchronizer, as well as its state. 1629 * The state, in brackets, includes the String {@code "State ="} 1630 * followed by the current value of {@link #getState}, and either 1631 * {@code "nonempty"} or {@code "empty"} depending on whether the 1632 * queue is empty. 1633 * 1634 * @return a string identifying this synchronizer, as well as its state 1635 */ toString()1636 public String toString() { 1637 return super.toString() 1638 + "[State = " + getState() + ", " 1639 + (hasQueuedThreads() ? "non" : "") + "empty queue]"; 1640 } 1641 1642 1643 // Internal support methods for Conditions 1644 1645 /** 1646 * Returns true if a node, always one that was initially placed on 1647 * a condition queue, is now waiting to reacquire on sync queue. 1648 * @param node the node 1649 * @return true if is reacquiring 1650 */ isOnSyncQueue(Node node)1651 final boolean isOnSyncQueue(Node node) { 1652 if (node.waitStatus == Node.CONDITION || node.prev == null) 1653 return false; 1654 if (node.next != null) // If has successor, it must be on queue 1655 return true; 1656 /* 1657 * node.prev can be non-null, but not yet on queue because 1658 * the CAS to place it on queue can fail. So we have to 1659 * traverse from tail to make sure it actually made it. It 1660 * will always be near the tail in calls to this method, and 1661 * unless the CAS failed (which is unlikely), it will be 1662 * there, so we hardly ever traverse much. 1663 */ 1664 return findNodeFromTail(node); 1665 } 1666 1667 /** 1668 * Returns true if node is on sync queue by searching backwards from tail. 1669 * Called only when needed by isOnSyncQueue. 1670 * @return true if present 1671 */ findNodeFromTail(Node node)1672 private boolean findNodeFromTail(Node node) { 1673 // We check for node first, since it's likely to be at or near tail. 1674 // tail is known to be non-null, so we could re-order to "save" 1675 // one null check, but we leave it this way to help the VM. 1676 for (Node p = tail;;) { 1677 if (p == node) 1678 return true; 1679 if (p == null) 1680 return false; 1681 p = p.prev; 1682 } 1683 } 1684 1685 /** 1686 * Transfers a node from a condition queue onto sync queue. 1687 * Returns true if successful. 1688 * @param node the node 1689 * @return true if successfully transferred (else the node was 1690 * cancelled before signal) 1691 */ transferForSignal(Node node)1692 final boolean transferForSignal(Node node) { 1693 /* 1694 * If cannot change waitStatus, the node has been cancelled. 1695 */ 1696 if (!node.compareAndSetWaitStatus(Node.CONDITION, 0)) 1697 return false; 1698 1699 /* 1700 * Splice onto queue and try to set waitStatus of predecessor to 1701 * indicate that thread is (probably) waiting. If cancelled or 1702 * attempt to set waitStatus fails, wake up to resync (in which 1703 * case the waitStatus can be transiently and harmlessly wrong). 1704 */ 1705 Node p = enq(node); 1706 int ws = p.waitStatus; 1707 if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL)) 1708 LockSupport.unpark(node.thread); 1709 return true; 1710 } 1711 1712 /** 1713 * Transfers node, if necessary, to sync queue after a cancelled wait. 1714 * Returns true if thread was cancelled before being signalled. 1715 * 1716 * @param node the node 1717 * @return true if cancelled before the node was signalled 1718 */ transferAfterCancelledWait(Node node)1719 final boolean transferAfterCancelledWait(Node node) { 1720 if (node.compareAndSetWaitStatus(Node.CONDITION, 0)) { 1721 enq(node); 1722 return true; 1723 } 1724 /* 1725 * If we lost out to a signal(), then we can't proceed 1726 * until it finishes its enq(). Cancelling during an 1727 * incomplete transfer is both rare and transient, so just 1728 * spin. 1729 */ 1730 while (!isOnSyncQueue(node)) 1731 Thread.yield(); 1732 return false; 1733 } 1734 1735 /** 1736 * Invokes release with current state value; returns saved state. 1737 * Cancels node and throws exception on failure. 1738 * @param node the condition node for this wait 1739 * @return previous sync state 1740 */ fullyRelease(Node node)1741 final int fullyRelease(Node node) { 1742 try { 1743 int savedState = getState(); 1744 if (release(savedState)) 1745 return savedState; 1746 throw new IllegalMonitorStateException(); 1747 } catch (Throwable t) { 1748 node.waitStatus = Node.CANCELLED; 1749 throw t; 1750 } 1751 } 1752 1753 // Instrumentation methods for conditions 1754 1755 /** 1756 * Queries whether the given ConditionObject 1757 * uses this synchronizer as its lock. 1758 * 1759 * @param condition the condition 1760 * @return {@code true} if owned 1761 * @throws NullPointerException if the condition is null 1762 */ owns(ConditionObject condition)1763 public final boolean owns(ConditionObject condition) { 1764 return condition.isOwnedBy(this); 1765 } 1766 1767 /** 1768 * Queries whether any threads are waiting on the given condition 1769 * associated with this synchronizer. Note that because timeouts 1770 * and interrupts may occur at any time, a {@code true} return 1771 * does not guarantee that a future {@code signal} will awaken 1772 * any threads. This method is designed primarily for use in 1773 * monitoring of the system state. 1774 * 1775 * @param condition the condition 1776 * @return {@code true} if there are any waiting threads 1777 * @throws IllegalMonitorStateException if exclusive synchronization 1778 * is not held 1779 * @throws IllegalArgumentException if the given condition is 1780 * not associated with this synchronizer 1781 * @throws NullPointerException if the condition is null 1782 */ hasWaiters(ConditionObject condition)1783 public final boolean hasWaiters(ConditionObject condition) { 1784 if (!owns(condition)) 1785 throw new IllegalArgumentException("Not owner"); 1786 return condition.hasWaiters(); 1787 } 1788 1789 /** 1790 * Returns an estimate of the number of threads waiting on the 1791 * given condition associated with this synchronizer. Note that 1792 * because timeouts and interrupts may occur at any time, the 1793 * estimate serves only as an upper bound on the actual number of 1794 * waiters. This method is designed for use in monitoring system 1795 * state, not for synchronization control. 1796 * 1797 * @param condition the condition 1798 * @return the estimated number of waiting threads 1799 * @throws IllegalMonitorStateException if exclusive synchronization 1800 * is not held 1801 * @throws IllegalArgumentException if the given condition is 1802 * not associated with this synchronizer 1803 * @throws NullPointerException if the condition is null 1804 */ getWaitQueueLength(ConditionObject condition)1805 public final int getWaitQueueLength(ConditionObject condition) { 1806 if (!owns(condition)) 1807 throw new IllegalArgumentException("Not owner"); 1808 return condition.getWaitQueueLength(); 1809 } 1810 1811 /** 1812 * Returns a collection containing those threads that may be 1813 * waiting on the given condition associated with this 1814 * synchronizer. Because the actual set of threads may change 1815 * dynamically while constructing this result, the returned 1816 * collection is only a best-effort estimate. The elements of the 1817 * returned collection are in no particular order. 1818 * 1819 * @param condition the condition 1820 * @return the collection of threads 1821 * @throws IllegalMonitorStateException if exclusive synchronization 1822 * is not held 1823 * @throws IllegalArgumentException if the given condition is 1824 * not associated with this synchronizer 1825 * @throws NullPointerException if the condition is null 1826 */ getWaitingThreads(ConditionObject condition)1827 public final Collection<Thread> getWaitingThreads(ConditionObject condition) { 1828 if (!owns(condition)) 1829 throw new IllegalArgumentException("Not owner"); 1830 return condition.getWaitingThreads(); 1831 } 1832 1833 /** 1834 * Condition implementation for a {@link 1835 * AbstractQueuedSynchronizer} serving as the basis of a {@link 1836 * Lock} implementation. 1837 * 1838 * <p>Method documentation for this class describes mechanics, 1839 * not behavioral specifications from the point of view of Lock 1840 * and Condition users. Exported versions of this class will in 1841 * general need to be accompanied by documentation describing 1842 * condition semantics that rely on those of the associated 1843 * {@code AbstractQueuedSynchronizer}. 1844 * 1845 * <p>This class is Serializable, but all fields are transient, 1846 * so deserialized conditions have no waiters. 1847 */ 1848 public class ConditionObject implements Condition, java.io.Serializable { 1849 private static final long serialVersionUID = 1173984872572414699L; 1850 /** First node of condition queue. */ 1851 private transient Node firstWaiter; 1852 /** Last node of condition queue. */ 1853 private transient Node lastWaiter; 1854 1855 /** 1856 * Creates a new {@code ConditionObject} instance. 1857 */ ConditionObject()1858 public ConditionObject() { } 1859 1860 // Internal methods 1861 1862 /** 1863 * Adds a new waiter to wait queue. 1864 * @return its new wait node 1865 */ addConditionWaiter()1866 private Node addConditionWaiter() { 1867 Node t = lastWaiter; 1868 // If lastWaiter is cancelled, clean out. 1869 if (t != null && t.waitStatus != Node.CONDITION) { 1870 unlinkCancelledWaiters(); 1871 t = lastWaiter; 1872 } 1873 1874 Node node = new Node(Node.CONDITION); 1875 1876 if (t == null) 1877 firstWaiter = node; 1878 else 1879 t.nextWaiter = node; 1880 lastWaiter = node; 1881 return node; 1882 } 1883 1884 /** 1885 * Removes and transfers nodes until hit non-cancelled one or 1886 * null. Split out from signal in part to encourage compilers 1887 * to inline the case of no waiters. 1888 * @param first (non-null) the first node on condition queue 1889 */ doSignal(Node first)1890 private void doSignal(Node first) { 1891 do { 1892 if ( (firstWaiter = first.nextWaiter) == null) 1893 lastWaiter = null; 1894 first.nextWaiter = null; 1895 } while (!transferForSignal(first) && 1896 (first = firstWaiter) != null); 1897 } 1898 1899 /** 1900 * Removes and transfers all nodes. 1901 * @param first (non-null) the first node on condition queue 1902 */ doSignalAll(Node first)1903 private void doSignalAll(Node first) { 1904 lastWaiter = firstWaiter = null; 1905 do { 1906 Node next = first.nextWaiter; 1907 first.nextWaiter = null; 1908 transferForSignal(first); 1909 first = next; 1910 } while (first != null); 1911 } 1912 1913 /** 1914 * Unlinks cancelled waiter nodes from condition queue. 1915 * Called only while holding lock. This is called when 1916 * cancellation occurred during condition wait, and upon 1917 * insertion of a new waiter when lastWaiter is seen to have 1918 * been cancelled. This method is needed to avoid garbage 1919 * retention in the absence of signals. So even though it may 1920 * require a full traversal, it comes into play only when 1921 * timeouts or cancellations occur in the absence of 1922 * signals. It traverses all nodes rather than stopping at a 1923 * particular target to unlink all pointers to garbage nodes 1924 * without requiring many re-traversals during cancellation 1925 * storms. 1926 */ unlinkCancelledWaiters()1927 private void unlinkCancelledWaiters() { 1928 Node t = firstWaiter; 1929 Node trail = null; 1930 while (t != null) { 1931 Node next = t.nextWaiter; 1932 if (t.waitStatus != Node.CONDITION) { 1933 t.nextWaiter = null; 1934 if (trail == null) 1935 firstWaiter = next; 1936 else 1937 trail.nextWaiter = next; 1938 if (next == null) 1939 lastWaiter = trail; 1940 } 1941 else 1942 trail = t; 1943 t = next; 1944 } 1945 } 1946 1947 // public methods 1948 1949 /** 1950 * Moves the longest-waiting thread, if one exists, from the 1951 * wait queue for this condition to the wait queue for the 1952 * owning lock. 1953 * 1954 * @throws IllegalMonitorStateException if {@link #isHeldExclusively} 1955 * returns {@code false} 1956 */ signal()1957 public final void signal() { 1958 if (!isHeldExclusively()) 1959 throw new IllegalMonitorStateException(); 1960 Node first = firstWaiter; 1961 if (first != null) 1962 doSignal(first); 1963 } 1964 1965 /** 1966 * Moves all threads from the wait queue for this condition to 1967 * the wait queue for the owning lock. 1968 * 1969 * @throws IllegalMonitorStateException if {@link #isHeldExclusively} 1970 * returns {@code false} 1971 */ signalAll()1972 public final void signalAll() { 1973 if (!isHeldExclusively()) 1974 throw new IllegalMonitorStateException(); 1975 Node first = firstWaiter; 1976 if (first != null) 1977 doSignalAll(first); 1978 } 1979 1980 /** 1981 * Implements uninterruptible condition wait. 1982 * <ol> 1983 * <li>Save lock state returned by {@link #getState}. 1984 * <li>Invoke {@link #release} with saved state as argument, 1985 * throwing IllegalMonitorStateException if it fails. 1986 * <li>Block until signalled. 1987 * <li>Reacquire by invoking specialized version of 1988 * {@link #acquire} with saved state as argument. 1989 * </ol> 1990 */ awaitUninterruptibly()1991 public final void awaitUninterruptibly() { 1992 Node node = addConditionWaiter(); 1993 int savedState = fullyRelease(node); 1994 boolean interrupted = false; 1995 while (!isOnSyncQueue(node)) { 1996 LockSupport.park(this); 1997 if (Thread.interrupted()) 1998 interrupted = true; 1999 } 2000 if (acquireQueued(node, savedState) || interrupted) 2001 selfInterrupt(); 2002 } 2003 2004 /* 2005 * For interruptible waits, we need to track whether to throw 2006 * InterruptedException, if interrupted while blocked on 2007 * condition, versus reinterrupt current thread, if 2008 * interrupted while blocked waiting to re-acquire. 2009 */ 2010 2011 /** Mode meaning to reinterrupt on exit from wait */ 2012 private static final int REINTERRUPT = 1; 2013 /** Mode meaning to throw InterruptedException on exit from wait */ 2014 private static final int THROW_IE = -1; 2015 2016 /** 2017 * Checks for interrupt, returning THROW_IE if interrupted 2018 * before signalled, REINTERRUPT if after signalled, or 2019 * 0 if not interrupted. 2020 */ checkInterruptWhileWaiting(Node node)2021 private int checkInterruptWhileWaiting(Node node) { 2022 return Thread.interrupted() ? 2023 (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) : 2024 0; 2025 } 2026 2027 /** 2028 * Throws InterruptedException, reinterrupts current thread, or 2029 * does nothing, depending on mode. 2030 */ reportInterruptAfterWait(int interruptMode)2031 private void reportInterruptAfterWait(int interruptMode) 2032 throws InterruptedException { 2033 if (interruptMode == THROW_IE) 2034 throw new InterruptedException(); 2035 else if (interruptMode == REINTERRUPT) 2036 selfInterrupt(); 2037 } 2038 2039 /** 2040 * Implements interruptible condition wait. 2041 * <ol> 2042 * <li>If current thread is interrupted, throw InterruptedException. 2043 * <li>Save lock state returned by {@link #getState}. 2044 * <li>Invoke {@link #release} with saved state as argument, 2045 * throwing IllegalMonitorStateException if it fails. 2046 * <li>Block until signalled or interrupted. 2047 * <li>Reacquire by invoking specialized version of 2048 * {@link #acquire} with saved state as argument. 2049 * <li>If interrupted while blocked in step 4, throw InterruptedException. 2050 * </ol> 2051 */ await()2052 public final void await() throws InterruptedException { 2053 if (Thread.interrupted()) 2054 throw new InterruptedException(); 2055 Node node = addConditionWaiter(); 2056 int savedState = fullyRelease(node); 2057 int interruptMode = 0; 2058 while (!isOnSyncQueue(node)) { 2059 LockSupport.park(this); 2060 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) 2061 break; 2062 } 2063 if (acquireQueued(node, savedState) && interruptMode != THROW_IE) 2064 interruptMode = REINTERRUPT; 2065 if (node.nextWaiter != null) // clean up if cancelled 2066 unlinkCancelledWaiters(); 2067 if (interruptMode != 0) 2068 reportInterruptAfterWait(interruptMode); 2069 } 2070 2071 /** 2072 * Implements timed condition wait. 2073 * <ol> 2074 * <li>If current thread is interrupted, throw InterruptedException. 2075 * <li>Save lock state returned by {@link #getState}. 2076 * <li>Invoke {@link #release} with saved state as argument, 2077 * throwing IllegalMonitorStateException if it fails. 2078 * <li>Block until signalled, interrupted, or timed out. 2079 * <li>Reacquire by invoking specialized version of 2080 * {@link #acquire} with saved state as argument. 2081 * <li>If interrupted while blocked in step 4, throw InterruptedException. 2082 * </ol> 2083 */ awaitNanos(long nanosTimeout)2084 public final long awaitNanos(long nanosTimeout) 2085 throws InterruptedException { 2086 if (Thread.interrupted()) 2087 throw new InterruptedException(); 2088 // We don't check for nanosTimeout <= 0L here, to allow 2089 // awaitNanos(0) as a way to "yield the lock". 2090 final long deadline = System.nanoTime() + nanosTimeout; 2091 long initialNanos = nanosTimeout; 2092 Node node = addConditionWaiter(); 2093 int savedState = fullyRelease(node); 2094 int interruptMode = 0; 2095 while (!isOnSyncQueue(node)) { 2096 if (nanosTimeout <= 0L) { 2097 transferAfterCancelledWait(node); 2098 break; 2099 } 2100 if (nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD) 2101 LockSupport.parkNanos(this, nanosTimeout); 2102 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) 2103 break; 2104 nanosTimeout = deadline - System.nanoTime(); 2105 } 2106 if (acquireQueued(node, savedState) && interruptMode != THROW_IE) 2107 interruptMode = REINTERRUPT; 2108 if (node.nextWaiter != null) 2109 unlinkCancelledWaiters(); 2110 if (interruptMode != 0) 2111 reportInterruptAfterWait(interruptMode); 2112 long remaining = deadline - System.nanoTime(); // avoid overflow 2113 return (remaining <= initialNanos) ? remaining : Long.MIN_VALUE; 2114 } 2115 2116 /** 2117 * Implements absolute timed condition wait. 2118 * <ol> 2119 * <li>If current thread is interrupted, throw InterruptedException. 2120 * <li>Save lock state returned by {@link #getState}. 2121 * <li>Invoke {@link #release} with saved state as argument, 2122 * throwing IllegalMonitorStateException if it fails. 2123 * <li>Block until signalled, interrupted, or timed out. 2124 * <li>Reacquire by invoking specialized version of 2125 * {@link #acquire} with saved state as argument. 2126 * <li>If interrupted while blocked in step 4, throw InterruptedException. 2127 * <li>If timed out while blocked in step 4, return false, else true. 2128 * </ol> 2129 */ awaitUntil(Date deadline)2130 public final boolean awaitUntil(Date deadline) 2131 throws InterruptedException { 2132 long abstime = deadline.getTime(); 2133 if (Thread.interrupted()) 2134 throw new InterruptedException(); 2135 Node node = addConditionWaiter(); 2136 int savedState = fullyRelease(node); 2137 boolean timedout = false; 2138 int interruptMode = 0; 2139 while (!isOnSyncQueue(node)) { 2140 if (System.currentTimeMillis() >= abstime) { 2141 timedout = transferAfterCancelledWait(node); 2142 break; 2143 } 2144 LockSupport.parkUntil(this, abstime); 2145 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) 2146 break; 2147 } 2148 if (acquireQueued(node, savedState) && interruptMode != THROW_IE) 2149 interruptMode = REINTERRUPT; 2150 if (node.nextWaiter != null) 2151 unlinkCancelledWaiters(); 2152 if (interruptMode != 0) 2153 reportInterruptAfterWait(interruptMode); 2154 return !timedout; 2155 } 2156 2157 /** 2158 * Implements timed condition wait. 2159 * <ol> 2160 * <li>If current thread is interrupted, throw InterruptedException. 2161 * <li>Save lock state returned by {@link #getState}. 2162 * <li>Invoke {@link #release} with saved state as argument, 2163 * throwing IllegalMonitorStateException if it fails. 2164 * <li>Block until signalled, interrupted, or timed out. 2165 * <li>Reacquire by invoking specialized version of 2166 * {@link #acquire} with saved state as argument. 2167 * <li>If interrupted while blocked in step 4, throw InterruptedException. 2168 * <li>If timed out while blocked in step 4, return false, else true. 2169 * </ol> 2170 */ await(long time, TimeUnit unit)2171 public final boolean await(long time, TimeUnit unit) 2172 throws InterruptedException { 2173 long nanosTimeout = unit.toNanos(time); 2174 if (Thread.interrupted()) 2175 throw new InterruptedException(); 2176 // We don't check for nanosTimeout <= 0L here, to allow 2177 // await(0, unit) as a way to "yield the lock". 2178 final long deadline = System.nanoTime() + nanosTimeout; 2179 Node node = addConditionWaiter(); 2180 int savedState = fullyRelease(node); 2181 boolean timedout = false; 2182 int interruptMode = 0; 2183 while (!isOnSyncQueue(node)) { 2184 if (nanosTimeout <= 0L) { 2185 timedout = transferAfterCancelledWait(node); 2186 break; 2187 } 2188 if (nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD) 2189 LockSupport.parkNanos(this, nanosTimeout); 2190 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) 2191 break; 2192 nanosTimeout = deadline - System.nanoTime(); 2193 } 2194 if (acquireQueued(node, savedState) && interruptMode != THROW_IE) 2195 interruptMode = REINTERRUPT; 2196 if (node.nextWaiter != null) 2197 unlinkCancelledWaiters(); 2198 if (interruptMode != 0) 2199 reportInterruptAfterWait(interruptMode); 2200 return !timedout; 2201 } 2202 2203 // support for instrumentation 2204 2205 /** 2206 * Returns true if this condition was created by the given 2207 * synchronization object. 2208 * 2209 * @return {@code true} if owned 2210 */ isOwnedBy(AbstractQueuedSynchronizer sync)2211 final boolean isOwnedBy(AbstractQueuedSynchronizer sync) { 2212 return sync == AbstractQueuedSynchronizer.this; 2213 } 2214 2215 /** 2216 * Queries whether any threads are waiting on this condition. 2217 * Implements {@link AbstractQueuedSynchronizer#hasWaiters(ConditionObject)}. 2218 * 2219 * @return {@code true} if there are any waiting threads 2220 * @throws IllegalMonitorStateException if {@link #isHeldExclusively} 2221 * returns {@code false} 2222 */ hasWaiters()2223 protected final boolean hasWaiters() { 2224 if (!isHeldExclusively()) 2225 throw new IllegalMonitorStateException(); 2226 for (Node w = firstWaiter; w != null; w = w.nextWaiter) { 2227 if (w.waitStatus == Node.CONDITION) 2228 return true; 2229 } 2230 return false; 2231 } 2232 2233 /** 2234 * Returns an estimate of the number of threads waiting on 2235 * this condition. 2236 * Implements {@link AbstractQueuedSynchronizer#getWaitQueueLength(ConditionObject)}. 2237 * 2238 * @return the estimated number of waiting threads 2239 * @throws IllegalMonitorStateException if {@link #isHeldExclusively} 2240 * returns {@code false} 2241 */ getWaitQueueLength()2242 protected final int getWaitQueueLength() { 2243 if (!isHeldExclusively()) 2244 throw new IllegalMonitorStateException(); 2245 int n = 0; 2246 for (Node w = firstWaiter; w != null; w = w.nextWaiter) { 2247 if (w.waitStatus == Node.CONDITION) 2248 ++n; 2249 } 2250 return n; 2251 } 2252 2253 /** 2254 * Returns a collection containing those threads that may be 2255 * waiting on this Condition. 2256 * Implements {@link AbstractQueuedSynchronizer#getWaitingThreads(ConditionObject)}. 2257 * 2258 * @return the collection of threads 2259 * @throws IllegalMonitorStateException if {@link #isHeldExclusively} 2260 * returns {@code false} 2261 */ getWaitingThreads()2262 protected final Collection<Thread> getWaitingThreads() { 2263 if (!isHeldExclusively()) 2264 throw new IllegalMonitorStateException(); 2265 ArrayList<Thread> list = new ArrayList<>(); 2266 for (Node w = firstWaiter; w != null; w = w.nextWaiter) { 2267 if (w.waitStatus == Node.CONDITION) { 2268 Thread t = w.thread; 2269 if (t != null) 2270 list.add(t); 2271 } 2272 } 2273 return list; 2274 } 2275 } 2276 2277 /** 2278 * Setup to support compareAndSet. We need to natively implement 2279 * this here: For the sake of permitting future enhancements, we 2280 * cannot explicitly subclass AtomicInteger, which would be 2281 * efficient and useful otherwise. So, as the lesser of evils, we 2282 * natively implement using hotspot intrinsics API. And while we 2283 * are at it, we do the same for other CASable fields (which could 2284 * otherwise be done with atomic field updaters). 2285 */ 2286 private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe(); 2287 private static final long STATE; 2288 private static final long HEAD; 2289 private static final long TAIL; 2290 2291 static { 2292 try { 2293 STATE = U.objectFieldOffset 2294 (AbstractQueuedSynchronizer.class.getDeclaredField("state")); 2295 HEAD = U.objectFieldOffset 2296 (AbstractQueuedSynchronizer.class.getDeclaredField("head")); 2297 TAIL = U.objectFieldOffset 2298 (AbstractQueuedSynchronizer.class.getDeclaredField("tail")); 2299 } catch (ReflectiveOperationException e) { 2300 throw new Error(e); 2301 } 2302 2303 // Reduce the risk of rare disastrous classloading in first call to 2304 // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773 2305 Class<?> ensureLoaded = LockSupport.class; 2306 } 2307 2308 /** 2309 * Initializes head and tail fields on first contention. 2310 */ initializeSyncQueue()2311 private final void initializeSyncQueue() { 2312 Node h; 2313 if (U.compareAndSwapObject(this, HEAD, null, (h = new Node()))) 2314 tail = h; 2315 } 2316 2317 /** 2318 * CASes tail field. 2319 */ compareAndSetTail(Node expect, Node update)2320 private final boolean compareAndSetTail(Node expect, Node update) { 2321 return U.compareAndSwapObject(this, TAIL, expect, update); 2322 } 2323 } 2324