/*

 * 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.

     */

    /**

     * 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

    }

    /**

     * Inserts the specified element into this priority queue.