util/concurrent/PriorityBlockingQueue.java

/*
 * 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
 * <tt>OutOfMemoryError</tt>).
 * @since 1.5
 * @author Doug Lea
*/
public class PriorityBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {

    private final PriorityQueue<E> q;
    private final ReentrantLock lock = new ReentrantLock(true);
    private final Condition notEmpty = lock.newCondition();

    /**
     * Creates a <tt>PriorityBlockingQueue</tt> with the default initial 
     * capacity
     * (11) that orders its elements according to their natural
     * ordering (using <tt>Comparable</tt>.)
     */
    public PriorityBlockingQueue() {
        q = new PriorityQueue<E>();
    }

    /**
     * Creates a <tt>PriorityBlockingQueue</tt> with the specified initial
     * capacity
     * that orders its elements according to their natural ordering
     * (using <tt>Comparable</tt>.)
     *
     * @param initialCapacity the initial capacity for this priority queue.
     * @throws IllegalArgumentException if <tt>initialCapacity</tt> is less
     * than 1
     */
    public PriorityBlockingQueue(int initialCapacity) {
        q = new PriorityQueue<E>(initialCapacity, null);
    }

    /**
     * Creates a <tt>PriorityBlockingQueue</tt> 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 <tt>null</tt> then the order depends on the elements' natural
     * ordering.
     * @throws IllegalArgumentException if <tt>initialCapacity</tt> is less
     * than 1
     */
    public PriorityBlockingQueue(int initialCapacity,
                                 Comparator<? super E> comparator) {
        q = new PriorityQueue<E>(initialCapacity, comparator);
    }

    /**
     * Creates a <tt>PriorityBlockingQueue</tt> 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 <tt>c</tt> or any element within it
     * is <tt>null</tt>
     */
    public PriorityBlockingQueue(Collection<? extends E> c) {
        q = new PriorityQueue<E>(c);
    }


    // these first few override just to update doc comments

    /**
     * Adds the specified element to this queue.
     * @return <tt>true</tt> (as per the general contract of
     * <tt>Collection.add</tt>).
     *
     * @throws NullPointerException {@inheritDoc}
     * @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);
    }

    /**
     * Adds all of the elements in the specified collection to this queue.
     * The behavior of this operation is undefined if
     * the specified collection is modified while the operation is in
     * progress.  (This implies that the behavior of this call is undefined if
     * the specified collection is this queue, and this queue is nonempty.)
     * <p>
     * This implementation iterates over the specified collection, and adds
     * each object returned by the iterator to this collection, in turn.
     * @throws NullPointerException {@inheritDoc}
     * @throws ClassCastException if any element cannot be compared
     * with elements currently in the priority queue according
     * to the priority queue's ordering.
     */
    public boolean addAll(Collection<? extends E> c) {
        return super.addAll(c);
    }

    /**
     * Returns the comparator used to order this collection, or <tt>null</tt>
     * if this collection is sorted according to its elements natural ordering
     * (using <tt>Comparable</tt>.)
     *
     * @return the comparator used to order this collection, or <tt>null</tt>
     * if this collection is sorted according to its elements natural ordering.
     */
    public Comparator comparator() {
        return q.comparator();
    }

    /**
     * Adds the specified element to this priority queue.
     *
     * @return <tt>true</tt>
     * @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 {@inheritDoc}
     */
    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.
     * @throws ClassCastException if the element cannot be compared
     * with elements currently in the priority queue according
     * to the priority queue's ordering.
     * @throws NullPointerException {@inheritDoc}
     */
    public void put(E o) {
        offer(o); // never need to block
    }

    /**
     * Adds the specified element to this priority queue. As the queue is
     * unbounded this method will never block.
     * @param o {@inheritDoc}
     * @param timeout This parameter is ignored as the method never blocks
     * @param unit This parameter is ignored as the method never blocks
     * @throws ClassCastException if the element cannot be compared
     * with elements currently in the priority queue according
     * to the priority queue's ordering.
     * @throws NullPointerException {@inheritDoc}
     * @return <tt>true</tt>
     */
    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 <tt>Integer.MAX_VALUE</tt> because
     * a <tt>PriorityBlockingQueue</tt> is not capacity constrained.
     * @return <tt>Integer.MAX_VALUE</tt>
     */
    public int remainingCapacity() {
        return Integer.MAX_VALUE;
    }

    /**
     * Removes a single instance of the specified element from this
     * queue, if it is present.  More formally,
     * removes an element <tt>e</tt> such that <tt>(o==null ? e==null :
     * o.equals(e))</tt>, if the queue contains one or more such
     * elements.  Returns <tt>true</tt> if the queue contained the
     * specified element (or equivalently, if the queue changed as a
     * result of the call).
     *
     * <p>This implementation iterates over the queue looking for the
     * specified element.  If it finds the element, it removes the element
     * from the queue using the iterator's remove method.<p>
     *
     */
    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();
        }
    }

    public <T> 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.
     *
     * @return an iterator over the elements in this queue.
     */
    public Iterator<E> iterator() {
        lock.lock();
        try {
            return new Itr(q.iterator());
        }
        finally {
            lock.unlock();
        }
    }

    private class Itr<E> implements Iterator<E> {
        private final Iterator<E> iter;
        Itr(Iterator<E> 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();
        }
    }

}
Revision:	1.16
Committed:	Wed Aug 6 01:57:53 2003 UTC (20 years, 10 months ago) by dholmes
Branch:	MAIN
Changes since 1.15:	+94 -19 lines
Log Message:	Final major updates to Collection related classes.
#	User	Rev	Content
1	dl	1.2	/*
2			* Written by Doug Lea with assistance from members of JCP JSR-166
3			* Expert Group and released to the public domain. Use, modify, and
4			* redistribute this code in any way without acknowledgement.
5			*/
6
7	tim	1.1	package java.util.concurrent;
8	tim	1.13
9	dl	1.8	import java.util.concurrent.locks.*;
10	tim	1.1	import java.util.*;
11
12			/**
13	dholmes	1.16	* An unbounded {@linkplain BlockingQueue blocking queue} based on a
14			* {@link PriorityQueue},
15			* obeying its ordering rules and implementation characteristics.
16			* While this queue is logically unbounded,
17	dl	1.15	* attempted additions may fail due to resource exhaustion (causing
18	dholmes	1.16	* <tt>OutOfMemoryError</tt>).
19	dl	1.6	* @since 1.5
20			* @author Doug Lea
21	dholmes	1.14	*/
22	dl	1.5	public class PriorityBlockingQueue<E> extends AbstractQueue<E>
23	dl	1.15	implements BlockingQueue<E>, java.io.Serializable {
24	tim	1.1
25	dl	1.5	private final PriorityQueue<E> q;
26	dl	1.9	private final ReentrantLock lock = new ReentrantLock(true);
27	dl	1.5	private final Condition notEmpty = lock.newCondition();
28
29	dl	1.2	/**
30	dholmes	1.16	* Creates a <tt>PriorityBlockingQueue</tt> with the default initial
31	dholmes	1.14	* capacity
32	dholmes	1.10	* (11) that orders its elements according to their natural
33			* ordering (using <tt>Comparable</tt>.)
34	dl	1.2	*/
35			public PriorityBlockingQueue() {
36	dl	1.5	q = new PriorityQueue<E>();
37	dl	1.2	}
38
39			/**
40	dholmes	1.16	* Creates a <tt>PriorityBlockingQueue</tt> with the specified initial
41	dholmes	1.10	* capacity
42			* that orders its elements according to their natural ordering
43			* (using <tt>Comparable</tt>.)
44	dl	1.2	*
45			* @param initialCapacity the initial capacity for this priority queue.
46	dholmes	1.16	* @throws IllegalArgumentException if <tt>initialCapacity</tt> is less
47			* than 1
48	dl	1.2	*/
49			public PriorityBlockingQueue(int initialCapacity) {
50	dl	1.5	q = new PriorityQueue<E>(initialCapacity, null);
51	dl	1.2	}
52
53			/**
54	dholmes	1.16	* Creates a <tt>PriorityBlockingQueue</tt> with the specified initial
55	dholmes	1.10	* capacity
56	dl	1.2	* that orders its elements according to the specified comparator.
57			*
58			* @param initialCapacity the initial capacity for this priority queue.
59			* @param comparator the comparator used to order this priority queue.
60	dholmes	1.10	* If <tt>null</tt> then the order depends on the elements' natural
61			* ordering.
62	dholmes	1.16	* @throws IllegalArgumentException if <tt>initialCapacity</tt> is less
63			* than 1
64	dl	1.2	*/
65	tim	1.13	public PriorityBlockingQueue(int initialCapacity,
66	dholmes	1.14	Comparator<? super E> comparator) {
67	dl	1.5	q = new PriorityQueue<E>(initialCapacity, comparator);
68	dl	1.2	}
69
70			/**
71	dholmes	1.16	* Creates a <tt>PriorityBlockingQueue</tt> containing the elements
72	dl	1.15	* in the specified collection. The priority queue has an initial
73			* capacity of 110% of the size of the specified collection. If
74			* the specified collection is a {@link SortedSet} or a {@link
75			* PriorityQueue}, this priority queue will be sorted according to
76			* the same comparator, or according to its elements' natural
77			* order if the collection is sorted according to its elements'
78			* natural order. Otherwise, this priority queue is ordered
79			* according to its elements' natural order.
80	dl	1.2	*
81	dholmes	1.14	* @param c the collection whose elements are to be placed
82	dl	1.2	* into this priority queue.
83			* @throws ClassCastException if elements of the specified collection
84			* cannot be compared to one another according to the priority
85			* queue's ordering.
86	dholmes	1.14	* @throws NullPointerException if <tt>c</tt> or any element within it
87			* is <tt>null</tt>
88	dl	1.2	*/
89	dholmes	1.14	public PriorityBlockingQueue(Collection<? extends E> c) {
90			q = new PriorityQueue<E>(c);
91	dl	1.7	}
92
93	dholmes	1.10
94	dholmes	1.16	// these first few override just to update doc comments
95	dholmes	1.10
96			/**
97	dholmes	1.16	* Adds the specified element to this queue.
98			* @return <tt>true</tt> (as per the general contract of
99			* <tt>Collection.add</tt>).
100			*
101	dholmes	1.14	* @throws NullPointerException {@inheritDoc}
102	dholmes	1.16	* @throws ClassCastException if the specified element cannot be compared
103			* with elements currently in the priority queue according
104			* to the priority queue's ordering.
105	dholmes	1.10	*/
106	dholmes	1.16	public boolean add(E o) {
107			return super.add(o);
108	dholmes	1.10	}
109
110	tim	1.13	/**
111	dholmes	1.16	* Adds all of the elements in the specified collection to this queue.
112			* The behavior of this operation is undefined if
113			* the specified collection is modified while the operation is in
114			* progress. (This implies that the behavior of this call is undefined if
115			* the specified collection is this queue, and this queue is nonempty.)
116			* <p>
117			* This implementation iterates over the specified collection, and adds
118			* each object returned by the iterator to this collection, in turn.
119	dholmes	1.14	* @throws NullPointerException {@inheritDoc}
120	dholmes	1.16	* @throws ClassCastException if any element cannot be compared
121			* with elements currently in the priority queue according
122			* to the priority queue's ordering.
123	tim	1.13	*/
124			public boolean addAll(Collection<? extends E> c) {
125			return super.addAll(c);
126			}
127	dholmes	1.10
128	dholmes	1.16	/**
129			* Returns the comparator used to order this collection, or <tt>null</tt>
130			* if this collection is sorted according to its elements natural ordering
131			* (using <tt>Comparable</tt>.)
132			*
133			* @return the comparator used to order this collection, or <tt>null</tt>
134			* if this collection is sorted according to its elements natural ordering.
135			*/
136	dl	1.7	public Comparator comparator() {
137			return q.comparator();
138	dl	1.5	}
139
140	dholmes	1.16	/**
141			* Adds the specified element to this priority queue.
142			*
143			* @return <tt>true</tt>
144			* @throws ClassCastException if the specified element cannot be compared
145			* with elements currently in the priority queue according
146			* to the priority queue's ordering.
147			* @throws NullPointerException {@inheritDoc}
148			*/
149	dholmes	1.14	public boolean offer(E o) {
150			if (o == null) throw new NullPointerException();
151	dl	1.5	lock.lock();
152			try {
153	dholmes	1.14	boolean ok = q.offer(o);
154	dl	1.5	assert ok;
155			notEmpty.signal();
156			return true;
157			}
158			finally {
159	tim	1.13	lock.unlock();
160	dl	1.5	}
161			}
162
163	dholmes	1.16	/**
164			* Adds the specified element to this priority queue. As the queue is
165			* unbounded this method will never block.
166			* @throws ClassCastException if the element cannot be compared
167			* with elements currently in the priority queue according
168			* to the priority queue's ordering.
169			* @throws NullPointerException {@inheritDoc}
170			*/
171			public void put(E o) {
172	dholmes	1.14	offer(o); // never need to block
173	dl	1.5	}
174
175	dholmes	1.16	/**
176			* Adds the specified element to this priority queue. As the queue is
177			* unbounded this method will never block.
178			* @param o {@inheritDoc}
179			* @param timeout This parameter is ignored as the method never blocks
180			* @param unit This parameter is ignored as the method never blocks
181			* @throws ClassCastException if the element cannot be compared
182			* with elements currently in the priority queue according
183			* to the priority queue's ordering.
184			* @throws NullPointerException {@inheritDoc}
185			* @return <tt>true</tt>
186			*/
187			public boolean offer(E o, long timeout, TimeUnit unit) {
188	dholmes	1.14	return offer(o); // never need to block
189	dl	1.5	}
190
191			public E take() throws InterruptedException {
192			lock.lockInterruptibly();
193			try {
194			try {
195			while (q.size() == 0)
196			notEmpty.await();
197			}
198			catch (InterruptedException ie) {
199			notEmpty.signal(); // propagate to non-interrupted thread
200			throw ie;
201			}
202			E x = q.poll();
203			assert x != null;
204			return x;
205			}
206			finally {
207			lock.unlock();
208			}
209			}
210
211
212			public E poll() {
213			lock.lock();
214			try {
215			return q.poll();
216			}
217			finally {
218	tim	1.13	lock.unlock();
219	dl	1.5	}
220			}
221
222			public E poll(long timeout, TimeUnit unit) throws InterruptedException {
223	dholmes	1.10	long nanos = unit.toNanos(timeout);
224	dl	1.5	lock.lockInterruptibly();
225			try {
226			for (;;) {
227			E x = q.poll();
228	tim	1.13	if (x != null)
229	dl	1.5	return x;
230			if (nanos <= 0)
231			return null;
232			try {
233			nanos = notEmpty.awaitNanos(nanos);
234			}
235			catch (InterruptedException ie) {
236			notEmpty.signal(); // propagate to non-interrupted thread
237			throw ie;
238			}
239			}
240			}
241			finally {
242			lock.unlock();
243			}
244			}
245
246			public E peek() {
247			lock.lock();
248			try {
249			return q.peek();
250			}
251			finally {
252	tim	1.13	lock.unlock();
253	dl	1.5	}
254			}
255
256			public int size() {
257			lock.lock();
258			try {
259			return q.size();
260			}
261			finally {
262			lock.unlock();
263			}
264			}
265
266			/**
267			* Always returns <tt>Integer.MAX_VALUE</tt> because
268	dholmes	1.16	* a <tt>PriorityBlockingQueue</tt> is not capacity constrained.
269	dl	1.5	* @return <tt>Integer.MAX_VALUE</tt>
270			*/
271			public int remainingCapacity() {
272			return Integer.MAX_VALUE;
273			}
274
275	dholmes	1.16	/**
276			* Removes a single instance of the specified element from this
277			* queue, if it is present. More formally,
278			* removes an element <tt>e</tt> such that <tt>(o==null ? e==null :
279			* o.equals(e))</tt>, if the queue contains one or more such
280			* elements. Returns <tt>true</tt> if the queue contained the
281			* specified element (or equivalently, if the queue changed as a
282			* result of the call).
283			*
284			* <p>This implementation iterates over the queue looking for the
285			* specified element. If it finds the element, it removes the element
286			* from the queue using the iterator's remove method.<p>
287			*
288			*/
289	dholmes	1.14	public boolean remove(Object o) {
290	dl	1.5	lock.lock();
291			try {
292	dholmes	1.14	return q.remove(o);
293	dl	1.5	}
294			finally {
295			lock.unlock();
296			}
297			}
298
299	dholmes	1.14	public boolean contains(Object o) {
300	dl	1.5	lock.lock();
301			try {
302	dholmes	1.14	return q.contains(o);
303	dl	1.5	}
304			finally {
305			lock.unlock();
306			}
307			}
308
309			public Object[] toArray() {
310			lock.lock();
311			try {
312			return q.toArray();
313			}
314			finally {
315			lock.unlock();
316			}
317			}
318
319
320			public String toString() {
321			lock.lock();
322			try {
323			return q.toString();
324			}
325			finally {
326			lock.unlock();
327			}
328			}
329
330			public <T> T[] toArray(T[] a) {
331			lock.lock();
332			try {
333			return q.toArray(a);
334			}
335			finally {
336			lock.unlock();
337			}
338			}
339
340	dholmes	1.16	/**
341			* Returns an iterator over the elements in this queue. The iterator
342			* does not return the elements in any particular order.
343			*
344			* @return an iterator over the elements in this queue.
345			*/
346	dl	1.5	public Iterator<E> iterator() {
347			lock.lock();
348			try {
349			return new Itr(q.iterator());
350			}
351			finally {
352			lock.unlock();
353			}
354			}
355
356			private class Itr<E> implements Iterator<E> {
357			private final Iterator<E> iter;
358	tim	1.13	Itr(Iterator<E> i) {
359			iter = i;
360	dl	1.5	}
361
362	tim	1.13	public boolean hasNext() {
363	dl	1.5	/*
364			* No sync -- we rely on underlying hasNext to be
365			* stateless, in which case we can return true by mistake
366			* only when next() willl subsequently throw
367			* ConcurrentModificationException.
368			*/
369			return iter.hasNext();
370	tim	1.13	}
371
372			public E next() {
373	dl	1.5	lock.lock();
374			try {
375			return iter.next();
376			}
377			finally {
378			lock.unlock();
379			}
380	tim	1.13	}
381
382			public void remove() {
383	dl	1.5	lock.lock();
384			try {
385			iter.remove();
386			}
387			finally {
388			lock.unlock();
389			}
390	tim	1.13	}
391	dl	1.5	}
392
393			/**
394			* Save the state to a stream (that is, serialize it). This
395			* merely wraps default serialization within lock. The
396			* serialization strategy for items is left to underlying
397			* Queue. Note that locking is not needed on deserialization, so
398			* readObject is not defined, just relying on default.
399			*/
400			private void writeObject(java.io.ObjectOutputStream s)
401			throws java.io.IOException {
402			lock.lock();
403			try {
404			s.defaultWriteObject();
405			}
406			finally {
407			lock.unlock();
408			}
409	tim	1.1	}
410
411			}