ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/DelayQueue.java
Revision: 1.78
Committed: Mon Oct 1 00:10:53 2018 UTC (5 years, 7 months ago) by jsr166
Branch: MAIN
CVS Tags: HEAD
Changes since 1.77: +1 -1 lines
Log Message:
update to using jdk11 by default, except link to jdk10 javadocs;
sync @docRoot references in javadoc with upstream

File Contents

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