util/concurrent/ArrayBlockingQueue.java

/*
 * 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/licenses/publicdomain
 */

package java.util.concurrent;
import java.util.concurrent.locks.*;
import java.util.*;

/**
 * A bounded {@linkplain BlockingQueue blocking queue} backed by an
 * array.  This queue orders elements FIFO (first-in-first-out).  The
 * <em>head</em> of the queue is that element that has been on the
 * queue the longest time.  The <em>tail</em> of the queue is that
 * element that has been on the queue the shortest time. New elements
 * are inserted at the tail of the queue, and the queue retrieval
 * operations obtain elements at the head of the queue.
 *
 * <p>This is a classic &quot;bounded buffer&quot;, in which a
 * fixed-sized array holds elements inserted by producers and
 * extracted by consumers.  Once created, the capacity cannot be
 * increased.  Attempts to offer an element to a full queue will
 * result in the offer operation blocking; attempts to retrieve an
 * element from an empty queue will similarly block.
 *
 * <p> This class supports an optional fairness policy for ordering
 * waiting producer and consumer threads.  By default, this ordering
 * is not guaranteed. However, a queue constructed with fairness set
 * to <tt>true</tt> grants threads access in FIFO order. Fairness
 * generally decreases throughput but reduces variability and avoids
 * starvation.
 *
 * <p>This class and its iterator implement all of the
 * <em>optional</em> methods of the {@link Collection} and {@link
 * Iterator} interfaces.
 *
 * <p>This class is a member of the
 * <a href="{@docRoot}/../guide/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 ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    /**
     * Serialization ID. This class relies on default serialization
     * even for the items array, which is default-serialized, even if
     * it is empty. Otherwise it could not be declared final, which is
     * necessary here.
     */
    private static final long serialVersionUID = -817911632652898426L;

    /** The queued items  */
    private final E[] items;
    /** items index for next take, poll or remove */
    private transient int takeIndex;
    /** items index for next put, offer, or add. */
    private transient int putIndex;
    /** Number of items in the queue */
    private int count;

    /*
     * Concurrency control uses the classic two-condition algorithm
     * found in any textbook.
     */

    /** Main lock guarding all access */
    private final ReentrantLock lock;
    /** Condition for waiting takes */
    private final Condition notEmpty;
    /** Condition for waiting puts */
    private final Condition notFull;

    // Internal helper methods

    /**
     * Circularly increment i.
     */
    final int inc(int i) {
        return (++i == items.length)? 0 : i;
    }

    /**
     * Insert element at current put position, advance, and signal.
     * Call only when holding lock.
     */
    private void insert(E x) {
        items[putIndex] = x;
        putIndex = inc(putIndex);
        ++count;
        notEmpty.signal();
    }

    /**
     * Extract element at current take position, advance, and signal.
     * Call only when holding lock.
     */
    private E extract() {
        final E[] items = this.items;
        E x = items[takeIndex];
        items[takeIndex] = null;
        takeIndex = inc(takeIndex);
        --count;
        notFull.signal();
        return x;
    }

    /**
     * Utility for remove and iterator.remove: Delete item at position i.
     * Call only when holding lock.
     */
    void removeAt(int i) {
        final E[] items = this.items;
        // if removing front item, just advance
        if (i == takeIndex) {
            items[takeIndex] = null;
            takeIndex = inc(takeIndex);
        } else {
            // slide over all others up through putIndex.
            for (;;) {
                int nexti = inc(i);
                if (nexti != putIndex) {
                    items[i] = items[nexti];
                    i = nexti;
                } else {
                    items[i] = null;
                    putIndex = i;
                    break;
                }
            }
        }
        --count;
        notFull.signal();
    }

    /**
     * Creates an <tt>ArrayBlockingQueue</tt> with the given (fixed)
     * capacity and default access policy.
     * @param capacity the capacity of this queue
     * @throws IllegalArgumentException if <tt>capacity</tt> is less than 1
     */
    public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }

    /**
     * Creates an <tt>ArrayBlockingQueue</tt> with the given (fixed)
     * capacity and the specified access policy.
     * @param capacity the capacity of this queue
     * @param fair if <tt>true</tt> then queue accesses for threads blocked
     * on insertion or removal, are processed in FIFO order; if <tt>false</tt>
     * the access order is unspecified.
     * @throws IllegalArgumentException if <tt>capacity</tt> is less than 1
     */
    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = (E[]) new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

    /**
     * Creates an <tt>ArrayBlockingQueue</tt> with the given (fixed)
     * capacity, the specified access policy and initially containing the
     * elements of the given collection,
     * added in traversal order of the collection's iterator.
     * @param capacity the capacity of this queue
     * @param fair if <tt>true</tt> then queue accesses for threads blocked
     * on insertion or removal, are processed in FIFO order; if <tt>false</tt>
     * the access order is unspecified.
     * @param c the collection of elements to initially contain
     * @throws IllegalArgumentException if <tt>capacity</tt> is less than
     * <tt>c.size()</tt>, or less than 1.
     * @throws NullPointerException if <tt>c</tt> or any element within it
     * is <tt>null</tt>
     */
    public ArrayBlockingQueue(int capacity, boolean fair,
                              Collection<? extends E> c) {
        this(capacity, fair);
        if (capacity < c.size())
            throw new IllegalArgumentException();

        for (Iterator<? extends E> it = c.iterator(); it.hasNext();)
            add(it.next());
    }

    /**
     * Inserts the specified element at the tail of this queue if possible,
     * returning immediately if this queue is full.
     *
     * @param o the element to add.
     * @return <tt>true</tt> if it was possible to add the element to
     *         this queue, else <tt>false</tt>
     * @throws NullPointerException if the specified element is <tt>null</tt>
     */
    public boolean offer(E o) {
        if (o == null) throw new NullPointerException();
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false;
            else {
                insert(o);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * Inserts the specified element at the tail of this queue, waiting if
     * necessary up to the specified wait time for space to become available.
     * @param o the element to add
     * @param timeout how long to wait before giving up, in units of
     * <tt>unit</tt>
     * @param unit a <tt>TimeUnit</tt> determining how to interpret the
     * <tt>timeout</tt> parameter
     * @return <tt>true</tt> if successful, or <tt>false</tt> if
     * the specified waiting time elapses before space is available.
     * @throws InterruptedException if interrupted while waiting.
     * @throws NullPointerException if the specified element is <tt>null</tt>.
     */
    public boolean offer(E o, long timeout, TimeUnit unit)
        throws InterruptedException {

        if (o == null) throw new NullPointerException();
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            long nanos = unit.toNanos(timeout);
            for (;;) {
                if (count != items.length) {
                    insert(o);
                    return true;
                }
                if (nanos <= 0)
                    return false;
                try {
                    nanos = notFull.awaitNanos(nanos);
                } catch (InterruptedException ie) {
                    notFull.signal(); // propagate to non-interrupted thread
                    throw ie;
                }
            }
        } finally {
            lock.unlock();
        }
    }


    public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == 0)
                return null;
            E x = extract();
            return x;
        } finally {
            lock.unlock();
        }
    }

    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            long nanos = unit.toNanos(timeout);
            for (;;) {
                if (count != 0) {
                    E x = extract();
                    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 boolean remove(Object o) {
        if (o == null) return false;
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            int i = takeIndex;
            int k = 0;
            for (;;) {
                if (k++ >= count)
                    return false;
                if (o.equals(items[i])) {
                    removeAt(i);
                    return true;
                }
                i = inc(i);
            }

        } finally {
            lock.unlock();
        }
    }

    public E peek() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return (count == 0) ? null : items[takeIndex];
        } finally {
            lock.unlock();
        }
    }

    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            try {
                while (count == 0)
                    notEmpty.await();
            } catch (InterruptedException ie) {
                notEmpty.signal(); // propagate to non-interrupted thread
                throw ie;
            }
            E x = extract();
            return x;
        } finally {
            lock.unlock();
        }
    }

    /**
     * Adds the specified element to the tail of this queue, waiting if
     * necessary for space to become available.
     * @param o the element to add
     * @throws InterruptedException if interrupted while waiting.
     * @throws NullPointerException if the specified element is <tt>null</tt>.
     */
    public void put(E o) throws InterruptedException {
        if (o == null) throw new NullPointerException();
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            try {
                while (count == items.length)
                    notFull.await();
            } catch (InterruptedException ie) {
                notFull.signal(); // propagate to non-interrupted thread
                throw ie;
            }
            insert(o);
        } finally {
            lock.unlock();
        }
    }

    // this doc comment is overridden to remove the reference to collections
    // greater in size than Integer.MAX_VALUE
    /**
     * Returns the number of elements in this queue.
     *
     * @return  the number of elements in this queue.
     */
    public int size() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return count;
        } finally {
            lock.unlock();
        }
    }

    // this doc comment is a modified copy of the inherited doc comment,
    // without the reference to unlimited queues.
    /**
     * Returns the number of elements that this queue can ideally (in
     * the absence of memory or resource constraints) accept without
     * blocking. This is always equal to the initial capacity of this queue
     * less the current <tt>size</tt> of this queue.
     * <p>Note that you <em>cannot</em> always tell if
     * an attempt to <tt>add</tt> an element will succeed by
     * inspecting <tt>remainingCapacity</tt> because it may be the
     * case that a waiting consumer is ready to <tt>take</tt> an
     * element out of an otherwise full queue.
     */
    public int remainingCapacity() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return items.length - count;
        } finally {
            lock.unlock();
        }
    }


    public boolean contains(Object o) {
        if (o == null) return false;
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            int i = takeIndex;
            int k = 0;
            while (k++ < count) {
                if (o.equals(items[i]))
                    return true;
                i = inc(i);
            }
            return false;
        } finally {
            lock.unlock();
        }
    }

    public Object[] toArray() {
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            Object[] a = new Object[count];
            int k = 0;
            int i = takeIndex;
            while (k < count) {
                a[k++] = items[i];
                i = inc(i);
            }
            return a;
        } finally {
            lock.unlock();
        }
    }

    public <T> T[] toArray(T[] a) {
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (a.length < count)
                a = (T[])java.lang.reflect.Array.newInstance(
                    a.getClass().getComponentType(),
                    count
                    );

            int k = 0;
            int i = takeIndex;
            while (k < count) {
                a[k++] = (T)items[i];
                i = inc(i);
            }
            if (a.length > count)
                a[count] = null;
            return a;
        } finally {
            lock.unlock();
        }
    }

    public String toString() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return super.toString();
        } finally {
            lock.unlock();
        }
    }


    public void clear() {
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            int i = takeIndex;
            int k = count;
            while (k-- > 0) {
                items[i] = null;
                i = inc(i);
            }
            count = 0;
            putIndex = 0;
            takeIndex = 0;
            notFull.signalAll();
        } finally {
            lock.unlock();
        }
    }

    public int drainTo(Collection<? super E> c) {
        if (c == null)
            throw new NullPointerException();
        if (c == this)
            throw new IllegalArgumentException();
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            int i = takeIndex;
            int n = 0;
            int max = count;
            while (n < max) {
                c.add(items[i]);
                items[i] = null;
                i = inc(i);
                ++n;
            }
            if (n > 0) {
                count = 0;
                putIndex = 0;
                takeIndex = 0;
                notFull.signalAll();
            }
            return n;
        } finally {
            lock.unlock();
        }
    }


    public int drainTo(Collection<? super E> c, int maxElements) {
        if (c == null)
            throw new NullPointerException();
        if (c == this)
            throw new IllegalArgumentException();
        if (maxElements <= 0)
            return 0;
        final E[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            int i = takeIndex;
            int n = 0;
            int sz = count;
            int max = (maxElements < count)? maxElements : count;
            while (n < max) {
                c.add(items[i]);
                items[i] = null;
                i = inc(i);
                ++n;
            }
            if (n > 0) {
                count -= n;
                takeIndex = i;
                notFull.signalAll();
            }
            return n;
        } finally {
            lock.unlock();
        }
    }


    /**
     * Returns an iterator over the elements in this queue in proper sequence.
     * The returned <tt>Iterator</tt> is a "weakly consistent" iterator that
     * will never throw {@link java.util.ConcurrentModificationException},
     * and guarantees to traverse elements as they existed upon
     * construction of the iterator, and may (but is not guaranteed to)
     * reflect any modifications subsequent to construction.
     *
     * @return an iterator over the elements in this queue in proper sequence.
     */
    public Iterator<E> iterator() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return new Itr();
        } finally {
            lock.unlock();
        }
    }

    /**
     * Iterator for ArrayBlockingQueue
     */
    private class Itr implements Iterator<E> {
        /**
         * Index of element to be returned by next,
         * or a negative number if no such.
         */
        private int nextIndex;

        /**
         * nextItem holds on to item fields because once we claim
         * that an element exists in hasNext(), we must return it in
         * the following next() call even if it was in the process of
         * being removed when hasNext() was called.
         **/
        private E nextItem;

        /**
         * Index of element returned by most recent call to next.
         * Reset to -1 if this element is deleted by a call to remove.
         */
        private int lastRet;

        Itr() {
            lastRet = -1;
            if (count == 0)
                nextIndex = -1;
            else {
                nextIndex = takeIndex;
                nextItem = items[takeIndex];
            }
        }

        public boolean hasNext() {
            /*
             * No sync. We can return true by mistake here
             * only if this iterator passed across threads,
             * which we don't support anyway.
             */
            return nextIndex >= 0;
        }

        /**
         * Check whether nextIndex is valid; if so setting nextItem.
         * Stops iterator when either hits putIndex or sees null item.
         */
        private void checkNext() {
            if (nextIndex == putIndex) {
                nextIndex = -1;
                nextItem = null;
            } else {
                nextItem = items[nextIndex];
                if (nextItem == null)
                    nextIndex = -1;
            }
        }

        public E next() {
            final ReentrantLock lock = ArrayBlockingQueue.this.lock;
            lock.lock();
            try {
                if (nextIndex < 0)
                    throw new NoSuchElementException();
                lastRet = nextIndex;
                E x = nextItem;
                nextIndex = inc(nextIndex);
                checkNext();
                return x;
            } finally {
                lock.unlock();
            }
        }

        public void remove() {
            final ReentrantLock lock = ArrayBlockingQueue.this.lock;
            lock.lock();
            try {
                int i = lastRet;
                if (i == -1)
                    throw new IllegalStateException();
                lastRet = -1;

                int ti = takeIndex;
                removeAt(i);
                // back up cursor (reset to front if was first element)
                nextIndex = (i == ti) ? takeIndex : i;
                checkNext();
            } finally {
                lock.unlock();
            }
        }
    }
}
Revision:	1.43
Committed:	Sat Apr 10 14:25:24 2004 UTC (20 years, 2 months ago) by dl
Branch:	MAIN
Changes since 1.42:	+3 -2 lines
Log Message:	Documentation improvements
#	User	Rev	Content
1	dl	1.2	/*
2			* Written by Doug Lea with assistance from members of JCP JSR-166
3	dl	1.38	* Expert Group and released to the public domain, as explained at
4			* http://creativecommons.org/licenses/publicdomain
5	dl	1.2	*/
6
7	tim	1.1	package java.util.concurrent;
8	dl	1.11	import java.util.concurrent.locks.*;
9	tim	1.1	import java.util.*;
10
11			/**
12	dl	1.25	* A bounded {@linkplain BlockingQueue blocking queue} backed by an
13			* array. This queue orders elements FIFO (first-in-first-out). The
14			* <em>head</em> of the queue is that element that has been on the
15			* queue the longest time. The <em>tail</em> of the queue is that
16			* element that has been on the queue the shortest time. New elements
17			* are inserted at the tail of the queue, and the queue retrieval
18			* operations obtain elements at the head of the queue.
19	dholmes	1.13	*
20	dl	1.40	* <p>This is a classic "bounded buffer", in which a
21			* fixed-sized array holds elements inserted by producers and
22			* extracted by consumers. Once created, the capacity cannot be
23			* increased. Attempts to offer an element to a full queue will
24			* result in the offer operation blocking; attempts to retrieve an
25			* element from an empty queue will similarly block.
26	dl	1.11	*
27			* <p> This class supports an optional fairness policy for ordering
28	dl	1.42	* waiting producer and consumer threads. By default, this ordering
29			* is not guaranteed. However, a queue constructed with fairness set
30			* to <tt>true</tt> grants threads access in FIFO order. Fairness
31			* generally decreases throughput but reduces variability and avoids
32			* starvation.
33	brian	1.7	*
34	dl	1.43	* <p>This class and its iterator implement all of the
35			* <em>optional</em> methods of the {@link Collection} and {@link
36			* Iterator} interfaces.
37	dl	1.26	*
38	dl	1.41	* <p>This class is a member of the
39			* <a href="{@docRoot}/../guide/collections/index.html">
40			* Java Collections Framework</a>.
41			*
42	dl	1.8	* @since 1.5
43			* @author Doug Lea
44	dl	1.33	* @param <E> the type of elements held in this collection
45	dl	1.8	*/
46	dl	1.5	public class ArrayBlockingQueue<E> extends AbstractQueue<E>
47	tim	1.1	implements BlockingQueue<E>, java.io.Serializable {
48
49	dl	1.36	/**
50	dl	1.42	* Serialization ID. This class relies on default serialization
51			* even for the items array, which is default-serialized, even if
52			* it is empty. Otherwise it could not be declared final, which is
53	dl	1.36	* necessary here.
54			*/
55			private static final long serialVersionUID = -817911632652898426L;
56
57			/** The queued items */
58			private final E[] items;
59	dl	1.8	/** items index for next take, poll or remove */
60	tim	1.12	private transient int takeIndex;
61	dl	1.8	/** items index for next put, offer, or add. */
62	tim	1.12	private transient int putIndex;
63	dl	1.8	/** Number of items in the queue */
64	dl	1.5	private int count;
65	tim	1.12
66	dl	1.5	/*
67	dl	1.36	* Concurrency control uses the classic two-condition algorithm
68	dl	1.5	* found in any textbook.
69			*/
70
71	dl	1.11	/** Main lock guarding all access */
72			private final ReentrantLock lock;
73	dholmes	1.13	/** Condition for waiting takes */
74	dl	1.37	private final Condition notEmpty;
75	dl	1.35	/** Condition for waiting puts */
76	dl	1.37	private final Condition notFull;
77	dl	1.5
78			// Internal helper methods
79
80			/**
81			* Circularly increment i.
82			*/
83	dl	1.40	final int inc(int i) {
84	dl	1.5	return (++i == items.length)? 0 : i;
85			}
86
87			/**
88	dl	1.9	* Insert element at current put position, advance, and signal.
89			* Call only when holding lock.
90	dl	1.5	*/
91			private void insert(E x) {
92			items[putIndex] = x;
93			putIndex = inc(putIndex);
94			++count;
95	dl	1.9	notEmpty.signal();
96	tim	1.1	}
97	tim	1.12
98	dl	1.5	/**
99	dl	1.9	* Extract element at current take position, advance, and signal.
100			* Call only when holding lock.
101	dl	1.5	*/
102	dholmes	1.13	private E extract() {
103	dl	1.36	final E[] items = this.items;
104	dl	1.5	E x = items[takeIndex];
105			items[takeIndex] = null;
106			takeIndex = inc(takeIndex);
107			--count;
108	dl	1.9	notFull.signal();
109	dl	1.5	return x;
110			}
111
112			/**
113	tim	1.12	* Utility for remove and iterator.remove: Delete item at position i.
114	dl	1.9	* Call only when holding lock.
115	dl	1.5	*/
116			void removeAt(int i) {
117	dl	1.36	final E[] items = this.items;
118	dl	1.9	// if removing front item, just advance
119			if (i == takeIndex) {
120			items[takeIndex] = null;
121			takeIndex = inc(takeIndex);
122	tim	1.23	} else {
123	dl	1.9	// slide over all others up through putIndex.
124			for (;;) {
125			int nexti = inc(i);
126			if (nexti != putIndex) {
127			items[i] = items[nexti];
128			i = nexti;
129	tim	1.23	} else {
130	dl	1.9	items[i] = null;
131			putIndex = i;
132			break;
133			}
134	dl	1.5	}
135			}
136	dl	1.9	--count;
137			notFull.signal();
138	tim	1.1	}
139
140			/**
141	dholmes	1.21	* Creates an <tt>ArrayBlockingQueue</tt> with the given (fixed)
142	dholmes	1.13	* capacity and default access policy.
143			* @param capacity the capacity of this queue
144	dholmes	1.16	* @throws IllegalArgumentException if <tt>capacity</tt> is less than 1
145	dl	1.5	*/
146	dholmes	1.13	public ArrayBlockingQueue(int capacity) {
147	dl	1.36	this(capacity, false);
148	dl	1.5	}
149	dl	1.2
150	dl	1.5	/**
151	dholmes	1.21	* Creates an <tt>ArrayBlockingQueue</tt> with the given (fixed)
152	dholmes	1.13	* capacity and the specified access policy.
153			* @param capacity the capacity of this queue
154			* @param fair if <tt>true</tt> then queue accesses for threads blocked
155			* on insertion or removal, are processed in FIFO order; if <tt>false</tt>
156			* the access order is unspecified.
157	dholmes	1.16	* @throws IllegalArgumentException if <tt>capacity</tt> is less than 1
158	dl	1.11	*/
159	dholmes	1.13	public ArrayBlockingQueue(int capacity, boolean fair) {
160	dl	1.36	if (capacity <= 0)
161			throw new IllegalArgumentException();
162			this.items = (E[]) new Object[capacity];
163			lock = new ReentrantLock(fair);
164			notEmpty = lock.newCondition();
165			notFull = lock.newCondition();
166	dl	1.5	}
167
168	dholmes	1.16	/**
169	dholmes	1.21	* Creates an <tt>ArrayBlockingQueue</tt> with the given (fixed)
170			* capacity, the specified access policy and initially containing the
171	tim	1.17	* elements of the given collection,
172	dholmes	1.16	* added in traversal order of the collection's iterator.
173			* @param capacity the capacity of this queue
174			* @param fair if <tt>true</tt> then queue accesses for threads blocked
175			* on insertion or removal, are processed in FIFO order; if <tt>false</tt>
176			* the access order is unspecified.
177			* @param c the collection of elements to initially contain
178			* @throws IllegalArgumentException if <tt>capacity</tt> is less than
179			* <tt>c.size()</tt>, or less than 1.
180			* @throws NullPointerException if <tt>c</tt> or any element within it
181			* is <tt>null</tt>
182			*/
183	tim	1.20	public ArrayBlockingQueue(int capacity, boolean fair,
184	dholmes	1.18	Collection<? extends E> c) {
185	dl	1.36	this(capacity, fair);
186	dholmes	1.16	if (capacity < c.size())
187			throw new IllegalArgumentException();
188
189	tim	1.20	for (Iterator<? extends E> it = c.iterator(); it.hasNext();)
190			add(it.next());
191	dholmes	1.16	}
192	dl	1.2
193	dholmes	1.13	/**
194	dholmes	1.27	* Inserts the specified element at the tail of this queue if possible,
195			* returning immediately if this queue is full.
196	dl	1.25	*
197	dl	1.28	* @param o the element to add.
198			* @return <tt>true</tt> if it was possible to add the element to
199			* this queue, else <tt>false</tt>
200	dl	1.25	* @throws NullPointerException if the specified element is <tt>null</tt>
201	dholmes	1.13	*/
202	dholmes	1.21	public boolean offer(E o) {
203			if (o == null) throw new NullPointerException();
204	dl	1.36	final ReentrantLock lock = this.lock;
205	dl	1.5	lock.lock();
206			try {
207	tim	1.12	if (count == items.length)
208	dl	1.2	return false;
209	dl	1.5	else {
210	dholmes	1.21	insert(o);
211	dl	1.5	return true;
212			}
213	tim	1.23	} finally {
214	tim	1.12	lock.unlock();
215	dl	1.2	}
216	dl	1.5	}
217	dl	1.2
218	dholmes	1.13	/**
219	dholmes	1.27	* Inserts the specified element at the tail of this queue, waiting if
220	dholmes	1.13	* necessary up to the specified wait time for space to become available.
221	dl	1.28	* @param o the element to add
222			* @param timeout how long to wait before giving up, in units of
223			* <tt>unit</tt>
224			* @param unit a <tt>TimeUnit</tt> determining how to interpret the
225			* <tt>timeout</tt> parameter
226			* @return <tt>true</tt> if successful, or <tt>false</tt> if
227			* the specified waiting time elapses before space is available.
228			* @throws InterruptedException if interrupted while waiting.
229	dl	1.25	* @throws NullPointerException if the specified element is <tt>null</tt>.
230	brian	1.7	*/
231	dholmes	1.21	public boolean offer(E o, long timeout, TimeUnit unit)
232	dholmes	1.13	throws InterruptedException {
233	dl	1.2
234	dholmes	1.21	if (o == null) throw new NullPointerException();
235	dl	1.36	final ReentrantLock lock = this.lock;
236	dl	1.5	lock.lockInterruptibly();
237			try {
238	dholmes	1.13	long nanos = unit.toNanos(timeout);
239	dl	1.5	for (;;) {
240			if (count != items.length) {
241	dholmes	1.21	insert(o);
242	dl	1.5	return true;
243			}
244			if (nanos <= 0)
245			return false;
246			try {
247			nanos = notFull.awaitNanos(nanos);
248	tim	1.23	} catch (InterruptedException ie) {
249	dl	1.5	notFull.signal(); // propagate to non-interrupted thread
250			throw ie;
251			}
252			}
253	tim	1.23	} finally {
254	dl	1.5	lock.unlock();
255	dl	1.2	}
256	dl	1.5	}
257	dl	1.2
258	dholmes	1.13
259			public E poll() {
260	dl	1.36	final ReentrantLock lock = this.lock;
261	dholmes	1.13	lock.lock();
262			try {
263			if (count == 0)
264			return null;
265			E x = extract();
266			return x;
267	tim	1.23	} finally {
268	tim	1.15	lock.unlock();
269	dholmes	1.13	}
270			}
271
272	dl	1.5	public E poll(long timeout, TimeUnit unit) throws InterruptedException {
273	dl	1.36	final ReentrantLock lock = this.lock;
274	dl	1.5	lock.lockInterruptibly();
275			try {
276	dholmes	1.13	long nanos = unit.toNanos(timeout);
277	dl	1.2	for (;;) {
278	dl	1.5	if (count != 0) {
279			E x = extract();
280			return x;
281	dl	1.2	}
282	dl	1.5	if (nanos <= 0)
283			return null;
284			try {
285			nanos = notEmpty.awaitNanos(nanos);
286	tim	1.23	} catch (InterruptedException ie) {
287	dl	1.5	notEmpty.signal(); // propagate to non-interrupted thread
288			throw ie;
289			}
290
291	dl	1.2	}
292	tim	1.23	} finally {
293	dl	1.5	lock.unlock();
294			}
295			}
296	dl	1.2
297	brian	1.7
298	dholmes	1.21	public boolean remove(Object o) {
299			if (o == null) return false;
300	dl	1.36	final E[] items = this.items;
301			final ReentrantLock lock = this.lock;
302	dl	1.5	lock.lock();
303			try {
304			int i = takeIndex;
305			int k = 0;
306			for (;;) {
307			if (k++ >= count)
308			return false;
309	dholmes	1.21	if (o.equals(items[i])) {
310	dl	1.5	removeAt(i);
311			return true;
312			}
313	dl	1.2	i = inc(i);
314	dl	1.5	}
315	tim	1.12
316	tim	1.23	} finally {
317	dl	1.5	lock.unlock();
318			}
319			}
320	brian	1.7
321	dholmes	1.13	public E peek() {
322	dl	1.36	final ReentrantLock lock = this.lock;
323	dholmes	1.13	lock.lock();
324			try {
325			return (count == 0) ? null : items[takeIndex];
326	tim	1.23	} finally {
327	dholmes	1.13	lock.unlock();
328			}
329			}
330
331			public E take() throws InterruptedException {
332	dl	1.36	final ReentrantLock lock = this.lock;
333	dholmes	1.13	lock.lockInterruptibly();
334			try {
335			try {
336			while (count == 0)
337			notEmpty.await();
338	tim	1.23	} catch (InterruptedException ie) {
339	dholmes	1.13	notEmpty.signal(); // propagate to non-interrupted thread
340			throw ie;
341			}
342			E x = extract();
343			return x;
344	tim	1.23	} finally {
345	dholmes	1.13	lock.unlock();
346			}
347			}
348
349			/**
350	dholmes	1.21	* Adds the specified element to the tail of this queue, waiting if
351	dholmes	1.13	* necessary for space to become available.
352	dl	1.28	* @param o the element to add
353			* @throws InterruptedException if interrupted while waiting.
354	dl	1.25	* @throws NullPointerException if the specified element is <tt>null</tt>.
355	dholmes	1.13	*/
356	dholmes	1.21	public void put(E o) throws InterruptedException {
357			if (o == null) throw new NullPointerException();
358	dl	1.36	final E[] items = this.items;
359			final ReentrantLock lock = this.lock;
360	dholmes	1.13	lock.lockInterruptibly();
361			try {
362			try {
363			while (count == items.length)
364			notFull.await();
365	tim	1.23	} catch (InterruptedException ie) {
366	dholmes	1.13	notFull.signal(); // propagate to non-interrupted thread
367			throw ie;
368			}
369	dholmes	1.21	insert(o);
370	tim	1.23	} finally {
371	dholmes	1.13	lock.unlock();
372			}
373			}
374
375			// this doc comment is overridden to remove the reference to collections
376			// greater in size than Integer.MAX_VALUE
377	tim	1.15	/**
378	dl	1.25	* Returns the number of elements in this queue.
379			*
380			* @return the number of elements in this queue.
381	dholmes	1.13	*/
382			public int size() {
383	dl	1.36	final ReentrantLock lock = this.lock;
384	dholmes	1.13	lock.lock();
385			try {
386			return count;
387	tim	1.23	} finally {
388	dholmes	1.13	lock.unlock();
389			}
390			}
391
392			// this doc comment is a modified copy of the inherited doc comment,
393			// without the reference to unlimited queues.
394	tim	1.15	/**
395	dholmes	1.21	* Returns the number of elements that this queue can ideally (in
396	dholmes	1.13	* the absence of memory or resource constraints) accept without
397			* blocking. This is always equal to the initial capacity of this queue
398			* less the current <tt>size</tt> of this queue.
399			* <p>Note that you <em>cannot</em> always tell if
400			* an attempt to <tt>add</tt> an element will succeed by
401			* inspecting <tt>remainingCapacity</tt> because it may be the
402			* case that a waiting consumer is ready to <tt>take</tt> an
403			* element out of an otherwise full queue.
404			*/
405			public int remainingCapacity() {
406	dl	1.36	final ReentrantLock lock = this.lock;
407	dholmes	1.13	lock.lock();
408			try {
409			return items.length - count;
410	tim	1.23	} finally {
411	dholmes	1.13	lock.unlock();
412			}
413			}
414
415
416	dholmes	1.21	public boolean contains(Object o) {
417			if (o == null) return false;
418	dl	1.36	final E[] items = this.items;
419			final ReentrantLock lock = this.lock;
420	dl	1.5	lock.lock();
421			try {
422			int i = takeIndex;
423			int k = 0;
424			while (k++ < count) {
425	dholmes	1.21	if (o.equals(items[i]))
426	dl	1.2	return true;
427	dl	1.5	i = inc(i);
428			}
429	dl	1.2	return false;
430	tim	1.23	} finally {
431	dl	1.5	lock.unlock();
432			}
433			}
434	brian	1.7
435	dl	1.5	public Object[] toArray() {
436	dl	1.36	final E[] items = this.items;
437			final ReentrantLock lock = this.lock;
438	dl	1.5	lock.lock();
439			try {
440	dl	1.34	Object[] a = new Object[count];
441	dl	1.5	int k = 0;
442			int i = takeIndex;
443			while (k < count) {
444	dl	1.2	a[k++] = items[i];
445	dl	1.5	i = inc(i);
446			}
447	dl	1.2	return a;
448	tim	1.23	} finally {
449	dl	1.5	lock.unlock();
450			}
451			}
452	brian	1.7
453	dl	1.5	public <T> T[] toArray(T[] a) {
454	dl	1.36	final E[] items = this.items;
455			final ReentrantLock lock = this.lock;
456	dl	1.5	lock.lock();
457			try {
458			if (a.length < count)
459	dholmes	1.16	a = (T[])java.lang.reflect.Array.newInstance(
460	tim	1.17	a.getClass().getComponentType(),
461	dholmes	1.16	count
462			);
463	tim	1.12
464	dl	1.5	int k = 0;
465			int i = takeIndex;
466			while (k < count) {
467	dl	1.2	a[k++] = (T)items[i];
468	dl	1.5	i = inc(i);
469			}
470			if (a.length > count)
471			a[count] = null;
472	dl	1.2	return a;
473	tim	1.23	} finally {
474	dl	1.5	lock.unlock();
475			}
476			}
477	dl	1.6
478			public String toString() {
479	dl	1.36	final ReentrantLock lock = this.lock;
480	dl	1.6	lock.lock();
481			try {
482			return super.toString();
483	tim	1.23	} finally {
484	dl	1.6	lock.unlock();
485			}
486			}
487	tim	1.12
488	dl	1.30
489			public void clear() {
490	dl	1.36	final E[] items = this.items;
491			final ReentrantLock lock = this.lock;
492	dl	1.30	lock.lock();
493			try {
494			int i = takeIndex;
495			int k = count;
496			while (k-- > 0) {
497			items[i] = null;
498			i = inc(i);
499			}
500			count = 0;
501			putIndex = 0;
502			takeIndex = 0;
503			notFull.signalAll();
504			} finally {
505			lock.unlock();
506			}
507			}
508
509			public int drainTo(Collection<? super E> c) {
510			if (c == null)
511			throw new NullPointerException();
512			if (c == this)
513			throw new IllegalArgumentException();
514	dl	1.36	final E[] items = this.items;
515			final ReentrantLock lock = this.lock;
516	dl	1.30	lock.lock();
517			try {
518			int i = takeIndex;
519			int n = 0;
520			int max = count;
521			while (n < max) {
522			c.add(items[i]);
523			items[i] = null;
524			i = inc(i);
525			++n;
526			}
527			if (n > 0) {
528			count = 0;
529			putIndex = 0;
530			takeIndex = 0;
531			notFull.signalAll();
532			}
533			return n;
534			} finally {
535			lock.unlock();
536			}
537			}
538
539
540			public int drainTo(Collection<? super E> c, int maxElements) {
541			if (c == null)
542			throw new NullPointerException();
543			if (c == this)
544			throw new IllegalArgumentException();
545			if (maxElements <= 0)
546			return 0;
547	dl	1.36	final E[] items = this.items;
548			final ReentrantLock lock = this.lock;
549	dl	1.30	lock.lock();
550			try {
551			int i = takeIndex;
552			int n = 0;
553			int sz = count;
554			int max = (maxElements < count)? maxElements : count;
555			while (n < max) {
556			c.add(items[i]);
557			items[i] = null;
558			i = inc(i);
559			++n;
560			}
561			if (n > 0) {
562			count -= n;
563			takeIndex = i;
564			notFull.signalAll();
565			}
566			return n;
567			} finally {
568			lock.unlock();
569			}
570			}
571
572
573	brian	1.7	/**
574			* Returns an iterator over the elements in this queue in proper sequence.
575	dl	1.22	* The returned <tt>Iterator</tt> is a "weakly consistent" iterator that
576			* will never throw {@link java.util.ConcurrentModificationException},
577			* and guarantees to traverse elements as they existed upon
578			* construction of the iterator, and may (but is not guaranteed to)
579			* reflect any modifications subsequent to construction.
580	brian	1.7	*
581			* @return an iterator over the elements in this queue in proper sequence.
582			*/
583	dl	1.5	public Iterator<E> iterator() {
584	dl	1.36	final ReentrantLock lock = this.lock;
585	dl	1.5	lock.lock();
586			try {
587	dl	1.2	return new Itr();
588	tim	1.23	} finally {
589	dl	1.5	lock.unlock();
590			}
591			}
592	dl	1.8
593			/**
594			* Iterator for ArrayBlockingQueue
595			*/
596	dl	1.5	private class Itr implements Iterator<E> {
597			/**
598			* Index of element to be returned by next,
599			* or a negative number if no such.
600			*/
601	dl	1.8	private int nextIndex;
602	dl	1.2
603	tim	1.12	/**
604	dl	1.5	* nextItem holds on to item fields because once we claim
605			* that an element exists in hasNext(), we must return it in
606			* the following next() call even if it was in the process of
607			* being removed when hasNext() was called.
608			**/
609	dl	1.8	private E nextItem;
610	dl	1.5
611			/**
612			* Index of element returned by most recent call to next.
613			* Reset to -1 if this element is deleted by a call to remove.
614			*/
615	dl	1.8	private int lastRet;
616	tim	1.12
617	dl	1.5	Itr() {
618			lastRet = -1;
619	tim	1.12	if (count == 0)
620	dl	1.5	nextIndex = -1;
621			else {
622			nextIndex = takeIndex;
623			nextItem = items[takeIndex];
624			}
625			}
626	tim	1.12
627	dl	1.5	public boolean hasNext() {
628			/*
629			* No sync. We can return true by mistake here
630			* only if this iterator passed across threads,
631			* which we don't support anyway.
632	dl	1.2	*/
633	dl	1.5	return nextIndex >= 0;
634			}
635
636			/**
637	dholmes	1.13	* Check whether nextIndex is valid; if so setting nextItem.
638	dl	1.5	* Stops iterator when either hits putIndex or sees null item.
639			*/
640			private void checkNext() {
641			if (nextIndex == putIndex) {
642			nextIndex = -1;
643			nextItem = null;
644	tim	1.23	} else {
645	dl	1.5	nextItem = items[nextIndex];
646			if (nextItem == null)
647			nextIndex = -1;
648	dl	1.2	}
649	dl	1.5	}
650	tim	1.12
651	dl	1.5	public E next() {
652	dl	1.36	final ReentrantLock lock = ArrayBlockingQueue.this.lock;
653	dl	1.5	lock.lock();
654			try {
655			if (nextIndex < 0)
656	dl	1.2	throw new NoSuchElementException();
657	dl	1.5	lastRet = nextIndex;
658			E x = nextItem;
659			nextIndex = inc(nextIndex);
660			checkNext();
661			return x;
662	tim	1.23	} finally {
663	tim	1.12	lock.unlock();
664	dl	1.2	}
665	dl	1.5	}
666	tim	1.12
667	dl	1.5	public void remove() {
668	dl	1.36	final ReentrantLock lock = ArrayBlockingQueue.this.lock;
669	dl	1.5	lock.lock();
670			try {
671	dl	1.2	int i = lastRet;
672			if (i == -1)
673			throw new IllegalStateException();
674			lastRet = -1;
675	tim	1.12
676	dl	1.9	int ti = takeIndex;
677	dl	1.2	removeAt(i);
678	dl	1.9	// back up cursor (reset to front if was first element)
679	tim	1.12	nextIndex = (i == ti) ? takeIndex : i;
680	dl	1.5	checkNext();
681	tim	1.23	} finally {
682	dl	1.5	lock.unlock();
683			}
684			}
685	tim	1.1	}
686			}