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