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 with assistance from members of JCP JSR-166
32 * Expert Group and released to the public domain, as explained at
33 * http://creativecommons.org/publicdomain/zero/1.0/
34 */
35
36 package java.util.concurrent;
37
38 import java.util.AbstractQueue;
39 import java.util.Collection;
40 import java.util.Iterator;
41 import java.util.NoSuchElementException;
42 import java.util.Objects;
43 import java.util.Spliterator;
44 import java.util.Spliterators;
45 import java.util.concurrent.locks.Condition;
46 import java.util.concurrent.locks.ReentrantLock;
47 import java.util.function.Consumer;
48 import java.util.function.Predicate;
49
50 /**
51 * An optionally-bounded {@linkplain BlockingDeque blocking deque} based on
52 * linked nodes.
53 *
54 * <p>The optional capacity bound constructor argument serves as a
55 * way to prevent excessive expansion. The capacity, if unspecified,
56 * is equal to {@link Integer#MAX_VALUE}. Linked nodes are
57 * dynamically created upon each insertion unless this would bring the
58 * deque above capacity.
59 *
60 * <p>Most operations run in constant time (ignoring time spent
61 * blocking). Exceptions include {@link #remove(Object) remove},
62 * {@link #removeFirstOccurrence removeFirstOccurrence}, {@link
63 * #removeLastOccurrence removeLastOccurrence}, {@link #contains
64 * contains}, {@link #iterator iterator.remove()}, and the bulk
65 * operations, all of which run in linear time.
66 *
67 * <p>This class and its iterator implement all of the <em>optional</em>
68 * methods of the {@link Collection} and {@link Iterator} interfaces.
69 *
70 * <p>This class is a member of the
71 * <a href="{@docRoot}/../technotes/guides/collections/index.html">
72 * Java Collections Framework</a>.
73 *
74 * @since 1.6
75 * @author Doug Lea
76 * @param <E> the type of elements held in this deque
77 */
78 public class LinkedBlockingDeque<E>
79 extends AbstractQueue<E>
80 implements BlockingDeque<E>, java.io.Serializable {
81
82 /*
83 * Implemented as a simple doubly-linked list protected by a
84 * single lock and using conditions to manage blocking.
85 *
86 * To implement weakly consistent iterators, it appears we need to
87 * keep all Nodes GC-reachable from a predecessor dequeued Node.
88 * That would cause two problems:
89 * - allow a rogue Iterator to cause unbounded memory retention
90 * - cause cross-generational linking of old Nodes to new Nodes if
91 * a Node was tenured while live, which generational GCs have a
92 * hard time dealing with, causing repeated major collections.
93 * However, only non-deleted Nodes need to be reachable from
94 * dequeued Nodes, and reachability does not necessarily have to
95 * be of the kind understood by the GC. We use the trick of
96 * linking a Node that has just been dequeued to itself. Such a
97 * self-link implicitly means to jump to "first" (for next links)
98 * or "last" (for prev links).
99 */
100
101 /*
102 * We have "diamond" multiple interface/abstract class inheritance
103 * here, and that introduces ambiguities. Often we want the
104 * BlockingDeque javadoc combined with the AbstractQueue
105 * implementation, so a lot of method specs are duplicated here.
106 */
107
108 private static final long serialVersionUID = -387911632671998426L;
109
110 /** Doubly-linked list node class */
111 static final class Node<E> {
112 /**
113 * The item, or null if this node has been removed.
114 */
115 E item;
116
117 /**
118 * One of:
119 * - the real predecessor Node
120 * - this Node, meaning the predecessor is tail
121 * - null, meaning there is no predecessor
122 */
123 Node<E> prev;
124
125 /**
126 * One of:
127 * - the real successor Node
128 * - this Node, meaning the successor is head
129 * - null, meaning there is no successor
130 */
131 Node<E> next;
132
133 Node(E x) {
134 item = x;
135 }
136 }
137
138 /**
139 * Pointer to first node.
140 * Invariant: (first == null && last == null) ||
141 * (first.prev == null && first.item != null)
142 */
143 transient Node<E> first;
144
145 /**
146 * Pointer to last node.
147 * Invariant: (first == null && last == null) ||
148 * (last.next == null && last.item != null)
149 */
150 transient Node<E> last;
151
152 /** Number of items in the deque */
153 private transient int count;
154
155 /** Maximum number of items in the deque */
156 private final int capacity;
157
158 /** Main lock guarding all access */
159 final ReentrantLock lock = new ReentrantLock();
160
161 /** Condition for waiting takes */
162 private final Condition notEmpty = lock.newCondition();
163
164 /** Condition for waiting puts */
165 private final Condition notFull = lock.newCondition();
166
167 /**
168 * Creates a {@code LinkedBlockingDeque} with a capacity of
169 * {@link Integer#MAX_VALUE}.
170 */
171 public LinkedBlockingDeque() {
172 this(Integer.MAX_VALUE);
173 }
174
175 /**
176 * Creates a {@code LinkedBlockingDeque} with the given (fixed) capacity.
177 *
178 * @param capacity the capacity of this deque
179 * @throws IllegalArgumentException if {@code capacity} is less than 1
180 */
181 public LinkedBlockingDeque(int capacity) {
182 if (capacity <= 0) throw new IllegalArgumentException();
183 this.capacity = capacity;
184 }
185
186 /**
187 * Creates a {@code LinkedBlockingDeque} with a capacity of
188 * {@link Integer#MAX_VALUE}, initially containing the elements of
189 * the given collection, added in traversal order of the
190 * collection's iterator.
191 *
192 * @param c the collection of elements to initially contain
193 * @throws NullPointerException if the specified collection or any
194 * of its elements are null
195 */
196 public LinkedBlockingDeque(Collection<? extends E> c) {
197 this(Integer.MAX_VALUE);
198 addAll(c);
199 }
200
201
202 // Basic linking and unlinking operations, called only while holding lock
203
204 /**
205 * Links node as first element, or returns false if full.
206 */
207 private boolean linkFirst(Node<E> node) {
208 // assert lock.isHeldByCurrentThread();
209 if (count >= capacity)
210 return false;
211 Node<E> f = first;
212 node.next = f;
213 first = node;
214 if (last == null)
215 last = node;
216 else
217 f.prev = node;
218 ++count;
219 notEmpty.signal();
220 return true;
221 }
222
223 /**
224 * Links node as last element, or returns false if full.
225 */
226 private boolean linkLast(Node<E> node) {
227 // assert lock.isHeldByCurrentThread();
228 if (count >= capacity)
229 return false;
230 Node<E> l = last;
231 node.prev = l;
232 last = node;
233 if (first == null)
234 first = node;
235 else
236 l.next = node;
237 ++count;
238 notEmpty.signal();
239 return true;
240 }
241
242 /**
243 * Removes and returns first element, or null if empty.
244 */
245 private E unlinkFirst() {
246 // assert lock.isHeldByCurrentThread();
247 Node<E> f = first;
248 if (f == null)
249 return null;
250 Node<E> n = f.next;
251 E item = f.item;
252 f.item = null;
253 f.next = f; // help GC
254 first = n;
255 if (n == null)
256 last = null;
257 else
258 n.prev = null;
259 --count;
260 notFull.signal();
261 return item;
262 }
263
264 /**
265 * Removes and returns last element, or null if empty.
266 */
267 private E unlinkLast() {
268 // assert lock.isHeldByCurrentThread();
269 Node<E> l = last;
270 if (l == null)
271 return null;
272 Node<E> p = l.prev;
273 E item = l.item;
274 l.item = null;
275 l.prev = l; // help GC
276 last = p;
277 if (p == null)
278 first = null;
279 else
280 p.next = null;
281 --count;
282 notFull.signal();
283 return item;
284 }
285
286 /**
287 * Unlinks x.
288 */
289 void unlink(Node<E> x) {
290 // assert lock.isHeldByCurrentThread();
291 // assert x.item != null;
292 Node<E> p = x.prev;
293 Node<E> n = x.next;
294 if (p == null) {
295 unlinkFirst();
296 } else if (n == null) {
297 unlinkLast();
298 } else {
299 p.next = n;
300 n.prev = p;
301 x.item = null;
302 // Don't mess with x's links. They may still be in use by
303 // an iterator.
304 --count;
305 notFull.signal();
306 }
307 }
308
309 // BlockingDeque methods
310
311 /**
312 * @throws IllegalStateException if this deque is full
313 * @throws NullPointerException {@inheritDoc}
314 */
315 public void addFirst(E e) {
316 if (!offerFirst(e))
317 throw new IllegalStateException("Deque full");
318 }
319
320 /**
321 * @throws IllegalStateException if this deque is full
322 * @throws NullPointerException {@inheritDoc}
323 */
324 public void addLast(E e) {
325 if (!offerLast(e))
326 throw new IllegalStateException("Deque full");
327 }
328
329 /**
330 * @throws NullPointerException {@inheritDoc}
331 */
332 public boolean offerFirst(E e) {
333 if (e == null) throw new NullPointerException();
334 Node<E> node = new Node<E>(e);
335 final ReentrantLock lock = this.lock;
336 lock.lock();
337 try {
338 return linkFirst(node);
339 } finally {
340 lock.unlock();
341 }
342 }
343
344 /**
345 * @throws NullPointerException {@inheritDoc}
346 */
347 public boolean offerLast(E e) {
348 if (e == null) throw new NullPointerException();
349 Node<E> node = new Node<E>(e);
350 final ReentrantLock lock = this.lock;
351 lock.lock();
352 try {
353 return linkLast(node);
354 } finally {
355 lock.unlock();
356 }
357 }
358
359 /**
360 * @throws NullPointerException {@inheritDoc}
361 * @throws InterruptedException {@inheritDoc}
362 */
363 public void putFirst(E e) throws InterruptedException {
364 if (e == null) throw new NullPointerException();
365 Node<E> node = new Node<E>(e);
366 final ReentrantLock lock = this.lock;
367 lock.lock();
368 try {
369 while (!linkFirst(node))
370 notFull.await();
371 } finally {
372 lock.unlock();
373 }
374 }
375
376 /**
377 * @throws NullPointerException {@inheritDoc}
378 * @throws InterruptedException {@inheritDoc}
379 */
380 public void putLast(E e) throws InterruptedException {
381 if (e == null) throw new NullPointerException();
382 Node<E> node = new Node<E>(e);
383 final ReentrantLock lock = this.lock;
384 lock.lock();
385 try {
386 while (!linkLast(node))
387 notFull.await();
388 } finally {
389 lock.unlock();
390 }
391 }
392
393 /**
394 * @throws NullPointerException {@inheritDoc}
395 * @throws InterruptedException {@inheritDoc}
396 */
397 public boolean offerFirst(E e, long timeout, TimeUnit unit)
398 throws InterruptedException {
399 if (e == null) throw new NullPointerException();
400 Node<E> node = new Node<E>(e);
401 long nanos = unit.toNanos(timeout);
402 final ReentrantLock lock = this.lock;
403 lock.lockInterruptibly();
404 try {
405 while (!linkFirst(node)) {
406 if (nanos <= 0L)
407 return false;
408 nanos = notFull.awaitNanos(nanos);
409 }
410 return true;
411 } finally {
412 lock.unlock();
413 }
414 }
415
416 /**
417 * @throws NullPointerException {@inheritDoc}
418 * @throws InterruptedException {@inheritDoc}
419 */
420 public boolean offerLast(E e, long timeout, TimeUnit unit)
421 throws InterruptedException {
422 if (e == null) throw new NullPointerException();
423 Node<E> node = new Node<E>(e);
424 long nanos = unit.toNanos(timeout);
425 final ReentrantLock lock = this.lock;
426 lock.lockInterruptibly();
427 try {
428 while (!linkLast(node)) {
429 if (nanos <= 0L)
430 return false;
431 nanos = notFull.awaitNanos(nanos);
432 }
433 return true;
434 } finally {
435 lock.unlock();
436 }
437 }
438
439 /**
440 * @throws NoSuchElementException {@inheritDoc}
441 */
442 public E removeFirst() {
443 E x = pollFirst();
444 if (x == null) throw new NoSuchElementException();
445 return x;
446 }
447
448 /**
449 * @throws NoSuchElementException {@inheritDoc}
450 */
451 public E removeLast() {
452 E x = pollLast();
453 if (x == null) throw new NoSuchElementException();
454 return x;
455 }
456
457 public E pollFirst() {
458 final ReentrantLock lock = this.lock;
459 lock.lock();
460 try {
461 return unlinkFirst();
462 } finally {
463 lock.unlock();
464 }
465 }
466
467 public E pollLast() {
468 final ReentrantLock lock = this.lock;
469 lock.lock();
470 try {
471 return unlinkLast();
472 } finally {
473 lock.unlock();
474 }
475 }
476
477 public E takeFirst() throws InterruptedException {
478 final ReentrantLock lock = this.lock;
479 lock.lock();
480 try {
481 E x;
482 while ( (x = unlinkFirst()) == null)
483 notEmpty.await();
484 return x;
485 } finally {
486 lock.unlock();
487 }
488 }
489
490 public E takeLast() throws InterruptedException {
491 final ReentrantLock lock = this.lock;
492 lock.lock();
493 try {
494 E x;
495 while ( (x = unlinkLast()) == null)
496 notEmpty.await();
497 return x;
498 } finally {
499 lock.unlock();
500 }
501 }
502
503 public E pollFirst(long timeout, TimeUnit unit)
504 throws InterruptedException {
505 long nanos = unit.toNanos(timeout);
506 final ReentrantLock lock = this.lock;
507 lock.lockInterruptibly();
508 try {
509 E x;
510 while ( (x = unlinkFirst()) == null) {
511 if (nanos <= 0L)
512 return null;
513 nanos = notEmpty.awaitNanos(nanos);
514 }
515 return x;
516 } finally {
517 lock.unlock();
518 }
519 }
520
521 public E pollLast(long timeout, TimeUnit unit)
522 throws InterruptedException {
523 long nanos = unit.toNanos(timeout);
524 final ReentrantLock lock = this.lock;
525 lock.lockInterruptibly();
526 try {
527 E x;
528 while ( (x = unlinkLast()) == null) {
529 if (nanos <= 0L)
530 return null;
531 nanos = notEmpty.awaitNanos(nanos);
532 }
533 return x;
534 } finally {
535 lock.unlock();
536 }
537 }
538
539 /**
540 * @throws NoSuchElementException {@inheritDoc}
541 */
542 public E getFirst() {
543 E x = peekFirst();
544 if (x == null) throw new NoSuchElementException();
545 return x;
546 }
547
548 /**
549 * @throws NoSuchElementException {@inheritDoc}
550 */
551 public E getLast() {
552 E x = peekLast();
553 if (x == null) throw new NoSuchElementException();
554 return x;
555 }
556
557 public E peekFirst() {
558 final ReentrantLock lock = this.lock;
559 lock.lock();
560 try {
561 return (first == null) ? null : first.item;
562 } finally {
563 lock.unlock();
564 }
565 }
566
567 public E peekLast() {
568 final ReentrantLock lock = this.lock;
569 lock.lock();
570 try {
571 return (last == null) ? null : last.item;
572 } finally {
573 lock.unlock();
574 }
575 }
576
577 public boolean removeFirstOccurrence(Object o) {
578 if (o == null) return false;
579 final ReentrantLock lock = this.lock;
580 lock.lock();
581 try {
582 for (Node<E> p = first; p != null; p = p.next) {
583 if (o.equals(p.item)) {
584 unlink(p);
585 return true;
586 }
587 }
588 return false;
589 } finally {
590 lock.unlock();
591 }
592 }
593
594 public boolean removeLastOccurrence(Object o) {
595 if (o == null) return false;
596 final ReentrantLock lock = this.lock;
597 lock.lock();
598 try {
599 for (Node<E> p = last; p != null; p = p.prev) {
600 if (o.equals(p.item)) {
601 unlink(p);
602 return true;
603 }
604 }
605 return false;
606 } finally {
607 lock.unlock();
608 }
609 }
610
611 // BlockingQueue methods
612
613 /**
614 * Inserts the specified element at the end of this deque unless it would
615 * violate capacity restrictions. When using a capacity-restricted deque,
616 * it is generally preferable to use method {@link #offer(Object) offer}.
617 *
618 * <p>This method is equivalent to {@link #addLast}.
619 *
620 * @throws IllegalStateException if this deque is full
621 * @throws NullPointerException if the specified element is null
622 */
623 public boolean add(E e) {
624 addLast(e);
625 return true;
626 }
627
628 /**
629 * @throws NullPointerException if the specified element is null
630 */
631 public boolean offer(E e) {
632 return offerLast(e);
633 }
634
635 /**
636 * @throws NullPointerException {@inheritDoc}
637 * @throws InterruptedException {@inheritDoc}
638 */
639 public void put(E e) throws InterruptedException {
640 putLast(e);
641 }
642
643 /**
644 * @throws NullPointerException {@inheritDoc}
645 * @throws InterruptedException {@inheritDoc}
646 */
647 public boolean offer(E e, long timeout, TimeUnit unit)
648 throws InterruptedException {
649 return offerLast(e, timeout, unit);
650 }
651
652 /**
653 * Retrieves and removes the head of the queue represented by this deque.
654 * This method differs from {@link #poll poll} only in that it throws an
655 * exception if this deque is empty.
656 *
657 * <p>This method is equivalent to {@link #removeFirst() removeFirst}.
658 *
659 * @return the head of the queue represented by this deque
660 * @throws NoSuchElementException if this deque is empty
661 */
662 public E remove() {
663 return removeFirst();
664 }
665
666 public E poll() {
667 return pollFirst();
668 }
669
670 public E take() throws InterruptedException {
671 return takeFirst();
672 }
673
674 public E poll(long timeout, TimeUnit unit) throws InterruptedException {
675 return pollFirst(timeout, unit);
676 }
677
678 /**
679 * Retrieves, but does not remove, the head of the queue represented by
680 * this deque. This method differs from {@link #peek peek} only in that
681 * it throws an exception if this deque is empty.
682 *
683 * <p>This method is equivalent to {@link #getFirst() getFirst}.
684 *
685 * @return the head of the queue represented by this deque
686 * @throws NoSuchElementException if this deque is empty
687 */
688 public E element() {
689 return getFirst();
690 }
691
692 public E peek() {
693 return peekFirst();
694 }
695
696 /**
697 * Returns the number of additional elements that this deque can ideally
698 * (in the absence of memory or resource constraints) accept without
699 * blocking. This is always equal to the initial capacity of this deque
700 * less the current {@code size} of this deque.
701 *
702 * <p>Note that you <em>cannot</em> always tell if an attempt to insert
703 * an element will succeed by inspecting {@code remainingCapacity}
704 * because it may be the case that another thread is about to
705 * insert or remove an element.
706 */
707 public int remainingCapacity() {
708 final ReentrantLock lock = this.lock;
709 lock.lock();
710 try {
711 return capacity - count;
712 } finally {
713 lock.unlock();
714 }
715 }
716
717 /**
718 * @throws UnsupportedOperationException {@inheritDoc}
719 * @throws ClassCastException {@inheritDoc}
720 * @throws NullPointerException {@inheritDoc}
721 * @throws IllegalArgumentException {@inheritDoc}
722 */
723 public int drainTo(Collection<? super E> c) {
724 return drainTo(c, Integer.MAX_VALUE);
725 }
726
727 /**
728 * @throws UnsupportedOperationException {@inheritDoc}
729 * @throws ClassCastException {@inheritDoc}
730 * @throws NullPointerException {@inheritDoc}
731 * @throws IllegalArgumentException {@inheritDoc}
732 */
733 public int drainTo(Collection<? super E> c, int maxElements) {
734 Objects.requireNonNull(c);
735 if (c == this)
736 throw new IllegalArgumentException();
737 if (maxElements <= 0)
738 return 0;
739 final ReentrantLock lock = this.lock;
740 lock.lock();
741 try {
742 int n = Math.min(maxElements, count);
743 for (int i = 0; i < n; i++) {
744 c.add(first.item); // In this order, in case add() throws.
745 unlinkFirst();
746 }
747 return n;
748 } finally {
749 lock.unlock();
750 }
751 }
752
753 // Stack methods
754
755 /**
756 * @throws IllegalStateException if this deque is full
757 * @throws NullPointerException {@inheritDoc}
758 */
759 public void push(E e) {
760 addFirst(e);
761 }
762
763 /**
764 * @throws NoSuchElementException {@inheritDoc}
765 */
766 public E pop() {
767 return removeFirst();
768 }
769
770 // Collection methods
771
772 /**
773 * Removes the first occurrence of the specified element from this deque.
774 * If the deque does not contain the element, it is unchanged.
775 * More formally, removes the first element {@code e} such that
776 * {@code o.equals(e)} (if such an element exists).
777 * Returns {@code true} if this deque contained the specified element
778 * (or equivalently, if this deque changed as a result of the call).
779 *
780 * <p>This method is equivalent to
781 * {@link #removeFirstOccurrence(Object) removeFirstOccurrence}.
782 *
783 * @param o element to be removed from this deque, if present
784 * @return {@code true} if this deque changed as a result of the call
785 */
786 public boolean remove(Object o) {
787 return removeFirstOccurrence(o);
788 }
789
790 /**
791 * Returns the number of elements in this deque.
792 *
793 * @return the number of elements in this deque
794 */
795 public int size() {
796 final ReentrantLock lock = this.lock;
797 lock.lock();
798 try {
799 return count;
800 } finally {
801 lock.unlock();
802 }
803 }
804
805 /**
806 * Returns {@code true} if this deque contains the specified element.
807 * More formally, returns {@code true} if and only if this deque contains
808 * at least one element {@code e} such that {@code o.equals(e)}.
809 *
810 * @param o object to be checked for containment in this deque
811 * @return {@code true} if this deque contains the specified element
812 */
813 public boolean contains(Object o) {
814 if (o == null) return false;
815 final ReentrantLock lock = this.lock;
816 lock.lock();
817 try {
818 for (Node<E> p = first; p != null; p = p.next)
819 if (o.equals(p.item))
820 return true;
821 return false;
822 } finally {
823 lock.unlock();
824 }
825 }
826
827 /**
828 * Appends all of the elements in the specified collection to the end of
829 * this deque, in the order that they are returned by the specified
830 * collection's iterator. Attempts to {@code addAll} of a deque to
831 * itself result in {@code IllegalArgumentException}.
832 *
833 * @param c the elements to be inserted into this deque
834 * @return {@code true} if this deque changed as a result of the call
835 * @throws NullPointerException if the specified collection or any
836 * of its elements are null
837 * @throws IllegalArgumentException if the collection is this deque
838 * @throws IllegalStateException if this deque is full
839 * @see #add(Object)
840 * @since 9
841 */
842 public boolean addAll(Collection<? extends E> c) {
843 if (c == this)
844 // As historically specified in AbstractQueue#addAll
845 throw new IllegalArgumentException();
846
847 // Copy c into a private chain of Nodes
848 Node<E> beg = null, end = null;
849 int n = 0;
850 for (E e : c) {
851 Objects.requireNonNull(e);
852 n++;
853 Node<E> newNode = new Node<E>(e);
854 if (beg == null)
855 beg = end = newNode;
856 else {
857 end.next = newNode;
858 newNode.prev = end;
859 end = newNode;
860 }
861 }
862 if (beg == null)
863 return false;
864
865 // Atomically append the chain at the end
866 final ReentrantLock lock = this.lock;
867 lock.lock();
868 try {
869 if (count + n <= capacity) {
870 beg.prev = last;
871 if (first == null)
872 first = beg;
873 else
874 last.next = beg;
875 last = end;
876 count += n;
877 notEmpty.signalAll();
878 return true;
879 }
880 } finally {
881 lock.unlock();
882 }
883 // Fall back to historic non-atomic implementation, failing
884 // with IllegalStateException when the capacity is exceeded.
885 return super.addAll(c);
886 }
887
888 /**
889 * Returns an array containing all of the elements in this deque, in
890 * proper sequence (from first to last element).
891 *
892 * <p>The returned array will be "safe" in that no references to it are
893 * maintained by this deque. (In other words, this method must allocate
894 * a new array). The caller is thus free to modify the returned array.
895 *
896 * <p>This method acts as bridge between array-based and collection-based
897 * APIs.
898 *
899 * @return an array containing all of the elements in this deque
900 */
901 @SuppressWarnings("unchecked")
902 public Object[] toArray() {
903 final ReentrantLock lock = this.lock;
904 lock.lock();
905 try {
906 Object[] a = new Object[count];
907 int k = 0;
908 for (Node<E> p = first; p != null; p = p.next)
909 a[k++] = p.item;
910 return a;
911 } finally {
912 lock.unlock();
913 }
914 }
915
916 /**
917 * Returns an array containing all of the elements in this deque, in
918 * proper sequence; the runtime type of the returned array is that of
919 * the specified array. If the deque fits in the specified array, it
920 * is returned therein. Otherwise, a new array is allocated with the
921 * runtime type of the specified array and the size of this deque.
922 *
923 * <p>If this deque fits in the specified array with room to spare
924 * (i.e., the array has more elements than this deque), the element in
925 * the array immediately following the end of the deque is set to
926 * {@code null}.
927 *
928 * <p>Like the {@link #toArray()} method, this method acts as bridge between
929 * array-based and collection-based APIs. Further, this method allows
930 * precise control over the runtime type of the output array, and may,
931 * under certain circumstances, be used to save allocation costs.
932 *
933 * <p>Suppose {@code x} is a deque known to contain only strings.
934 * The following code can be used to dump the deque into a newly
935 * allocated array of {@code String}:
936 *
937 * <pre> {@code String[] y = x.toArray(new String[0]);}</pre>
938 *
939 * Note that {@code toArray(new Object[0])} is identical in function to
940 * {@code toArray()}.
941 *
942 * @param a the array into which the elements of the deque are to
943 * be stored, if it is big enough; otherwise, a new array of the
944 * same runtime type is allocated for this purpose
945 * @return an array containing all of the elements in this deque
946 * @throws ArrayStoreException if the runtime type of the specified array
947 * is not a supertype of the runtime type of every element in
948 * this deque
949 * @throws NullPointerException if the specified array is null
950 */
951 @SuppressWarnings("unchecked")
952 public <T> T[] toArray(T[] a) {
953 final ReentrantLock lock = this.lock;
954 lock.lock();
955 try {
956 if (a.length < count)
957 a = (T[])java.lang.reflect.Array.newInstance
958 (a.getClass().getComponentType(), count);
959
960 int k = 0;
961 for (Node<E> p = first; p != null; p = p.next)
962 a[k++] = (T)p.item;
963 if (a.length > k)
964 a[k] = null;
965 return a;
966 } finally {
967 lock.unlock();
968 }
969 }
970
971 public String toString() {
972 return Helpers.collectionToString(this);
973 }
974
975 /**
976 * Atomically removes all of the elements from this deque.
977 * The deque will be empty after this call returns.
978 */
979 public void clear() {
980 final ReentrantLock lock = this.lock;
981 lock.lock();
982 try {
983 for (Node<E> f = first; f != null; ) {
984 f.item = null;
985 Node<E> n = f.next;
986 f.prev = null;
987 f.next = null;
988 f = n;
989 }
990 first = last = null;
991 count = 0;
992 notFull.signalAll();
993 } finally {
994 lock.unlock();
995 }
996 }
997
998 /**
999 * Used for any element traversal that is not entirely under lock.
1000 * Such traversals must handle both:
1001 * - dequeued nodes (p.next == p)
1002 * - (possibly multiple) interior removed nodes (p.item == null)
1003 */
1004 Node<E> succ(Node<E> p) {
1005 if (p == (p = p.next))
1006 p = first;
1007 return p;
1008 }
1009
1010 /**
1011 * Returns an iterator over the elements in this deque in proper sequence.
1012 * The elements will be returned in order from first (head) to last (tail).
1013 *
1014 * <p>The returned iterator is
1015 * <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>.
1016 *
1017 * @return an iterator over the elements in this deque in proper sequence
1018 */
1019 public Iterator<E> iterator() {
1020 return new Itr();
1021 }
1022
1023 /**
1024 * Returns an iterator over the elements in this deque in reverse
1025 * sequential order. The elements will be returned in order from
1026 * last (tail) to first (head).
1027 *
1028 * <p>The returned iterator is
1029 * <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>.
1030 *
1031 * @return an iterator over the elements in this deque in reverse order
1032 */
1033 public Iterator<E> descendingIterator() {
1034 return new DescendingItr();
1035 }
1036
1037 /**
1038 * Base class for LinkedBlockingDeque iterators.
1039 */
1040 private abstract class AbstractItr implements Iterator<E> {
1041 /**
1042 * The next node to return in next().
1043 */
1044 Node<E> next;
1045
1046 /**
1047 * nextItem holds on to item fields because once we claim that
1048 * an element exists in hasNext(), we must return item read
1049 * under lock even if it was in the process of being removed
1050 * when hasNext() was called.
1051 */
1052 E nextItem;
1053
1054 /**
1055 * Node returned by most recent call to next. Needed by remove.
1056 * Reset to null if this element is deleted by a call to remove.
1057 */
1058 private Node<E> lastRet;
1059
1060 abstract Node<E> firstNode();
1061 abstract Node<E> nextNode(Node<E> n);
1062
1063 private Node<E> succ(Node<E> p) {
1064 if (p == (p = nextNode(p)))
1065 p = firstNode();
1066 return p;
1067 }
1068
1069 AbstractItr() {
1070 // set to initial position
1071 final ReentrantLock lock = LinkedBlockingDeque.this.lock;
1072 lock.lock();
1073 try {
1074 if ((next = firstNode()) != null)
1075 nextItem = next.item;
1076 } finally {
1077 lock.unlock();
1078 }
1079 }
1080
1081 public boolean hasNext() {
1082 return next != null;
1083 }
1084
1085 public E next() {
1086 Node<E> p;
1087 if ((p = next) == null)
1088 throw new NoSuchElementException();
1089 lastRet = p;
1090 E x = nextItem;
1091 final ReentrantLock lock = LinkedBlockingDeque.this.lock;
1092 lock.lock();
1093 try {
1094 E e = null;
1095 for (p = nextNode(p); p != null && (e = p.item) == null; )
1096 p = succ(p);
1097 next = p;
1098 nextItem = e;
1099 } finally {
1100 lock.unlock();
1101 }
1102 return x;
1103 }
1104
1105 public void forEachRemaining(Consumer<? super E> action) {
1106 // A variant of forEachFrom
1107 Objects.requireNonNull(action);
1108 Node<E> p;
1109 if ((p = next) == null) return;
1110 lastRet = p;
1111 next = null;
1112 final ReentrantLock lock = LinkedBlockingDeque.this.lock;
1113 final int batchSize = 64;
1114 Object[] es = null;
1115 int n, len = 1;
1116 do {
1117 lock.lock();
1118 try {
1119 if (es == null) {
1120 p = nextNode(p);
1121 for (Node<E> q = p; q != null; q = succ(q))
1122 if (q.item != null && ++len == batchSize)
1123 break;
1124 es = new Object[len];
1125 es[0] = nextItem;
1126 nextItem = null;
1127 n = 1;
1128 } else
1129 n = 0;
1130 for (; p != null && n < len; p = succ(p))
1131 if ((es[n] = p.item) != null) {
1132 lastRet = p;
1133 n++;
1134 }
1135 } finally {
1136 lock.unlock();
1137 }
1138 for (int i = 0; i < n; i++) {
1139 @SuppressWarnings("unchecked") E e = (E) es[i];
1140 action.accept(e);
1141 }
1142 } while (n > 0 && p != null);
1143 }
1144
1145 public void remove() {
1146 Node<E> n = lastRet;
1147 if (n == null)
1148 throw new IllegalStateException();
1149 lastRet = null;
1150 final ReentrantLock lock = LinkedBlockingDeque.this.lock;
1151 lock.lock();
1152 try {
1153 if (n.item != null)
1154 unlink(n);
1155 } finally {
1156 lock.unlock();
1157 }
1158 }
1159 }
1160
1161 /** Forward iterator */
1162 private class Itr extends AbstractItr {
1163 Itr() {} // prevent access constructor creation
1164 Node<E> firstNode() { return first; }
1165 Node<E> nextNode(Node<E> n) { return n.next; }
1166 }
1167
1168 /** Descending iterator */
1169 private class DescendingItr extends AbstractItr {
1170 DescendingItr() {} // prevent access constructor creation
1171 Node<E> firstNode() { return last; }
1172 Node<E> nextNode(Node<E> n) { return n.prev; }
1173 }
1174
1175 /**
1176 * A customized variant of Spliterators.IteratorSpliterator.
1177 * Keep this class in sync with (very similar) LBQSpliterator.
1178 */
1179 private final class LBDSpliterator implements Spliterator<E> {
1180 static final int MAX_BATCH = 1 << 25; // max batch array size;
1181 Node<E> current; // current node; null until initialized
1182 int batch; // batch size for splits
1183 boolean exhausted; // true when no more nodes
1184 long est = size(); // size estimate
1185
1186 LBDSpliterator() {}
1187
1188 public long estimateSize() { return est; }
1189
1190 public Spliterator<E> trySplit() {
1191 Node<E> h;
1192 if (!exhausted &&
1193 ((h = current) != null || (h = first) != null)
1194 && h.next != null) {
1195 int n = batch = Math.min(batch + 1, MAX_BATCH);
1196 Object[] a = new Object[n];
1197 final ReentrantLock lock = LinkedBlockingDeque.this.lock;
1198 int i = 0;
1199 Node<E> p = current;
1200 lock.lock();
1201 try {
1202 if (p != null || (p = first) != null)
1203 for (; p != null && i < n; p = succ(p))
1204 if ((a[i] = p.item) != null)
1205 i++;
1206 } finally {
1207 lock.unlock();
1208 }
1209 if ((current = p) == null) {
1210 est = 0L;
1211 exhausted = true;
1212 }
1213 else if ((est -= i) < 0L)
1214 est = 0L;
1215 if (i > 0)
1216 return Spliterators.spliterator
1217 (a, 0, i, (Spliterator.ORDERED |
1218 Spliterator.NONNULL |
1219 Spliterator.CONCURRENT));
1220 }
1221 return null;
1222 }
1223
1224 public boolean tryAdvance(Consumer<? super E> action) {
1225 Objects.requireNonNull(action);
1226 if (!exhausted) {
1227 E e = null;
1228 final ReentrantLock lock = LinkedBlockingDeque.this.lock;
1229 lock.lock();
1230 try {
1231 Node<E> p;
1232 if ((p = current) != null || (p = first) != null)
1233 do {
1234 e = p.item;
1235 p = succ(p);
1236 } while (e == null && p != null);
1237 if ((current = p) == null)
1238 exhausted = true;
1239 } finally {
1240 lock.unlock();
1241 }
1242 if (e != null) {
1243 action.accept(e);
1244 return true;
1245 }
1246 }
1247 return false;
1248 }
1249
1250 public void forEachRemaining(Consumer<? super E> action) {
1251 Objects.requireNonNull(action);
1252 if (!exhausted) {
1253 exhausted = true;
1254 Node<E> p = current;
1255 current = null;
1256 forEachFrom(action, p);
1257 }
1258 }
1259
1260 public int characteristics() {
1261 return (Spliterator.ORDERED |
1262 Spliterator.NONNULL |
1263 Spliterator.CONCURRENT);
1264 }
1265 }
1266
1267 /**
1268 * Returns a {@link Spliterator} over the elements in this deque.
1269 *
1270 * <p>The returned spliterator is
1271 * <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>.
1272 *
1273 * <p>The {@code Spliterator} reports {@link Spliterator#CONCURRENT},
1274 * {@link Spliterator#ORDERED}, and {@link Spliterator#NONNULL}.
1275 *
1276 * @implNote
1277 * The {@code Spliterator} implements {@code trySplit} to permit limited
1278 * parallelism.
1279 *
1280 * @return a {@code Spliterator} over the elements in this deque
1281 * @since 1.8
1282 */
1283 public Spliterator<E> spliterator() {
1284 return new LBDSpliterator();
1285 }
1286
1287 /**
1288 * @throws NullPointerException {@inheritDoc}
1289 * @since 9
1290 */
1291 public void forEach(Consumer<? super E> action) {
1292 Objects.requireNonNull(action);
1293 forEachFrom(action, null);
1294 }
1295
1296 /**
1297 * Runs action on each element found during a traversal starting at p.
1298 * If p is null, traversal starts at head.
1299 */
1300 void forEachFrom(Consumer<? super E> action, Node<E> p) {
1301 // Extract batches of elements while holding the lock; then
1302 // run the action on the elements while not
1303 final ReentrantLock lock = this.lock;
1304 final int batchSize = 64; // max number of elements per batch
1305 Object[] es = null; // container for batch of elements
1306 int n, len = 0;
1307 do {
1308 lock.lock();
1309 try {
1310 if (es == null) {
1311 if (p == null) p = first;
1312 for (Node<E> q = p; q != null; q = succ(q))
1313 if (q.item != null && ++len == batchSize)
1314 break;
1315 es = new Object[len];
1316 }
1317 for (n = 0; p != null && n < len; p = succ(p))
1318 if ((es[n] = p.item) != null)
1319 n++;
1320 } finally {
1321 lock.unlock();
1322 }
1323 for (int i = 0; i < n; i++) {
1324 @SuppressWarnings("unchecked") E e = (E) es[i];
1325 action.accept(e);
1326 }
1327 } while (n > 0 && p != null);
1328 }
1329
1330 /**
1331 * @throws NullPointerException {@inheritDoc}
1332 * @since 9
1333 */
1334 public boolean removeIf(Predicate<? super E> filter) {
1335 Objects.requireNonNull(filter);
1336 return bulkRemove(filter);
1337 }
1338
1339 /**
1340 * @throws NullPointerException {@inheritDoc}
1341 * @since 9
1342 */
1343 public boolean removeAll(Collection<?> c) {
1344 Objects.requireNonNull(c);
1345 return bulkRemove(e -> c.contains(e));
1346 }
1347
1348 /**
1349 * @throws NullPointerException {@inheritDoc}
1350 * @since 9
1351 */
1352 public boolean retainAll(Collection<?> c) {
1353 Objects.requireNonNull(c);
1354 return bulkRemove(e -> !c.contains(e));
1355 }
1356
1357 /** Implementation of bulk remove methods. */
1358 @SuppressWarnings("unchecked")
1359 private boolean bulkRemove(Predicate<? super E> filter) {
1360 boolean removed = false;
1361 Node<E> p = null;
1362 final ReentrantLock lock = this.lock;
1363 Node<E>[] nodes = null;
1364 int n, len = 0;
1365 do {
1366 // 1. Extract batch of up to 64 elements while holding the lock.
1367 long deathRow = 0; // "bitset" of size 64
1368 lock.lock();
1369 try {
1370 if (nodes == null) {
1371 if (p == null) p = first;
1372 for (Node<E> q = p; q != null; q = succ(q))
1373 if (q.item != null && ++len == 64)
1374 break;
1375 nodes = (Node<E>[]) new Node<?>[len];
1376 }
1377 for (n = 0; p != null && n < len; p = succ(p))
1378 nodes[n++] = p;
1379 } finally {
1380 lock.unlock();
1381 }
1382
1383 // 2. Run the filter on the elements while lock is free.
1384 for (int i = 0; i < n; i++) {
1385 final E e;
1386 if ((e = nodes[i].item) != null && filter.test(e))
1387 deathRow |= 1L << i;
1388 }
1389
1390 // 3. Remove any filtered elements while holding the lock.
1391 if (deathRow != 0) {
1392 lock.lock();
1393 try {
1394 for (int i = 0; i < n; i++) {
1395 final Node<E> q;
1396 if ((deathRow & (1L << i)) != 0L
1397 && (q = nodes[i]).item != null) {
1398 unlink(q);
1399 removed = true;
1400 }
1401 }
1402 } finally {
1403 lock.unlock();
1404 }
1405 }
1406 } while (n > 0 && p != null);
1407 return removed;
1408 }
1409
1410 /**
1411 * Saves this deque to a stream (that is, serializes it).
1412 *
1413 * @param s the stream
1414 * @throws java.io.IOException if an I/O error occurs
1415 * @serialData The capacity (int), followed by elements (each an
1416 * {@code Object}) in the proper order, followed by a null
1417 */
1418 private void writeObject(java.io.ObjectOutputStream s)
1419 throws java.io.IOException {
1420 final ReentrantLock lock = this.lock;
1421 lock.lock();
1422 try {
1423 // Write out capacity and any hidden stuff
1424 s.defaultWriteObject();
1425 // Write out all elements in the proper order.
1426 for (Node<E> p = first; p != null; p = p.next)
1427 s.writeObject(p.item);
1428 // Use trailing null as sentinel
1429 s.writeObject(null);
1430 } finally {
1431 lock.unlock();
1432 }
1433 }
1434
1435 /**
1436 * Reconstitutes this deque from a stream (that is, deserializes it).
1437 * @param s the stream
1438 * @throws ClassNotFoundException if the class of a serialized object
1439 * could not be found
1440 * @throws java.io.IOException if an I/O error occurs
1441 */
1442 private void readObject(java.io.ObjectInputStream s)
1443 throws java.io.IOException, ClassNotFoundException {
1444 s.defaultReadObject();
1445 count = 0;
1446 first = null;
1447 last = null;
1448 // Read in all elements and place in queue
1449 for (;;) {
1450 @SuppressWarnings("unchecked") E item = (E)s.readObject();
1451 if (item == null)
1452 break;
1453 add(item);
1454 }
1455 }
1456
1457 void checkInvariants() {
1458 // assert lock.isHeldByCurrentThread();
1459 // Nodes may get self-linked or lose their item, but only
1460 // after being unlinked and becoming unreachable from first.
1461 for (Node<E> p = first; p != null; p = p.next) {
1462 // assert p.next != p;
1463 // assert p.item != null;
1464 }
1465 }
1466
1467 }