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 */ transfer(E e, boolean timed, long nanos)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. */ isFulfilling(int m)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 SNode(Object item)241 SNode(Object item) { 242 this.item = item; 243 } 244 casNext(SNode cmp, SNode val)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 */ tryMatch(SNode s)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 */ tryCancel()275 boolean tryCancel() { 276 return SMATCH.compareAndSet(this, null, this); 277 } 278 isCancelled()279 boolean isCancelled() { 280 return match == this; 281 } 282 isReleasable()283 public final boolean isReleasable() { 284 return match != null || Thread.currentThread().isInterrupted(); 285 } 286 block()287 public final boolean block() { 288 while (!isReleasable()) LockSupport.park(); 289 return true; 290 } 291 forgetWaiter()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 casHead(SNode h, SNode nh)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 */ snode(SNode s, Object e, SNode next, int mode)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") transfer(E e, boolean timed, long nanos)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 */ clean(SNode s)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 QNode(Object item, boolean isData)511 QNode(Object item, boolean isData) { 512 this.item = item; 513 this.isData = isData; 514 } 515 casNext(QNode cmp, QNode val)516 boolean casNext(QNode cmp, QNode val) { 517 return next == cmp && 518 QNEXT.compareAndSet(this, cmp, val); 519 } 520 casItem(Object cmp, Object val)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 */ tryCancel(Object cmp)529 boolean tryCancel(Object cmp) { 530 return QITEM.compareAndSet(this, cmp, this); 531 } 532 isCancelled()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 */ isOffList()542 boolean isOffList() { 543 return next == this; 544 } 545 forgetWaiter()546 void forgetWaiter() { 547 QWAITER.setOpaque(this, null); 548 } 549 isFulfilled()550 boolean isFulfilled() { 551 Object x; 552 return isData == ((x = item) == null) || x == this; 553 } 554 isReleasable()555 public final boolean isReleasable() { 556 Object x; 557 return isData == ((x = item) == null) || x == this || 558 Thread.currentThread().isInterrupted(); 559 } 560 block()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 TransferQueue()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 */ advanceHead(QNode h, QNode nh)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 */ advanceTail(QNode t, QNode nt)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 */ casCleanMe(QNode cmp, QNode val)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") transfer(E e, boolean timed, long nanos)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 */ clean(QNode pred, QNode s)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 */ SynchronousQueue()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 */ SynchronousQueue(boolean fair)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 */ put(E e)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 */ offer(E e, long timeout, TimeUnit unit)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 */ offer(E e)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 */ take()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 */ poll(long timeout, TimeUnit unit)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 */ poll()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 */ isEmpty()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 */ size()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 */ remainingCapacity()946 public int remainingCapacity() { 947 return 0; 948 } 949 950 /** 951 * Does nothing. 952 * A {@code SynchronousQueue} has no internal capacity. 953 */ clear()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 */ contains(Object o)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 */ remove(Object o)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 */ containsAll(Collection<?> c)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 */ removeAll(Collection<?> c)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 */ retainAll(Collection<?> c)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 */ peek()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 */ iterator()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 */ spliterator()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 */ toArray()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 */ toArray(T[] a)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 */ toString()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 */ drainTo(Collection<? super E> c)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 */ drainTo(Collection<? super E> c, int maxElements)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 */ writeObject(java.io.ObjectOutputStream s)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 */ readObject(java.io.ObjectInputStream s)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 } 1169