Java example source code file (ArrayBlockingQueue.java)
The ArrayBlockingQueue.java Java example source code/*
* 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.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.AbstractQueue;
import java.util.Collection;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.lang.ref.WeakReference;
import java.util.Spliterators;
import java.util.Spliterator;
/**
* A bounded {@linkplain BlockingQueue blocking queue} backed by an
* array. This queue orders elements FIFO (first-in-first-out). The
* <em>head of the queue is that element that has been on the
* queue the longest time. The <em>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.
*
* <p>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
* changed. Attempts to {@code put} an element into a full queue
* will result in the operation blocking; attempts to {@code take} 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 {@code true} 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 methods of the {@link Collection} and {@link
* Iterator} interfaces.
*
* <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 ArrayBlockingQueue<E> extends AbstractQueue
*
* Note that {@code toArray(new Object[0])} is identical in function to
* {@code toArray()}.
*
* @param a the array into which the elements of the queue are to
* be stored, if it is big enough; otherwise, a new array of the
* same runtime type is allocated for this purpose
* @return an array containing all of the elements in this queue
* @throws ArrayStoreException if the runtime type of the specified array
* is not a supertype of the runtime type of every element in
* this queue
* @throws NullPointerException if the specified array is null
*/
@SuppressWarnings("unchecked")
public <T> T[] toArray(T[] a) {
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
final int count = this.count;
final int len = a.length;
if (len < count)
a = (T[])java.lang.reflect.Array.newInstance(
a.getClass().getComponentType(), count);
int n = items.length - takeIndex;
if (count <= n)
System.arraycopy(items, takeIndex, a, 0, count);
else {
System.arraycopy(items, takeIndex, a, 0, n);
System.arraycopy(items, 0, a, n, count - n);
}
if (len > count)
a[count] = null;
} finally {
lock.unlock();
}
return a;
}
public String toString() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
int k = count;
if (k == 0)
return "[]";
final Object[] items = this.items;
StringBuilder sb = new StringBuilder();
sb.append('[');
for (int i = takeIndex; ; ) {
Object e = items[i];
sb.append(e == this ? "(this Collection)" : e);
if (--k == 0)
return sb.append(']').toString();
sb.append(',').append(' ');
if (++i == items.length)
i = 0;
}
} finally {
lock.unlock();
}
}
/**
* Atomically removes all of the elements from this queue.
* The queue will be empty after this call returns.
*/
public void clear() {
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
int k = count;
if (k > 0) {
final int putIndex = this.putIndex;
int i = takeIndex;
do {
items[i] = null;
if (++i == items.length)
i = 0;
} while (i != putIndex);
takeIndex = putIndex;
count = 0;
if (itrs != null)
itrs.queueIsEmpty();
for (; k > 0 && lock.hasWaiters(notFull); k--)
notFull.signal();
}
} finally {
lock.unlock();
}
}
/**
* @throws UnsupportedOperationException {@inheritDoc}
* @throws ClassCastException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
* @throws IllegalArgumentException {@inheritDoc}
*/
public int drainTo(Collection<? super E> c) {
return drainTo(c, Integer.MAX_VALUE);
}
/**
* @throws UnsupportedOperationException {@inheritDoc}
* @throws ClassCastException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
* @throws IllegalArgumentException {@inheritDoc}
*/
public int drainTo(Collection<? super E> c, int maxElements) {
checkNotNull(c);
if (c == this)
throw new IllegalArgumentException();
if (maxElements <= 0)
return 0;
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
int n = Math.min(maxElements, count);
int take = takeIndex;
int i = 0;
try {
while (i < n) {
@SuppressWarnings("unchecked")
E x = (E) items[take];
c.add(x);
items[take] = null;
if (++take == items.length)
take = 0;
i++;
}
return n;
} finally {
// Restore invariants even if c.add() threw
if (i > 0) {
count -= i;
takeIndex = take;
if (itrs != null) {
if (count == 0)
itrs.queueIsEmpty();
else if (i > take)
itrs.takeIndexWrapped();
}
for (; i > 0 && lock.hasWaiters(notFull); i--)
notFull.signal();
}
}
} finally {
lock.unlock();
}
}
/**
* Returns an iterator over the elements in this queue in proper sequence.
* The elements will be returned in order from first (head) to last (tail).
*
* <p>The returned iterator is
* <a href="package-summary.html#Weakly">weakly consistent.
*
* @return an iterator over the elements in this queue in proper sequence
*/
public Iterator<E> iterator() {
return new Itr();
}
/**
* Shared data between iterators and their queue, allowing queue
* modifications to update iterators when elements are removed.
*
* This adds a lot of complexity for the sake of correctly
* handling some uncommon operations, but the combination of
* circular-arrays and supporting interior removes (i.e., those
* not at head) would cause iterators to sometimes lose their
* places and/or (re)report elements they shouldn't. To avoid
* this, when a queue has one or more iterators, it keeps iterator
* state consistent by:
*
* (1) keeping track of the number of "cycles", that is, the
* number of times takeIndex has wrapped around to 0.
* (2) notifying all iterators via the callback removedAt whenever
* an interior element is removed (and thus other elements may
* be shifted).
*
* These suffice to eliminate iterator inconsistencies, but
* unfortunately add the secondary responsibility of maintaining
* the list of iterators. We track all active iterators in a
* simple linked list (accessed only when the queue's lock is
* held) of weak references to Itr. The list is cleaned up using
* 3 different mechanisms:
*
* (1) Whenever a new iterator is created, do some O(1) checking for
* stale list elements.
*
* (2) Whenever takeIndex wraps around to 0, check for iterators
* that have been unused for more than one wrap-around cycle.
*
* (3) Whenever the queue becomes empty, all iterators are notified
* and this entire data structure is discarded.
*
* So in addition to the removedAt callback that is necessary for
* correctness, iterators have the shutdown and takeIndexWrapped
* callbacks that help remove stale iterators from the list.
*
* Whenever a list element is examined, it is expunged if either
* the GC has determined that the iterator is discarded, or if the
* iterator reports that it is "detached" (does not need any
* further state updates). Overhead is maximal when takeIndex
* never advances, iterators are discarded before they are
* exhausted, and all removals are interior removes, in which case
* all stale iterators are discovered by the GC. But even in this
* case we don't increase the amortized complexity.
*
* Care must be taken to keep list sweeping methods from
* reentrantly invoking another such method, causing subtle
* corruption bugs.
*/
class Itrs {
/**
* Node in a linked list of weak iterator references.
*/
private class Node extends WeakReference<Itr> {
Node next;
Node(Itr iterator, Node next) {
super(iterator);
this.next = next;
}
}
/** Incremented whenever takeIndex wraps around to 0 */
int cycles = 0;
/** Linked list of weak iterator references */
private Node head;
/** Used to expunge stale iterators */
private Node sweeper = null;
private static final int SHORT_SWEEP_PROBES = 4;
private static final int LONG_SWEEP_PROBES = 16;
Itrs(Itr initial) {
register(initial);
}
/**
* Sweeps itrs, looking for and expunging stale iterators.
* If at least one was found, tries harder to find more.
* Called only from iterating thread.
*
* @param tryHarder whether to start in try-harder mode, because
* there is known to be at least one iterator to collect
*/
void doSomeSweeping(boolean tryHarder) {
// assert lock.getHoldCount() == 1;
// assert head != null;
int probes = tryHarder ? LONG_SWEEP_PROBES : SHORT_SWEEP_PROBES;
Node o, p;
final Node sweeper = this.sweeper;
boolean passedGo; // to limit search to one full sweep
if (sweeper == null) {
o = null;
p = head;
passedGo = true;
} else {
o = sweeper;
p = o.next;
passedGo = false;
}
for (; probes > 0; probes--) {
if (p == null) {
if (passedGo)
break;
o = null;
p = head;
passedGo = true;
}
final Itr it = p.get();
final Node next = p.next;
if (it == null || it.isDetached()) {
// found a discarded/exhausted iterator
probes = LONG_SWEEP_PROBES; // "try harder"
// unlink p
p.clear();
p.next = null;
if (o == null) {
head = next;
if (next == null) {
// We've run out of iterators to track; retire
itrs = null;
return;
}
}
else
o.next = next;
} else {
o = p;
}
p = next;
}
this.sweeper = (p == null) ? null : o;
}
/**
* Adds a new iterator to the linked list of tracked iterators.
*/
void register(Itr itr) {
// assert lock.getHoldCount() == 1;
head = new Node(itr, head);
}
/**
* Called whenever takeIndex wraps around to 0.
*
* Notifies all iterators, and expunges any that are now stale.
*/
void takeIndexWrapped() {
// assert lock.getHoldCount() == 1;
cycles++;
for (Node o = null, p = head; p != null;) {
final Itr it = p.get();
final Node next = p.next;
if (it == null || it.takeIndexWrapped()) {
// unlink p
// assert it == null || it.isDetached();
p.clear();
p.next = null;
if (o == null)
head = next;
else
o.next = next;
} else {
o = p;
}
p = next;
}
if (head == null) // no more iterators to track
itrs = null;
}
/**
* Called whenever an interior remove (not at takeIndex) occurred.
*
* Notifies all iterators, and expunges any that are now stale.
*/
void removedAt(int removedIndex) {
for (Node o = null, p = head; p != null;) {
final Itr it = p.get();
final Node next = p.next;
if (it == null || it.removedAt(removedIndex)) {
// unlink p
// assert it == null || it.isDetached();
p.clear();
p.next = null;
if (o == null)
head = next;
else
o.next = next;
} else {
o = p;
}
p = next;
}
if (head == null) // no more iterators to track
itrs = null;
}
/**
* Called whenever the queue becomes empty.
*
* Notifies all active iterators that the queue is empty,
* clears all weak refs, and unlinks the itrs datastructure.
*/
void queueIsEmpty() {
// assert lock.getHoldCount() == 1;
for (Node p = head; p != null; p = p.next) {
Itr it = p.get();
if (it != null) {
p.clear();
it.shutdown();
}
}
head = null;
itrs = null;
}
/**
* Called whenever an element has been dequeued (at takeIndex).
*/
void elementDequeued() {
// assert lock.getHoldCount() == 1;
if (count == 0)
queueIsEmpty();
else if (takeIndex == 0)
takeIndexWrapped();
}
}
/**
* Iterator for ArrayBlockingQueue.
*
* To maintain weak consistency with respect to puts and takes, we
* read ahead one slot, so as to not report hasNext true but then
* not have an element to return.
*
* We switch into "detached" mode (allowing prompt unlinking from
* itrs without help from the GC) when all indices are negative, or
* when hasNext returns false for the first time. This allows the
* iterator to track concurrent updates completely accurately,
* except for the corner case of the user calling Iterator.remove()
* after hasNext() returned false. Even in this case, we ensure
* that we don't remove the wrong element by keeping track of the
* expected element to remove, in lastItem. Yes, we may fail to
* remove lastItem from the queue if it moved due to an interleaved
* interior remove while in detached mode.
*/
private class Itr implements Iterator<E> {
/** Index to look for new nextItem; NONE at end */
private int cursor;
/** Element to be returned by next call to next(); null if none */
private E nextItem;
/** Index of nextItem; NONE if none, REMOVED if removed elsewhere */
private int nextIndex;
/** Last element returned; null if none or not detached. */
private E lastItem;
/** Index of lastItem, NONE if none, REMOVED if removed elsewhere */
private int lastRet;
/** Previous value of takeIndex, or DETACHED when detached */
private int prevTakeIndex;
/** Previous value of iters.cycles */
private int prevCycles;
/** Special index value indicating "not available" or "undefined" */
private static final int NONE = -1;
/**
* Special index value indicating "removed elsewhere", that is,
* removed by some operation other than a call to this.remove().
*/
private static final int REMOVED = -2;
/** Special value for prevTakeIndex indicating "detached mode" */
private static final int DETACHED = -3;
Itr() {
// assert lock.getHoldCount() == 0;
lastRet = NONE;
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
if (count == 0) {
// assert itrs == null;
cursor = NONE;
nextIndex = NONE;
prevTakeIndex = DETACHED;
} else {
final int takeIndex = ArrayBlockingQueue.this.takeIndex;
prevTakeIndex = takeIndex;
nextItem = itemAt(nextIndex = takeIndex);
cursor = incCursor(takeIndex);
if (itrs == null) {
itrs = new Itrs(this);
} else {
itrs.register(this); // in this order
itrs.doSomeSweeping(false);
}
prevCycles = itrs.cycles;
// assert takeIndex >= 0;
// assert prevTakeIndex == takeIndex;
// assert nextIndex >= 0;
// assert nextItem != null;
}
} finally {
lock.unlock();
}
}
boolean isDetached() {
// assert lock.getHoldCount() == 1;
return prevTakeIndex < 0;
}
private int incCursor(int index) {
// assert lock.getHoldCount() == 1;
if (++index == items.length)
index = 0;
if (index == putIndex)
index = NONE;
return index;
}
/**
* Returns true if index is invalidated by the given number of
* dequeues, starting from prevTakeIndex.
*/
private boolean invalidated(int index, int prevTakeIndex,
long dequeues, int length) {
if (index < 0)
return false;
int distance = index - prevTakeIndex;
if (distance < 0)
distance += length;
return dequeues > distance;
}
/**
* Adjusts indices to incorporate all dequeues since the last
* operation on this iterator. Call only from iterating thread.
*/
private void incorporateDequeues() {
// assert lock.getHoldCount() == 1;
// assert itrs != null;
// assert !isDetached();
// assert count > 0;
final int cycles = itrs.cycles;
final int takeIndex = ArrayBlockingQueue.this.takeIndex;
final int prevCycles = this.prevCycles;
final int prevTakeIndex = this.prevTakeIndex;
if (cycles != prevCycles || takeIndex != prevTakeIndex) {
final int len = items.length;
// how far takeIndex has advanced since the previous
// operation of this iterator
long dequeues = (cycles - prevCycles) * len
+ (takeIndex - prevTakeIndex);
// Check indices for invalidation
if (invalidated(lastRet, prevTakeIndex, dequeues, len))
lastRet = REMOVED;
if (invalidated(nextIndex, prevTakeIndex, dequeues, len))
nextIndex = REMOVED;
if (invalidated(cursor, prevTakeIndex, dequeues, len))
cursor = takeIndex;
if (cursor < 0 && nextIndex < 0 && lastRet < 0)
detach();
else {
this.prevCycles = cycles;
this.prevTakeIndex = takeIndex;
}
}
}
/**
* Called when itrs should stop tracking this iterator, either
* because there are no more indices to update (cursor < 0 &&
* nextIndex < 0 && lastRet < 0) or as a special exception, when
* lastRet >= 0, because hasNext() is about to return false for the
* first time. Call only from iterating thread.
*/
private void detach() {
// Switch to detached mode
// assert lock.getHoldCount() == 1;
// assert cursor == NONE;
// assert nextIndex < 0;
// assert lastRet < 0 || nextItem == null;
// assert lastRet < 0 ^ lastItem != null;
if (prevTakeIndex >= 0) {
// assert itrs != null;
prevTakeIndex = DETACHED;
// try to unlink from itrs (but not too hard)
itrs.doSomeSweeping(true);
}
}
/**
* For performance reasons, we would like not to acquire a lock in
* hasNext in the common case. To allow for this, we only access
* fields (i.e. nextItem) that are not modified by update operations
* triggered by queue modifications.
*/
public boolean hasNext() {
// assert lock.getHoldCount() == 0;
if (nextItem != null)
return true;
noNext();
return false;
}
private void noNext() {
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
// assert cursor == NONE;
// assert nextIndex == NONE;
if (!isDetached()) {
// assert lastRet >= 0;
incorporateDequeues(); // might update lastRet
if (lastRet >= 0) {
lastItem = itemAt(lastRet);
// assert lastItem != null;
detach();
}
}
// assert isDetached();
// assert lastRet < 0 ^ lastItem != null;
} finally {
lock.unlock();
}
}
public E next() {
// assert lock.getHoldCount() == 0;
final E x = nextItem;
if (x == null)
throw new NoSuchElementException();
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
if (!isDetached())
incorporateDequeues();
// assert nextIndex != NONE;
// assert lastItem == null;
lastRet = nextIndex;
final int cursor = this.cursor;
if (cursor >= 0) {
nextItem = itemAt(nextIndex = cursor);
// assert nextItem != null;
this.cursor = incCursor(cursor);
} else {
nextIndex = NONE;
nextItem = null;
}
} finally {
lock.unlock();
}
return x;
}
public void remove() {
// assert lock.getHoldCount() == 0;
final ReentrantLock lock = ArrayBlockingQueue.this.lock;
lock.lock();
try {
if (!isDetached())
incorporateDequeues(); // might update lastRet or detach
final int lastRet = this.lastRet;
this.lastRet = NONE;
if (lastRet >= 0) {
if (!isDetached())
removeAt(lastRet);
else {
final E lastItem = this.lastItem;
// assert lastItem != null;
this.lastItem = null;
if (itemAt(lastRet) == lastItem)
removeAt(lastRet);
}
} else if (lastRet == NONE)
throw new IllegalStateException();
// else lastRet == REMOVED and the last returned element was
// previously asynchronously removed via an operation other
// than this.remove(), so nothing to do.
if (cursor < 0 && nextIndex < 0)
detach();
} finally {
lock.unlock();
// assert lastRet == NONE;
// assert lastItem == null;
}
}
/**
* Called to notify the iterator that the queue is empty, or that it
* has fallen hopelessly behind, so that it should abandon any
* further iteration, except possibly to return one more element
* from next(), as promised by returning true from hasNext().
*/
void shutdown() {
// assert lock.getHoldCount() == 1;
cursor = NONE;
if (nextIndex >= 0)
nextIndex = REMOVED;
if (lastRet >= 0) {
lastRet = REMOVED;
lastItem = null;
}
prevTakeIndex = DETACHED;
// Don't set nextItem to null because we must continue to be
// able to return it on next().
//
// Caller will unlink from itrs when convenient.
}
private int distance(int index, int prevTakeIndex, int length) {
int distance = index - prevTakeIndex;
if (distance < 0)
distance += length;
return distance;
}
/**
* Called whenever an interior remove (not at takeIndex) occurred.
*
* @return true if this iterator should be unlinked from itrs
*/
boolean removedAt(int removedIndex) {
// assert lock.getHoldCount() == 1;
if (isDetached())
return true;
final int cycles = itrs.cycles;
final int takeIndex = ArrayBlockingQueue.this.takeIndex;
final int prevCycles = this.prevCycles;
final int prevTakeIndex = this.prevTakeIndex;
final int len = items.length;
int cycleDiff = cycles - prevCycles;
if (removedIndex < takeIndex)
cycleDiff++;
final int removedDistance =
(cycleDiff * len) + (removedIndex - prevTakeIndex);
// assert removedDistance >= 0;
int cursor = this.cursor;
if (cursor >= 0) {
int x = distance(cursor, prevTakeIndex, len);
if (x == removedDistance) {
if (cursor == putIndex)
this.cursor = cursor = NONE;
}
else if (x > removedDistance) {
// assert cursor != prevTakeIndex;
this.cursor = cursor = dec(cursor);
}
}
int lastRet = this.lastRet;
if (lastRet >= 0) {
int x = distance(lastRet, prevTakeIndex, len);
if (x == removedDistance)
this.lastRet = lastRet = REMOVED;
else if (x > removedDistance)
this.lastRet = lastRet = dec(lastRet);
}
int nextIndex = this.nextIndex;
if (nextIndex >= 0) {
int x = distance(nextIndex, prevTakeIndex, len);
if (x == removedDistance)
this.nextIndex = nextIndex = REMOVED;
else if (x > removedDistance)
this.nextIndex = nextIndex = dec(nextIndex);
}
else if (cursor < 0 && nextIndex < 0 && lastRet < 0) {
this.prevTakeIndex = DETACHED;
return true;
}
return false;
}
/**
* Called whenever takeIndex wraps around to zero.
*
* @return true if this iterator should be unlinked from itrs
*/
boolean takeIndexWrapped() {
// assert lock.getHoldCount() == 1;
if (isDetached())
return true;
if (itrs.cycles - prevCycles > 1) {
// All the elements that existed at the time of the last
// operation are gone, so abandon further iteration.
shutdown();
return true;
}
return false;
}
// /** Uncomment for debugging. */
// public String toString() {
// return ("cursor=" + cursor + " " +
// "nextIndex=" + nextIndex + " " +
// "lastRet=" + lastRet + " " +
// "nextItem=" + nextItem + " " +
// "lastItem=" + lastItem + " " +
// "prevCycles=" + prevCycles + " " +
// "prevTakeIndex=" + prevTakeIndex + " " +
// "size()=" + size() + " " +
// "remainingCapacity()=" + remainingCapacity());
// }
}
/**
* Returns a {@link Spliterator} over the elements in this queue.
*
* <p>The returned spliterator is
* <a href="package-summary.html#Weakly">weakly consistent.
*
* <p>The {@code Spliterator} reports {@link Spliterator#CONCURRENT},
* {@link Spliterator#ORDERED}, and {@link Spliterator#NONNULL}.
*
* @implNote
* The {@code Spliterator} implements {@code trySplit} to permit limited
* parallelism.
*
* @return a {@code Spliterator} over the elements in this queue
* @since 1.8
*/
public Spliterator<E> spliterator() {
return Spliterators.spliterator
(this, Spliterator.ORDERED | Spliterator.NONNULL |
Spliterator.CONCURRENT);
}
}
Other Java examples (source code examples)Here is a short list of links related to this Java ArrayBlockingQueue.java source code file: |
The search page
Other Java source code examples at this package level
Click here to learn more about this project
