/* * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain. Use, modify, and * redistribute this code in any way without acknowledgement. */ package java.util.concurrent; import java.util.concurrent.locks.*; import java.util.*; /** * An unbounded {@linkplain BlockingQueue blocking queue} based on a * {@link PriorityQueue}, obeying its ordering rules and * implementation characteristics. While this queue is logically * unbounded, attempted additions may fail due to resource exhaustion * (causing OutOfMemoryError). A priority queue does not * permit null elements. A priority queue relying on natural * ordering also does not permit insertion of non-comparable objects * (doing so results in ClassCastException). * *

This class implements all of the optional methods * of the {@link Collection} and {@link Iterator} interfaces. *

The Iterator provided in method {@link #iterator()} is * not guaranteed to traverse the elements of the * PriorityBlockingQueue in any particular order. If you need ordered * traversal, consider using Arrays.sort(pq.toArray()). * * @since 1.5 * @author Doug Lea */ public class PriorityBlockingQueue extends AbstractQueue implements BlockingQueue, java.io.Serializable { private static final long serialVersionUID = 5595510919245408276L; private final PriorityQueue q; private final ReentrantLock lock = new ReentrantLock(true); private final Condition notEmpty = lock.newCondition(); /** * Creates a PriorityBlockingQueue with the default initial * capacity * (11) that orders its elements according to their natural * ordering (using Comparable). */ public PriorityBlockingQueue() { q = new PriorityQueue(); } /** * Creates a PriorityBlockingQueue with the specified initial * capacity * that orders its elements according to their natural ordering * (using Comparable). * * @param initialCapacity the initial capacity for this priority queue. * @throws IllegalArgumentException if initialCapacity is less * than 1 */ public PriorityBlockingQueue(int initialCapacity) { q = new PriorityQueue(initialCapacity, null); } /** * Creates a PriorityBlockingQueue with the specified initial * capacity * that orders its elements according to the specified comparator. * * @param initialCapacity the initial capacity for this priority queue. * @param comparator the comparator used to order this priority queue. * If null then the order depends on the elements' natural * ordering. * @throws IllegalArgumentException if initialCapacity is less * than 1 */ public PriorityBlockingQueue(int initialCapacity, Comparator comparator) { q = new PriorityQueue(initialCapacity, comparator); } /** * Creates a PriorityBlockingQueue containing the elements * in the specified collection. The priority queue has an initial * capacity of 110% of the size of the specified collection. If * the specified collection is a {@link SortedSet} or a {@link * PriorityQueue}, this priority queue will be sorted according to * the same comparator, or according to its elements' natural * order if the collection is sorted according to its elements' * natural order. Otherwise, this priority queue is ordered * according to its elements' natural order. * * @param c the collection whose elements are to be placed * into this priority queue. * @throws ClassCastException if elements of the specified collection * cannot be compared to one another according to the priority * queue's ordering. * @throws NullPointerException if c or any element within it * is null */ public PriorityBlockingQueue(Collection c) { q = new PriorityQueue(c); } // these first few override just to update doc comments /** * Adds the specified element to this queue. * @return true (as per the general contract of * Collection.add). * * @throws NullPointerException if the specified element is null. * @throws ClassCastException if the specified element cannot be compared * with elements currently in the priority queue according * to the priority queue's ordering. */ public boolean add(E o) { return super.add(o); } /** * Returns the comparator used to order this collection, or null * if this collection is sorted according to its elements natural ordering * (using Comparable). * * @return the comparator used to order this collection, or null * if this collection is sorted according to its elements natural ordering. */ public Comparator comparator() { return q.comparator(); } /** * Inserts the specified element into this priority queue. * * @param o the element to add * @return true * @throws ClassCastException if the specified element cannot be compared * with elements currently in the priority queue according * to the priority queue's ordering. * @throws NullPointerException if the specified element is null. */ public boolean offer(E o) { if (o == null) throw new NullPointerException(); lock.lock(); try { boolean ok = q.offer(o); assert ok; notEmpty.signal(); return true; } finally { lock.unlock(); } } /** * Adds the specified element to this priority queue. As the queue is * unbounded this method will never block. * @param o the element to add * @throws ClassCastException if the element cannot be compared * with elements currently in the priority queue according * to the priority queue's ordering. * @throws NullPointerException if the specified element is null. */ public void put(E o) { offer(o); // never need to block } /** * Inserts the specified element into this priority queue. As the queue is * unbounded this method will never block. * @param o 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 true * @throws ClassCastException if the element cannot be compared * with elements currently in the priority queue according * to the priority queue's ordering. * @throws NullPointerException if the specified element is null. */ public boolean offer(E o, long timeout, TimeUnit unit) { return offer(o); // never need to block } public E take() throws InterruptedException { lock.lockInterruptibly(); try { try { while (q.size() == 0) notEmpty.await(); } catch (InterruptedException ie) { notEmpty.signal(); // propagate to non-interrupted thread throw ie; } E x = q.poll(); assert x != null; return x; } finally { lock.unlock(); } } public E poll() { lock.lock(); try { return q.poll(); } finally { lock.unlock(); } } public E poll(long timeout, TimeUnit unit) throws InterruptedException { long nanos = unit.toNanos(timeout); lock.lockInterruptibly(); try { for (;;) { E x = q.poll(); if (x != null) return x; if (nanos <= 0) return null; try { nanos = notEmpty.awaitNanos(nanos); } catch (InterruptedException ie) { notEmpty.signal(); // propagate to non-interrupted thread throw ie; } } } finally { lock.unlock(); } } public E peek() { lock.lock(); try { return q.peek(); } finally { lock.unlock(); } } public int size() { lock.lock(); try { return q.size(); } finally { lock.unlock(); } } /** * Always returns Integer.MAX_VALUE because * a PriorityBlockingQueue is not capacity constrained. * @return Integer.MAX_VALUE */ public int remainingCapacity() { return Integer.MAX_VALUE; } public boolean remove(Object o) { lock.lock(); try { return q.remove(o); } finally { lock.unlock(); } } public boolean contains(Object o) { lock.lock(); try { return q.contains(o); } finally { lock.unlock(); } } public Object[] toArray() { lock.lock(); try { return q.toArray(); } finally { lock.unlock(); } } public String toString() { lock.lock(); try { return q.toString(); } finally { lock.unlock(); } } /** * Atomically removes all of the elements from this delay queue. * The queue will be empty after this call returns. */ public void clear() { lock.lock(); try { q.clear(); } finally { lock.unlock(); } } public T[] toArray(T[] a) { lock.lock(); try { return q.toArray(a); } finally { lock.unlock(); } } /** * Returns an iterator over the elements in this queue. The * iterator does not return the elements in any particular order. * The returned iterator is a thread-safe "fast-fail" iterator * that will throw {@link * java.util.ConcurrentModificationException} upon detected * interference. * * @return an iterator over the elements in this queue. */ public Iterator iterator() { lock.lock(); try { return new Itr(q.iterator()); } finally { lock.unlock(); } } private class Itr implements Iterator { private final Iterator iter; Itr(Iterator i) { iter = i; } public boolean hasNext() { /* * No sync -- we rely on underlying hasNext to be * stateless, in which case we can return true by mistake * only when next() willl subsequently throw * ConcurrentModificationException. */ return iter.hasNext(); } public E next() { lock.lock(); try { return iter.next(); } finally { lock.unlock(); } } public void remove() { lock.lock(); try { iter.remove(); } finally { lock.unlock(); } } } /** * Save the state to a stream (that is, serialize it). This * merely wraps default serialization within lock. The * serialization strategy for items is left to underlying * Queue. Note that locking is not needed on deserialization, so * readObject is not defined, just relying on default. */ private void writeObject(java.io.ObjectOutputStream s) throws java.io.IOException { lock.lock(); try { s.defaultWriteObject(); } finally { lock.unlock(); } } }