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
171 /**
172 * Shared internal API for dual stacks and queues.
173 */
174 abstract static class Transferer<E> {
175 /**
176 * Performs a put or take.
177 *
178 * @param e if non-null, the item to be handed to a consumer;
179 * if null, requests that transfer return an item
180 * offered by producer.
181 * @param timed if this operation should timeout
182 * @param nanos the timeout, in nanoseconds
183 * @return if non-null, the item provided or received; if null,
184 * the operation failed due to timeout or interrupt --
185 * the caller can distinguish which of these occurred
186 * by checking Thread.interrupted.
187 */
188 abstract E transfer(E e, boolean timed, long nanos);
189 }
190
191 /**
192 * The number of times to spin before blocking in timed waits.
193 * The value is empirically derived -- it works well across a
194 * variety of processors and OSes. Empirically, the best value
195 * seems not to vary with number of CPUs (beyond 2) so is just
196 * a constant.
197 */
198 static final int MAX_TIMED_SPINS =
199 (Runtime.getRuntime().availableProcessors() < 2) ? 0 : 32;
200
201 /**
202 * The number of times to spin before blocking in untimed waits.
203 * This is greater than timed value because untimed waits spin
204 * faster since they don't need to check times on each spin.
205 */
206 static final int MAX_UNTIMED_SPINS = MAX_TIMED_SPINS * 16;
207
208 /**
209 * The number of nanoseconds for which it is faster to spin
210 * rather than to use timed park. A rough estimate suffices.
211 */
212 static final long SPIN_FOR_TIMEOUT_THRESHOLD = 1000L;
213
214 /** Dual stack */
215 static final class TransferStack<E> extends Transferer<E> {
216 /*
217 * This extends Scherer-Scott dual stack algorithm, differing,
218 * among other ways, by using "covering" nodes rather than
219 * bit-marked pointers: Fulfilling operations push on marker
220 * nodes (with FULFILLING bit set in mode) to reserve a spot
221 * to match a waiting node.
222 */
223
224 /* Modes for SNodes, ORed together in node fields */
225 /** Node represents an unfulfilled consumer */
226 static final int REQUEST = 0;
227 /** Node represents an unfulfilled producer */
228 static final int DATA = 1;
229 /** Node is fulfilling another unfulfilled DATA or REQUEST */
230 static final int FULFILLING = 2;
231
232 /** Returns true if m has fulfilling bit set. */
233 static boolean isFulfilling(int m) { return (m & FULFILLING) != 0; }
234
235 /** Node class for TransferStacks. */
236 static final class SNode {
237 volatile SNode next; // next node in stack
238 volatile SNode match; // the node matched to this
239 volatile Thread waiter; // to control park/unpark
240 Object item; // data; or null for REQUESTs
241 int mode;
242 // Note: item and mode fields don't need to be volatile
243 // since they are always written before, and read after,
244 // other volatile/atomic operations.
245
246 SNode(Object item) {
247 this.item = item;
248 }
249
250 boolean casNext(SNode cmp, SNode val) {
251 return cmp == next &&
252 SNEXT.compareAndSet(this, cmp, val);
253 }
254
255 /**
256 * Tries to match node s to this node, if so, waking up thread.
257 * Fulfillers call tryMatch to identify their waiters.
258 * Waiters block until they have been matched.
259 *
260 * @param s the node to match
261 * @return true if successfully matched to s
262 */
263 boolean tryMatch(SNode s) {
264 if (match == null &&
265 SMATCH.compareAndSet(this, null, s)) {
266 Thread w = waiter;
267 if (w != null) { // waiters need at most one unpark
268 waiter = null;
269 LockSupport.unpark(w);
270 }
271 return true;
272 }
273 return match == s;
274 }
275
276 /**
277 * Tries to cancel a wait by matching node to itself.
278 */
279 void tryCancel() {
280 SMATCH.compareAndSet(this, null, this);
281 }
282
283 boolean isCancelled() {
284 return match == this;
285 }
286
287 // VarHandle mechanics
288 private static final VarHandle SMATCH;
289 private static final VarHandle SNEXT;
290 static {
291 try {
292 MethodHandles.Lookup l = MethodHandles.lookup();
293 SMATCH = l.findVarHandle(SNode.class, "match", SNode.class);
294 SNEXT = l.findVarHandle(SNode.class, "next", SNode.class);
295 } catch (ReflectiveOperationException e) {
296 throw new ExceptionInInitializerError(e);
297 }
298 }
299 }
300
301 /** The head (top) of the stack */
302 volatile SNode head;
303
304 boolean casHead(SNode h, SNode nh) {
305 return h == head &&
306 SHEAD.compareAndSet(this, h, nh);
307 }
308
309 /**
310 * Creates or resets fields of a node. Called only from transfer
311 * where the node to push on stack is lazily created and
312 * reused when possible to help reduce intervals between reads
313 * and CASes of head and to avoid surges of garbage when CASes
314 * to push nodes fail due to contention.
315 */
316 static SNode snode(SNode s, Object e, SNode next, int mode) {
317 if (s == null) s = new SNode(e);
318 s.mode = mode;
319 s.next = next;
320 return s;
321 }
322
323 /**
324 * Puts or takes an item.
325 */
326 @SuppressWarnings("unchecked")
327 E transfer(E e, boolean timed, long nanos) {
328 /*
329 * Basic algorithm is to loop trying one of three actions:
330 *
331 * 1. If apparently empty or already containing nodes of same
332 * mode, try to push node on stack and wait for a match,
333 * returning it, or null if cancelled.
334 *
335 * 2. If apparently containing node of complementary mode,
336 * try to push a fulfilling node on to stack, match
337 * with corresponding waiting node, pop both from
338 * stack, and return matched item. The matching or
339 * unlinking might not actually be necessary because of
340 * other threads performing action 3:
341 *
342 * 3. If top of stack already holds another fulfilling node,
343 * help it out by doing its match and/or pop
344 * operations, and then continue. The code for helping
345 * is essentially the same as for fulfilling, except
346 * that it doesn't return the item.
347 */
348
349 SNode s = null; // constructed/reused as needed
350 int mode = (e == null) ? REQUEST : DATA;
351
352 for (;;) {
353 SNode h = head;
354 if (h == null || h.mode == mode) { // empty or same-mode
355 if (timed && nanos <= 0L) { // can't wait
356 if (h != null && h.isCancelled())
357 casHead(h, h.next); // pop cancelled node
358 else
359 return null;
360 } else if (casHead(h, s = snode(s, e, h, mode))) {
361 SNode m = awaitFulfill(s, timed, nanos);
362 if (m == s) { // wait was cancelled
363 clean(s);
364 return null;
365 }
366 if ((h = head) != null && h.next == s)
367 casHead(h, s.next); // help s's fulfiller
368 return (E) ((mode == REQUEST) ? m.item : s.item);
369 }
370 } else if (!isFulfilling(h.mode)) { // try to fulfill
371 if (h.isCancelled()) // already cancelled
372 casHead(h, h.next); // pop and retry
373 else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
374 for (;;) { // loop until matched or waiters disappear
375 SNode m = s.next; // m is s's match
376 if (m == null) { // all waiters are gone
377 casHead(s, null); // pop fulfill node
378 s = null; // use new node next time
379 break; // restart main loop
380 }
381 SNode mn = m.next;
382 if (m.tryMatch(s)) {
383 casHead(s, mn); // pop both s and m
384 return (E) ((mode == REQUEST) ? m.item : s.item);
385 } else // lost match
386 s.casNext(m, mn); // help unlink
387 }
388 }
389 } else { // help a fulfiller
390 SNode m = h.next; // m is h's match
391 if (m == null) // waiter is gone
392 casHead(h, null); // pop fulfilling node
393 else {
394 SNode mn = m.next;
395 if (m.tryMatch(h)) // help match
396 casHead(h, mn); // pop both h and m
397 else // lost match
398 h.casNext(m, mn); // help unlink
399 }
400 }
401 }
402 }
403
404 /**
405 * Spins/blocks until node s is matched by a fulfill operation.
406 *
407 * @param s the waiting node
408 * @param timed true if timed wait
409 * @param nanos timeout value
410 * @return matched node, or s if cancelled
411 */
412 SNode awaitFulfill(SNode s, boolean timed, long nanos) {
413 /*
414 * When a node/thread is about to block, it sets its waiter
415 * field and then rechecks state at least one more time
416 * before actually parking, thus covering race vs
417 * fulfiller noticing that waiter is non-null so should be
418 * woken.
419 *
420 * When invoked by nodes that appear at the point of call
421 * to be at the head of the stack, calls to park are
422 * preceded by spins to avoid blocking when producers and
423 * consumers are arriving very close in time. This can
424 * happen enough to bother only on multiprocessors.
425 *
426 * The order of checks for returning out of main loop
427 * reflects fact that interrupts have precedence over
428 * normal returns, which have precedence over
429 * timeouts. (So, on timeout, one last check for match is
430 * done before giving up.) Except that calls from untimed
431 * SynchronousQueue.{poll/offer} don't check interrupts
432 * and don't wait at all, so are trapped in transfer
433 * method rather than calling awaitFulfill.
434 */
435 final long deadline = timed ? System.nanoTime() + nanos : 0L;
436 Thread w = Thread.currentThread();
437 int spins = shouldSpin(s)
438 ? (timed ? MAX_TIMED_SPINS : MAX_UNTIMED_SPINS)
439 : 0;
440 for (;;) {
441 if (w.isInterrupted())
442 s.tryCancel();
443 SNode m = s.match;
444 if (m != null)
445 return m;
446 if (timed) {
447 nanos = deadline - System.nanoTime();
448 if (nanos <= 0L) {
449 s.tryCancel();
450 continue;
451 }
452 }
453 if (spins > 0) {
454 Thread.onSpinWait();
455 spins = shouldSpin(s) ? (spins - 1) : 0;
456 }
457 else if (s.waiter == null)
458 s.waiter = w; // establish waiter so can park next iter
459 else if (!timed)
460 LockSupport.park(this);
461 else if (nanos > SPIN_FOR_TIMEOUT_THRESHOLD)
462 LockSupport.parkNanos(this, nanos);
463 }
464 }
465
466 /**
467 * Returns true if node s is at head or there is an active
468 * fulfiller.
469 */
470 boolean shouldSpin(SNode s) {
471 SNode h = head;
472 return (h == s || h == null || isFulfilling(h.mode));
473 }
474
475 /**
476 * Unlinks s from the stack.
477 */
478 void clean(SNode s) {
479 s.item = null; // forget item
480 s.waiter = null; // forget thread
481
482 /*
483 * At worst we may need to traverse entire stack to unlink
484 * s. If there are multiple concurrent calls to clean, we
485 * might not see s if another thread has already removed
486 * it. But we can stop when we see any node known to
487 * follow s. We use s.next unless it too is cancelled, in
488 * which case we try the node one past. We don't check any
489 * further because we don't want to doubly traverse just to
490 * find sentinel.
491 */
492
493 SNode past = s.next;
494 if (past != null && past.isCancelled())
495 past = past.next;
496
497 // Absorb cancelled nodes at head
498 SNode p;
499 while ((p = head) != null && p != past && p.isCancelled())
500 casHead(p, p.next);
501
502 // Unsplice embedded nodes
503 while (p != null && p != past) {
504 SNode n = p.next;
505 if (n != null && n.isCancelled())
506 p.casNext(n, n.next);
507 else
508 p = n;
509 }
510 }
511
512 // VarHandle mechanics
513 private static final VarHandle SHEAD;
514 static {
515 try {
516 MethodHandles.Lookup l = MethodHandles.lookup();
517 SHEAD = l.findVarHandle(TransferStack.class, "head", SNode.class);
518 } catch (ReflectiveOperationException e) {
519 throw new ExceptionInInitializerError(e);
520 }
521 }
522 }
523
524 /** Dual Queue */
525 static final class TransferQueue<E> extends Transferer<E> {
526 /*
527 * This extends Scherer-Scott dual queue algorithm, differing,
528 * among other ways, by using modes within nodes rather than
529 * marked pointers. The algorithm is a little simpler than
530 * that for stacks because fulfillers do not need explicit
531 * nodes, and matching is done by CAS'ing QNode.item field
532 * from non-null to null (for put) or vice versa (for take).
533 */
534
535 /** Node class for TransferQueue. */
536 static final class QNode {
537 volatile QNode next; // next node in queue
538 volatile Object item; // CAS'ed to or from null
539 volatile Thread waiter; // to control park/unpark
540 final boolean isData;
541
542 QNode(Object item, boolean isData) {
543 this.item = item;
544 this.isData = isData;
545 }
546
547 boolean casNext(QNode cmp, QNode val) {
548 return next == cmp &&
549 QNEXT.compareAndSet(this, cmp, val);
550 }
551
552 boolean casItem(Object cmp, Object val) {
553 return item == cmp &&
554 QITEM.compareAndSet(this, cmp, val);
555 }
556
557 /**
558 * Tries to cancel by CAS'ing ref to this as item.
559 */
560 void tryCancel(Object cmp) {
561 QITEM.compareAndSet(this, cmp, this);
562 }
563
564 boolean isCancelled() {
565 return item == this;
566 }
567
568 /**
569 * Returns true if this node is known to be off the queue
570 * because its next pointer has been forgotten due to
571 * an advanceHead operation.
572 */
573 boolean isOffList() {
574 return next == this;
575 }
576
577 // VarHandle mechanics
578 private static final VarHandle QITEM;
579 private static final VarHandle QNEXT;
580 static {
581 try {
582 MethodHandles.Lookup l = MethodHandles.lookup();
583 QITEM = l.findVarHandle(QNode.class, "item", Object.class);
584 QNEXT = l.findVarHandle(QNode.class, "next", QNode.class);
585 } catch (ReflectiveOperationException e) {
586 throw new ExceptionInInitializerError(e);
587 }
588 }
589 }
590
591 /** Head of queue */
592 transient volatile QNode head;
593 /** Tail of queue */
594 transient volatile QNode tail;
595 /**
596 * Reference to a cancelled node that might not yet have been
597 * unlinked from queue because it was the last inserted node
598 * when it was cancelled.
599 */
600 transient volatile QNode cleanMe;
601
602 TransferQueue() {
603 QNode h = new QNode(null, false); // initialize to dummy node.
604 head = h;
605 tail = h;
606 }
607
608 /**
609 * Tries to cas nh as new head; if successful, unlink
610 * old head's next node to avoid garbage retention.
611 */
612 void advanceHead(QNode h, QNode nh) {
613 if (h == head &&
614 QHEAD.compareAndSet(this, h, nh))
615 h.next = h; // forget old next
616 }
617
618 /**
619 * Tries to cas nt as new tail.
620 */
621 void advanceTail(QNode t, QNode nt) {
622 if (tail == t)
623 QTAIL.compareAndSet(this, t, nt);
624 }
625
626 /**
627 * Tries to CAS cleanMe slot.
628 */
629 boolean casCleanMe(QNode cmp, QNode val) {
630 return cleanMe == cmp &&
631 QCLEANME.compareAndSet(this, cmp, val);
632 }
633
634 /**
635 * Puts or takes an item.
636 */
637 @SuppressWarnings("unchecked")
638 E transfer(E e, boolean timed, long nanos) {
639 /* Basic algorithm is to loop trying to take either of
640 * two actions:
641 *
642 * 1. If queue apparently empty or holding same-mode nodes,
643 * try to add node to queue of waiters, wait to be
644 * fulfilled (or cancelled) and return matching item.
645 *
646 * 2. If queue apparently contains waiting items, and this
647 * call is of complementary mode, try to fulfill by CAS'ing
648 * item field of waiting node and dequeuing it, and then
649 * returning matching item.
650 *
651 * In each case, along the way, check for and try to help
652 * advance head and tail on behalf of other stalled/slow
653 * threads.
654 *
655 * The loop starts off with a null check guarding against
656 * seeing uninitialized head or tail values. This never
657 * happens in current SynchronousQueue, but could if
658 * callers held non-volatile/final ref to the
659 * transferer. The check is here anyway because it places
660 * null checks at top of loop, which is usually faster
661 * than having them implicitly interspersed.
662 */
663
664 QNode s = null; // constructed/reused as needed
665 boolean isData = (e != null);
666
667 for (;;) {
668 QNode t = tail;
669 QNode h = head;
670 if (t == null || h == null) // saw uninitialized value
671 continue; // spin
672
673 if (h == t || t.isData == isData) { // empty or same-mode
674 QNode tn = t.next;
675 if (t != tail) // inconsistent read
676 continue;
677 if (tn != null) { // lagging tail
678 advanceTail(t, tn);
679 continue;
680 }
681 if (timed && nanos <= 0L) // can't wait
682 return null;
683 if (s == null)
684 s = new QNode(e, isData);
685 if (!t.casNext(null, s)) // failed to link in
686 continue;
687
688 advanceTail(t, s); // swing tail and wait
689 Object x = awaitFulfill(s, e, timed, nanos);
690 if (x == s) { // wait was cancelled
691 clean(t, s);
692 return null;
693 }
694
695 if (!s.isOffList()) { // not already unlinked
696 advanceHead(t, s); // unlink if head
697 if (x != null) // and forget fields
698 s.item = s;
699 s.waiter = null;
700 }
701 return (x != null) ? (E)x : e;
702
703 } else { // complementary-mode
704 QNode m = h.next; // node to fulfill
705 if (t != tail || m == null || h != head)
706 continue; // inconsistent read
707
708 Object x = m.item;
709 if (isData == (x != null) || // m already fulfilled
710 x == m || // m cancelled
711 !m.casItem(x, e)) { // lost CAS
712 advanceHead(h, m); // dequeue and retry
713 continue;
714 }
715
716 advanceHead(h, m); // successfully fulfilled
717 LockSupport.unpark(m.waiter);
718 return (x != null) ? (E)x : e;
719 }
720 }
721 }
722
723 /**
724 * Spins/blocks until node s is fulfilled.
725 *
726 * @param s the waiting node
727 * @param e the comparison value for checking match
728 * @param timed true if timed wait
729 * @param nanos timeout value
730 * @return matched item, or s if cancelled
731 */
732 Object awaitFulfill(QNode s, E e, boolean timed, long nanos) {
733 /* Same idea as TransferStack.awaitFulfill */
734 final long deadline = timed ? System.nanoTime() + nanos : 0L;
735 Thread w = Thread.currentThread();
736 int spins = (head.next == s)
737 ? (timed ? MAX_TIMED_SPINS : MAX_UNTIMED_SPINS)
738 : 0;
739 for (;;) {
740 if (w.isInterrupted())
741 s.tryCancel(e);
742 Object x = s.item;
743 if (x != e)
744 return x;
745 if (timed) {
746 nanos = deadline - System.nanoTime();
747 if (nanos <= 0L) {
748 s.tryCancel(e);
749 continue;
750 }
751 }
752 if (spins > 0) {
753 --spins;
754 Thread.onSpinWait();
755 }
756 else if (s.waiter == null)
757 s.waiter = w;
758 else if (!timed)
759 LockSupport.park(this);
760 else if (nanos > SPIN_FOR_TIMEOUT_THRESHOLD)
761 LockSupport.parkNanos(this, nanos);
762 }
763 }
764
765 /**
766 * Gets rid of cancelled node s with original predecessor pred.
767 */
768 void clean(QNode pred, QNode s) {
769 s.waiter = null; // forget thread
770 /*
771 * At any given time, exactly one node on list cannot be
772 * deleted -- the last inserted node. To accommodate this,
773 * if we cannot delete s, we save its predecessor as
774 * "cleanMe", deleting the previously saved version
775 * first. At least one of node s or the node previously
776 * saved can always be deleted, so this always terminates.
777 */
778 while (pred.next == s) { // Return early if already unlinked
779 QNode h = head;
780 QNode hn = h.next; // Absorb cancelled first node as head
781 if (hn != null && hn.isCancelled()) {
782 advanceHead(h, hn);
783 continue;
784 }
785 QNode t = tail; // Ensure consistent read for tail
786 if (t == h)
787 return;
788 QNode tn = t.next;
789 if (t != tail)
790 continue;
791 if (tn != null) {
792 advanceTail(t, tn);
793 continue;
794 }
795 if (s != t) { // If not tail, try to unsplice
796 QNode sn = s.next;
797 if (sn == s || pred.casNext(s, sn))
798 return;
799 }
800 QNode dp = cleanMe;
801 if (dp != null) { // Try unlinking previous cancelled node
802 QNode d = dp.next;
803 QNode dn;
804 if (d == null || // d is gone or
805 d == dp || // d is off list or
806 !d.isCancelled() || // d not cancelled or
807 (d != t && // d not tail and
808 (dn = d.next) != null && // has successor
809 dn != d && // that is on list
810 dp.casNext(d, dn))) // d unspliced
811 casCleanMe(dp, null);
812 if (dp == pred)
813 return; // s is already saved node
814 } else if (casCleanMe(null, pred))
815 return; // Postpone cleaning s
816 }
817 }
818
819 // VarHandle mechanics
820 private static final VarHandle QHEAD;
821 private static final VarHandle QTAIL;
822 private static final VarHandle QCLEANME;
823 static {
824 try {
825 MethodHandles.Lookup l = MethodHandles.lookup();
826 QHEAD = l.findVarHandle(TransferQueue.class, "head",
827 QNode.class);
828 QTAIL = l.findVarHandle(TransferQueue.class, "tail",
829 QNode.class);
830 QCLEANME = l.findVarHandle(TransferQueue.class, "cleanMe",
831 QNode.class);
832 } catch (ReflectiveOperationException e) {
833 throw new ExceptionInInitializerError(e);
834 }
835 }
836 }
837
838 /**
839 * The transferer. Set only in constructor, but cannot be declared
840 * as final without further complicating serialization. Since
841 * this is accessed only at most once per public method, there
842 * isn't a noticeable performance penalty for using volatile
843 * instead of final here.
844 */
845 private transient volatile Transferer<E> transferer;
846
847 /**
848 * Creates a {@code SynchronousQueue} with nonfair access policy.
849 */
850 public SynchronousQueue() {
851 this(false);
852 }
853
854 /**
855 * Creates a {@code SynchronousQueue} with the specified fairness policy.
856 *
857 * @param fair if true, waiting threads contend in FIFO order for
858 * access; otherwise the order is unspecified.
859 */
860 public SynchronousQueue(boolean fair) {
861 transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
862 }
863
864 /**
865 * Adds the specified element to this queue, waiting if necessary for
866 * another thread to receive it.
867 *
868 * @throws InterruptedException {@inheritDoc}
869 * @throws NullPointerException {@inheritDoc}
870 */
871 public void put(E e) throws InterruptedException {
872 if (e == null) throw new NullPointerException();
873 if (transferer.transfer(e, false, 0) == null) {
874 Thread.interrupted();
875 throw new InterruptedException();
876 }
877 }
878
879 /**
880 * Inserts the specified element into this queue, waiting if necessary
881 * up to the specified wait time for another thread to receive it.
882 *
883 * @return {@code true} if successful, or {@code false} if the
884 * specified waiting time elapses before a consumer appears
885 * @throws InterruptedException {@inheritDoc}
886 * @throws NullPointerException {@inheritDoc}
887 */
888 public boolean offer(E e, long timeout, TimeUnit unit)
889 throws InterruptedException {
890 if (e == null) throw new NullPointerException();
891 if (transferer.transfer(e, true, unit.toNanos(timeout)) != null)
892 return true;
893 if (!Thread.interrupted())
894 return false;
895 throw new InterruptedException();
896 }
897
898 /**
899 * Inserts the specified element into this queue, if another thread is
900 * waiting to receive it.
901 *
902 * @param e the element to add
903 * @return {@code true} if the element was added to this queue, else
904 * {@code false}
905 * @throws NullPointerException if the specified element is null
906 */
907 public boolean offer(E e) {
908 if (e == null) throw new NullPointerException();
909 return transferer.transfer(e, true, 0) != null;
910 }
911
912 /**
913 * Retrieves and removes the head of this queue, waiting if necessary
914 * for another thread to insert it.
915 *
916 * @return the head of this queue
917 * @throws InterruptedException {@inheritDoc}
918 */
919 public E take() throws InterruptedException {
920 E e = transferer.transfer(null, false, 0);
921 if (e != null)
922 return e;
923 Thread.interrupted();
924 throw new InterruptedException();
925 }
926
927 /**
928 * Retrieves and removes the head of this queue, waiting
929 * if necessary up to the specified wait time, for another thread
930 * to insert it.
931 *
932 * @return the head of this queue, or {@code null} if the
933 * specified waiting time elapses before an element is present
934 * @throws InterruptedException {@inheritDoc}
935 */
936 public E poll(long timeout, TimeUnit unit) throws InterruptedException {
937 E e = transferer.transfer(null, true, unit.toNanos(timeout));
938 if (e != null || !Thread.interrupted())
939 return e;
940 throw new InterruptedException();
941 }
942
943 /**
944 * Retrieves and removes the head of this queue, if another thread
945 * is currently making an element available.
946 *
947 * @return the head of this queue, or {@code null} if no
948 * element is available
949 */
950 public E poll() {
951 return transferer.transfer(null, true, 0);
952 }
953
954 /**
955 * Always returns {@code true}.
956 * A {@code SynchronousQueue} has no internal capacity.
957 *
958 * @return {@code true}
959 */
960 public boolean isEmpty() {
961 return true;
962 }
963
964 /**
965 * Always returns zero.
966 * A {@code SynchronousQueue} has no internal capacity.
967 *
968 * @return zero
969 */
970 public int size() {
971 return 0;
972 }
973
974 /**
975 * Always returns zero.
976 * A {@code SynchronousQueue} has no internal capacity.
977 *
978 * @return zero
979 */
980 public int remainingCapacity() {
981 return 0;
982 }
983
984 /**
985 * Does nothing.
986 * A {@code SynchronousQueue} has no internal capacity.
987 */
988 public void clear() {
989 }
990
991 /**
992 * Always returns {@code false}.
993 * A {@code SynchronousQueue} has no internal capacity.
994 *
995 * @param o the element
996 * @return {@code false}
997 */
998 public boolean contains(Object o) {
999 return false;
1000 }
1001
1002 /**
1003 * Always returns {@code false}.
1004 * A {@code SynchronousQueue} has no internal capacity.
1005 *
1006 * @param o the element to remove
1007 * @return {@code false}
1008 */
1009 public boolean remove(Object o) {
1010 return false;
1011 }
1012
1013 /**
1014 * Returns {@code false} unless the given collection is empty.
1015 * A {@code SynchronousQueue} has no internal capacity.
1016 *
1017 * @param c the collection
1018 * @return {@code false} unless given collection is empty
1019 */
1020 public boolean containsAll(Collection<?> c) {
1021 return c.isEmpty();
1022 }
1023
1024 /**
1025 * Always returns {@code false}.
1026 * A {@code SynchronousQueue} has no internal capacity.
1027 *
1028 * @param c the collection
1029 * @return {@code false}
1030 */
1031 public boolean removeAll(Collection<?> c) {
1032 return false;
1033 }
1034
1035 /**
1036 * Always returns {@code false}.
1037 * A {@code SynchronousQueue} has no internal capacity.
1038 *
1039 * @param c the collection
1040 * @return {@code false}
1041 */
1042 public boolean retainAll(Collection<?> c) {
1043 return false;
1044 }
1045
1046 /**
1047 * Always returns {@code null}.
1048 * A {@code SynchronousQueue} does not return elements
1049 * unless actively waited on.
1050 *
1051 * @return {@code null}
1052 */
1053 public E peek() {
1054 return null;
1055 }
1056
1057 /**
1058 * Returns an empty iterator in which {@code hasNext} always returns
1059 * {@code false}.
1060 *
1061 * @return an empty iterator
1062 */
1063 public Iterator<E> iterator() {
1064 return Collections.emptyIterator();
1065 }
1066
1067 /**
1068 * Returns an empty spliterator in which calls to
1069 * {@link Spliterator#trySplit() trySplit} always return {@code null}.
1070 *
1071 * @return an empty spliterator
1072 * @since 1.8
1073 */
1074 public Spliterator<E> spliterator() {
1075 return Spliterators.emptySpliterator();
1076 }
1077
1078 /**
1079 * Returns a zero-length array.
1080 * @return a zero-length array
1081 */
1082 public Object[] toArray() {
1083 return new Object[0];
1084 }
1085
1086 /**
1087 * Sets the zeroth element of the specified array to {@code null}
1088 * (if the array has non-zero length) and returns it.
1089 *
1090 * @param a the array
1091 * @return the specified array
1092 * @throws NullPointerException if the specified array is null
1093 */
1094 public <T> T[] toArray(T[] a) {
1095 if (a.length > 0)
1096 a[0] = null;
1097 return a;
1098 }
1099
1100 /**
1101 * Always returns {@code "[]"}.
1102 * @return {@code "[]"}
1103 */
1104 public String toString() {
1105 return "[]";
1106 }
1107
1108 /**
1109 * @throws UnsupportedOperationException {@inheritDoc}
1110 * @throws ClassCastException {@inheritDoc}
1111 * @throws NullPointerException {@inheritDoc}
1112 * @throws IllegalArgumentException {@inheritDoc}
1113 */
1114 public int drainTo(Collection<? super E> c) {
1115 Objects.requireNonNull(c);
1116 if (c == this)
1117 throw new IllegalArgumentException();
1118 int n = 0;
1119 for (E e; (e = poll()) != null; n++)
1120 c.add(e);
1121 return n;
1122 }
1123
1124 /**
1125 * @throws UnsupportedOperationException {@inheritDoc}
1126 * @throws ClassCastException {@inheritDoc}
1127 * @throws NullPointerException {@inheritDoc}
1128 * @throws IllegalArgumentException {@inheritDoc}
1129 */
1130 public int drainTo(Collection<? super E> c, int maxElements) {
1131 Objects.requireNonNull(c);
1132 if (c == this)
1133 throw new IllegalArgumentException();
1134 int n = 0;
1135 for (E e; n < maxElements && (e = poll()) != null; n++)
1136 c.add(e);
1137 return n;
1138 }
1139
1140 /*
1141 * To cope with serialization strategy in the 1.5 version of
1142 * SynchronousQueue, we declare some unused classes and fields
1143 * that exist solely to enable serializability across versions.
1144 * These fields are never used, so are initialized only if this
1145 * object is ever serialized or deserialized.
1146 */
1147
1148 @SuppressWarnings("serial")
1149 static class WaitQueue implements java.io.Serializable { }
1150 static class LifoWaitQueue extends WaitQueue {
1151 private static final long serialVersionUID = -3633113410248163686L;
1152 }
1153 static class FifoWaitQueue extends WaitQueue {
1154 private static final long serialVersionUID = -3623113410248163686L;
1155 }
1156 private ReentrantLock qlock;
1157 private WaitQueue waitingProducers;
1158 private WaitQueue waitingConsumers;
1159
1160 /**
1161 * Saves this queue to a stream (that is, serializes it).
1162 * @param s the stream
1163 * @throws java.io.IOException if an I/O error occurs
1164 */
1165 private void writeObject(java.io.ObjectOutputStream s)
1166 throws java.io.IOException {
1167 boolean fair = transferer instanceof TransferQueue;
1168 if (fair) {
1169 qlock = new ReentrantLock(true);
1170 waitingProducers = new FifoWaitQueue();
1171 waitingConsumers = new FifoWaitQueue();
1172 }
1173 else {
1174 qlock = new ReentrantLock();
1175 waitingProducers = new LifoWaitQueue();
1176 waitingConsumers = new LifoWaitQueue();
1177 }
1178 s.defaultWriteObject();
1179 }
1180
1181 /**
1182 * Reconstitutes this queue from a stream (that is, deserializes it).
1183 * @param s the stream
1184 * @throws ClassNotFoundException if the class of a serialized object
1185 * could not be found
1186 * @throws java.io.IOException if an I/O error occurs
1187 */
1188 private void readObject(java.io.ObjectInputStream s)
1189 throws java.io.IOException, ClassNotFoundException {
1190 s.defaultReadObject();
1191 if (waitingProducers instanceof FifoWaitQueue)
1192 transferer = new TransferQueue<E>();
1193 else
1194 transferer = new TransferStack<E>();
1195 }
1196
1197 static {
1198 // Reduce the risk of rare disastrous classloading in first call to
1199 // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
1200 Class<?> ensureLoaded = LockSupport.class;
1201 }
1202 }