util/concurrent/DelayQueue.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/publicdomain/zero/1.0/
 */


package java.util.concurrent;
import java.util.concurrent.locks.*;
import java.util.*;

/**
 * An unbounded {@linkplain BlockingQueue blocking queue} of
 * <tt>Delayed</tt> elements, in which an element can only be taken
 * when its delay has expired.  The <em>head</em> of the queue is that
 * <tt>Delayed</tt> element whose delay expired furthest in the
 * past.  If no delay has expired there is no head and <tt>poll</tt>
 * will return <tt>null</tt>. Expiration occurs when an element's
 * <tt>getDelay(TimeUnit.NANOSECONDS)</tt> method returns a value less
 * than or equal to zero.  Even though unexpired elements cannot be
 * removed using <tt>take</tt> or <tt>poll</tt>, they are otherwise
 * treated as normal elements. For example, the <tt>size</tt> method
 * returns the count of both expired and unexpired elements.
 * This queue does not permit null elements.
 *
 * <p>This class and its iterator implement all of the
 * <em>optional</em> methods of the {@link Collection} and {@link
 * Iterator} interfaces.  The Iterator provided in method {@link
 * #iterator()} is <em>not</em> guaranteed to traverse the elements of
 * the DelayQueue in any particular order.
 *
 * <p>This class is a member of the
 * <a href="{@docRoot}/../technotes/guides/collections/index.html">
 * Java Collections Framework</a>.
 *
 * @since 1.5
 * @author Doug Lea
 * @param <E> the type of elements held in this collection
 */

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
    implements BlockingQueue<E> {

    private transient final ReentrantLock lock = new ReentrantLock();
    private final PriorityQueue<E> q = new PriorityQueue<E>();

    /**
     * Thread designated to wait for the element at the head of
     * the queue.  This variant of the Leader-Follower pattern
     * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to
     * minimize unnecessary timed waiting.  When a thread becomes
     * the leader, it waits only for the next delay to elapse, but
     * other threads await indefinitely.  The leader thread must
     * signal some other thread before returning from take() or
     * poll(...), unless some other thread becomes leader in the
     * interim.  Whenever the head of the queue is replaced with
     * an element with an earlier expiration time, the leader
     * field is invalidated by being reset to null, and some
     * waiting thread, but not necessarily the current leader, is
     * signalled.  So waiting threads must be prepared to acquire
     * and lose leadership while waiting.
     */
    private Thread leader = null;

    /**
     * Condition signalled when a newer element becomes available
     * at the head of the queue or a new thread may need to
     * become leader.
     */
    private final Condition available = lock.newCondition();

    /**
     * Creates a new <tt>DelayQueue</tt> that is initially empty.
     */
    public DelayQueue() {}

    /**
     * Creates a <tt>DelayQueue</tt> initially containing the elements of the
     * given collection of {@link Delayed} instances.
     *
     * @param c the collection of elements to initially contain
     * @throws NullPointerException if the specified collection or any
     *         of its elements are null
     */
    public DelayQueue(Collection<? extends E> c) {
        this.addAll(c);
    }

    /**
     * Inserts the specified element into this delay queue.
     *
     * @param e the element to add
     * @return <tt>true</tt> (as specified by {@link Collection#add})
     * @throws NullPointerException if the specified element is null
     */
    public boolean add(E e) {
        return offer(e);
    }

    /**
     * Inserts the specified element into this delay queue.
     *
     * @param e the element to add
     * @return <tt>true</tt>
     * @throws NullPointerException if the specified element is null
     */
    public boolean offer(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            q.offer(e);
            if (q.peek() == e) {
                leader = null;
                available.signal();
            }
            return true;
        } finally {
            lock.unlock();
        }
    }

    /**
     * Inserts the specified element into this delay queue. As the queue is
     * unbounded this method will never block.
     *
     * @param e the element to add
     * @throws NullPointerException {@inheritDoc}
     */
    public void put(E e) {
        offer(e);
    }

    /**
     * Inserts the specified element into this delay queue. As the queue is
     * unbounded this method will never block.
     *
     * @param e the element to add
     * @param timeout This parameter is ignored as the method never blocks
     * @param unit This parameter is ignored as the method never blocks
     * @return <tt>true</tt>
     * @throws NullPointerException {@inheritDoc}
     */
    public boolean offer(E e, long timeout, TimeUnit unit) {
        return offer(e);
    }

    /**
     * Retrieves and removes the head of this queue, or returns <tt>null</tt>
     * if this queue has no elements with an expired delay.
     *
     * @return the head of this queue, or <tt>null</tt> if this
     *         queue has no elements with an expired delay
     */
    public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            E first = q.peek();
            if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)
                return null;
            else
                return q.poll();
        } finally {
            lock.unlock();
        }
    }

    /**
     * Retrieves and removes the head of this queue, waiting if necessary
     * until an element with an expired delay is available on this queue.
     *
     * @return the head of this queue
     * @throws InterruptedException {@inheritDoc}
     */
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            for (;;) {
                E first = q.peek();
                if (first == null)
                    available.await();
                else {
                    long delay = first.getDelay(TimeUnit.NANOSECONDS);
                    if (delay <= 0)
                        return q.poll();
                    else if (leader != null)
                        available.await();
                    else {
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try {
                            available.awaitNanos(delay);
                        } finally {
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            if (leader == null && q.peek() != null)
                available.signal();
            lock.unlock();
        }
    }

    /**
     * Retrieves and removes the head of this queue, waiting if necessary
     * until an element with an expired delay is available on this queue,
     * or the specified wait time expires.
     *
     * @return the head of this queue, or <tt>null</tt> if the
     *         specified waiting time elapses before an element with
     *         an expired delay becomes available
     * @throws InterruptedException {@inheritDoc}
     */
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            for (;;) {
                E first = q.peek();
                if (first == null) {
                    if (nanos <= 0)
                        return null;
                    else
                        nanos = available.awaitNanos(nanos);
                } else {
                    long delay = first.getDelay(TimeUnit.NANOSECONDS);
                    if (delay <= 0)
                        return q.poll();
                    if (nanos <= 0)
                        return null;
                    if (nanos < delay || leader != null)
                        nanos = available.awaitNanos(nanos);
                    else {
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try {
                            long timeLeft = available.awaitNanos(delay);
                            nanos -= delay - timeLeft;
                        } finally {
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            if (leader == null && q.peek() != null)
                available.signal();
            lock.unlock();
        }
    }

    /**
     * Retrieves, but does not remove, the head of this queue, or
     * returns <tt>null</tt> if this queue is empty.  Unlike
     * <tt>poll</tt>, if no expired elements are available in the queue,
     * this method returns the element that will expire next,
     * if one exists.
     *
     * @return the head of this queue, or <tt>null</tt> if this
     *         queue is empty.
     */
    public E peek() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.peek();
        } finally {
            lock.unlock();
        }
    }

    public int size() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.size();
        } finally {
            lock.unlock();
        }
    }

    /**
     * @throws UnsupportedOperationException {@inheritDoc}
     * @throws ClassCastException            {@inheritDoc}
     * @throws NullPointerException          {@inheritDoc}
     * @throws IllegalArgumentException      {@inheritDoc}
     */
    public int drainTo(Collection<? super E> c) {
        if (c == null)
            throw new NullPointerException();
        if (c == this)
            throw new IllegalArgumentException();
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            int n = 0;
            for (;;) {
                E first = q.peek();
                if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)
                    break;
                c.add(q.poll());
                ++n;
            }
            return n;
        } finally {
            lock.unlock();
        }
    }

    /**
     * @throws UnsupportedOperationException {@inheritDoc}
     * @throws ClassCastException            {@inheritDoc}
     * @throws NullPointerException          {@inheritDoc}
     * @throws IllegalArgumentException      {@inheritDoc}
     */
    public int drainTo(Collection<? super E> c, int maxElements) {
        if (c == null)
            throw new NullPointerException();
        if (c == this)
            throw new IllegalArgumentException();
        if (maxElements <= 0)
            return 0;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            int n = 0;
            while (n < maxElements) {
                E first = q.peek();
                if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)
                    break;
                c.add(q.poll());
                ++n;
            }
            return n;
        } finally {
            lock.unlock();
        }
    }

    /**
     * Atomically removes all of the elements from this delay queue.
     * The queue will be empty after this call returns.
     * Elements with an unexpired delay are not waited for; they are
     * simply discarded from the queue.
     */
    public void clear() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            q.clear();
        } finally {
            lock.unlock();
        }
    }

    /**
     * Always returns <tt>Integer.MAX_VALUE</tt> because
     * a <tt>DelayQueue</tt> is not capacity constrained.
     *
     * @return <tt>Integer.MAX_VALUE</tt>
     */
    public int remainingCapacity() {
        return Integer.MAX_VALUE;
    }

    /**
     * Returns an array containing all of the elements in this queue.
     * The returned array elements are in no particular order.
     *
     * <p>The returned array will be "safe" in that no references to it are
     * maintained by this queue.  (In other words, this method must allocate
     * a new array).  The caller is thus free to modify the returned array.
     *
     * <p>This method acts as bridge between array-based and collection-based
     * APIs.
     *
     * @return an array containing all of the elements in this queue
     */
    public Object[] toArray() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.toArray();
        } finally {
            lock.unlock();
        }
    }

    /**
     * Returns an array containing all of the elements in this queue; the
     * runtime type of the returned array is that of the specified array.
     * The returned array elements are in no particular order.
     * If the queue fits in the specified array, it is returned therein.
     * Otherwise, a new array is allocated with the runtime type of the
     * specified array and the size of this queue.
     *
     * <p>If this queue fits in the specified array with room to spare
     * (i.e., the array has more elements than this queue), the element in
     * the array immediately following the end of the queue is set to
     * <tt>null</tt>.
     *
     * <p>Like the {@link #toArray()} method, this method acts as bridge between
     * array-based and collection-based APIs.  Further, this method allows
     * precise control over the runtime type of the output array, and may,
     * under certain circumstances, be used to save allocation costs.
     *
     * <p>The following code can be used to dump a delay queue into a newly
     * allocated array of <tt>Delayed</tt>:
     *
     * <pre>
     *     Delayed[] a = q.toArray(new Delayed[0]);</pre>
     *
     * Note that <tt>toArray(new Object[0])</tt> is identical in function to
     * <tt>toArray()</tt>.
     *
     * @param a the array into which the elements of the queue are to
     *          be stored, if it is big enough; otherwise, a new array of the
     *          same runtime type is allocated for this purpose
     * @return an array containing all of the elements in this queue
     * @throws ArrayStoreException if the runtime type of the specified array
     *         is not a supertype of the runtime type of every element in
     *         this queue
     * @throws NullPointerException if the specified array is null
     */
    public <T> T[] toArray(T[] a) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.toArray(a);
        } finally {
            lock.unlock();
        }
    }

    /**
     * Removes a single instance of the specified element from this
     * queue, if it is present, whether or not it has expired.
     */
    public boolean remove(Object o) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.remove(o);
        } finally {
            lock.unlock();
        }
    }

    /**
     * Returns an iterator over all the elements (both expired and
     * unexpired) in this queue. The iterator does not return the
     * elements in any particular order.
     *
     * <p>The returned iterator is a "weakly consistent" iterator that
     * will never throw {@link java.util.ConcurrentModificationException
     * ConcurrentModificationException}, and guarantees to traverse
     * elements as they existed upon construction of the iterator, and
     * may (but is not guaranteed to) reflect any modifications
     * subsequent to construction.
     *
     * @return an iterator over the elements in this queue
     */
    public Iterator<E> iterator() {
        return new Itr(toArray());
    }

    /**
     * Snapshot iterator that works off copy of underlying q array.
     */
    private class Itr implements Iterator<E> {
        final Object[] array; // Array of all elements
        int cursor;           // index of next element to return;
        int lastRet;          // index of last element, or -1 if no such

        Itr(Object[] array) {
            lastRet = -1;
            this.array = array;
        }

        public boolean hasNext() {
            return cursor < array.length;
        }

        @SuppressWarnings("unchecked")
        public E next() {
            if (cursor >= array.length)
                throw new NoSuchElementException();
            lastRet = cursor;
            return (E)array[cursor++];
        }

        public void remove() {
            if (lastRet < 0)
                throw new IllegalStateException();
            Object x = array[lastRet];
            lastRet = -1;
            // Traverse underlying queue to find == element,
            // not just a .equals element.
            lock.lock();
            try {
                for (Iterator<E> it = q.iterator(); it.hasNext(); ) {
                    if (it.next() == x) {
                        it.remove();
                        return;
                    }
                }
            } finally {
                lock.unlock();
            }
        }
    }

}
Revision:	1.54
Committed:	Fri Jun 3 02:28:05 2011 UTC (13 years ago) by jsr166
Branch:	MAIN
Changes since 1.53:	+1 -1 lines
Log Message:	fix javac 7 [rawtypes] warnings
#	Content
1	/*
2	* Written by Doug Lea with assistance from members of JCP JSR-166
3	* Expert Group and released to the public domain, as explained at
4	* http://creativecommons.org/publicdomain/zero/1.0/
5	*/
6
7
8	package java.util.concurrent;
9	import java.util.concurrent.locks.*;
10	import java.util.*;
11
12	/**
13	* An unbounded {@linkplain BlockingQueue blocking queue} of
14	* <tt>Delayed</tt> elements, in which an element can only be taken
15	* when its delay has expired. The <em>head</em> of the queue is that
16	* <tt>Delayed</tt> element whose delay expired furthest in the
17	* past. If no delay has expired there is no head and <tt>poll</tt>
18	* will return <tt>null</tt>. Expiration occurs when an element's
19	* <tt>getDelay(TimeUnit.NANOSECONDS)</tt> method returns a value less
20	* than or equal to zero. Even though unexpired elements cannot be
21	* removed using <tt>take</tt> or <tt>poll</tt>, they are otherwise
22	* treated as normal elements. For example, the <tt>size</tt> method
23	* returns the count of both expired and unexpired elements.
24	* This queue does not permit null elements.
25	*
26	* <p>This class and its iterator implement all of the
27	* <em>optional</em> methods of the {@link Collection} and {@link
28	* Iterator} interfaces. The Iterator provided in method {@link
29	* #iterator()} is <em>not</em> guaranteed to traverse the elements of
30	* the DelayQueue in any particular order.
31	*
32	* <p>This class is a member of the
33	* <a href="{@docRoot}/../technotes/guides/collections/index.html">
34	* Java Collections Framework</a>.
35	*
36	* @since 1.5
37	* @author Doug Lea
38	* @param <E> the type of elements held in this collection
39	*/
40
41	public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
42	implements BlockingQueue<E> {
43
44	private transient final ReentrantLock lock = new ReentrantLock();
45	private final PriorityQueue<E> q = new PriorityQueue<E>();
46
47	/**
48	* Thread designated to wait for the element at the head of
49	* the queue. This variant of the Leader-Follower pattern
50	* (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to
51	* minimize unnecessary timed waiting. When a thread becomes
52	* the leader, it waits only for the next delay to elapse, but
53	* other threads await indefinitely. The leader thread must
54	* signal some other thread before returning from take() or
55	* poll(...), unless some other thread becomes leader in the
56	* interim. Whenever the head of the queue is replaced with
57	* an element with an earlier expiration time, the leader
58	* field is invalidated by being reset to null, and some
59	* waiting thread, but not necessarily the current leader, is
60	* signalled. So waiting threads must be prepared to acquire
61	* and lose leadership while waiting.
62	*/
63	private Thread leader = null;
64
65	/**
66	* Condition signalled when a newer element becomes available
67	* at the head of the queue or a new thread may need to
68	* become leader.
69	*/
70	private final Condition available = lock.newCondition();
71
72	/**
73	* Creates a new <tt>DelayQueue</tt> that is initially empty.
74	*/
75	public DelayQueue() {}
76
77	/**
78	* Creates a <tt>DelayQueue</tt> initially containing the elements of the
79	* given collection of {@link Delayed} instances.
80	*
81	* @param c the collection of elements to initially contain
82	* @throws NullPointerException if the specified collection or any
83	* of its elements are null
84	*/
85	public DelayQueue(Collection<? extends E> c) {
86	this.addAll(c);
87	}
88
89	/**
90	* Inserts the specified element into this delay queue.
91	*
92	* @param e the element to add
93	* @return <tt>true</tt> (as specified by {@link Collection#add})
94	* @throws NullPointerException if the specified element is null
95	*/
96	public boolean add(E e) {
97	return offer(e);
98	}
99
100	/**
101	* Inserts the specified element into this delay queue.
102	*
103	* @param e the element to add
104	* @return <tt>true</tt>
105	* @throws NullPointerException if the specified element is null
106	*/
107	public boolean offer(E e) {
108	final ReentrantLock lock = this.lock;
109	lock.lock();
110	try {
111	q.offer(e);
112	if (q.peek() == e) {
113	leader = null;
114	available.signal();
115	}
116	return true;
117	} finally {
118	lock.unlock();
119	}
120	}
121
122	/**
123	* Inserts the specified element into this delay queue. As the queue is
124	* unbounded this method will never block.
125	*
126	* @param e the element to add
127	* @throws NullPointerException {@inheritDoc}
128	*/
129	public void put(E e) {
130	offer(e);
131	}
132
133	/**
134	* Inserts the specified element into this delay queue. As the queue is
135	* unbounded this method will never block.
136	*
137	* @param e the element to add
138	* @param timeout This parameter is ignored as the method never blocks
139	* @param unit This parameter is ignored as the method never blocks
140	* @return <tt>true</tt>
141	* @throws NullPointerException {@inheritDoc}
142	*/
143	public boolean offer(E e, long timeout, TimeUnit unit) {
144	return offer(e);
145	}
146
147	/**
148	* Retrieves and removes the head of this queue, or returns <tt>null</tt>
149	* if this queue has no elements with an expired delay.
150	*
151	* @return the head of this queue, or <tt>null</tt> if this
152	* queue has no elements with an expired delay
153	*/
154	public E poll() {
155	final ReentrantLock lock = this.lock;
156	lock.lock();
157	try {
158	E first = q.peek();
159	if (first == null \|\| first.getDelay(TimeUnit.NANOSECONDS) > 0)
160	return null;
161	else
162	return q.poll();
163	} finally {
164	lock.unlock();
165	}
166	}
167
168	/**
169	* Retrieves and removes the head of this queue, waiting if necessary
170	* until an element with an expired delay is available on this queue.
171	*
172	* @return the head of this queue
173	* @throws InterruptedException {@inheritDoc}
174	*/
175	public E take() throws InterruptedException {
176	final ReentrantLock lock = this.lock;
177	lock.lockInterruptibly();
178	try {
179	for (;;) {
180	E first = q.peek();
181	if (first == null)
182	available.await();
183	else {
184	long delay = first.getDelay(TimeUnit.NANOSECONDS);
185	if (delay <= 0)
186	return q.poll();
187	else if (leader != null)
188	available.await();
189	else {
190	Thread thisThread = Thread.currentThread();
191	leader = thisThread;
192	try {
193	available.awaitNanos(delay);
194	} finally {
195	if (leader == thisThread)
196	leader = null;
197	}
198	}
199	}
200	}
201	} finally {
202	if (leader == null && q.peek() != null)
203	available.signal();
204	lock.unlock();
205	}
206	}
207
208	/**
209	* Retrieves and removes the head of this queue, waiting if necessary
210	* until an element with an expired delay is available on this queue,
211	* or the specified wait time expires.
212	*
213	* @return the head of this queue, or <tt>null</tt> if the
214	* specified waiting time elapses before an element with
215	* an expired delay becomes available
216	* @throws InterruptedException {@inheritDoc}
217	*/
218	public E poll(long timeout, TimeUnit unit) throws InterruptedException {
219	long nanos = unit.toNanos(timeout);
220	final ReentrantLock lock = this.lock;
221	lock.lockInterruptibly();
222	try {
223	for (;;) {
224	E first = q.peek();
225	if (first == null) {
226	if (nanos <= 0)
227	return null;
228	else
229	nanos = available.awaitNanos(nanos);
230	} else {
231	long delay = first.getDelay(TimeUnit.NANOSECONDS);
232	if (delay <= 0)
233	return q.poll();
234	if (nanos <= 0)
235	return null;
236	if (nanos < delay \|\| leader != null)
237	nanos = available.awaitNanos(nanos);
238	else {
239	Thread thisThread = Thread.currentThread();
240	leader = thisThread;
241	try {
242	long timeLeft = available.awaitNanos(delay);
243	nanos -= delay - timeLeft;
244	} finally {
245	if (leader == thisThread)
246	leader = null;
247	}
248	}
249	}
250	}
251	} finally {
252	if (leader == null && q.peek() != null)
253	available.signal();
254	lock.unlock();
255	}
256	}
257
258	/**
259	* Retrieves, but does not remove, the head of this queue, or
260	* returns <tt>null</tt> if this queue is empty. Unlike
261	* <tt>poll</tt>, if no expired elements are available in the queue,
262	* this method returns the element that will expire next,
263	* if one exists.
264	*
265	* @return the head of this queue, or <tt>null</tt> if this
266	* queue is empty.
267	*/
268	public E peek() {
269	final ReentrantLock lock = this.lock;
270	lock.lock();
271	try {
272	return q.peek();
273	} finally {
274	lock.unlock();
275	}
276	}
277
278	public int size() {
279	final ReentrantLock lock = this.lock;
280	lock.lock();
281	try {
282	return q.size();
283	} finally {
284	lock.unlock();
285	}
286	}
287
288	/**
289	* @throws UnsupportedOperationException {@inheritDoc}
290	* @throws ClassCastException {@inheritDoc}
291	* @throws NullPointerException {@inheritDoc}
292	* @throws IllegalArgumentException {@inheritDoc}
293	*/
294	public int drainTo(Collection<? super E> c) {
295	if (c == null)
296	throw new NullPointerException();
297	if (c == this)
298	throw new IllegalArgumentException();
299	final ReentrantLock lock = this.lock;
300	lock.lock();
301	try {
302	int n = 0;
303	for (;;) {
304	E first = q.peek();
305	if (first == null \|\| first.getDelay(TimeUnit.NANOSECONDS) > 0)
306	break;
307	c.add(q.poll());
308	++n;
309	}
310	return n;
311	} finally {
312	lock.unlock();
313	}
314	}
315
316	/**
317	* @throws UnsupportedOperationException {@inheritDoc}
318	* @throws ClassCastException {@inheritDoc}
319	* @throws NullPointerException {@inheritDoc}
320	* @throws IllegalArgumentException {@inheritDoc}
321	*/
322	public int drainTo(Collection<? super E> c, int maxElements) {
323	if (c == null)
324	throw new NullPointerException();
325	if (c == this)
326	throw new IllegalArgumentException();
327	if (maxElements <= 0)
328	return 0;
329	final ReentrantLock lock = this.lock;
330	lock.lock();
331	try {
332	int n = 0;
333	while (n < maxElements) {
334	E first = q.peek();
335	if (first == null \|\| first.getDelay(TimeUnit.NANOSECONDS) > 0)
336	break;
337	c.add(q.poll());
338	++n;
339	}
340	return n;
341	} finally {
342	lock.unlock();
343	}
344	}
345
346	/**
347	* Atomically removes all of the elements from this delay queue.
348	* The queue will be empty after this call returns.
349	* Elements with an unexpired delay are not waited for; they are
350	* simply discarded from the queue.
351	*/
352	public void clear() {
353	final ReentrantLock lock = this.lock;
354	lock.lock();
355	try {
356	q.clear();
357	} finally {
358	lock.unlock();
359	}
360	}
361
362	/**
363	* Always returns <tt>Integer.MAX_VALUE</tt> because
364	* a <tt>DelayQueue</tt> is not capacity constrained.
365	*
366	* @return <tt>Integer.MAX_VALUE</tt>
367	*/
368	public int remainingCapacity() {
369	return Integer.MAX_VALUE;
370	}
371
372	/**
373	* Returns an array containing all of the elements in this queue.
374	* The returned array elements are in no particular order.
375	*
376	* <p>The returned array will be "safe" in that no references to it are
377	* maintained by this queue. (In other words, this method must allocate
378	* a new array). The caller is thus free to modify the returned array.
379	*
380	* <p>This method acts as bridge between array-based and collection-based
381	* APIs.
382	*
383	* @return an array containing all of the elements in this queue
384	*/
385	public Object[] toArray() {
386	final ReentrantLock lock = this.lock;
387	lock.lock();
388	try {
389	return q.toArray();
390	} finally {
391	lock.unlock();
392	}
393	}
394
395	/**
396	* Returns an array containing all of the elements in this queue; the
397	* runtime type of the returned array is that of the specified array.
398	* The returned array elements are in no particular order.
399	* If the queue fits in the specified array, it is returned therein.
400	* Otherwise, a new array is allocated with the runtime type of the
401	* specified array and the size of this queue.
402	*
403	* <p>If this queue fits in the specified array with room to spare
404	* (i.e., the array has more elements than this queue), the element in
405	* the array immediately following the end of the queue is set to
406	* <tt>null</tt>.
407	*
408	* <p>Like the {@link #toArray()} method, this method acts as bridge between
409	* array-based and collection-based APIs. Further, this method allows
410	* precise control over the runtime type of the output array, and may,
411	* under certain circumstances, be used to save allocation costs.
412	*
413	* <p>The following code can be used to dump a delay queue into a newly
414	* allocated array of <tt>Delayed</tt>:
415	*
416	* <pre>
417	* Delayed[] a = q.toArray(new Delayed[0]);</pre>
418	*
419	* Note that <tt>toArray(new Object[0])</tt> is identical in function to
420	* <tt>toArray()</tt>.
421	*
422	* @param a the array into which the elements of the queue are to
423	* be stored, if it is big enough; otherwise, a new array of the
424	* same runtime type is allocated for this purpose
425	* @return an array containing all of the elements in this queue
426	* @throws ArrayStoreException if the runtime type of the specified array
427	* is not a supertype of the runtime type of every element in
428	* this queue
429	* @throws NullPointerException if the specified array is null
430	*/
431	public <T> T[] toArray(T[] a) {
432	final ReentrantLock lock = this.lock;
433	lock.lock();
434	try {
435	return q.toArray(a);
436	} finally {
437	lock.unlock();
438	}
439	}
440
441	/**
442	* Removes a single instance of the specified element from this
443	* queue, if it is present, whether or not it has expired.
444	*/
445	public boolean remove(Object o) {
446	final ReentrantLock lock = this.lock;
447	lock.lock();
448	try {
449	return q.remove(o);
450	} finally {
451	lock.unlock();
452	}
453	}
454
455	/**
456	* Returns an iterator over all the elements (both expired and
457	* unexpired) in this queue. The iterator does not return the
458	* elements in any particular order.
459	*
460	* <p>The returned iterator is a "weakly consistent" iterator that
461	* will never throw {@link java.util.ConcurrentModificationException
462	* ConcurrentModificationException}, and guarantees to traverse
463	* elements as they existed upon construction of the iterator, and
464	* may (but is not guaranteed to) reflect any modifications
465	* subsequent to construction.
466	*
467	* @return an iterator over the elements in this queue
468	*/
469	public Iterator<E> iterator() {
470	return new Itr(toArray());
471	}
472
473	/**
474	* Snapshot iterator that works off copy of underlying q array.
475	*/
476	private class Itr implements Iterator<E> {
477	final Object[] array; // Array of all elements
478	int cursor; // index of next element to return;
479	int lastRet; // index of last element, or -1 if no such
480
481	Itr(Object[] array) {
482	lastRet = -1;
483	this.array = array;
484	}
485
486	public boolean hasNext() {
487	return cursor < array.length;
488	}
489
490	@SuppressWarnings("unchecked")
491	public E next() {
492	if (cursor >= array.length)
493	throw new NoSuchElementException();
494	lastRet = cursor;
495	return (E)array[cursor++];
496	}
497
498	public void remove() {
499	if (lastRet < 0)
500	throw new IllegalStateException();
501	Object x = array[lastRet];
502	lastRet = -1;
503	// Traverse underlying queue to find == element,
504	// not just a .equals element.
505	lock.lock();
506	try {
507	for (Iterator<E> it = q.iterator(); it.hasNext(); ) {
508	if (it.next() == x) {
509	it.remove();
510	return;
511	}
512	}
513	} finally {
514	lock.unlock();
515	}
516	}
517	}
518
519	}