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.lang.invoke.MethodHandles;
39 import java.lang.invoke.VarHandle;
40 import java.util.ArrayList;
41 import java.util.Collection;
42 import java.util.Date;
43 import java.util.concurrent.TimeUnit;
44 import java.util.concurrent.locks.AbstractQueuedSynchronizer.Node;
45
46 /**
47 * A version of {@link AbstractQueuedSynchronizer} in
48 * which synchronization state is maintained as a {@code long}.
49 * This class has exactly the same structure, properties, and methods
50 * as {@code AbstractQueuedSynchronizer} with the exception
51 * that all state-related parameters and results are defined
52 * as {@code long} rather than {@code int}. This class
53 * may be useful when creating synchronizers such as
54 * multilevel locks and barriers that require
55 * 64 bits of state.
56 *
57 * <p>See {@link AbstractQueuedSynchronizer} for usage
58 * notes and examples.
59 *
60 * @since 1.6
61 * @author Doug Lea
62 */
63 public abstract class AbstractQueuedLongSynchronizer
64 extends AbstractOwnableSynchronizer
65 implements java.io.Serializable {
66
67 private static final long serialVersionUID = 7373984972572414692L;
68
69 /*
70 * To keep sources in sync, the remainder of this source file is
71 * exactly cloned from AbstractQueuedSynchronizer, replacing class
72 * name and changing ints related with sync state to longs. Please
73 * keep it that way.
74 */
75
76 /**
77 * Creates a new {@code AbstractQueuedLongSynchronizer} instance
78 * with initial synchronization state of zero.
79 */
80 protected AbstractQueuedLongSynchronizer() { }
81
82 /**
83 * Head of the wait queue, lazily initialized. Except for
84 * initialization, it is modified only via method setHead. Note:
85 * If head exists, its waitStatus is guaranteed not to be
86 * CANCELLED.
87 */
88 private transient volatile Node head;
89
90 /**
91 * Tail of the wait queue, lazily initialized. Modified only via
92 * method enq to add new wait node.
93 */
94 private transient volatile Node tail;
95
96 /**
97 * The synchronization state.
98 */
99 private volatile long state;
100
101 /**
102 * Returns the current value of synchronization state.
103 * This operation has memory semantics of a {@code volatile} read.
104 * @return current state value
105 */
106 protected final long getState() {
107 return state;
108 }
109
110 /**
111 * Sets the value of synchronization state.
112 * This operation has memory semantics of a {@code volatile} write.
113 * @param newState the new state value
114 */
115 protected final void setState(long newState) {
116 // See JDK-8180620: Clarify VarHandle mixed-access subtleties
117 STATE.setVolatile(this, newState);
118 }
119
120 /**
121 * Atomically sets synchronization state to the given updated
122 * value if the current state value equals the expected value.
123 * This operation has memory semantics of a {@code volatile} read
124 * and write.
125 *
126 * @param expect the expected value
127 * @param update the new value
128 * @return {@code true} if successful. False return indicates that the actual
129 * value was not equal to the expected value.
130 */
131 protected final boolean compareAndSetState(long expect, long update) {
132 return STATE.compareAndSet(this, expect, update);
133 }
134
135 // Queuing utilities
136
137 /**
138 * The number of nanoseconds for which it is faster to spin
139 * rather than to use timed park. A rough estimate suffices
140 * to improve responsiveness with very short timeouts.
141 */
142 static final long SPIN_FOR_TIMEOUT_THRESHOLD = 1000L;
143
144 /**
145 * Inserts node into queue, initializing if necessary. See picture above.
146 * @param node the node to insert
147 * @return node's predecessor
148 */
149 private Node enq(Node node) {
150 for (;;) {
151 Node oldTail = tail;
152 if (oldTail != null) {
153 node.setPrevRelaxed(oldTail);
154 if (compareAndSetTail(oldTail, node)) {
155 oldTail.next = node;
156 return oldTail;
157 }
158 } else {
159 initializeSyncQueue();
160 }
161 }
162 }
163
164 /**
165 * Creates and enqueues node for current thread and given mode.
166 *
167 * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
168 * @return the new node
169 */
170 private Node addWaiter(Node mode) {
171 Node node = new Node(mode);
172
173 for (;;) {
174 Node oldTail = tail;
175 if (oldTail != null) {
176 node.setPrevRelaxed(oldTail);
177 if (compareAndSetTail(oldTail, node)) {
178 oldTail.next = node;
179 return node;
180 }
181 } else {
182 initializeSyncQueue();
183 }
184 }
185 }
186
187 /**
188 * Sets head of queue to be node, thus dequeuing. Called only by
189 * acquire methods. Also nulls out unused fields for sake of GC
190 * and to suppress unnecessary signals and traversals.
191 *
192 * @param node the node
193 */
194 private void setHead(Node node) {
195 head = node;
196 node.thread = null;
197 node.prev = null;
198 }
199
200 /**
201 * Wakes up node's successor, if one exists.
202 *
203 * @param node the node
204 */
205 private void unparkSuccessor(Node node) {
206 /*
207 * If status is negative (i.e., possibly needing signal) try
208 * to clear in anticipation of signalling. It is OK if this
209 * fails or if status is changed by waiting thread.
210 */
211 int ws = node.waitStatus;
212 if (ws < 0)
213 node.compareAndSetWaitStatus(ws, 0);
214
215 /*
216 * Thread to unpark is held in successor, which is normally
217 * just the next node. But if cancelled or apparently null,
218 * traverse backwards from tail to find the actual
219 * non-cancelled successor.
220 */
221 Node s = node.next;
222 if (s == null || s.waitStatus > 0) {
223 s = null;
224 for (Node p = tail; p != node && p != null; p = p.prev)
225 if (p.waitStatus <= 0)
226 s = p;
227 }
228 if (s != null)
229 LockSupport.unpark(s.thread);
230 }
231
232 /**
233 * Release action for shared mode -- signals successor and ensures
234 * propagation. (Note: For exclusive mode, release just amounts
235 * to calling unparkSuccessor of head if it needs signal.)
236 */
237 private void doReleaseShared() {
238 /*
239 * Ensure that a release propagates, even if there are other
240 * in-progress acquires/releases. This proceeds in the usual
241 * way of trying to unparkSuccessor of head if it needs
242 * signal. But if it does not, status is set to PROPAGATE to
243 * ensure that upon release, propagation continues.
244 * Additionally, we must loop in case a new node is added
245 * while we are doing this. Also, unlike other uses of
246 * unparkSuccessor, we need to know if CAS to reset status
247 * fails, if so rechecking.
248 */
249 for (;;) {
250 Node h = head;
251 if (h != null && h != tail) {
252 int ws = h.waitStatus;
253 if (ws == Node.SIGNAL) {
254 if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0))
255 continue; // loop to recheck cases
256 unparkSuccessor(h);
257 }
258 else if (ws == 0 &&
259 !h.compareAndSetWaitStatus(0, Node.PROPAGATE))
260 continue; // loop on failed CAS
261 }
262 if (h == head) // loop if head changed
263 break;
264 }
265 }
266
267 /**
268 * Sets head of queue, and checks if successor may be waiting
269 * in shared mode, if so propagating if either propagate > 0 or
270 * PROPAGATE status was set.
271 *
272 * @param node the node
273 * @param propagate the return value from a tryAcquireShared
274 */
275 private void setHeadAndPropagate(Node node, long propagate) {
276 Node h = head; // Record old head for check below
277 setHead(node);
278 /*
279 * Try to signal next queued node if:
280 * Propagation was indicated by caller,
281 * or was recorded (as h.waitStatus either before
282 * or after setHead) by a previous operation
283 * (note: this uses sign-check of waitStatus because
284 * PROPAGATE status may transition to SIGNAL.)
285 * and
286 * The next node is waiting in shared mode,
287 * or we don't know, because it appears null
288 *
289 * The conservatism in both of these checks may cause
290 * unnecessary wake-ups, but only when there are multiple
291 * racing acquires/releases, so most need signals now or soon
292 * anyway.
293 */
294 if (propagate > 0 || h == null || h.waitStatus < 0 ||
295 (h = head) == null || h.waitStatus < 0) {
296 Node s = node.next;
297 if (s == null || s.isShared())
298 doReleaseShared();
299 }
300 }
301
302 // Utilities for various versions of acquire
303
304 /**
305 * Cancels an ongoing attempt to acquire.
306 *
307 * @param node the node
308 */
309 private void cancelAcquire(Node node) {
310 // Ignore if node doesn't exist
311 if (node == null)
312 return;
313
314 node.thread = null;
315
316 // Skip cancelled predecessors
317 Node pred = node.prev;
318 while (pred.waitStatus > 0)
319 node.prev = pred = pred.prev;
320
321 // predNext is the apparent node to unsplice. CASes below will
322 // fail if not, in which case, we lost race vs another cancel
323 // or signal, so no further action is necessary, although with
324 // a possibility that a cancelled node may transiently remain
325 // reachable.
326 Node predNext = pred.next;
327
328 // Can use unconditional write instead of CAS here.
329 // After this atomic step, other Nodes can skip past us.
330 // Before, we are free of interference from other threads.
331 node.waitStatus = Node.CANCELLED;
332
333 // If we are the tail, remove ourselves.
334 if (node == tail && compareAndSetTail(node, pred)) {
335 pred.compareAndSetNext(predNext, null);
336 } else {
337 // If successor needs signal, try to set pred's next-link
338 // so it will get one. Otherwise wake it up to propagate.
339 int ws;
340 if (pred != head &&
341 ((ws = pred.waitStatus) == Node.SIGNAL ||
342 (ws <= 0 && pred.compareAndSetWaitStatus(ws, Node.SIGNAL))) &&
343 pred.thread != null) {
344 Node next = node.next;
345 if (next != null && next.waitStatus <= 0)
346 pred.compareAndSetNext(predNext, next);
347 } else {
348 unparkSuccessor(node);
349 }
350
351 node.next = node; // help GC
352 }
353 }
354
355 /**
356 * Checks and updates status for a node that failed to acquire.
357 * Returns true if thread should block. This is the main signal
358 * control in all acquire loops. Requires that pred == node.prev.
359 *
360 * @param pred node's predecessor holding status
361 * @param node the node
362 * @return {@code true} if thread should block
363 */
364 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
365 int ws = pred.waitStatus;
366 if (ws == Node.SIGNAL)
367 /*
368 * This node has already set status asking a release
369 * to signal it, so it can safely park.
370 */
371 return true;
372 if (ws > 0) {
373 /*
374 * Predecessor was cancelled. Skip over predecessors and
375 * indicate retry.
376 */
377 do {
378 node.prev = pred = pred.prev;
379 } while (pred.waitStatus > 0);
380 pred.next = node;
381 } else {
382 /*
383 * waitStatus must be 0 or PROPAGATE. Indicate that we
384 * need a signal, but don't park yet. Caller will need to
385 * retry to make sure it cannot acquire before parking.
386 */
387 pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
388 }
389 return false;
390 }
391
392 /**
393 * Convenience method to interrupt current thread.
394 */
395 static void selfInterrupt() {
396 Thread.currentThread().interrupt();
397 }
398
399 /**
400 * Convenience method to park and then check if interrupted.
401 *
402 * @return {@code true} if interrupted
403 */
404 private final boolean parkAndCheckInterrupt() {
405 LockSupport.park(this);
406 return Thread.interrupted();
407 }
408
409 /*
410 * Various flavors of acquire, varying in exclusive/shared and
411 * control modes. Each is mostly the same, but annoyingly
412 * different. Only a little bit of factoring is possible due to
413 * interactions of exception mechanics (including ensuring that we
414 * cancel if tryAcquire throws exception) and other control, at
415 * least not without hurting performance too much.
416 */
417
418 /**
419 * Acquires in exclusive uninterruptible mode for thread already in
420 * queue. Used by condition wait methods as well as acquire.
421 *
422 * @param node the node
423 * @param arg the acquire argument
424 * @return {@code true} if interrupted while waiting
425 */
426 final boolean acquireQueued(final Node node, long arg) {
427 boolean interrupted = false;
428 try {
429 for (;;) {
430 final Node p = node.predecessor();
431 if (p == head && tryAcquire(arg)) {
432 setHead(node);
433 p.next = null; // help GC
434 return interrupted;
435 }
436 if (shouldParkAfterFailedAcquire(p, node))
437 interrupted |= parkAndCheckInterrupt();
438 }
439 } catch (Throwable t) {
440 cancelAcquire(node);
441 if (interrupted)
442 selfInterrupt();
443 throw t;
444 }
445 }
446
447 /**
448 * Acquires in exclusive interruptible mode.
449 * @param arg the acquire argument
450 */
451 private void doAcquireInterruptibly(long arg)
452 throws InterruptedException {
453 final Node node = addWaiter(Node.EXCLUSIVE);
454 try {
455 for (;;) {
456 final Node p = node.predecessor();
457 if (p == head && tryAcquire(arg)) {
458 setHead(node);
459 p.next = null; // help GC
460 return;
461 }
462 if (shouldParkAfterFailedAcquire(p, node) &&
463 parkAndCheckInterrupt())
464 throw new InterruptedException();
465 }
466 } catch (Throwable t) {
467 cancelAcquire(node);
468 throw t;
469 }
470 }
471
472 /**
473 * Acquires in exclusive timed mode.
474 *
475 * @param arg the acquire argument
476 * @param nanosTimeout max wait time
477 * @return {@code true} if acquired
478 */
479 private boolean doAcquireNanos(long arg, long nanosTimeout)
480 throws InterruptedException {
481 if (nanosTimeout <= 0L)
482 return false;
483 final long deadline = System.nanoTime() + nanosTimeout;
484 final Node node = addWaiter(Node.EXCLUSIVE);
485 try {
486 for (;;) {
487 final Node p = node.predecessor();
488 if (p == head && tryAcquire(arg)) {
489 setHead(node);
490 p.next = null; // help GC
491 return true;
492 }
493 nanosTimeout = deadline - System.nanoTime();
494 if (nanosTimeout <= 0L) {
495 cancelAcquire(node);
496 return false;
497 }
498 if (shouldParkAfterFailedAcquire(p, node) &&
499 nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD)
500 LockSupport.parkNanos(this, nanosTimeout);
501 if (Thread.interrupted())
502 throw new InterruptedException();
503 }
504 } catch (Throwable t) {
505 cancelAcquire(node);
506 throw t;
507 }
508 }
509
510 /**
511 * Acquires in shared uninterruptible mode.
512 * @param arg the acquire argument
513 */
514 private void doAcquireShared(long arg) {
515 final Node node = addWaiter(Node.SHARED);
516 boolean interrupted = false;
517 try {
518 for (;;) {
519 final Node p = node.predecessor();
520 if (p == head) {
521 long r = tryAcquireShared(arg);
522 if (r >= 0) {
523 setHeadAndPropagate(node, r);
524 p.next = null; // help GC
525 return;
526 }
527 }
528 if (shouldParkAfterFailedAcquire(p, node))
529 interrupted |= parkAndCheckInterrupt();
530 }
531 } catch (Throwable t) {
532 cancelAcquire(node);
533 throw t;
534 } finally {
535 if (interrupted)
536 selfInterrupt();
537 }
538 }
539
540 /**
541 * Acquires in shared interruptible mode.
542 * @param arg the acquire argument
543 */
544 private void doAcquireSharedInterruptibly(long arg)
545 throws InterruptedException {
546 final Node node = addWaiter(Node.SHARED);
547 try {
548 for (;;) {
549 final Node p = node.predecessor();
550 if (p == head) {
551 long r = tryAcquireShared(arg);
552 if (r >= 0) {
553 setHeadAndPropagate(node, r);
554 p.next = null; // help GC
555 return;
556 }
557 }
558 if (shouldParkAfterFailedAcquire(p, node) &&
559 parkAndCheckInterrupt())
560 throw new InterruptedException();
561 }
562 } catch (Throwable t) {
563 cancelAcquire(node);
564 throw t;
565 }
566 }
567
568 /**
569 * Acquires in shared timed mode.
570 *
571 * @param arg the acquire argument
572 * @param nanosTimeout max wait time
573 * @return {@code true} if acquired
574 */
575 private boolean doAcquireSharedNanos(long arg, long nanosTimeout)
576 throws InterruptedException {
577 if (nanosTimeout <= 0L)
578 return false;
579 final long deadline = System.nanoTime() + nanosTimeout;
580 final Node node = addWaiter(Node.SHARED);
581 try {
582 for (;;) {
583 final Node p = node.predecessor();
584 if (p == head) {
585 long r = tryAcquireShared(arg);
586 if (r >= 0) {
587 setHeadAndPropagate(node, r);
588 p.next = null; // help GC
589 return true;
590 }
591 }
592 nanosTimeout = deadline - System.nanoTime();
593 if (nanosTimeout <= 0L) {
594 cancelAcquire(node);
595 return false;
596 }
597 if (shouldParkAfterFailedAcquire(p, node) &&
598 nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD)
599 LockSupport.parkNanos(this, nanosTimeout);
600 if (Thread.interrupted())
601 throw new InterruptedException();
602 }
603 } catch (Throwable t) {
604 cancelAcquire(node);
605 throw t;
606 }
607 }
608
609 // Main exported methods
610
611 /**
612 * Attempts to acquire in exclusive mode. This method should query
613 * if the state of the object permits it to be acquired in the
614 * exclusive mode, and if so to acquire it.
615 *
616 * <p>This method is always invoked by the thread performing
617 * acquire. If this method reports failure, the acquire method
618 * may queue the thread, if it is not already queued, until it is
619 * signalled by a release from some other thread. This can be used
620 * to implement method {@link Lock#tryLock()}.
621 *
622 * <p>The default
623 * implementation throws {@link UnsupportedOperationException}.
624 *
625 * @param arg the acquire argument. This value is always the one
626 * passed to an acquire method, or is the value saved on entry
627 * to a condition wait. The value is otherwise uninterpreted
628 * and can represent anything you like.
629 * @return {@code true} if successful. Upon success, this object has
630 * been acquired.
631 * @throws IllegalMonitorStateException if acquiring would place this
632 * synchronizer in an illegal state. This exception must be
633 * thrown in a consistent fashion for synchronization to work
634 * correctly.
635 * @throws UnsupportedOperationException if exclusive mode is not supported
636 */
637 protected boolean tryAcquire(long arg) {
638 throw new UnsupportedOperationException();
639 }
640
641 /**
642 * Attempts to set the state to reflect a release in exclusive
643 * mode.
644 *
645 * <p>This method is always invoked by the thread performing release.
646 *
647 * <p>The default implementation throws
648 * {@link UnsupportedOperationException}.
649 *
650 * @param arg the release argument. This value is always the one
651 * passed to a release method, or the current state value upon
652 * entry to a condition wait. The value is otherwise
653 * uninterpreted and can represent anything you like.
654 * @return {@code true} if this object is now in a fully released
655 * state, so that any waiting threads may attempt to acquire;
656 * and {@code false} otherwise.
657 * @throws IllegalMonitorStateException if releasing would place this
658 * synchronizer in an illegal state. This exception must be
659 * thrown in a consistent fashion for synchronization to work
660 * correctly.
661 * @throws UnsupportedOperationException if exclusive mode is not supported
662 */
663 protected boolean tryRelease(long arg) {
664 throw new UnsupportedOperationException();
665 }
666
667 /**
668 * Attempts to acquire in shared mode. This method should query if
669 * the state of the object permits it to be acquired in the shared
670 * mode, and if so to acquire it.
671 *
672 * <p>This method is always invoked by the thread performing
673 * acquire. If this method reports failure, the acquire method
674 * may queue the thread, if it is not already queued, until it is
675 * signalled by a release from some other thread.
676 *
677 * <p>The default implementation throws {@link
678 * UnsupportedOperationException}.
679 *
680 * @param arg the acquire argument. This value is always the one
681 * passed to an acquire method, or is the value saved on entry
682 * to a condition wait. The value is otherwise uninterpreted
683 * and can represent anything you like.
684 * @return a negative value on failure; zero if acquisition in shared
685 * mode succeeded but no subsequent shared-mode acquire can
686 * succeed; and a positive value if acquisition in shared
687 * mode succeeded and subsequent shared-mode acquires might
688 * also succeed, in which case a subsequent waiting thread
689 * must check availability. (Support for three different
690 * return values enables this method to be used in contexts
691 * where acquires only sometimes act exclusively.) Upon
692 * success, this object has been acquired.
693 * @throws IllegalMonitorStateException if acquiring would place this
694 * synchronizer in an illegal state. This exception must be
695 * thrown in a consistent fashion for synchronization to work
696 * correctly.
697 * @throws UnsupportedOperationException if shared mode is not supported
698 */
699 protected long tryAcquireShared(long arg) {
700 throw new UnsupportedOperationException();
701 }
702
703 /**
704 * Attempts to set the state to reflect a release in shared mode.
705 *
706 * <p>This method is always invoked by the thread performing release.
707 *
708 * <p>The default implementation throws
709 * {@link UnsupportedOperationException}.
710 *
711 * @param arg the release argument. This value is always the one
712 * passed to a release method, or the current state value upon
713 * entry to a condition wait. The value is otherwise
714 * uninterpreted and can represent anything you like.
715 * @return {@code true} if this release of shared mode may permit a
716 * waiting acquire (shared or exclusive) to succeed; and
717 * {@code false} otherwise
718 * @throws IllegalMonitorStateException if releasing would place this
719 * synchronizer in an illegal state. This exception must be
720 * thrown in a consistent fashion for synchronization to work
721 * correctly.
722 * @throws UnsupportedOperationException if shared mode is not supported
723 */
724 protected boolean tryReleaseShared(long arg) {
725 throw new UnsupportedOperationException();
726 }
727
728 /**
729 * Returns {@code true} if synchronization is held exclusively with
730 * respect to the current (calling) thread. This method is invoked
731 * upon each call to a {@link ConditionObject} method.
732 *
733 * <p>The default implementation throws {@link
734 * UnsupportedOperationException}. This method is invoked
735 * internally only within {@link ConditionObject} methods, so need
736 * not be defined if conditions are not used.
737 *
738 * @return {@code true} if synchronization is held exclusively;
739 * {@code false} otherwise
740 * @throws UnsupportedOperationException if conditions are not supported
741 */
742 protected boolean isHeldExclusively() {
743 throw new UnsupportedOperationException();
744 }
745
746 /**
747 * Acquires in exclusive mode, ignoring interrupts. Implemented
748 * by invoking at least once {@link #tryAcquire},
749 * returning on success. Otherwise the thread is queued, possibly
750 * repeatedly blocking and unblocking, invoking {@link
751 * #tryAcquire} until success. This method can be used
752 * to implement method {@link Lock#lock}.
753 *
754 * @param arg the acquire argument. This value is conveyed to
755 * {@link #tryAcquire} but is otherwise uninterpreted and
756 * can represent anything you like.
757 */
758 public final void acquire(long arg) {
759 if (!tryAcquire(arg) &&
760 acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
761 selfInterrupt();
762 }
763
764 /**
765 * Acquires in exclusive mode, aborting if interrupted.
766 * Implemented by first checking interrupt status, then invoking
767 * at least once {@link #tryAcquire}, returning on
768 * success. Otherwise the thread is queued, possibly repeatedly
769 * blocking and unblocking, invoking {@link #tryAcquire}
770 * until success or the thread is interrupted. This method can be
771 * used to implement method {@link Lock#lockInterruptibly}.
772 *
773 * @param arg the acquire argument. This value is conveyed to
774 * {@link #tryAcquire} but is otherwise uninterpreted and
775 * can represent anything you like.
776 * @throws InterruptedException if the current thread is interrupted
777 */
778 public final void acquireInterruptibly(long arg)
779 throws InterruptedException {
780 if (Thread.interrupted())
781 throw new InterruptedException();
782 if (!tryAcquire(arg))
783 doAcquireInterruptibly(arg);
784 }
785
786 /**
787 * Attempts to acquire in exclusive mode, aborting if interrupted,
788 * and failing if the given timeout elapses. Implemented by first
789 * checking interrupt status, then invoking at least once {@link
790 * #tryAcquire}, returning on success. Otherwise, the thread is
791 * queued, possibly repeatedly blocking and unblocking, invoking
792 * {@link #tryAcquire} until success or the thread is interrupted
793 * or the timeout elapses. This method can be used to implement
794 * method {@link Lock#tryLock(long, TimeUnit)}.
795 *
796 * @param arg the acquire argument. This value is conveyed to
797 * {@link #tryAcquire} but is otherwise uninterpreted and
798 * can represent anything you like.
799 * @param nanosTimeout the maximum number of nanoseconds to wait
800 * @return {@code true} if acquired; {@code false} if timed out
801 * @throws InterruptedException if the current thread is interrupted
802 */
803 public final boolean tryAcquireNanos(long arg, long nanosTimeout)
804 throws InterruptedException {
805 if (Thread.interrupted())
806 throw new InterruptedException();
807 return tryAcquire(arg) ||
808 doAcquireNanos(arg, nanosTimeout);
809 }
810
811 /**
812 * Releases in exclusive mode. Implemented by unblocking one or
813 * more threads if {@link #tryRelease} returns true.
814 * This method can be used to implement method {@link Lock#unlock}.
815 *
816 * @param arg the release argument. This value is conveyed to
817 * {@link #tryRelease} but is otherwise uninterpreted and
818 * can represent anything you like.
819 * @return the value returned from {@link #tryRelease}
820 */
821 public final boolean release(long arg) {
822 if (tryRelease(arg)) {
823 Node h = head;
824 if (h != null && h.waitStatus != 0)
825 unparkSuccessor(h);
826 return true;
827 }
828 return false;
829 }
830
831 /**
832 * Acquires in shared mode, ignoring interrupts. Implemented by
833 * first invoking at least once {@link #tryAcquireShared},
834 * returning on success. Otherwise the thread is queued, possibly
835 * repeatedly blocking and unblocking, invoking {@link
836 * #tryAcquireShared} until success.
837 *
838 * @param arg the acquire argument. This value is conveyed to
839 * {@link #tryAcquireShared} but is otherwise uninterpreted
840 * and can represent anything you like.
841 */
842 public final void acquireShared(long arg) {
843 if (tryAcquireShared(arg) < 0)
844 doAcquireShared(arg);
845 }
846
847 /**
848 * Acquires in shared mode, aborting if interrupted. Implemented
849 * by first checking interrupt status, then invoking at least once
850 * {@link #tryAcquireShared}, returning on success. Otherwise the
851 * thread is queued, possibly repeatedly blocking and unblocking,
852 * invoking {@link #tryAcquireShared} until success or the thread
853 * is interrupted.
854 * @param arg the acquire argument.
855 * This value is conveyed to {@link #tryAcquireShared} but is
856 * otherwise uninterpreted and can represent anything
857 * you like.
858 * @throws InterruptedException if the current thread is interrupted
859 */
860 public final void acquireSharedInterruptibly(long arg)
861 throws InterruptedException {
862 if (Thread.interrupted())
863 throw new InterruptedException();
864 if (tryAcquireShared(arg) < 0)
865 doAcquireSharedInterruptibly(arg);
866 }
867
868 /**
869 * Attempts to acquire in shared mode, aborting if interrupted, and
870 * failing if the given timeout elapses. Implemented by first
871 * checking interrupt status, then invoking at least once {@link
872 * #tryAcquireShared}, returning on success. Otherwise, the
873 * thread is queued, possibly repeatedly blocking and unblocking,
874 * invoking {@link #tryAcquireShared} until success or the thread
875 * is interrupted or the timeout elapses.
876 *
877 * @param arg the acquire argument. This value is conveyed to
878 * {@link #tryAcquireShared} but is otherwise uninterpreted
879 * and can represent anything you like.
880 * @param nanosTimeout the maximum number of nanoseconds to wait
881 * @return {@code true} if acquired; {@code false} if timed out
882 * @throws InterruptedException if the current thread is interrupted
883 */
884 public final boolean tryAcquireSharedNanos(long arg, long nanosTimeout)
885 throws InterruptedException {
886 if (Thread.interrupted())
887 throw new InterruptedException();
888 return tryAcquireShared(arg) >= 0 ||
889 doAcquireSharedNanos(arg, nanosTimeout);
890 }
891
892 /**
893 * Releases in shared mode. Implemented by unblocking one or more
894 * threads if {@link #tryReleaseShared} returns true.
895 *
896 * @param arg the release argument. This value is conveyed to
897 * {@link #tryReleaseShared} but is otherwise uninterpreted
898 * and can represent anything you like.
899 * @return the value returned from {@link #tryReleaseShared}
900 */
901 public final boolean releaseShared(long arg) {
902 if (tryReleaseShared(arg)) {
903 doReleaseShared();
904 return true;
905 }
906 return false;
907 }
908
909 // Queue inspection methods
910
911 /**
912 * Queries whether any threads are waiting to acquire. Note that
913 * because cancellations due to interrupts and timeouts may occur
914 * at any time, a {@code true} return does not guarantee that any
915 * other thread will ever acquire.
916 *
917 * @return {@code true} if there may be other threads waiting to acquire
918 */
919 public final boolean hasQueuedThreads() {
920 for (Node p = tail, h = head; p != h && p != null; p = p.prev)
921 if (p.waitStatus <= 0)
922 return true;
923 return false;
924 }
925
926 /**
927 * Queries whether any threads have ever contended to acquire this
928 * synchronizer; that is, if an acquire method has ever blocked.
929 *
930 * <p>In this implementation, this operation returns in
931 * constant time.
932 *
933 * @return {@code true} if there has ever been contention
934 */
935 public final boolean hasContended() {
936 return head != null;
937 }
938
939 /**
940 * Returns the first (longest-waiting) thread in the queue, or
941 * {@code null} if no threads are currently queued.
942 *
943 * <p>In this implementation, this operation normally returns in
944 * constant time, but may iterate upon contention if other threads are
945 * concurrently modifying the queue.
946 *
947 * @return the first (longest-waiting) thread in the queue, or
948 * {@code null} if no threads are currently queued
949 */
950 public final Thread getFirstQueuedThread() {
951 // handle only fast path, else relay
952 return (head == tail) ? null : fullGetFirstQueuedThread();
953 }
954
955 /**
956 * Version of getFirstQueuedThread called when fastpath fails.
957 */
958 private Thread fullGetFirstQueuedThread() {
959 /*
960 * The first node is normally head.next. Try to get its
961 * thread field, ensuring consistent reads: If thread
962 * field is nulled out or s.prev is no longer head, then
963 * some other thread(s) concurrently performed setHead in
964 * between some of our reads. We try this twice before
965 * resorting to traversal.
966 */
967 Node h, s;
968 Thread st;
969 if (((h = head) != null && (s = h.next) != null &&
970 s.prev == head && (st = s.thread) != null) ||
971 ((h = head) != null && (s = h.next) != null &&
972 s.prev == head && (st = s.thread) != null))
973 return st;
974
975 /*
976 * Head's next field might not have been set yet, or may have
977 * been unset after setHead. So we must check to see if tail
978 * is actually first node. If not, we continue on, safely
979 * traversing from tail back to head to find first,
980 * guaranteeing termination.
981 */
982
983 Thread firstThread = null;
984 for (Node p = tail; p != null && p != head; p = p.prev) {
985 Thread t = p.thread;
986 if (t != null)
987 firstThread = t;
988 }
989 return firstThread;
990 }
991
992 /**
993 * Returns true if the given thread is currently queued.
994 *
995 * <p>This implementation traverses the queue to determine
996 * presence of the given thread.
997 *
998 * @param thread the thread
999 * @return {@code true} if the given thread is on the queue
1000 * @throws NullPointerException if the thread is null
1001 */
1002 public final boolean isQueued(Thread thread) {
1003 if (thread == null)
1004 throw new NullPointerException();
1005 for (Node p = tail; p != null; p = p.prev)
1006 if (p.thread == thread)
1007 return true;
1008 return false;
1009 }
1010
1011 /**
1012 * Returns {@code true} if the apparent first queued thread, if one
1013 * exists, is waiting in exclusive mode. If this method returns
1014 * {@code true}, and the current thread is attempting to acquire in
1015 * shared mode (that is, this method is invoked from {@link
1016 * #tryAcquireShared}) then it is guaranteed that the current thread
1017 * is not the first queued thread. Used only as a heuristic in
1018 * ReentrantReadWriteLock.
1019 */
1020 final boolean apparentlyFirstQueuedIsExclusive() {
1021 Node h, s;
1022 return (h = head) != null &&
1023 (s = h.next) != null &&
1024 !s.isShared() &&
1025 s.thread != null;
1026 }
1027
1028 /**
1029 * Queries whether any threads have been waiting to acquire longer
1030 * than the current thread.
1031 *
1032 * <p>An invocation of this method is equivalent to (but may be
1033 * more efficient than):
1034 * <pre> {@code
1035 * getFirstQueuedThread() != Thread.currentThread()
1036 * && hasQueuedThreads()}</pre>
1037 *
1038 * <p>Note that because cancellations due to interrupts and
1039 * timeouts may occur at any time, a {@code true} return does not
1040 * guarantee that some other thread will acquire before the current
1041 * thread. Likewise, it is possible for another thread to win a
1042 * race to enqueue after this method has returned {@code false},
1043 * due to the queue being empty.
1044 *
1045 * <p>This method is designed to be used by a fair synchronizer to
1046 * avoid <a href="AbstractQueuedSynchronizer.html#barging">barging</a>.
1047 * Such a synchronizer's {@link #tryAcquire} method should return
1048 * {@code false}, and its {@link #tryAcquireShared} method should
1049 * return a negative value, if this method returns {@code true}
1050 * (unless this is a reentrant acquire). For example, the {@code
1051 * tryAcquire} method for a fair, reentrant, exclusive mode
1052 * synchronizer might look like this:
1053 *
1054 * <pre> {@code
1055 * protected boolean tryAcquire(int arg) {
1056 * if (isHeldExclusively()) {
1057 * // A reentrant acquire; increment hold count
1058 * return true;
1059 * } else if (hasQueuedPredecessors()) {
1060 * return false;
1061 * } else {
1062 * // try to acquire normally
1063 * }
1064 * }}</pre>
1065 *
1066 * @return {@code true} if there is a queued thread preceding the
1067 * current thread, and {@code false} if the current thread
1068 * is at the head of the queue or the queue is empty
1069 * @since 1.7
1070 */
1071 public final boolean hasQueuedPredecessors() {
1072 Node h, s;
1073 if ((h = head) != null) {
1074 if ((s = h.next) == null || s.waitStatus > 0) {
1075 s = null; // traverse in case of concurrent cancellation
1076 for (Node p = tail; p != h && p != null; p = p.prev) {
1077 if (p.waitStatus <= 0)
1078 s = p;
1079 }
1080 }
1081 if (s != null && s.thread != Thread.currentThread())
1082 return true;
1083 }
1084 return false;
1085 }
1086
1087 // Instrumentation and monitoring methods
1088
1089 /**
1090 * Returns an estimate of the number of threads waiting to
1091 * acquire. The value is only an estimate because the number of
1092 * threads may change dynamically while this method traverses
1093 * internal data structures. This method is designed for use in
1094 * monitoring system state, not for synchronization control.
1095 *
1096 * @return the estimated number of threads waiting to acquire
1097 */
1098 public final int getQueueLength() {
1099 int n = 0;
1100 for (Node p = tail; p != null; p = p.prev) {
1101 if (p.thread != null)
1102 ++n;
1103 }
1104 return n;
1105 }
1106
1107 /**
1108 * Returns a collection containing threads that may be waiting to
1109 * acquire. Because the actual set of threads may change
1110 * dynamically while constructing this result, the returned
1111 * collection is only a best-effort estimate. The elements of the
1112 * returned collection are in no particular order. This method is
1113 * designed to facilitate construction of subclasses that provide
1114 * more extensive monitoring facilities.
1115 *
1116 * @return the collection of threads
1117 */
1118 public final Collection<Thread> getQueuedThreads() {
1119 ArrayList<Thread> list = new ArrayList<>();
1120 for (Node p = tail; p != null; p = p.prev) {
1121 Thread t = p.thread;
1122 if (t != null)
1123 list.add(t);
1124 }
1125 return list;
1126 }
1127
1128 /**
1129 * Returns a collection containing threads that may be waiting to
1130 * acquire in exclusive mode. This has the same properties
1131 * as {@link #getQueuedThreads} except that it only returns
1132 * those threads waiting due to an exclusive acquire.
1133 *
1134 * @return the collection of threads
1135 */
1136 public final Collection<Thread> getExclusiveQueuedThreads() {
1137 ArrayList<Thread> list = new ArrayList<>();
1138 for (Node p = tail; p != null; p = p.prev) {
1139 if (!p.isShared()) {
1140 Thread t = p.thread;
1141 if (t != null)
1142 list.add(t);
1143 }
1144 }
1145 return list;
1146 }
1147
1148 /**
1149 * Returns a collection containing threads that may be waiting to
1150 * acquire in shared mode. This has the same properties
1151 * as {@link #getQueuedThreads} except that it only returns
1152 * those threads waiting due to a shared acquire.
1153 *
1154 * @return the collection of threads
1155 */
1156 public final Collection<Thread> getSharedQueuedThreads() {
1157 ArrayList<Thread> list = new ArrayList<>();
1158 for (Node p = tail; p != null; p = p.prev) {
1159 if (p.isShared()) {
1160 Thread t = p.thread;
1161 if (t != null)
1162 list.add(t);
1163 }
1164 }
1165 return list;
1166 }
1167
1168 /**
1169 * Returns a string identifying this synchronizer, as well as its state.
1170 * The state, in brackets, includes the String {@code "State ="}
1171 * followed by the current value of {@link #getState}, and either
1172 * {@code "nonempty"} or {@code "empty"} depending on whether the
1173 * queue is empty.
1174 *
1175 * @return a string identifying this synchronizer, as well as its state
1176 */
1177 public String toString() {
1178 return super.toString()
1179 + "[State = " + getState() + ", "
1180 + (hasQueuedThreads() ? "non" : "") + "empty queue]";
1181 }
1182
1183
1184 // Internal support methods for Conditions
1185
1186 /**
1187 * Returns true if a node, always one that was initially placed on
1188 * a condition queue, is now waiting to reacquire on sync queue.
1189 * @param node the node
1190 * @return true if is reacquiring
1191 */
1192 final boolean isOnSyncQueue(Node node) {
1193 if (node.waitStatus == Node.CONDITION || node.prev == null)
1194 return false;
1195 if (node.next != null) // If has successor, it must be on queue
1196 return true;
1197 /*
1198 * node.prev can be non-null, but not yet on queue because
1199 * the CAS to place it on queue can fail. So we have to
1200 * traverse from tail to make sure it actually made it. It
1201 * will always be near the tail in calls to this method, and
1202 * unless the CAS failed (which is unlikely), it will be
1203 * there, so we hardly ever traverse much.
1204 */
1205 return findNodeFromTail(node);
1206 }
1207
1208 /**
1209 * Returns true if node is on sync queue by searching backwards from tail.
1210 * Called only when needed by isOnSyncQueue.
1211 * @return true if present
1212 */
1213 private boolean findNodeFromTail(Node node) {
1214 // We check for node first, since it's likely to be at or near tail.
1215 // tail is known to be non-null, so we could re-order to "save"
1216 // one null check, but we leave it this way to help the VM.
1217 for (Node p = tail;;) {
1218 if (p == node)
1219 return true;
1220 if (p == null)
1221 return false;
1222 p = p.prev;
1223 }
1224 }
1225
1226 /**
1227 * Transfers a node from a condition queue onto sync queue.
1228 * Returns true if successful.
1229 * @param node the node
1230 * @return true if successfully transferred (else the node was
1231 * cancelled before signal)
1232 */
1233 final boolean transferForSignal(Node node) {
1234 /*
1235 * If cannot change waitStatus, the node has been cancelled.
1236 */
1237 if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
1238 return false;
1239
1240 /*
1241 * Splice onto queue and try to set waitStatus of predecessor to
1242 * indicate that thread is (probably) waiting. If cancelled or
1243 * attempt to set waitStatus fails, wake up to resync (in which
1244 * case the waitStatus can be transiently and harmlessly wrong).
1245 */
1246 Node p = enq(node);
1247 int ws = p.waitStatus;
1248 if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
1249 LockSupport.unpark(node.thread);
1250 return true;
1251 }
1252
1253 /**
1254 * Transfers node, if necessary, to sync queue after a cancelled wait.
1255 * Returns true if thread was cancelled before being signalled.
1256 *
1257 * @param node the node
1258 * @return true if cancelled before the node was signalled
1259 */
1260 final boolean transferAfterCancelledWait(Node node) {
1261 if (node.compareAndSetWaitStatus(Node.CONDITION, 0)) {
1262 enq(node);
1263 return true;
1264 }
1265 /*
1266 * If we lost out to a signal(), then we can't proceed
1267 * until it finishes its enq(). Cancelling during an
1268 * incomplete transfer is both rare and transient, so just
1269 * spin.
1270 */
1271 while (!isOnSyncQueue(node))
1272 Thread.yield();
1273 return false;
1274 }
1275
1276 /**
1277 * Invokes release with current state value; returns saved state.
1278 * Cancels node and throws exception on failure.
1279 * @param node the condition node for this wait
1280 * @return previous sync state
1281 */
1282 final long fullyRelease(Node node) {
1283 try {
1284 long savedState = getState();
1285 if (release(savedState))
1286 return savedState;
1287 throw new IllegalMonitorStateException();
1288 } catch (Throwable t) {
1289 node.waitStatus = Node.CANCELLED;
1290 throw t;
1291 }
1292 }
1293
1294 // Instrumentation methods for conditions
1295
1296 /**
1297 * Queries whether the given ConditionObject
1298 * uses this synchronizer as its lock.
1299 *
1300 * @param condition the condition
1301 * @return {@code true} if owned
1302 * @throws NullPointerException if the condition is null
1303 */
1304 public final boolean owns(ConditionObject condition) {
1305 return condition.isOwnedBy(this);
1306 }
1307
1308 /**
1309 * Queries whether any threads are waiting on the given condition
1310 * associated with this synchronizer. Note that because timeouts
1311 * and interrupts may occur at any time, a {@code true} return
1312 * does not guarantee that a future {@code signal} will awaken
1313 * any threads. This method is designed primarily for use in
1314 * monitoring of the system state.
1315 *
1316 * @param condition the condition
1317 * @return {@code true} if there are any waiting threads
1318 * @throws IllegalMonitorStateException if exclusive synchronization
1319 * is not held
1320 * @throws IllegalArgumentException if the given condition is
1321 * not associated with this synchronizer
1322 * @throws NullPointerException if the condition is null
1323 */
1324 public final boolean hasWaiters(ConditionObject condition) {
1325 if (!owns(condition))
1326 throw new IllegalArgumentException("Not owner");
1327 return condition.hasWaiters();
1328 }
1329
1330 /**
1331 * Returns an estimate of the number of threads waiting on the
1332 * given condition associated with this synchronizer. Note that
1333 * because timeouts and interrupts may occur at any time, the
1334 * estimate serves only as an upper bound on the actual number of
1335 * waiters. This method is designed for use in monitoring system
1336 * state, not for synchronization control.
1337 *
1338 * @param condition the condition
1339 * @return the estimated number of waiting threads
1340 * @throws IllegalMonitorStateException if exclusive synchronization
1341 * is not held
1342 * @throws IllegalArgumentException if the given condition is
1343 * not associated with this synchronizer
1344 * @throws NullPointerException if the condition is null
1345 */
1346 public final int getWaitQueueLength(ConditionObject condition) {
1347 if (!owns(condition))
1348 throw new IllegalArgumentException("Not owner");
1349 return condition.getWaitQueueLength();
1350 }
1351
1352 /**
1353 * Returns a collection containing those threads that may be
1354 * waiting on the given condition associated with this
1355 * synchronizer. Because the actual set of threads may change
1356 * dynamically while constructing this result, the returned
1357 * collection is only a best-effort estimate. The elements of the
1358 * returned collection are in no particular order.
1359 *
1360 * @param condition the condition
1361 * @return the collection of threads
1362 * @throws IllegalMonitorStateException if exclusive synchronization
1363 * is not held
1364 * @throws IllegalArgumentException if the given condition is
1365 * not associated with this synchronizer
1366 * @throws NullPointerException if the condition is null
1367 */
1368 public final Collection<Thread> getWaitingThreads(ConditionObject condition) {
1369 if (!owns(condition))
1370 throw new IllegalArgumentException("Not owner");
1371 return condition.getWaitingThreads();
1372 }
1373
1374 /**
1375 * Condition implementation for a {@link AbstractQueuedLongSynchronizer}
1376 * serving as the basis of a {@link Lock} implementation.
1377 *
1378 * <p>Method documentation for this class describes mechanics,
1379 * not behavioral specifications from the point of view of Lock
1380 * and Condition users. Exported versions of this class will in
1381 * general need to be accompanied by documentation describing
1382 * condition semantics that rely on those of the associated
1383 * {@code AbstractQueuedLongSynchronizer}.
1384 *
1385 * <p>This class is Serializable, but all fields are transient,
1386 * so deserialized conditions have no waiters.
1387 *
1388 * @since 1.6
1389 */
1390 public class ConditionObject implements Condition, java.io.Serializable {
1391 private static final long serialVersionUID = 1173984872572414699L;
1392 /** First node of condition queue. */
1393 private transient Node firstWaiter;
1394 /** Last node of condition queue. */
1395 private transient Node lastWaiter;
1396
1397 /**
1398 * Creates a new {@code ConditionObject} instance.
1399 */
1400 public ConditionObject() { }
1401
1402 // Internal methods
1403
1404 /**
1405 * Adds a new waiter to wait queue.
1406 * @return its new wait node
1407 */
1408 private Node addConditionWaiter() {
1409 if (!isHeldExclusively())
1410 throw new IllegalMonitorStateException();
1411 Node t = lastWaiter;
1412 // If lastWaiter is cancelled, clean out.
1413 if (t != null && t.waitStatus != Node.CONDITION) {
1414 unlinkCancelledWaiters();
1415 t = lastWaiter;
1416 }
1417
1418 Node node = new Node(Node.CONDITION);
1419
1420 if (t == null)
1421 firstWaiter = node;
1422 else
1423 t.nextWaiter = node;
1424 lastWaiter = node;
1425 return node;
1426 }
1427
1428 /**
1429 * Removes and transfers nodes until hit non-cancelled one or
1430 * null. Split out from signal in part to encourage compilers
1431 * to inline the case of no waiters.
1432 * @param first (non-null) the first node on condition queue
1433 */
1434 private void doSignal(Node first) {
1435 do {
1436 if ( (firstWaiter = first.nextWaiter) == null)
1437 lastWaiter = null;
1438 first.nextWaiter = null;
1439 } while (!transferForSignal(first) &&
1440 (first = firstWaiter) != null);
1441 }
1442
1443 /**
1444 * Removes and transfers all nodes.
1445 * @param first (non-null) the first node on condition queue
1446 */
1447 private void doSignalAll(Node first) {
1448 lastWaiter = firstWaiter = null;
1449 do {
1450 Node next = first.nextWaiter;
1451 first.nextWaiter = null;
1452 transferForSignal(first);
1453 first = next;
1454 } while (first != null);
1455 }
1456
1457 /**
1458 * Unlinks cancelled waiter nodes from condition queue.
1459 * Called only while holding lock. This is called when
1460 * cancellation occurred during condition wait, and upon
1461 * insertion of a new waiter when lastWaiter is seen to have
1462 * been cancelled. This method is needed to avoid garbage
1463 * retention in the absence of signals. So even though it may
1464 * require a full traversal, it comes into play only when
1465 * timeouts or cancellations occur in the absence of
1466 * signals. It traverses all nodes rather than stopping at a
1467 * particular target to unlink all pointers to garbage nodes
1468 * without requiring many re-traversals during cancellation
1469 * storms.
1470 */
1471 private void unlinkCancelledWaiters() {
1472 Node t = firstWaiter;
1473 Node trail = null;
1474 while (t != null) {
1475 Node next = t.nextWaiter;
1476 if (t.waitStatus != Node.CONDITION) {
1477 t.nextWaiter = null;
1478 if (trail == null)
1479 firstWaiter = next;
1480 else
1481 trail.nextWaiter = next;
1482 if (next == null)
1483 lastWaiter = trail;
1484 }
1485 else
1486 trail = t;
1487 t = next;
1488 }
1489 }
1490
1491 // public methods
1492
1493 /**
1494 * Moves the longest-waiting thread, if one exists, from the
1495 * wait queue for this condition to the wait queue for the
1496 * owning lock.
1497 *
1498 * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
1499 * returns {@code false}
1500 */
1501 public final void signal() {
1502 if (!isHeldExclusively())
1503 throw new IllegalMonitorStateException();
1504 Node first = firstWaiter;
1505 if (first != null)
1506 doSignal(first);
1507 }
1508
1509 /**
1510 * Moves all threads from the wait queue for this condition to
1511 * the wait queue for the owning lock.
1512 *
1513 * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
1514 * returns {@code false}
1515 */
1516 public final void signalAll() {
1517 if (!isHeldExclusively())
1518 throw new IllegalMonitorStateException();
1519 Node first = firstWaiter;
1520 if (first != null)
1521 doSignalAll(first);
1522 }
1523
1524 /**
1525 * Implements uninterruptible condition wait.
1526 * <ol>
1527 * <li>Save lock state returned by {@link #getState}.
1528 * <li>Invoke {@link #release} with saved state as argument,
1529 * throwing IllegalMonitorStateException if it fails.
1530 * <li>Block until signalled.
1531 * <li>Reacquire by invoking specialized version of
1532 * {@link #acquire} with saved state as argument.
1533 * </ol>
1534 */
1535 public final void awaitUninterruptibly() {
1536 Node node = addConditionWaiter();
1537 long savedState = fullyRelease(node);
1538 boolean interrupted = false;
1539 while (!isOnSyncQueue(node)) {
1540 LockSupport.park(this);
1541 if (Thread.interrupted())
1542 interrupted = true;
1543 }
1544 if (acquireQueued(node, savedState) || interrupted)
1545 selfInterrupt();
1546 }
1547
1548 /*
1549 * For interruptible waits, we need to track whether to throw
1550 * InterruptedException, if interrupted while blocked on
1551 * condition, versus reinterrupt current thread, if
1552 * interrupted while blocked waiting to re-acquire.
1553 */
1554
1555 /** Mode meaning to reinterrupt on exit from wait */
1556 private static final int REINTERRUPT = 1;
1557 /** Mode meaning to throw InterruptedException on exit from wait */
1558 private static final int THROW_IE = -1;
1559
1560 /**
1561 * Checks for interrupt, returning THROW_IE if interrupted
1562 * before signalled, REINTERRUPT if after signalled, or
1563 * 0 if not interrupted.
1564 */
1565 private int checkInterruptWhileWaiting(Node node) {
1566 return Thread.interrupted() ?
1567 (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
1568 0;
1569 }
1570
1571 /**
1572 * Throws InterruptedException, reinterrupts current thread, or
1573 * does nothing, depending on mode.
1574 */
1575 private void reportInterruptAfterWait(int interruptMode)
1576 throws InterruptedException {
1577 if (interruptMode == THROW_IE)
1578 throw new InterruptedException();
1579 else if (interruptMode == REINTERRUPT)
1580 selfInterrupt();
1581 }
1582
1583 /**
1584 * Implements interruptible condition wait.
1585 * <ol>
1586 * <li>If current thread is interrupted, throw InterruptedException.
1587 * <li>Save lock state returned by {@link #getState}.
1588 * <li>Invoke {@link #release} with saved state as argument,
1589 * throwing IllegalMonitorStateException if it fails.
1590 * <li>Block until signalled or interrupted.
1591 * <li>Reacquire by invoking specialized version of
1592 * {@link #acquire} with saved state as argument.
1593 * <li>If interrupted while blocked in step 4, throw InterruptedException.
1594 * </ol>
1595 */
1596 public final void await() throws InterruptedException {
1597 if (Thread.interrupted())
1598 throw new InterruptedException();
1599 Node node = addConditionWaiter();
1600 long savedState = fullyRelease(node);
1601 int interruptMode = 0;
1602 while (!isOnSyncQueue(node)) {
1603 LockSupport.park(this);
1604 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
1605 break;
1606 }
1607 if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
1608 interruptMode = REINTERRUPT;
1609 if (node.nextWaiter != null) // clean up if cancelled
1610 unlinkCancelledWaiters();
1611 if (interruptMode != 0)
1612 reportInterruptAfterWait(interruptMode);
1613 }
1614
1615 /**
1616 * Implements timed condition wait.
1617 * <ol>
1618 * <li>If current thread is interrupted, throw InterruptedException.
1619 * <li>Save lock state returned by {@link #getState}.
1620 * <li>Invoke {@link #release} with saved state as argument,
1621 * throwing IllegalMonitorStateException if it fails.
1622 * <li>Block until signalled, interrupted, or timed out.
1623 * <li>Reacquire by invoking specialized version of
1624 * {@link #acquire} with saved state as argument.
1625 * <li>If interrupted while blocked in step 4, throw InterruptedException.
1626 * </ol>
1627 */
1628 public final long awaitNanos(long nanosTimeout)
1629 throws InterruptedException {
1630 if (Thread.interrupted())
1631 throw new InterruptedException();
1632 // We don't check for nanosTimeout <= 0L here, to allow
1633 // awaitNanos(0) as a way to "yield the lock".
1634 final long deadline = System.nanoTime() + nanosTimeout;
1635 long initialNanos = nanosTimeout;
1636 Node node = addConditionWaiter();
1637 long savedState = fullyRelease(node);
1638 int interruptMode = 0;
1639 while (!isOnSyncQueue(node)) {
1640 if (nanosTimeout <= 0L) {
1641 transferAfterCancelledWait(node);
1642 break;
1643 }
1644 if (nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD)
1645 LockSupport.parkNanos(this, nanosTimeout);
1646 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
1647 break;
1648 nanosTimeout = deadline - System.nanoTime();
1649 }
1650 if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
1651 interruptMode = REINTERRUPT;
1652 if (node.nextWaiter != null)
1653 unlinkCancelledWaiters();
1654 if (interruptMode != 0)
1655 reportInterruptAfterWait(interruptMode);
1656 long remaining = deadline - System.nanoTime(); // avoid overflow
1657 return (remaining <= initialNanos) ? remaining : Long.MIN_VALUE;
1658 }
1659
1660 /**
1661 * Implements absolute timed condition wait.
1662 * <ol>
1663 * <li>If current thread is interrupted, throw InterruptedException.
1664 * <li>Save lock state returned by {@link #getState}.
1665 * <li>Invoke {@link #release} with saved state as argument,
1666 * throwing IllegalMonitorStateException if it fails.
1667 * <li>Block until signalled, interrupted, or timed out.
1668 * <li>Reacquire by invoking specialized version of
1669 * {@link #acquire} with saved state as argument.
1670 * <li>If interrupted while blocked in step 4, throw InterruptedException.
1671 * <li>If timed out while blocked in step 4, return false, else true.
1672 * </ol>
1673 */
1674 public final boolean awaitUntil(Date deadline)
1675 throws InterruptedException {
1676 long abstime = deadline.getTime();
1677 if (Thread.interrupted())
1678 throw new InterruptedException();
1679 Node node = addConditionWaiter();
1680 long savedState = fullyRelease(node);
1681 boolean timedout = false;
1682 int interruptMode = 0;
1683 while (!isOnSyncQueue(node)) {
1684 if (System.currentTimeMillis() >= abstime) {
1685 timedout = transferAfterCancelledWait(node);
1686 break;
1687 }
1688 LockSupport.parkUntil(this, abstime);
1689 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
1690 break;
1691 }
1692 if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
1693 interruptMode = REINTERRUPT;
1694 if (node.nextWaiter != null)
1695 unlinkCancelledWaiters();
1696 if (interruptMode != 0)
1697 reportInterruptAfterWait(interruptMode);
1698 return !timedout;
1699 }
1700
1701 /**
1702 * Implements timed condition wait.
1703 * <ol>
1704 * <li>If current thread is interrupted, throw InterruptedException.
1705 * <li>Save lock state returned by {@link #getState}.
1706 * <li>Invoke {@link #release} with saved state as argument,
1707 * throwing IllegalMonitorStateException if it fails.
1708 * <li>Block until signalled, interrupted, or timed out.
1709 * <li>Reacquire by invoking specialized version of
1710 * {@link #acquire} with saved state as argument.
1711 * <li>If interrupted while blocked in step 4, throw InterruptedException.
1712 * <li>If timed out while blocked in step 4, return false, else true.
1713 * </ol>
1714 */
1715 public final boolean await(long time, TimeUnit unit)
1716 throws InterruptedException {
1717 long nanosTimeout = unit.toNanos(time);
1718 if (Thread.interrupted())
1719 throw new InterruptedException();
1720 // We don't check for nanosTimeout <= 0L here, to allow
1721 // await(0, unit) as a way to "yield the lock".
1722 final long deadline = System.nanoTime() + nanosTimeout;
1723 Node node = addConditionWaiter();
1724 long savedState = fullyRelease(node);
1725 boolean timedout = false;
1726 int interruptMode = 0;
1727 while (!isOnSyncQueue(node)) {
1728 if (nanosTimeout <= 0L) {
1729 timedout = transferAfterCancelledWait(node);
1730 break;
1731 }
1732 if (nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD)
1733 LockSupport.parkNanos(this, nanosTimeout);
1734 if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
1735 break;
1736 nanosTimeout = deadline - System.nanoTime();
1737 }
1738 if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
1739 interruptMode = REINTERRUPT;
1740 if (node.nextWaiter != null)
1741 unlinkCancelledWaiters();
1742 if (interruptMode != 0)
1743 reportInterruptAfterWait(interruptMode);
1744 return !timedout;
1745 }
1746
1747 // support for instrumentation
1748
1749 /**
1750 * Returns true if this condition was created by the given
1751 * synchronization object.
1752 *
1753 * @return {@code true} if owned
1754 */
1755 final boolean isOwnedBy(AbstractQueuedLongSynchronizer sync) {
1756 return sync == AbstractQueuedLongSynchronizer.this;
1757 }
1758
1759 /**
1760 * Queries whether any threads are waiting on this condition.
1761 * Implements {@link AbstractQueuedLongSynchronizer#hasWaiters(ConditionObject)}.
1762 *
1763 * @return {@code true} if there are any waiting threads
1764 * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
1765 * returns {@code false}
1766 */
1767 protected final boolean hasWaiters() {
1768 if (!isHeldExclusively())
1769 throw new IllegalMonitorStateException();
1770 for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
1771 if (w.waitStatus == Node.CONDITION)
1772 return true;
1773 }
1774 return false;
1775 }
1776
1777 /**
1778 * Returns an estimate of the number of threads waiting on
1779 * this condition.
1780 * Implements {@link AbstractQueuedLongSynchronizer#getWaitQueueLength(ConditionObject)}.
1781 *
1782 * @return the estimated number of waiting threads
1783 * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
1784 * returns {@code false}
1785 */
1786 protected final int getWaitQueueLength() {
1787 if (!isHeldExclusively())
1788 throw new IllegalMonitorStateException();
1789 int n = 0;
1790 for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
1791 if (w.waitStatus == Node.CONDITION)
1792 ++n;
1793 }
1794 return n;
1795 }
1796
1797 /**
1798 * Returns a collection containing those threads that may be
1799 * waiting on this Condition.
1800 * Implements {@link AbstractQueuedLongSynchronizer#getWaitingThreads(ConditionObject)}.
1801 *
1802 * @return the collection of threads
1803 * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
1804 * returns {@code false}
1805 */
1806 protected final Collection<Thread> getWaitingThreads() {
1807 if (!isHeldExclusively())
1808 throw new IllegalMonitorStateException();
1809 ArrayList<Thread> list = new ArrayList<>();
1810 for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
1811 if (w.waitStatus == Node.CONDITION) {
1812 Thread t = w.thread;
1813 if (t != null)
1814 list.add(t);
1815 }
1816 }
1817 return list;
1818 }
1819 }
1820
1821 // VarHandle mechanics
1822 private static final VarHandle STATE;
1823 private static final VarHandle HEAD;
1824 private static final VarHandle TAIL;
1825
1826 static {
1827 try {
1828 MethodHandles.Lookup l = MethodHandles.lookup();
1829 STATE = l.findVarHandle(AbstractQueuedLongSynchronizer.class, "state", long.class);
1830 HEAD = l.findVarHandle(AbstractQueuedLongSynchronizer.class, "head", Node.class);
1831 TAIL = l.findVarHandle(AbstractQueuedLongSynchronizer.class, "tail", Node.class);
1832 } catch (ReflectiveOperationException e) {
1833 throw new ExceptionInInitializerError(e);
1834 }
1835
1836 // Reduce the risk of rare disastrous classloading in first call to
1837 // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
1838 Class<?> ensureLoaded = LockSupport.class;
1839 }
1840
1841 /**
1842 * Initializes head and tail fields on first contention.
1843 */
1844 private final void initializeSyncQueue() {
1845 Node h;
1846 if (HEAD.compareAndSet(this, null, (h = new Node())))
1847 tail = h;
1848 }
1849
1850 /**
1851 * CASes tail field.
1852 */
1853 private final boolean compareAndSetTail(Node expect, Node update) {
1854 return TAIL.compareAndSet(this, expect, update);
1855 }
1856 }