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>). A priority queue does not
 * permit <tt>null</tt> elements.  A priority queue relying on natural
 * ordering also does not permit insertion of non-comparable objects
 * (doing so results in <tt>ClassCastException</tt>).
 *
 * <p>This class implements all of the <em>optional</em> methods
 * of the {@link Collection} and {@link Iterator} interfaces.
 * <p>The Iterator provided in method {@link #iterator()} is
 * <em>not</em> guaranteed to traverse the elements of the
 * PriorityBlockingQueue in any particular order. If you need ordered
 * traversal, consider using <tt>Arrays.sort(pq.toArray())</tt>.
 *
 * @since 1.5
 * @author Doug Lea
*/
public class PriorityBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {
    private static final long serialVersionUID = 5595510919245408276L;

    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 if the specified element is <tt>null</tt>.
     * @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 <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();
    }

    /**
     * Inserts the specified element into this priority queue.
     *
     * @param o the element to add
     * @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 if the specified element is <tt>null</tt>.
     */
    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 <tt>null</tt>.
     */
    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 <tt>true</tt>
     * @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 <tt>null</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;
    }

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