/*
* 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
* head of the queue is that element that has been on the
* queue the longest time. The tail 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.
*
* This is a classic "bounded buffer", 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.
*
*
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 true grants threads access in FIFO order. Fairness
* generally decreases throughput but reduces variability and avoids
* starvation.
*
*
This class implements all of the optional methods
* of the {@link Collection} and {@link Iterator} interfaces.
*
*
This class is a member of the
*
* Java Collections Framework.
*
* @since 1.5
* @author Doug Lea
* @param the type of elements held in this collection
*/
public class ArrayBlockingQueue extends AbstractQueue
implements BlockingQueue, 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;
//@ instance invariant takeIndex >= 0 && takeIndex < items.length;
//@ instance invariant putIndex >= 0 && putIndex < items.length;
//@ instance invariant count == (putIndex - takeIndex) % items.length;
/** The queued items */
private /*@ spec_public non_null rep @*/ final E[] items;
/** items index for next take, poll or remove */
private /*@ spec_public @*/ transient int takeIndex;
/** items index for next put, offer, or add. */
private /*@ spec_public @*/ transient int putIndex;
/** Number of items in the queue */
private /*@ spec_public @*/ int count;
/*
* Concurrency control uses the classic two-condition algorithm
* found in any textbook.
*/
/** Main lock guarding all access */
private /*@ spec_public readonly rep @*/ final ReentrantLock lock;
/** Condition for waiting takes */
private /*@ spec_public readonly rep @*/ final Condition notEmpty;
/** Condition for waiting puts */
private /*@ spec_public readonly rep @*/ final Condition notFull;
// Internal helper methods
/**
* Circularly increment i.
*/
/*@ normal_behavior
@ requires lock.isLocked() && i > 0 && i < items.length;
@ ensures \result == (\old(i) + 1) % items.length && \independent;
@*/
final /* atomic */ 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 normal_behavior
@ requires lock.isLocked();
@ assignable items[putIndex], putIndex, count;
@ ensure count == (\old(count) + 1) && \independent &&
@ putIndex == (\old(putIndex) + 1) % items.length;
@*/
private /* atomic */ 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 normal_behavior
@ requires lock.isLocked();
@ assignable items[takeIndex], takeIndex, count;
@ ensures \result == \old(items[takeIndex]) && takeIndex == \old(takeIndex) + 1 &&
@ count == \old(count) -1 && \independent;
@*/
private /* atomic */ 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.
*/
/*@ normal_behavior
@ requires lock.isLocked() && i == takeIndex;
@ assignable items[takeIndex], count, takeIndex;
@ ensures \independent && \old(items[takeIndex]) == null &&
@ takeIndex == \old(takeIndex) + 1;
@also
@ requires lock.isLocked() && i != takeIndex;
@ assignable items[*], count, takeIndex;
@ ensures \independent && (\forall int i; i >= i && i < putIndex;
@ items[i] == \old(items[i + 1]) &&
@ putIndex == \old(putIndex) - 1);
@*/
void /*@ atomic @*/ 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 ArrayBlockingQueue with the given (fixed)
* capacity and default access policy.
* @param capacity the capacity of this queue
* @throws IllegalArgumentException if capacity is less than 1
*/
/*@ public normal_behavior
@ requires capacity > 0;
@ ensures \fresh(items) && items.length == capacity && \independent;
@also
@ public exceptional_behavior
@ requires capacity <= 0;
@ signals (IllegalArgumentException);
@*/
public /*@ atomic @*/ ArrayBlockingQueue(int capacity) {
this(capacity, false);
}
/**
* Creates an ArrayBlockingQueue with the given (fixed)
* capacity and the specified access policy.
* @param capacity the capacity of this queue
* @param fair if true then queue accesses for threads blocked
* on insertion or removal, are processed in FIFO order; if false
* the access order is unspecified.
* @throws IllegalArgumentException if capacity is less than 1
*/
/*@ public normal_behavior
@ requires capacity > 0;
@ ensures \fresh(items) && items.length == capacity && \independent;
@also
@ public exceptional_behavior
@ requires capacity <= 0;
@ signals (IllegalArgumentException);
@*/
public /*@ atomic @*/ 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 ArrayBlockingQueue 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 true then queue accesses for threads blocked
* on insertion or removal, are processed in FIFO order; if false
* the access order is unspecified.
* @param c the collection of elements to initially contain
* @throws IllegalArgumentException if capacity is less than
* c.size(), or less than 1.
* @throws NullPointerException if c or any element within it
* is null
*/
/*@ public normal_behavior
@ requires \thread_safe(c);
@ assignable items[*], count, putIndex, takeIndex;
@ ensures \independent && (\forall int i; i >= 0 && i < items.length
@ && i < putIndex && i >= takeIndex;
@ c.contains(items[i]));
@*/
public /*@ atomic @*/ ArrayBlockingQueue(int capacity, boolean fair,
Collection c) {
this(capacity, fair);
if (capacity < c.size())
throw new IllegalArgumentException();
for (Iterator 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 true if it was possible to add the element to
* this queue, else false
* @throws NullPointerException if the specified element is null
*/
/*@ public normal_behavior
@ requires o != null;
@ locks this.lock;
@ assignable items[putIndex], putIndex, count;
@ ensures \old(items.length == count) ==> !(\result) &&
@ \old(items.length != count) ==> count == (\old(count) + 1) &&
@ putIndex == (\old(putIndex) + 1) % items.length &&
@ items[\old(putIndex)] == o;
@also
@ public exceptional_behavior
@ requires o == null;
@ assignable \nothing;
@ signals (NullPointerException)
@*/
public /*@ atomic @*/ 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
* unit
* @param unit a TimeUnit determining how to interpret the
* timeout parameter
* @return true if successful, or false if
* the specified waiting time elapses before space is available.
* @throws InterruptedException if interrupted while waiting.
* @throws NullPointerException if the specified element is null.
*/
/*@ public behavior
@ requires o != null;
@ locks this.lock;
@ ensures \old(items.length != count) ==>
@ count == (\old(count) + 1) &&
@ putIndex == (\old(putIndex) + 1) % items.length &&
@ items[\old(putIndex)] == o;
@ signals (Exception ie) ie instanceof InterruptedException;
@*/
public /*@ atomic @*/ 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 normal_behavior
@ locks this.lock;
@ assignable items[takeIndex], takeIndex, count;
@ ensures \old(count == 0) ==> \result == null &&
@ \old(count != 0) ==> \result == \old(items[takeIndex]) &&
@ takeIndex == \old(takeIndex) + 1 &&
@ count == \old(count) -1;
@*/
public /*@ atomic @*/ E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == 0)
return null;
E x = extract();
return x;
} finally {
lock.unlock();
}
}
/*@ public behavior
@ locks this.lock;
@ assignable items[takeIndex], takeIndex, count;
@ ensures \old(count != 0) ==> \result == \old(items[takeIndex]) &&
@ takeIndex == \old(takeIndex) + 1 &&
@ count == \old(count) -1;
@ signals (Exception ie) ie instanceof InterruptedException;
@*/
public /*@ atomic @*/ 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 normal_behavior
@ locks this.lock;
@ assignable \nothing;
@ ensures (!\old(\exists int i; i >= 0 && i < items.length; o.equals(items[i]))) ==>
@ !(\result);
@also
@ public normal_behavior
@ locks this.lock;
@ assignable items[takeIndex], count, takeIndex;
@ ensures \old(\exists int i; i >= 0 && i < items.length; o.equals(items[i])) ==>
@ \old(items[takeIndex]) == null &&
@ takeIndex == \old(takeIndex) + 1 && \result;
@*/
public /*@ atomic @*/ 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 normal_behavior
@ locks this.lock;
@ assignable \nothing;
@ ensures \old(count) == 0 ==> \result == null &&
@ \old(count) != 0 ==> \result == items[takeIndex];
@*/
public /*@ atomic @*/ E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : items[takeIndex];
} finally {
lock.unlock();
}
}
/*@ public normal_behavior
@ locks this.lock;
@ when count != 0;
@ assignable items[takeIndex], takeIndex, count;
@ ensures \result == \old(items[takeIndex]) && takeIndex == \old(takeIndex) + 1 &&
@ count == \old(count) -1;
@*/
public /*@ atomic @*/ 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;
}
//@ commit:
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 null.
*/
/*@ public normal_behavior
@ locks this.lock;
@ when count != items.length;
@ assignable items[putIndex], putIndex, count;
@ ensures count == (\old(count) + 1) && putIndex == (\old(putIndex) + 1) % items.length;
@*/
public /*@ atomic @*/ 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;
}
//@ commit:
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 normal_behavior
@ locks this.lock;
@ assignable \nothing;
@ ensures \result == count;
@*/
public /*@ weakly_pure atomic @*/ 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 size of this queue.
* Note that you cannot always tell if
* an attempt to add an element will succeed by
* inspecting remainingCapacity because it may be the
* case that a waiting consumer is ready to take an
* element out of an otherwise full queue.
*/
/*@ public normal_behavior
@ locks this.lock;
@ assignable \nothing;
@ ensures \result items.length - count;
@*/
public /*@ atomic @*/ int remainingCapacity() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return items.length - count;
} finally {
lock.unlock();
}
}
/*@ public normal_behavior
@ requires o != null;
@ locks this.lock;
@ assignable \nothing;
@ ensures \old(\exists int i; i >= 0 && i < items.length &&
@ i < putIndex && i >= takeIndex;
@ o.equals(items[i])) ==> \result
@ ||
@ !(\old(\exists int i; i >= 0 && i < items.length &&
@ i < putIndex && i >= takeIndex;
@ o.equals(items[i]))) ==> !(\result);
@also
@ requires o == null;
@ locks this.lock;
@ assignable \nothing;
@ ensures !(\result);
@*/
public /*@ weakly_pure atomic @*/ 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 normal_behavior
@ locks this.lock;
@ assignable \nothing;
@ ensures \result.length == count &&
@ (\forall int i; i >= 0 && i < \result.length;
@ \exists int j; j >= 0 && j < items.length;
@ \result[i].equals(items[j]));
@*/
public /*@ atomic @*/ 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 normal_behavior
@ requires a != null && \elemtype(\typeof(items)) <: \type(T);
@ locks this.lock;
@ ensures \result.length >= count &&
@ (\forall int i; i >= 0 && i < \result.length;
@ \exists int j; j >= 0 && j < items.length;
@ \result[i].equals(items[j]));
@also
@ public exceptional_behavior
@ requires a == null || !(\elemtype(\typeof(items)) <: \type(T));
@ locks this.lock;
@ signals (NullPointerException) a == null;
@ signals (ClassCastException) !(\elemtype(\typeof(items)) <: \type(T));
@*/
public /*@ atomic @*/ 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 normal_behavior
@ locks this.lock;
@ ensures (* The result is a string that represents the queue contents *);
@*/
public /*@ atomic @*/ String toString() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return super.toString();
} finally {
lock.unlock();
}
}
/*@ public normal_behavior
@ locks this.lock;
@ assignable items[*], takeIndex, putIndex, count;
@ ensures count == 0 && takeIndex == 0 && putIndex == 0 &&
@ (\forall int i; i >= 0 && i < items.length; items[i] == null);
@*/
public /*@ atomic @*/ 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 normal_behavior
@ requires c != null && c != this && \thread_safe(c);
@ locks this.lock;
@ ensures count == 0 && putIndex == 0 && takeIndex == 0 &&
@ (forall int i; i >= \old(takeIndex) && i < \old(putIndex);
@ c.contains(\old(items[i])));
@also
@ public exceptional_behavior
@ requires c == null && \thread_safe(c);
@ signals (Exception npe) npe instanceof NullPointerException;
@also
@ public exceptional_behavior
@ requires c == this;
@ signals (Exception iae) iae instanceof IllegalArgumentException;
@*/
public /*@ atomic @*/ int drainTo(Collection 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 normal_behavior
@ requires c != null && c != this && \thread_safe(c);
@ locks this.lock;
@ ensures (maxElements >= \old(count)) ==> (count == 0 && takeIndex == putIndex &&
@ (forall int i; i >= \old(takeIndex) && i < \old(putIndex);
@ c.contains(\old(items[i]))));
@ ensures (maxElements < \old(count)) ==>
@ (count == \old(count) - maxElements) &&
@ (takeIndex == \old(takeIndex) + maxElements % items.length) &&
@ (forall int i; i >= \old(takeIndex) && i < \old(takeIndex) + maxElements;
@ c.contains(\old(items[i])));
@also
@ public exceptional_behavior
@ requires c == null && \thread_safe(c);
@ signals (Exception npe) npe instanceof NullPointerException;
@also
@ public exceptional_behavior
@ requires c == this && \thread_safe(c);
@ signals (Exception iae) iae instanceof IllegalArgumentException;
@*/
public /*@ atomic @*/ int drainTo(Collection 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 Iterator 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 /*@ atomic @*/ Iterator iterator() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return new Itr();
} finally {
lock.unlock();
}
}
/**
* Iterator for ArrayBlockingQueue
*/
private class Itr implements Iterator {
/**
* 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();
}
}
}
}