lib/classpath/external/jsr166/java/util/concurrent/DelayQueue.java

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