< prev index next >

src/java.base/share/classes/java/util/concurrent/SynchronousQueue.java

Print this page

        

*** 46,55 **** --- 46,57 ---- import java.util.Spliterator; import java.util.Spliterators; import java.util.concurrent.locks.LockSupport; import java.util.concurrent.locks.ReentrantLock; + import jdk.internal.misc.Strands; + /** * A {@linkplain BlockingQueue blocking queue} in which each insert * operation must wait for a corresponding remove operation by another * thread, and vice versa. A synchronous queue does not have any * internal capacity, not even a capacity of one. You cannot
*** 234,244 **** /** Node class for TransferStacks. */ static final class SNode { volatile SNode next; // next node in stack volatile SNode match; // the node matched to this ! volatile Thread waiter; // to control park/unpark Object item; // data; or null for REQUESTs int mode; // Note: item and mode fields don't need to be volatile // since they are always written before, and read after, // other volatile/atomic operations. --- 236,246 ---- /** Node class for TransferStacks. */ static final class SNode { volatile SNode next; // next node in stack volatile SNode match; // the node matched to this ! volatile Object waiter; // to control park/unpark Object item; // data; or null for REQUESTs int mode; // Note: item and mode fields don't need to be volatile // since they are always written before, and read after, // other volatile/atomic operations.
*** 261,271 **** * @return true if successfully matched to s */ boolean tryMatch(SNode s) { if (match == null && SMATCH.compareAndSet(this, null, s)) { ! Thread w = waiter; if (w != null) { // waiters need at most one unpark waiter = null; LockSupport.unpark(w); } return true; --- 263,273 ---- * @return true if successfully matched to s */ boolean tryMatch(SNode s) { if (match == null && SMATCH.compareAndSet(this, null, s)) { ! Object w = waiter; if (w != null) { // waiters need at most one unpark waiter = null; LockSupport.unpark(w); } return true;
*** 431,446 **** * SynchronousQueue.{poll/offer} don't check interrupts * and don't wait at all, so are trapped in transfer * method rather than calling awaitFulfill. */ final long deadline = timed ? System.nanoTime() + nanos : 0L; ! Thread w = Thread.currentThread(); int spins = shouldSpin(s) ? (timed ? MAX_TIMED_SPINS : MAX_UNTIMED_SPINS) : 0; for (;;) { ! if (w.isInterrupted()) s.tryCancel(); SNode m = s.match; if (m != null) return m; if (timed) { --- 433,448 ---- * SynchronousQueue.{poll/offer} don't check interrupts * and don't wait at all, so are trapped in transfer * method rather than calling awaitFulfill. */ final long deadline = timed ? System.nanoTime() + nanos : 0L; ! Object w = Strands.currentStrand(); int spins = shouldSpin(s) ? (timed ? MAX_TIMED_SPINS : MAX_UNTIMED_SPINS) : 0; for (;;) { ! if (Strands.isInterrupted()) s.tryCancel(); SNode m = s.match; if (m != null) return m; if (timed) {
*** 534,544 **** /** Node class for TransferQueue. */ static final class QNode { volatile QNode next; // next node in queue volatile Object item; // CAS'ed to or from null ! volatile Thread waiter; // to control park/unpark final boolean isData; QNode(Object item, boolean isData) { this.item = item; this.isData = isData; --- 536,546 ---- /** Node class for TransferQueue. */ static final class QNode { volatile QNode next; // next node in queue volatile Object item; // CAS'ed to or from null ! volatile Object waiter; // to control park/unpark final boolean isData; QNode(Object item, boolean isData) { this.item = item; this.isData = isData;
*** 730,745 **** * @return matched item, or s if cancelled */ Object awaitFulfill(QNode s, E e, boolean timed, long nanos) { /* Same idea as TransferStack.awaitFulfill */ final long deadline = timed ? System.nanoTime() + nanos : 0L; ! Thread w = Thread.currentThread(); int spins = (head.next == s) ? (timed ? MAX_TIMED_SPINS : MAX_UNTIMED_SPINS) : 0; for (;;) { ! if (w.isInterrupted()) s.tryCancel(e); Object x = s.item; if (x != e) return x; if (timed) { --- 732,747 ---- * @return matched item, or s if cancelled */ Object awaitFulfill(QNode s, E e, boolean timed, long nanos) { /* Same idea as TransferStack.awaitFulfill */ final long deadline = timed ? System.nanoTime() + nanos : 0L; ! Object w = Strands.currentStrand(); int spins = (head.next == s) ? (timed ? MAX_TIMED_SPINS : MAX_UNTIMED_SPINS) : 0; for (;;) { ! if (Strands.isInterrupted()) s.tryCancel(e); Object x = s.item; if (x != e) return x; if (timed) {
< prev index next >