/* * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * * This code is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License version 2 only, as * published by the Free Software Foundation.  Oracle designates this * particular file as subject to the "Classpath" exception as provided * by Oracle in the LICENSE file that accompanied this code. * * This code is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License * version 2 for more details (a copy is included in the LICENSE file that * accompanied this code). * * You should have received a copy of the GNU General Public License version * 2 along with this work; if not, write to the Free Software Foundation, * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. * * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA * or visit www.oracle.com if you need additional information or have any * questions. */ /* * This file is available under and governed by the GNU General Public * License version 2 only, as published by the Free Software Foundation. * However, the following notice accompanied the original version of this * file: * * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at * http://creativecommons.org/publicdomain/zero/1.0/ */ package java.util.concurrent; import java.util.concurrent.locks.*;import java.util.*; /** * An unbounded {@linkplain BlockingQueue blocking queue} that uses * the same ordering rules as class {@link PriorityQueue} and supplies * blocking retrieval operations.  While this queue is logically * unbounded, attempted additions may fail due to resource exhaustion * (causing {@code OutOfMemoryError}). This class does not permit * {@code null} elements.  A priority queue relying on {@linkplain * Comparable natural ordering} also does not permit insertion of * non-comparable objects (doing so results in * {@code ClassCastException}). * * <p>This class and its iterator implement all of the * <em>optional</em> methods of the {@link Collection} and {@link * Iterator} interfaces.  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 * {@code Arrays.sort(pq.toArray())}.  Also, method {@code drainTo} * can be used to <em>remove</em> some or all elements in priority * order and place them in another collection. * * <p>Operations on this class make no guarantees about the ordering * of elements with equal priority. If you need to enforce an * ordering, you can define custom classes or comparators that use a * secondary key to break ties in primary priority values.  For * example, here is a class that applies first-in-first-out * tie-breaking to comparable elements. To use it, you would insert a * {@code new FIFOEntry(anEntry)} instead of a plain entry object. * *  <pre> {@code * class FIFOEntry<E extends Comparable<? super E>> *     implements Comparable<FIFOEntry<E>> { *   static final AtomicLong seq = new AtomicLong(0); *   final long seqNum; *   final E entry; *   public FIFOEntry(E entry) { *     seqNum = seq.getAndIncrement(); *     this.entry = entry; *   } *   public E getEntry() { return entry; } *   public int compareTo(FIFOEntry<E> other) { *     int res = entry.compareTo(other.entry); *     if (res == 0 && other.entry != this.entry) *       res = (seqNum < other.seqNum ? -1 : 1); *     return res; *   } * }}</pre> * * <p>This class is a member of the * <a href="{@docRoot}/../technotes/guides/collections/index.html"> * Java Collections Framework</a>. * * @since 1.5 * @author Doug Lea * @param <E> the type of elements held in this collection */public class PriorityBlockingQueue<E> extends AbstractQueue<E>    implements BlockingQueue<E>, java.io.Serializable {    private static final long serialVersionUID = 5595510919245408276L;     /*     * The implementation uses an array-based binary heap, with public     * operations protected with a single lock. However, allocation     * during resizing uses a simple spinlock (used only while not     * holding main lock) in order to allow takes to operate     * concurrently with allocation.  This avoids repeated     * postponement of waiting consumers and consequent element     * build-up. The need to back away from lock during allocation     * makes it impossible to simply wrap delegated     * java.util.PriorityQueue operations within a lock, as was done     * in a previous version of this class. To maintain     * interoperability, a plain PriorityQueue is still used during     * serialization, which maintains compatibility at the espense of     * transiently doubling overhead.     */     /**     * Default array capacity.     */    private static final int DEFAULT_INITIAL_CAPACITY = 11;     /**     * The maximum size of array to allocate.     * Some VMs reserve some header words in an array.     * Attempts to allocate larger arrays may result in     * OutOfMemoryError: Requested array size exceeds VM limit     */    private static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;     /**     * Priority queue represented as a balanced binary heap: the two     * children of queue[n] are queue[2*n+1] and queue[2*(n+1)].  The     * priority queue is ordered by comparator, or by the elements'     * natural ordering, if comparator is null: For each node n in the     * heap and each descendant d of n, n <= d.  The element with the     * lowest value is in queue[0], assuming the queue is nonempty.     */    private transient Object[] queue;     /**     * The number of elements in the priority queue.     */    private transient int size;     /**     * The comparator, or null if priority queue uses elements'     * natural ordering.     */    private transient Comparator<? super E> comparator;     /**     * Lock used for all public operations     */    private final ReentrantLock lock;     /**     * Condition for blocking when empty     */    private final Condition notEmpty;     /**     * Spinlock for allocation, acquired via CAS.     */    private transient volatile int allocationSpinLock;     /**     * A plain PriorityQueue used only for serialization,     * to maintain compatibility with previous versions     * of this class. Non-null only during serialization/deserialization.     */    private PriorityQueue q;     /**     * Creates a {@code PriorityBlockingQueue} with the default     * initial capacity (11) that orders its elements according to     * their {@linkplain Comparable natural ordering}.     */    public PriorityBlockingQueue() {        this(DEFAULT_INITIAL_CAPACITY, null);    }     /**     * Creates a {@code PriorityBlockingQueue} with the specified     * initial capacity that orders its elements according to their     * {@linkplain Comparable natural ordering}.     *     * @param initialCapacity the initial capacity for this priority queue     * @throws IllegalArgumentException if {@code initialCapacity} is less     *         than 1     */    public PriorityBlockingQueue(int initialCapacity) {        this(initialCapacity, null);    }     /**     * Creates a {@code 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 that will be used to order this     *         priority queue.  If {@code null}, the {@linkplain Comparable     *         natural ordering} of the elements will be used.     * @throws IllegalArgumentException if {@code initialCapacity} is less     *         than 1     */    public PriorityBlockingQueue(int initialCapacity,                                 Comparator<? super E> comparator) {        if (initialCapacity < 1)            throw new IllegalArgumentException();        this.lock = new ReentrantLock();        this.notEmpty = lock.newCondition();        this.comparator = comparator;        this.queue = new Object[initialCapacity];    }     /**     * Creates a {@code PriorityBlockingQueue} containing the elements     * in the specified collection.  If the specified collection is a     * {@link SortedSet} or a {@link PriorityQueue},  this     * priority queue will be ordered according to the same ordering.     * Otherwise, this priority queue will be ordered according to the     * {@linkplain Comparable natural ordering} of its elements.     *     * @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 the specified collection or any     *         of its elements are null     */    public PriorityBlockingQueue(Collection<? extends E> c) {        this.lock = new ReentrantLock();        this.notEmpty = lock.newCondition();        boolean heapify = true; // true if not known to be in heap order        boolean screen = true;  // true if must screen for nulls        if (c instanceof SortedSet<?>) {            SortedSet<? extends E> ss = (SortedSet<? extends E>) c;            this.comparator = (Comparator<? super E>) ss.comparator();            heapify = false;        }        else if (c instanceof PriorityBlockingQueue<?>) {            PriorityBlockingQueue<? extends E> pq =                (PriorityBlockingQueue<? extends E>) c;            this.comparator = (Comparator<? super E>) pq.comparator();            screen = false;            if (pq.getClass() == PriorityBlockingQueue.class) // exact match                heapify = false;        }        Object[] a = c.toArray();        int n = a.length;        // If c.toArray incorrectly doesn't return Object[], copy it.        if (a.getClass() != Object[].class)            a = Arrays.copyOf(a, n, Object[].class);        if (screen && (n == 1 || this.comparator != null)) {            for (int i = 0; i < n; ++i)                if (a[i] == null)                    throw new NullPointerException();        }        this.queue = a;        this.size = n;        if (heapify)            heapify();    }     /**     * Tries to grow array to accommodate at least one more element     * (but normally expand by about 50%), giving up (allowing retry)     * on contention (which we expect to be rare). Call only while     * holding lock.     *     * @param array the heap array     * @param oldCap the length of the array     */    private void tryGrow(Object[] array, int oldCap) {        lock.unlock(); // must release and then re-acquire main lock        Object[] newArray = null;        if (allocationSpinLock == 0 &&            UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,                                     0, 1)) {            try {                int newCap = oldCap + ((oldCap < 64) ?                                       (oldCap + 2) : // grow faster if small                                       (oldCap >> 1));                if (newCap - MAX_ARRAY_SIZE > 0) {    // possible overflow                    int minCap = oldCap + 1;                    if (minCap < 0 || minCap > MAX_ARRAY_SIZE)                        throw new OutOfMemoryError();                    newCap = MAX_ARRAY_SIZE;                }                if (newCap > oldCap && queue == array)                    newArray = new Object[newCap];            } finally {                allocationSpinLock = 0;            }        }        if (newArray == null) // back off if another thread is allocating            Thread.yield();        lock.lock();        if (newArray != null && queue == array) {            queue = newArray;            System.arraycopy(array, 0, newArray, 0, oldCap);        }    }     /**     * Mechanics for poll().  Call only while holding lock.     */    private E extract() {        E result;        int n = size - 1;        if (n < 0)            result = null;        else {            Object[] array = queue;            result = (E) array[0];            E x = (E) array[n];            array[n] = null;            Comparator<? super E> cmp = comparator;            if (cmp == null)                siftDownComparable(0, x, array, n);            else                siftDownUsingComparator(0, x, array, n, cmp);            size = n;        }        return result;    }     /**     * Inserts item x at position k, maintaining heap invariant by     * promoting x up the tree until it is greater than or equal to     * its parent, or is the root.     *     * To simplify and speed up coercions and comparisons. the     * Comparable and Comparator versions are separated into different     * methods that are otherwise identical. (Similarly for siftDown.)     * These methods are static, with heap state as arguments, to     * simplify use in light of possible comparator exceptions.     *     * @param k the position to fill     * @param x the item to insert     * @param array the heap array     * @param n heap size     */    private static <T> void siftUpComparable(int k, T x, Object[] array) {        Comparable<? super T> key = (Comparable<? super T>) x;        while (k > 0) {            int parent = (k - 1) >>> 1;            Object e = array[parent];            if (key.compareTo((T) e) >= 0)                break;            array[k] = e;            k = parent;        }        array[k] = key;    }     private static <T> void siftUpUsingComparator(int k, T x, Object[] array,                                       Comparator<? super T> cmp) {        while (k > 0) {            int parent = (k - 1) >>> 1;            Object e = array[parent];            if (cmp.compare(x, (T) e) >= 0)                break;            array[k] = e;            k = parent;        }        array[k] = x;    }     /**     * Inserts item x at position k, maintaining heap invariant by     * demoting x down the tree repeatedly until it is less than or     * equal to its children or is a leaf.     *     * @param k the position to fill     * @param x the item to insert     * @param array the heap array     * @param n heap size     */    private static <T> void siftDownComparable(int k, T x, Object[] array,                                               int n) {        Comparable<? super T> key = (Comparable<? super T>)x;        int half = n >>> 1;           // loop while a non-leaf        while (k < half) {            int child = (k << 1) + 1; // assume left child is least            Object c = array[child];            int right = child + 1;            if (right < n &&                ((Comparable<? super T>) c).compareTo((T) array[right]) > 0)                c = array[child = right];            if (key.compareTo((T) c) <= 0)                break;            array[k] = c;            k = child;        }        array[k] = key;    }     private static <T> void siftDownUsingComparator(int k, T x, Object[] array,                                                    int n,                                                    Comparator<? super T> cmp) {        int half = n >>> 1;        while (k < half) {            int child = (k << 1) + 1;            Object c = array[child];            int right = child + 1;            if (right < n && cmp.compare((T) c, (T) array[right]) > 0)                c = array[child = right];            if (cmp.compare(x, (T) c) <= 0)                break;            array[k] = c;            k = child;        }        array[k] = x;    }     /**     * Establishes the heap invariant (described above) in the entire tree,     * assuming nothing about the order of the elements prior to the call.     */    private void heapify() {        Object[] array = queue;        int n = size;        int half = (n >>> 1) - 1;        Comparator<? super E> cmp = comparator;        if (cmp == null) {            for (int i = half; i >= 0; i--)                siftDownComparable(i, (E) array[i], array, n);        }        else {            for (int i = half; i >= 0; i--)                siftDownUsingComparator(i, (E) array[i], array, n, cmp);        }    }     /**     * Inserts the specified element into this priority queue.     *     * @param e the element to add     * @return {@code true} (as specified by {@link Collection#add})     * @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 add(E e) {        return offer(e);    }     /**     * Inserts the specified element into this priority queue.     * As the queue is unbounded, this method will never return {@code false}.     *     * @param e the element to add     * @return {@code true} (as specified by {@link Queue#offer})     * @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 e) {        if (e == null)            throw new NullPointerException();        final ReentrantLock lock = this.lock;        lock.lock();        int n, cap;        Object[] array;        while ((n = size) >= (cap = (array = queue).length))            tryGrow(array, cap);        try {            Comparator<? super E> cmp = comparator;            if (cmp == null)                siftUpComparable(n, e, array);            else                siftUpUsingComparator(n, e, array, cmp);            size = n + 1;            notEmpty.signal();        } finally {            lock.unlock();        }        return true;    }     /**     * Inserts the specified element into this priority queue.     * As the queue is unbounded, this method will never block.     *     * @param e the element to add     * @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 void put(E e) {        offer(e); // never need to block    }     /**

Lines 1–500 of 978. Load more