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 and Martin Buchholz with assistance from members of 32 * JCP JSR-166 Expert Group and released to the public domain, as explained 33 * at http://creativecommons.org/publicdomain/zero/1.0/ 34 */ 35 36 package java.util.concurrent; 37 38 import java.util.AbstractQueue; 39 import java.util.Arrays; 40 import java.util.Collection; 41 import java.util.Iterator; 42 import java.util.NoSuchElementException; 43 import java.util.Objects; 44 import java.util.Queue; 45 import java.util.Spliterator; 46 import java.util.Spliterators; 47 import java.util.function.Consumer; 48 49 // BEGIN android-note 50 // removed link to collections framework docs 51 // END android-note 52 53 /** 54 * An unbounded thread-safe {@linkplain Queue queue} based on linked nodes. 55 * This queue orders elements FIFO (first-in-first-out). 56 * The <em>head</em> of the queue is that element that has been on the 57 * queue the longest time. 58 * The <em>tail</em> of the queue is that element that has been on the 59 * queue the shortest time. New elements 60 * are inserted at the tail of the queue, and the queue retrieval 61 * operations obtain elements at the head of the queue. 62 * A {@code ConcurrentLinkedQueue} is an appropriate choice when 63 * many threads will share access to a common collection. 64 * Like most other concurrent collection implementations, this class 65 * does not permit the use of {@code null} elements. 66 * 67 * <p>This implementation employs an efficient <em>non-blocking</em> 68 * algorithm based on one described in 69 * <a href="http://www.cs.rochester.edu/~scott/papers/1996_PODC_queues.pdf"> 70 * Simple, Fast, and Practical Non-Blocking and Blocking Concurrent Queue 71 * Algorithms</a> by Maged M. Michael and Michael L. Scott. 72 * 73 * <p>Iterators are <i>weakly consistent</i>, returning elements 74 * reflecting the state of the queue at some point at or since the 75 * creation of the iterator. They do <em>not</em> throw {@link 76 * java.util.ConcurrentModificationException}, and may proceed concurrently 77 * with other operations. Elements contained in the queue since the creation 78 * of the iterator will be returned exactly once. 79 * 80 * <p>Beware that, unlike in most collections, the {@code size} method 81 * is <em>NOT</em> a constant-time operation. Because of the 82 * asynchronous nature of these queues, determining the current number 83 * of elements requires a traversal of the elements, and so may report 84 * inaccurate results if this collection is modified during traversal. 85 * Additionally, the bulk operations {@code addAll}, 86 * {@code removeAll}, {@code retainAll}, {@code containsAll}, 87 * {@code equals}, and {@code toArray} are <em>not</em> guaranteed 88 * to be performed atomically. For example, an iterator operating 89 * concurrently with an {@code addAll} operation might view only some 90 * of the added elements. 91 * 92 * <p>This class and its iterator implement all of the <em>optional</em> 93 * methods of the {@link Queue} and {@link Iterator} interfaces. 94 * 95 * <p>Memory consistency effects: As with other concurrent 96 * collections, actions in a thread prior to placing an object into a 97 * {@code ConcurrentLinkedQueue} 98 * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a> 99 * actions subsequent to the access or removal of that element from 100 * the {@code ConcurrentLinkedQueue} in another thread. 101 * 102 * @since 1.5 103 * @author Doug Lea 104 * @param <E> the type of elements held in this queue 105 */ 106 public class ConcurrentLinkedQueue<E> extends AbstractQueue<E> 107 implements Queue<E>, java.io.Serializable { 108 private static final long serialVersionUID = 196745693267521676L; 109 110 /* 111 * This is a modification of the Michael & Scott algorithm, 112 * adapted for a garbage-collected environment, with support for 113 * interior node deletion (to support remove(Object)). For 114 * explanation, read the paper. 115 * 116 * Note that like most non-blocking algorithms in this package, 117 * this implementation relies on the fact that in garbage 118 * collected systems, there is no possibility of ABA problems due 119 * to recycled nodes, so there is no need to use "counted 120 * pointers" or related techniques seen in versions used in 121 * non-GC'ed settings. 122 * 123 * The fundamental invariants are: 124 * - There is exactly one (last) Node with a null next reference, 125 * which is CASed when enqueueing. This last Node can be 126 * reached in O(1) time from tail, but tail is merely an 127 * optimization - it can always be reached in O(N) time from 128 * head as well. 129 * - The elements contained in the queue are the non-null items in 130 * Nodes that are reachable from head. CASing the item 131 * reference of a Node to null atomically removes it from the 132 * queue. Reachability of all elements from head must remain 133 * true even in the case of concurrent modifications that cause 134 * head to advance. A dequeued Node may remain in use 135 * indefinitely due to creation of an Iterator or simply a 136 * poll() that has lost its time slice. 137 * 138 * The above might appear to imply that all Nodes are GC-reachable 139 * from a predecessor dequeued Node. That would cause two problems: 140 * - allow a rogue Iterator to cause unbounded memory retention 141 * - cause cross-generational linking of old Nodes to new Nodes if 142 * a Node was tenured while live, which generational GCs have a 143 * hard time dealing with, causing repeated major collections. 144 * However, only non-deleted Nodes need to be reachable from 145 * dequeued Nodes, and reachability does not necessarily have to 146 * be of the kind understood by the GC. We use the trick of 147 * linking a Node that has just been dequeued to itself. Such a 148 * self-link implicitly means to advance to head. 149 * 150 * Both head and tail are permitted to lag. In fact, failing to 151 * update them every time one could is a significant optimization 152 * (fewer CASes). As with LinkedTransferQueue (see the internal 153 * documentation for that class), we use a slack threshold of two; 154 * that is, we update head/tail when the current pointer appears 155 * to be two or more steps away from the first/last node. 156 * 157 * Since head and tail are updated concurrently and independently, 158 * it is possible for tail to lag behind head (why not)? 159 * 160 * CASing a Node's item reference to null atomically removes the 161 * element from the queue. Iterators skip over Nodes with null 162 * items. Prior implementations of this class had a race between 163 * poll() and remove(Object) where the same element would appear 164 * to be successfully removed by two concurrent operations. The 165 * method remove(Object) also lazily unlinks deleted Nodes, but 166 * this is merely an optimization. 167 * 168 * When constructing a Node (before enqueuing it) we avoid paying 169 * for a volatile write to item by using Unsafe.putObject instead 170 * of a normal write. This allows the cost of enqueue to be 171 * "one-and-a-half" CASes. 172 * 173 * Both head and tail may or may not point to a Node with a 174 * non-null item. If the queue is empty, all items must of course 175 * be null. Upon creation, both head and tail refer to a dummy 176 * Node with null item. Both head and tail are only updated using 177 * CAS, so they never regress, although again this is merely an 178 * optimization. 179 */ 180 181 private static class Node<E> { 182 volatile E item; 183 volatile Node<E> next; 184 } 185 186 /** 187 * Returns a new node holding item. Uses relaxed write because item 188 * can only be seen after piggy-backing publication via casNext. 189 */ newNode(E item)190 static <E> Node<E> newNode(E item) { 191 Node<E> node = new Node<E>(); 192 U.putObject(node, ITEM, item); 193 return node; 194 } 195 casItem(Node<E> node, E cmp, E val)196 static <E> boolean casItem(Node<E> node, E cmp, E val) { 197 return U.compareAndSwapObject(node, ITEM, cmp, val); 198 } 199 lazySetNext(Node<E> node, Node<E> val)200 static <E> void lazySetNext(Node<E> node, Node<E> val) { 201 U.putOrderedObject(node, NEXT, val); 202 } 203 casNext(Node<E> node, Node<E> cmp, Node<E> val)204 static <E> boolean casNext(Node<E> node, Node<E> cmp, Node<E> val) { 205 return U.compareAndSwapObject(node, NEXT, cmp, val); 206 } 207 208 /** 209 * A node from which the first live (non-deleted) node (if any) 210 * can be reached in O(1) time. 211 * Invariants: 212 * - all live nodes are reachable from head via succ() 213 * - head != null 214 * - (tmp = head).next != tmp || tmp != head 215 * Non-invariants: 216 * - head.item may or may not be null. 217 * - it is permitted for tail to lag behind head, that is, for tail 218 * to not be reachable from head! 219 */ 220 transient volatile Node<E> head; 221 222 /** 223 * A node from which the last node on list (that is, the unique 224 * node with node.next == null) can be reached in O(1) time. 225 * Invariants: 226 * - the last node is always reachable from tail via succ() 227 * - tail != null 228 * Non-invariants: 229 * - tail.item may or may not be null. 230 * - it is permitted for tail to lag behind head, that is, for tail 231 * to not be reachable from head! 232 * - tail.next may or may not be self-pointing to tail. 233 */ 234 private transient volatile Node<E> tail; 235 236 /** 237 * Creates a {@code ConcurrentLinkedQueue} that is initially empty. 238 */ ConcurrentLinkedQueue()239 public ConcurrentLinkedQueue() { 240 head = tail = newNode(null); 241 } 242 243 /** 244 * Creates a {@code ConcurrentLinkedQueue} 245 * initially containing the elements of the given collection, 246 * added in traversal order of the collection's iterator. 247 * 248 * @param c the collection of elements to initially contain 249 * @throws NullPointerException if the specified collection or any 250 * of its elements are null 251 */ ConcurrentLinkedQueue(Collection<? extends E> c)252 public ConcurrentLinkedQueue(Collection<? extends E> c) { 253 Node<E> h = null, t = null; 254 for (E e : c) { 255 Node<E> newNode = newNode(Objects.requireNonNull(e)); 256 if (h == null) 257 h = t = newNode; 258 else { 259 lazySetNext(t, newNode); 260 t = newNode; 261 } 262 } 263 if (h == null) 264 h = t = newNode(null); 265 head = h; 266 tail = t; 267 } 268 269 // Have to override just to update the javadoc 270 271 /** 272 * Inserts the specified element at the tail of this queue. 273 * As the queue is unbounded, this method will never throw 274 * {@link IllegalStateException} or return {@code false}. 275 * 276 * @return {@code true} (as specified by {@link Collection#add}) 277 * @throws NullPointerException if the specified element is null 278 */ add(E e)279 public boolean add(E e) { 280 return offer(e); 281 } 282 283 /** 284 * Tries to CAS head to p. If successful, repoint old head to itself 285 * as sentinel for succ(), below. 286 */ updateHead(Node<E> h, Node<E> p)287 final void updateHead(Node<E> h, Node<E> p) { 288 // assert h != null && p != null && (h == p || h.item == null); 289 if (h != p && casHead(h, p)) 290 lazySetNext(h, h); 291 } 292 293 /** 294 * Returns the successor of p, or the head node if p.next has been 295 * linked to self, which will only be true if traversing with a 296 * stale pointer that is now off the list. 297 */ succ(Node<E> p)298 final Node<E> succ(Node<E> p) { 299 Node<E> next = p.next; 300 return (p == next) ? head : next; 301 } 302 303 /** 304 * Inserts the specified element at the tail of this queue. 305 * As the queue is unbounded, this method will never return {@code false}. 306 * 307 * @return {@code true} (as specified by {@link Queue#offer}) 308 * @throws NullPointerException if the specified element is null 309 */ offer(E e)310 public boolean offer(E e) { 311 final Node<E> newNode = newNode(Objects.requireNonNull(e)); 312 313 for (Node<E> t = tail, p = t;;) { 314 Node<E> q = p.next; 315 if (q == null) { 316 // p is last node 317 if (casNext(p, null, newNode)) { 318 // Successful CAS is the linearization point 319 // for e to become an element of this queue, 320 // and for newNode to become "live". 321 if (p != t) // hop two nodes at a time 322 casTail(t, newNode); // Failure is OK. 323 return true; 324 } 325 // Lost CAS race to another thread; re-read next 326 } 327 else if (p == q) 328 // We have fallen off list. If tail is unchanged, it 329 // will also be off-list, in which case we need to 330 // jump to head, from which all live nodes are always 331 // reachable. Else the new tail is a better bet. 332 p = (t != (t = tail)) ? t : head; 333 else 334 // Check for tail updates after two hops. 335 p = (p != t && t != (t = tail)) ? t : q; 336 } 337 } 338 poll()339 public E poll() { 340 restartFromHead: 341 for (;;) { 342 for (Node<E> h = head, p = h, q;;) { 343 E item = p.item; 344 345 if (item != null && casItem(p, item, null)) { 346 // Successful CAS is the linearization point 347 // for item to be removed from this queue. 348 if (p != h) // hop two nodes at a time 349 updateHead(h, ((q = p.next) != null) ? q : p); 350 return item; 351 } 352 else if ((q = p.next) == null) { 353 updateHead(h, p); 354 return null; 355 } 356 else if (p == q) 357 continue restartFromHead; 358 else 359 p = q; 360 } 361 } 362 } 363 peek()364 public E peek() { 365 restartFromHead: 366 for (;;) { 367 for (Node<E> h = head, p = h, q;;) { 368 E item = p.item; 369 if (item != null || (q = p.next) == null) { 370 updateHead(h, p); 371 return item; 372 } 373 else if (p == q) 374 continue restartFromHead; 375 else 376 p = q; 377 } 378 } 379 } 380 381 /** 382 * Returns the first live (non-deleted) node on list, or null if none. 383 * This is yet another variant of poll/peek; here returning the 384 * first node, not element. We could make peek() a wrapper around 385 * first(), but that would cost an extra volatile read of item, 386 * and the need to add a retry loop to deal with the possibility 387 * of losing a race to a concurrent poll(). 388 */ first()389 Node<E> first() { 390 restartFromHead: 391 for (;;) { 392 for (Node<E> h = head, p = h, q;;) { 393 boolean hasItem = (p.item != null); 394 if (hasItem || (q = p.next) == null) { 395 updateHead(h, p); 396 return hasItem ? p : null; 397 } 398 else if (p == q) 399 continue restartFromHead; 400 else 401 p = q; 402 } 403 } 404 } 405 406 /** 407 * Returns {@code true} if this queue contains no elements. 408 * 409 * @return {@code true} if this queue contains no elements 410 */ isEmpty()411 public boolean isEmpty() { 412 return first() == null; 413 } 414 415 /** 416 * Returns the number of elements in this queue. If this queue 417 * contains more than {@code Integer.MAX_VALUE} elements, returns 418 * {@code Integer.MAX_VALUE}. 419 * 420 * <p>Beware that, unlike in most collections, this method is 421 * <em>NOT</em> a constant-time operation. Because of the 422 * asynchronous nature of these queues, determining the current 423 * number of elements requires an O(n) traversal. 424 * Additionally, if elements are added or removed during execution 425 * of this method, the returned result may be inaccurate. Thus, 426 * this method is typically not very useful in concurrent 427 * applications. 428 * 429 * @return the number of elements in this queue 430 */ size()431 public int size() { 432 restartFromHead: for (;;) { 433 int count = 0; 434 for (Node<E> p = first(); p != null;) { 435 if (p.item != null) 436 if (++count == Integer.MAX_VALUE) 437 break; // @see Collection.size() 438 if (p == (p = p.next)) 439 continue restartFromHead; 440 } 441 return count; 442 } 443 } 444 445 /** 446 * Returns {@code true} if this queue contains the specified element. 447 * More formally, returns {@code true} if and only if this queue contains 448 * at least one element {@code e} such that {@code o.equals(e)}. 449 * 450 * @param o object to be checked for containment in this queue 451 * @return {@code true} if this queue contains the specified element 452 */ contains(Object o)453 public boolean contains(Object o) { 454 if (o != null) { 455 for (Node<E> p = first(); p != null; p = succ(p)) { 456 E item = p.item; 457 if (item != null && o.equals(item)) 458 return true; 459 } 460 } 461 return false; 462 } 463 464 /** 465 * Removes a single instance of the specified element from this queue, 466 * if it is present. More formally, removes an element {@code e} such 467 * that {@code o.equals(e)}, if this queue contains one or more such 468 * elements. 469 * Returns {@code true} if this queue contained the specified element 470 * (or equivalently, if this queue changed as a result of the call). 471 * 472 * @param o element to be removed from this queue, if present 473 * @return {@code true} if this queue changed as a result of the call 474 */ remove(Object o)475 public boolean remove(Object o) { 476 if (o != null) { 477 Node<E> next, pred = null; 478 for (Node<E> p = first(); p != null; pred = p, p = next) { 479 boolean removed = false; 480 E item = p.item; 481 if (item != null) { 482 if (!o.equals(item)) { 483 next = succ(p); 484 continue; 485 } 486 removed = casItem(p, item, null); 487 } 488 489 next = succ(p); 490 if (pred != null && next != null) // unlink 491 casNext(pred, p, next); 492 if (removed) 493 return true; 494 } 495 } 496 return false; 497 } 498 499 /** 500 * Appends all of the elements in the specified collection to the end of 501 * this queue, in the order that they are returned by the specified 502 * collection's iterator. Attempts to {@code addAll} of a queue to 503 * itself result in {@code IllegalArgumentException}. 504 * 505 * @param c the elements to be inserted into this queue 506 * @return {@code true} if this queue changed as a result of the call 507 * @throws NullPointerException if the specified collection or any 508 * of its elements are null 509 * @throws IllegalArgumentException if the collection is this queue 510 */ addAll(Collection<? extends E> c)511 public boolean addAll(Collection<? extends E> c) { 512 if (c == this) 513 // As historically specified in AbstractQueue#addAll 514 throw new IllegalArgumentException(); 515 516 // Copy c into a private chain of Nodes 517 Node<E> beginningOfTheEnd = null, last = null; 518 for (E e : c) { 519 Node<E> newNode = newNode(Objects.requireNonNull(e)); 520 if (beginningOfTheEnd == null) 521 beginningOfTheEnd = last = newNode; 522 else { 523 lazySetNext(last, newNode); 524 last = newNode; 525 } 526 } 527 if (beginningOfTheEnd == null) 528 return false; 529 530 // Atomically append the chain at the tail of this collection 531 for (Node<E> t = tail, p = t;;) { 532 Node<E> q = p.next; 533 if (q == null) { 534 // p is last node 535 if (casNext(p, null, beginningOfTheEnd)) { 536 // Successful CAS is the linearization point 537 // for all elements to be added to this queue. 538 if (!casTail(t, last)) { 539 // Try a little harder to update tail, 540 // since we may be adding many elements. 541 t = tail; 542 if (last.next == null) 543 casTail(t, last); 544 } 545 return true; 546 } 547 // Lost CAS race to another thread; re-read next 548 } 549 else if (p == q) 550 // We have fallen off list. If tail is unchanged, it 551 // will also be off-list, in which case we need to 552 // jump to head, from which all live nodes are always 553 // reachable. Else the new tail is a better bet. 554 p = (t != (t = tail)) ? t : head; 555 else 556 // Check for tail updates after two hops. 557 p = (p != t && t != (t = tail)) ? t : q; 558 } 559 } 560 toString()561 public String toString() { 562 String[] a = null; 563 restartFromHead: for (;;) { 564 int charLength = 0; 565 int size = 0; 566 for (Node<E> p = first(); p != null;) { 567 E item = p.item; 568 if (item != null) { 569 if (a == null) 570 a = new String[4]; 571 else if (size == a.length) 572 a = Arrays.copyOf(a, 2 * size); 573 String s = item.toString(); 574 a[size++] = s; 575 charLength += s.length(); 576 } 577 if (p == (p = p.next)) 578 continue restartFromHead; 579 } 580 581 if (size == 0) 582 return "[]"; 583 584 return Helpers.toString(a, size, charLength); 585 } 586 } 587 toArrayInternal(Object[] a)588 private Object[] toArrayInternal(Object[] a) { 589 Object[] x = a; 590 restartFromHead: for (;;) { 591 int size = 0; 592 for (Node<E> p = first(); p != null;) { 593 E item = p.item; 594 if (item != null) { 595 if (x == null) 596 x = new Object[4]; 597 else if (size == x.length) 598 x = Arrays.copyOf(x, 2 * (size + 4)); 599 x[size++] = item; 600 } 601 if (p == (p = p.next)) 602 continue restartFromHead; 603 } 604 if (x == null) 605 return new Object[0]; 606 else if (a != null && size <= a.length) { 607 if (a != x) 608 System.arraycopy(x, 0, a, 0, size); 609 if (size < a.length) 610 a[size] = null; 611 return a; 612 } 613 return (size == x.length) ? x : Arrays.copyOf(x, size); 614 } 615 } 616 617 /** 618 * Returns an array containing all of the elements in this queue, in 619 * proper sequence. 620 * 621 * <p>The returned array will be "safe" in that no references to it are 622 * maintained by this queue. (In other words, this method must allocate 623 * a new array). The caller is thus free to modify the returned array. 624 * 625 * <p>This method acts as bridge between array-based and collection-based 626 * APIs. 627 * 628 * @return an array containing all of the elements in this queue 629 */ toArray()630 public Object[] toArray() { 631 return toArrayInternal(null); 632 } 633 634 /** 635 * Returns an array containing all of the elements in this queue, in 636 * proper sequence; the runtime type of the returned array is that of 637 * the specified array. If the queue fits in the specified array, it 638 * is returned therein. Otherwise, a new array is allocated with the 639 * runtime type of the specified array and the size of this queue. 640 * 641 * <p>If this queue fits in the specified array with room to spare 642 * (i.e., the array has more elements than this queue), the element in 643 * the array immediately following the end of the queue is set to 644 * {@code null}. 645 * 646 * <p>Like the {@link #toArray()} method, this method acts as bridge between 647 * array-based and collection-based APIs. Further, this method allows 648 * precise control over the runtime type of the output array, and may, 649 * under certain circumstances, be used to save allocation costs. 650 * 651 * <p>Suppose {@code x} is a queue known to contain only strings. 652 * The following code can be used to dump the queue into a newly 653 * allocated array of {@code String}: 654 * 655 * <pre> {@code String[] y = x.toArray(new String[0]);}</pre> 656 * 657 * Note that {@code toArray(new Object[0])} is identical in function to 658 * {@code toArray()}. 659 * 660 * @param a the array into which the elements of the queue are to 661 * be stored, if it is big enough; otherwise, a new array of the 662 * same runtime type is allocated for this purpose 663 * @return an array containing all of the elements in this queue 664 * @throws ArrayStoreException if the runtime type of the specified array 665 * is not a supertype of the runtime type of every element in 666 * this queue 667 * @throws NullPointerException if the specified array is null 668 */ 669 @SuppressWarnings("unchecked") toArray(T[] a)670 public <T> T[] toArray(T[] a) { 671 if (a == null) throw new NullPointerException(); 672 return (T[]) toArrayInternal(a); 673 } 674 675 /** 676 * Returns an iterator over the elements in this queue in proper sequence. 677 * The elements will be returned in order from first (head) to last (tail). 678 * 679 * <p>The returned iterator is 680 * <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>. 681 * 682 * @return an iterator over the elements in this queue in proper sequence 683 */ iterator()684 public Iterator<E> iterator() { 685 return new Itr(); 686 } 687 688 private class Itr implements Iterator<E> { 689 /** 690 * Next node to return item for. 691 */ 692 private Node<E> nextNode; 693 694 /** 695 * nextItem holds on to item fields because once we claim 696 * that an element exists in hasNext(), we must return it in 697 * the following next() call even if it was in the process of 698 * being removed when hasNext() was called. 699 */ 700 private E nextItem; 701 702 /** 703 * Node of the last returned item, to support remove. 704 */ 705 private Node<E> lastRet; 706 Itr()707 Itr() { 708 restartFromHead: for (;;) { 709 Node<E> h, p, q; 710 for (p = h = head;; p = q) { 711 E item; 712 if ((item = p.item) != null) { 713 nextNode = p; 714 nextItem = item; 715 break; 716 } 717 else if ((q = p.next) == null) 718 break; 719 else if (p == q) 720 continue restartFromHead; 721 } 722 updateHead(h, p); 723 return; 724 } 725 } 726 hasNext()727 public boolean hasNext() { 728 return nextItem != null; 729 } 730 next()731 public E next() { 732 final Node<E> pred = nextNode; 733 if (pred == null) throw new NoSuchElementException(); 734 // assert nextItem != null; 735 lastRet = pred; 736 E item = null; 737 738 for (Node<E> p = succ(pred), q;; p = q) { 739 if (p == null || (item = p.item) != null) { 740 nextNode = p; 741 E x = nextItem; 742 nextItem = item; 743 return x; 744 } 745 // unlink deleted nodes 746 if ((q = succ(p)) != null) 747 casNext(pred, p, q); 748 } 749 } 750 remove()751 public void remove() { 752 Node<E> l = lastRet; 753 if (l == null) throw new IllegalStateException(); 754 // rely on a future traversal to relink. 755 l.item = null; 756 lastRet = null; 757 } 758 } 759 760 /** 761 * Saves this queue to a stream (that is, serializes it). 762 * 763 * @param s the stream 764 * @throws java.io.IOException if an I/O error occurs 765 * @serialData All of the elements (each an {@code E}) in 766 * the proper order, followed by a null 767 */ writeObject(java.io.ObjectOutputStream s)768 private void writeObject(java.io.ObjectOutputStream s) 769 throws java.io.IOException { 770 771 // Write out any hidden stuff 772 s.defaultWriteObject(); 773 774 // Write out all elements in the proper order. 775 for (Node<E> p = first(); p != null; p = succ(p)) { 776 Object item = p.item; 777 if (item != null) 778 s.writeObject(item); 779 } 780 781 // Use trailing null as sentinel 782 s.writeObject(null); 783 } 784 785 /** 786 * Reconstitutes this queue from a stream (that is, deserializes it). 787 * @param s the stream 788 * @throws ClassNotFoundException if the class of a serialized object 789 * could not be found 790 * @throws java.io.IOException if an I/O error occurs 791 */ readObject(java.io.ObjectInputStream s)792 private void readObject(java.io.ObjectInputStream s) 793 throws java.io.IOException, ClassNotFoundException { 794 s.defaultReadObject(); 795 796 // Read in elements until trailing null sentinel found 797 Node<E> h = null, t = null; 798 for (Object item; (item = s.readObject()) != null; ) { 799 @SuppressWarnings("unchecked") 800 Node<E> newNode = newNode((E) item); 801 if (h == null) 802 h = t = newNode; 803 else { 804 lazySetNext(t, newNode); 805 t = newNode; 806 } 807 } 808 if (h == null) 809 h = t = newNode(null); 810 head = h; 811 tail = t; 812 } 813 814 /** A customized variant of Spliterators.IteratorSpliterator */ 815 static final class CLQSpliterator<E> implements Spliterator<E> { 816 static final int MAX_BATCH = 1 << 25; // max batch array size; 817 final ConcurrentLinkedQueue<E> queue; 818 Node<E> current; // current node; null until initialized 819 int batch; // batch size for splits 820 boolean exhausted; // true when no more nodes CLQSpliterator(ConcurrentLinkedQueue<E> queue)821 CLQSpliterator(ConcurrentLinkedQueue<E> queue) { 822 this.queue = queue; 823 } 824 trySplit()825 public Spliterator<E> trySplit() { 826 Node<E> p; 827 final ConcurrentLinkedQueue<E> q = this.queue; 828 int b = batch; 829 int n = (b <= 0) ? 1 : (b >= MAX_BATCH) ? MAX_BATCH : b + 1; 830 if (!exhausted && 831 ((p = current) != null || (p = q.first()) != null) && 832 p.next != null) { 833 Object[] a = new Object[n]; 834 int i = 0; 835 do { 836 if ((a[i] = p.item) != null) 837 ++i; 838 if (p == (p = p.next)) 839 p = q.first(); 840 } while (p != null && i < n); 841 if ((current = p) == null) 842 exhausted = true; 843 if (i > 0) { 844 batch = i; 845 return Spliterators.spliterator 846 (a, 0, i, (Spliterator.ORDERED | 847 Spliterator.NONNULL | 848 Spliterator.CONCURRENT)); 849 } 850 } 851 return null; 852 } 853 forEachRemaining(Consumer<? super E> action)854 public void forEachRemaining(Consumer<? super E> action) { 855 Node<E> p; 856 if (action == null) throw new NullPointerException(); 857 final ConcurrentLinkedQueue<E> q = this.queue; 858 if (!exhausted && 859 ((p = current) != null || (p = q.first()) != null)) { 860 exhausted = true; 861 do { 862 E e = p.item; 863 if (p == (p = p.next)) 864 p = q.first(); 865 if (e != null) 866 action.accept(e); 867 } while (p != null); 868 } 869 } 870 tryAdvance(Consumer<? super E> action)871 public boolean tryAdvance(Consumer<? super E> action) { 872 Node<E> p; 873 if (action == null) throw new NullPointerException(); 874 final ConcurrentLinkedQueue<E> q = this.queue; 875 if (!exhausted && 876 ((p = current) != null || (p = q.first()) != null)) { 877 E e; 878 do { 879 e = p.item; 880 if (p == (p = p.next)) 881 p = q.first(); 882 } while (e == null && p != null); 883 if ((current = p) == null) 884 exhausted = true; 885 if (e != null) { 886 action.accept(e); 887 return true; 888 } 889 } 890 return false; 891 } 892 estimateSize()893 public long estimateSize() { return Long.MAX_VALUE; } 894 characteristics()895 public int characteristics() { 896 return Spliterator.ORDERED | Spliterator.NONNULL | 897 Spliterator.CONCURRENT; 898 } 899 } 900 901 /** 902 * Returns a {@link Spliterator} over the elements in this queue. 903 * 904 * <p>The returned spliterator is 905 * <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>. 906 * 907 * <p>The {@code Spliterator} reports {@link Spliterator#CONCURRENT}, 908 * {@link Spliterator#ORDERED}, and {@link Spliterator#NONNULL}. 909 * 910 * @implNote 911 * The {@code Spliterator} implements {@code trySplit} to permit limited 912 * parallelism. 913 * 914 * @return a {@code Spliterator} over the elements in this queue 915 * @since 1.8 916 */ 917 @Override spliterator()918 public Spliterator<E> spliterator() { 919 return new CLQSpliterator<E>(this); 920 } 921 casTail(Node<E> cmp, Node<E> val)922 private boolean casTail(Node<E> cmp, Node<E> val) { 923 return U.compareAndSwapObject(this, TAIL, cmp, val); 924 } 925 casHead(Node<E> cmp, Node<E> val)926 private boolean casHead(Node<E> cmp, Node<E> val) { 927 return U.compareAndSwapObject(this, HEAD, cmp, val); 928 } 929 930 // Unsafe mechanics 931 932 private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe(); 933 private static final long HEAD; 934 private static final long TAIL; 935 private static final long ITEM; 936 private static final long NEXT; 937 static { 938 try { 939 HEAD = U.objectFieldOffset 940 (ConcurrentLinkedQueue.class.getDeclaredField("head")); 941 TAIL = U.objectFieldOffset 942 (ConcurrentLinkedQueue.class.getDeclaredField("tail")); 943 ITEM = U.objectFieldOffset 944 (Node.class.getDeclaredField("item")); 945 NEXT = U.objectFieldOffset 946 (Node.class.getDeclaredField("next")); 947 } catch (ReflectiveOperationException e) { 948 throw new Error(e); 949 } 950 } 951 } 952