2 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
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.
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).
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.
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
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
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/
37 package java.util.concurrent;
38 import java.util.concurrent.locks.*;
42 * An unbounded {@linkplain BlockingQueue blocking queue} of
43 * <tt>Delayed</tt> elements, in which an element can only be taken
44 * when its delay has expired. The <em>head</em> of the queue is that
45 * <tt>Delayed</tt> element whose delay expired furthest in the
46 * past. If no delay has expired there is no head and <tt>poll</tt>
47 * will return <tt>null</tt>. Expiration occurs when an element's
48 * <tt>getDelay(TimeUnit.NANOSECONDS)</tt> method returns a value less
49 * than or equal to zero. Even though unexpired elements cannot be
50 * removed using <tt>take</tt> or <tt>poll</tt>, they are otherwise
51 * treated as normal elements. For example, the <tt>size</tt> method
52 * returns the count of both expired and unexpired elements.
53 * This queue does not permit null elements.
55 * <p>This class and its iterator implement all of the
56 * <em>optional</em> methods of the {@link Collection} and {@link
57 * Iterator} interfaces.
59 * <p>This class is a member of the
60 * <a href="{@docRoot}/../technotes/guides/collections/index.html">
61 * Java Collections Framework</a>.
65 * @param <E> the type of elements held in this collection
68 public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
69 implements BlockingQueue<E> {
71 private transient final ReentrantLock lock = new ReentrantLock();
72 private final PriorityQueue<E> q = new PriorityQueue<E>();
75 * Thread designated to wait for the element at the head of
76 * the queue. This variant of the Leader-Follower pattern
77 * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to
78 * minimize unnecessary timed waiting. When a thread becomes
79 * the leader, it waits only for the next delay to elapse, but
80 * other threads await indefinitely. The leader thread must
81 * signal some other thread before returning from take() or
82 * poll(...), unless some other thread becomes leader in the
83 * interim. Whenever the head of the queue is replaced with
84 * an element with an earlier expiration time, the leader
85 * field is invalidated by being reset to null, and some
86 * waiting thread, but not necessarily the current leader, is
87 * signalled. So waiting threads must be prepared to acquire
88 * and lose leadership while waiting.
90 private Thread leader = null;
93 * Condition signalled when a newer element becomes available
94 * at the head of the queue or a new thread may need to
97 private final Condition available = lock.newCondition();
100 * Creates a new <tt>DelayQueue</tt> that is initially empty.
102 public DelayQueue() {}
105 * Creates a <tt>DelayQueue</tt> initially containing the elements of the
106 * given collection of {@link Delayed} instances.
108 * @param c the collection of elements to initially contain
109 * @throws NullPointerException if the specified collection or any
110 * of its elements are null
112 public DelayQueue(Collection<? extends E> c) {
117 * Inserts the specified element into this delay queue.
119 * @param e the element to add
120 * @return <tt>true</tt> (as specified by {@link Collection#add})
121 * @throws NullPointerException if the specified element is null
123 public boolean add(E e) {
128 * Inserts the specified element into this delay queue.
130 * @param e the element to add
131 * @return <tt>true</tt>
132 * @throws NullPointerException if the specified element is null
134 public boolean offer(E e) {
135 final ReentrantLock lock = this.lock;
150 * Inserts the specified element into this delay queue. As the queue is
151 * unbounded this method will never block.
153 * @param e the element to add
154 * @throws NullPointerException {@inheritDoc}
156 public void put(E e) {
161 * Inserts the specified element into this delay queue. As the queue is
162 * unbounded this method will never block.
164 * @param e the element to add
165 * @param timeout This parameter is ignored as the method never blocks
166 * @param unit This parameter is ignored as the method never blocks
167 * @return <tt>true</tt>
168 * @throws NullPointerException {@inheritDoc}
170 public boolean offer(E e, long timeout, TimeUnit unit) {
175 * Retrieves and removes the head of this queue, or returns <tt>null</tt>
176 * if this queue has no elements with an expired delay.
178 * @return the head of this queue, or <tt>null</tt> if this
179 * queue has no elements with an expired delay
182 final ReentrantLock lock = this.lock;
186 if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)
196 * Retrieves and removes the head of this queue, waiting if necessary
197 * until an element with an expired delay is available on this queue.
199 * @return the head of this queue
200 * @throws InterruptedException {@inheritDoc}
202 public E take() throws InterruptedException {
203 final ReentrantLock lock = this.lock;
204 lock.lockInterruptibly();
211 long delay = first.getDelay(TimeUnit.NANOSECONDS);
214 else if (leader != null)
217 Thread thisThread = Thread.currentThread();
220 available.awaitNanos(delay);
222 if (leader == thisThread)
229 if (leader == null && q.peek() != null)
236 * Retrieves and removes the head of this queue, waiting if necessary
237 * until an element with an expired delay is available on this queue,
238 * or the specified wait time expires.
240 * @return the head of this queue, or <tt>null</tt> if the
241 * specified waiting time elapses before an element with
242 * an expired delay becomes available
243 * @throws InterruptedException {@inheritDoc}
245 public E poll(long timeout, TimeUnit unit) throws InterruptedException {
246 long nanos = unit.toNanos(timeout);
247 final ReentrantLock lock = this.lock;
248 lock.lockInterruptibly();
256 nanos = available.awaitNanos(nanos);
258 long delay = first.getDelay(TimeUnit.NANOSECONDS);
263 if (nanos < delay || leader != null)
264 nanos = available.awaitNanos(nanos);
266 Thread thisThread = Thread.currentThread();
269 long timeLeft = available.awaitNanos(delay);
270 nanos -= delay - timeLeft;
272 if (leader == thisThread)
279 if (leader == null && q.peek() != null)
286 * Retrieves, but does not remove, the head of this queue, or
287 * returns <tt>null</tt> if this queue is empty. Unlike
288 * <tt>poll</tt>, if no expired elements are available in the queue,
289 * this method returns the element that will expire next,
292 * @return the head of this queue, or <tt>null</tt> if this
296 final ReentrantLock lock = this.lock;
306 final ReentrantLock lock = this.lock;
316 * @throws UnsupportedOperationException {@inheritDoc}
317 * @throws ClassCastException {@inheritDoc}
318 * @throws NullPointerException {@inheritDoc}
319 * @throws IllegalArgumentException {@inheritDoc}
321 public int drainTo(Collection<? super E> c) {
323 throw new NullPointerException();
325 throw new IllegalArgumentException();
326 final ReentrantLock lock = this.lock;
332 if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)
344 * @throws UnsupportedOperationException {@inheritDoc}
345 * @throws ClassCastException {@inheritDoc}
346 * @throws NullPointerException {@inheritDoc}
347 * @throws IllegalArgumentException {@inheritDoc}
349 public int drainTo(Collection<? super E> c, int maxElements) {
351 throw new NullPointerException();
353 throw new IllegalArgumentException();
354 if (maxElements <= 0)
356 final ReentrantLock lock = this.lock;
360 while (n < maxElements) {
362 if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)
374 * Atomically removes all of the elements from this delay queue.
375 * The queue will be empty after this call returns.
376 * Elements with an unexpired delay are not waited for; they are
377 * simply discarded from the queue.
379 public void clear() {
380 final ReentrantLock lock = this.lock;
390 * Always returns <tt>Integer.MAX_VALUE</tt> because
391 * a <tt>DelayQueue</tt> is not capacity constrained.
393 * @return <tt>Integer.MAX_VALUE</tt>
395 public int remainingCapacity() {
396 return Integer.MAX_VALUE;
400 * Returns an array containing all of the elements in this queue.
401 * The returned array elements are in no particular order.
403 * <p>The returned array will be "safe" in that no references to it are
404 * maintained by this queue. (In other words, this method must allocate
405 * a new array). The caller is thus free to modify the returned array.
407 * <p>This method acts as bridge between array-based and collection-based
410 * @return an array containing all of the elements in this queue
412 public Object[] toArray() {
413 final ReentrantLock lock = this.lock;
423 * Returns an array containing all of the elements in this queue; the
424 * runtime type of the returned array is that of the specified array.
425 * The returned array elements are in no particular order.
426 * If the queue fits in the specified array, it is returned therein.
427 * Otherwise, a new array is allocated with the runtime type of the
428 * specified array and the size of this queue.
430 * <p>If this queue fits in the specified array with room to spare
431 * (i.e., the array has more elements than this queue), the element in
432 * the array immediately following the end of the queue is set to
435 * <p>Like the {@link #toArray()} method, this method acts as bridge between
436 * array-based and collection-based APIs. Further, this method allows
437 * precise control over the runtime type of the output array, and may,
438 * under certain circumstances, be used to save allocation costs.
440 * <p>The following code can be used to dump a delay queue into a newly
441 * allocated array of <tt>Delayed</tt>:
444 * Delayed[] a = q.toArray(new Delayed[0]);</pre>
446 * Note that <tt>toArray(new Object[0])</tt> is identical in function to
447 * <tt>toArray()</tt>.
449 * @param a the array into which the elements of the queue are to
450 * be stored, if it is big enough; otherwise, a new array of the
451 * same runtime type is allocated for this purpose
452 * @return an array containing all of the elements in this queue
453 * @throws ArrayStoreException if the runtime type of the specified array
454 * is not a supertype of the runtime type of every element in
456 * @throws NullPointerException if the specified array is null
458 public <T> T[] toArray(T[] a) {
459 final ReentrantLock lock = this.lock;
469 * Removes a single instance of the specified element from this
470 * queue, if it is present, whether or not it has expired.
472 public boolean remove(Object o) {
473 final ReentrantLock lock = this.lock;
483 * Returns an iterator over all the elements (both expired and
484 * unexpired) in this queue. The iterator does not return the
485 * elements in any particular order.
487 * <p>The returned iterator is a "weakly consistent" iterator that
488 * will never throw {@link java.util.ConcurrentModificationException
489 * ConcurrentModificationException}, and guarantees to traverse
490 * elements as they existed upon construction of the iterator, and
491 * may (but is not guaranteed to) reflect any modifications
492 * subsequent to construction.
494 * @return an iterator over the elements in this queue
496 public Iterator<E> iterator() {
497 return new Itr(toArray());
501 * Snapshot iterator that works off copy of underlying q array.
503 private class Itr implements Iterator<E> {
504 final Object[] array; // Array of all elements
505 int cursor; // index of next element to return;
506 int lastRet; // index of last element, or -1 if no such
508 Itr(Object[] array) {
513 public boolean hasNext() {
514 return cursor < array.length;
517 @SuppressWarnings("unchecked")
519 if (cursor >= array.length)
520 throw new NoSuchElementException();
522 return (E)array[cursor++];
525 public void remove() {
527 throw new IllegalStateException();
528 Object x = array[lastRet];
530 // Traverse underlying queue to find == element,
531 // not just a .equals element.
534 for (Iterator it = q.iterator(); it.hasNext(); ) {
535 if (it.next() == x) {