1.1 --- /dev/null Thu Jan 01 00:00:00 1970 +0000
1.2 +++ b/rt/emul/compact/src/main/java/java/util/concurrent/ArrayBlockingQueue.java Sat Mar 19 10:46:31 2016 +0100
1.3 @@ -0,0 +1,805 @@
1.4 +/*
1.5 + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
1.6 + *
1.7 + * This code is free software; you can redistribute it and/or modify it
1.8 + * under the terms of the GNU General Public License version 2 only, as
1.9 + * published by the Free Software Foundation. Oracle designates this
1.10 + * particular file as subject to the "Classpath" exception as provided
1.11 + * by Oracle in the LICENSE file that accompanied this code.
1.12 + *
1.13 + * This code is distributed in the hope that it will be useful, but WITHOUT
1.14 + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
1.15 + * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
1.16 + * version 2 for more details (a copy is included in the LICENSE file that
1.17 + * accompanied this code).
1.18 + *
1.19 + * You should have received a copy of the GNU General Public License version
1.20 + * 2 along with this work; if not, write to the Free Software Foundation,
1.21 + * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
1.22 + *
1.23 + * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
1.24 + * or visit www.oracle.com if you need additional information or have any
1.25 + * questions.
1.26 + */
1.27 +
1.28 +/*
1.29 + * This file is available under and governed by the GNU General Public
1.30 + * License version 2 only, as published by the Free Software Foundation.
1.31 + * However, the following notice accompanied the original version of this
1.32 + * file:
1.33 + *
1.34 + * Written by Doug Lea with assistance from members of JCP JSR-166
1.35 + * Expert Group and released to the public domain, as explained at
1.36 + * http://creativecommons.org/publicdomain/zero/1.0/
1.37 + */
1.38 +
1.39 +package java.util.concurrent;
1.40 +import java.util.concurrent.locks.*;
1.41 +import java.util.*;
1.42 +
1.43 +/**
1.44 + * A bounded {@linkplain BlockingQueue blocking queue} backed by an
1.45 + * array. This queue orders elements FIFO (first-in-first-out). The
1.46 + * <em>head</em> of the queue is that element that has been on the
1.47 + * queue the longest time. The <em>tail</em> of the queue is that
1.48 + * element that has been on the queue the shortest time. New elements
1.49 + * are inserted at the tail of the queue, and the queue retrieval
1.50 + * operations obtain elements at the head of the queue.
1.51 + *
1.52 + * <p>This is a classic "bounded buffer", in which a
1.53 + * fixed-sized array holds elements inserted by producers and
1.54 + * extracted by consumers. Once created, the capacity cannot be
1.55 + * changed. Attempts to {@code put} an element into a full queue
1.56 + * will result in the operation blocking; attempts to {@code take} an
1.57 + * element from an empty queue will similarly block.
1.58 + *
1.59 + * <p>This class supports an optional fairness policy for ordering
1.60 + * waiting producer and consumer threads. By default, this ordering
1.61 + * is not guaranteed. However, a queue constructed with fairness set
1.62 + * to {@code true} grants threads access in FIFO order. Fairness
1.63 + * generally decreases throughput but reduces variability and avoids
1.64 + * starvation.
1.65 + *
1.66 + * <p>This class and its iterator implement all of the
1.67 + * <em>optional</em> methods of the {@link Collection} and {@link
1.68 + * Iterator} interfaces.
1.69 + *
1.70 + * <p>This class is a member of the
1.71 + * <a href="{@docRoot}/../technotes/guides/collections/index.html">
1.72 + * Java Collections Framework</a>.
1.73 + *
1.74 + * @since 1.5
1.75 + * @author Doug Lea
1.76 + * @param <E> the type of elements held in this collection
1.77 + */
1.78 +public class ArrayBlockingQueue<E> extends AbstractQueue<E>
1.79 + implements BlockingQueue<E>, java.io.Serializable {
1.80 +
1.81 + /**
1.82 + * Serialization ID. This class relies on default serialization
1.83 + * even for the items array, which is default-serialized, even if
1.84 + * it is empty. Otherwise it could not be declared final, which is
1.85 + * necessary here.
1.86 + */
1.87 + private static final long serialVersionUID = -817911632652898426L;
1.88 +
1.89 + /** The queued items */
1.90 + final Object[] items;
1.91 +
1.92 + /** items index for next take, poll, peek or remove */
1.93 + int takeIndex;
1.94 +
1.95 + /** items index for next put, offer, or add */
1.96 + int putIndex;
1.97 +
1.98 + /** Number of elements in the queue */
1.99 + int count;
1.100 +
1.101 + /*
1.102 + * Concurrency control uses the classic two-condition algorithm
1.103 + * found in any textbook.
1.104 + */
1.105 +
1.106 + /** Main lock guarding all access */
1.107 + final ReentrantLock lock;
1.108 + /** Condition for waiting takes */
1.109 + private final Condition notEmpty;
1.110 + /** Condition for waiting puts */
1.111 + private final Condition notFull;
1.112 +
1.113 + // Internal helper methods
1.114 +
1.115 + /**
1.116 + * Circularly increment i.
1.117 + */
1.118 + final int inc(int i) {
1.119 + return (++i == items.length) ? 0 : i;
1.120 + }
1.121 +
1.122 + /**
1.123 + * Circularly decrement i.
1.124 + */
1.125 + final int dec(int i) {
1.126 + return ((i == 0) ? items.length : i) - 1;
1.127 + }
1.128 +
1.129 + @SuppressWarnings("unchecked")
1.130 + static <E> E cast(Object item) {
1.131 + return (E) item;
1.132 + }
1.133 +
1.134 + /**
1.135 + * Returns item at index i.
1.136 + */
1.137 + final E itemAt(int i) {
1.138 + return this.<E>cast(items[i]);
1.139 + }
1.140 +
1.141 + /**
1.142 + * Throws NullPointerException if argument is null.
1.143 + *
1.144 + * @param v the element
1.145 + */
1.146 + private static void checkNotNull(Object v) {
1.147 + if (v == null)
1.148 + throw new NullPointerException();
1.149 + }
1.150 +
1.151 + /**
1.152 + * Inserts element at current put position, advances, and signals.
1.153 + * Call only when holding lock.
1.154 + */
1.155 + private void insert(E x) {
1.156 + items[putIndex] = x;
1.157 + putIndex = inc(putIndex);
1.158 + ++count;
1.159 + notEmpty.signal();
1.160 + }
1.161 +
1.162 + /**
1.163 + * Extracts element at current take position, advances, and signals.
1.164 + * Call only when holding lock.
1.165 + */
1.166 + private E extract() {
1.167 + final Object[] items = this.items;
1.168 + E x = this.<E>cast(items[takeIndex]);
1.169 + items[takeIndex] = null;
1.170 + takeIndex = inc(takeIndex);
1.171 + --count;
1.172 + notFull.signal();
1.173 + return x;
1.174 + }
1.175 +
1.176 + /**
1.177 + * Deletes item at position i.
1.178 + * Utility for remove and iterator.remove.
1.179 + * Call only when holding lock.
1.180 + */
1.181 + void removeAt(int i) {
1.182 + final Object[] items = this.items;
1.183 + // if removing front item, just advance
1.184 + if (i == takeIndex) {
1.185 + items[takeIndex] = null;
1.186 + takeIndex = inc(takeIndex);
1.187 + } else {
1.188 + // slide over all others up through putIndex.
1.189 + for (;;) {
1.190 + int nexti = inc(i);
1.191 + if (nexti != putIndex) {
1.192 + items[i] = items[nexti];
1.193 + i = nexti;
1.194 + } else {
1.195 + items[i] = null;
1.196 + putIndex = i;
1.197 + break;
1.198 + }
1.199 + }
1.200 + }
1.201 + --count;
1.202 + notFull.signal();
1.203 + }
1.204 +
1.205 + /**
1.206 + * Creates an {@code ArrayBlockingQueue} with the given (fixed)
1.207 + * capacity and default access policy.
1.208 + *
1.209 + * @param capacity the capacity of this queue
1.210 + * @throws IllegalArgumentException if {@code capacity < 1}
1.211 + */
1.212 + public ArrayBlockingQueue(int capacity) {
1.213 + this(capacity, false);
1.214 + }
1.215 +
1.216 + /**
1.217 + * Creates an {@code ArrayBlockingQueue} with the given (fixed)
1.218 + * capacity and the specified access policy.
1.219 + *
1.220 + * @param capacity the capacity of this queue
1.221 + * @param fair if {@code true} then queue accesses for threads blocked
1.222 + * on insertion or removal, are processed in FIFO order;
1.223 + * if {@code false} the access order is unspecified.
1.224 + * @throws IllegalArgumentException if {@code capacity < 1}
1.225 + */
1.226 + public ArrayBlockingQueue(int capacity, boolean fair) {
1.227 + if (capacity <= 0)
1.228 + throw new IllegalArgumentException();
1.229 + this.items = new Object[capacity];
1.230 + lock = new ReentrantLock(fair);
1.231 + notEmpty = lock.newCondition();
1.232 + notFull = lock.newCondition();
1.233 + }
1.234 +
1.235 + /**
1.236 + * Creates an {@code ArrayBlockingQueue} with the given (fixed)
1.237 + * capacity, the specified access policy and initially containing the
1.238 + * elements of the given collection,
1.239 + * added in traversal order of the collection's iterator.
1.240 + *
1.241 + * @param capacity the capacity of this queue
1.242 + * @param fair if {@code true} then queue accesses for threads blocked
1.243 + * on insertion or removal, are processed in FIFO order;
1.244 + * if {@code false} the access order is unspecified.
1.245 + * @param c the collection of elements to initially contain
1.246 + * @throws IllegalArgumentException if {@code capacity} is less than
1.247 + * {@code c.size()}, or less than 1.
1.248 + * @throws NullPointerException if the specified collection or any
1.249 + * of its elements are null
1.250 + */
1.251 + public ArrayBlockingQueue(int capacity, boolean fair,
1.252 + Collection<? extends E> c) {
1.253 + this(capacity, fair);
1.254 +
1.255 + final ReentrantLock lock = this.lock;
1.256 + lock.lock(); // Lock only for visibility, not mutual exclusion
1.257 + try {
1.258 + int i = 0;
1.259 + try {
1.260 + for (E e : c) {
1.261 + checkNotNull(e);
1.262 + items[i++] = e;
1.263 + }
1.264 + } catch (ArrayIndexOutOfBoundsException ex) {
1.265 + throw new IllegalArgumentException();
1.266 + }
1.267 + count = i;
1.268 + putIndex = (i == capacity) ? 0 : i;
1.269 + } finally {
1.270 + lock.unlock();
1.271 + }
1.272 + }
1.273 +
1.274 + /**
1.275 + * Inserts the specified element at the tail of this queue if it is
1.276 + * possible to do so immediately without exceeding the queue's capacity,
1.277 + * returning {@code true} upon success and throwing an
1.278 + * {@code IllegalStateException} if this queue is full.
1.279 + *
1.280 + * @param e the element to add
1.281 + * @return {@code true} (as specified by {@link Collection#add})
1.282 + * @throws IllegalStateException if this queue is full
1.283 + * @throws NullPointerException if the specified element is null
1.284 + */
1.285 + public boolean add(E e) {
1.286 + return super.add(e);
1.287 + }
1.288 +
1.289 + /**
1.290 + * Inserts the specified element at the tail of this queue if it is
1.291 + * possible to do so immediately without exceeding the queue's capacity,
1.292 + * returning {@code true} upon success and {@code false} if this queue
1.293 + * is full. This method is generally preferable to method {@link #add},
1.294 + * which can fail to insert an element only by throwing an exception.
1.295 + *
1.296 + * @throws NullPointerException if the specified element is null
1.297 + */
1.298 + public boolean offer(E e) {
1.299 + checkNotNull(e);
1.300 + final ReentrantLock lock = this.lock;
1.301 + lock.lock();
1.302 + try {
1.303 + if (count == items.length)
1.304 + return false;
1.305 + else {
1.306 + insert(e);
1.307 + return true;
1.308 + }
1.309 + } finally {
1.310 + lock.unlock();
1.311 + }
1.312 + }
1.313 +
1.314 + /**
1.315 + * Inserts the specified element at the tail of this queue, waiting
1.316 + * for space to become available if the queue is full.
1.317 + *
1.318 + * @throws InterruptedException {@inheritDoc}
1.319 + * @throws NullPointerException {@inheritDoc}
1.320 + */
1.321 + public void put(E e) throws InterruptedException {
1.322 + checkNotNull(e);
1.323 + final ReentrantLock lock = this.lock;
1.324 + lock.lockInterruptibly();
1.325 + try {
1.326 + while (count == items.length)
1.327 + notFull.await();
1.328 + insert(e);
1.329 + } finally {
1.330 + lock.unlock();
1.331 + }
1.332 + }
1.333 +
1.334 + /**
1.335 + * Inserts the specified element at the tail of this queue, waiting
1.336 + * up to the specified wait time for space to become available if
1.337 + * the queue is full.
1.338 + *
1.339 + * @throws InterruptedException {@inheritDoc}
1.340 + * @throws NullPointerException {@inheritDoc}
1.341 + */
1.342 + public boolean offer(E e, long timeout, TimeUnit unit)
1.343 + throws InterruptedException {
1.344 +
1.345 + checkNotNull(e);
1.346 + long nanos = unit.toNanos(timeout);
1.347 + final ReentrantLock lock = this.lock;
1.348 + lock.lockInterruptibly();
1.349 + try {
1.350 + while (count == items.length) {
1.351 + if (nanos <= 0)
1.352 + return false;
1.353 + nanos = notFull.awaitNanos(nanos);
1.354 + }
1.355 + insert(e);
1.356 + return true;
1.357 + } finally {
1.358 + lock.unlock();
1.359 + }
1.360 + }
1.361 +
1.362 + public E poll() {
1.363 + final ReentrantLock lock = this.lock;
1.364 + lock.lock();
1.365 + try {
1.366 + return (count == 0) ? null : extract();
1.367 + } finally {
1.368 + lock.unlock();
1.369 + }
1.370 + }
1.371 +
1.372 + public E take() throws InterruptedException {
1.373 + final ReentrantLock lock = this.lock;
1.374 + lock.lockInterruptibly();
1.375 + try {
1.376 + while (count == 0)
1.377 + notEmpty.await();
1.378 + return extract();
1.379 + } finally {
1.380 + lock.unlock();
1.381 + }
1.382 + }
1.383 +
1.384 + public E poll(long timeout, TimeUnit unit) throws InterruptedException {
1.385 + long nanos = unit.toNanos(timeout);
1.386 + final ReentrantLock lock = this.lock;
1.387 + lock.lockInterruptibly();
1.388 + try {
1.389 + while (count == 0) {
1.390 + if (nanos <= 0)
1.391 + return null;
1.392 + nanos = notEmpty.awaitNanos(nanos);
1.393 + }
1.394 + return extract();
1.395 + } finally {
1.396 + lock.unlock();
1.397 + }
1.398 + }
1.399 +
1.400 + public E peek() {
1.401 + final ReentrantLock lock = this.lock;
1.402 + lock.lock();
1.403 + try {
1.404 + return (count == 0) ? null : itemAt(takeIndex);
1.405 + } finally {
1.406 + lock.unlock();
1.407 + }
1.408 + }
1.409 +
1.410 + // this doc comment is overridden to remove the reference to collections
1.411 + // greater in size than Integer.MAX_VALUE
1.412 + /**
1.413 + * Returns the number of elements in this queue.
1.414 + *
1.415 + * @return the number of elements in this queue
1.416 + */
1.417 + public int size() {
1.418 + final ReentrantLock lock = this.lock;
1.419 + lock.lock();
1.420 + try {
1.421 + return count;
1.422 + } finally {
1.423 + lock.unlock();
1.424 + }
1.425 + }
1.426 +
1.427 + // this doc comment is a modified copy of the inherited doc comment,
1.428 + // without the reference to unlimited queues.
1.429 + /**
1.430 + * Returns the number of additional elements that this queue can ideally
1.431 + * (in the absence of memory or resource constraints) accept without
1.432 + * blocking. This is always equal to the initial capacity of this queue
1.433 + * less the current {@code size} of this queue.
1.434 + *
1.435 + * <p>Note that you <em>cannot</em> always tell if an attempt to insert
1.436 + * an element will succeed by inspecting {@code remainingCapacity}
1.437 + * because it may be the case that another thread is about to
1.438 + * insert or remove an element.
1.439 + */
1.440 + public int remainingCapacity() {
1.441 + final ReentrantLock lock = this.lock;
1.442 + lock.lock();
1.443 + try {
1.444 + return items.length - count;
1.445 + } finally {
1.446 + lock.unlock();
1.447 + }
1.448 + }
1.449 +
1.450 + /**
1.451 + * Removes a single instance of the specified element from this queue,
1.452 + * if it is present. More formally, removes an element {@code e} such
1.453 + * that {@code o.equals(e)}, if this queue contains one or more such
1.454 + * elements.
1.455 + * Returns {@code true} if this queue contained the specified element
1.456 + * (or equivalently, if this queue changed as a result of the call).
1.457 + *
1.458 + * <p>Removal of interior elements in circular array based queues
1.459 + * is an intrinsically slow and disruptive operation, so should
1.460 + * be undertaken only in exceptional circumstances, ideally
1.461 + * only when the queue is known not to be accessible by other
1.462 + * threads.
1.463 + *
1.464 + * @param o element to be removed from this queue, if present
1.465 + * @return {@code true} if this queue changed as a result of the call
1.466 + */
1.467 + public boolean remove(Object o) {
1.468 + if (o == null) return false;
1.469 + final Object[] items = this.items;
1.470 + final ReentrantLock lock = this.lock;
1.471 + lock.lock();
1.472 + try {
1.473 + for (int i = takeIndex, k = count; k > 0; i = inc(i), k--) {
1.474 + if (o.equals(items[i])) {
1.475 + removeAt(i);
1.476 + return true;
1.477 + }
1.478 + }
1.479 + return false;
1.480 + } finally {
1.481 + lock.unlock();
1.482 + }
1.483 + }
1.484 +
1.485 + /**
1.486 + * Returns {@code true} if this queue contains the specified element.
1.487 + * More formally, returns {@code true} if and only if this queue contains
1.488 + * at least one element {@code e} such that {@code o.equals(e)}.
1.489 + *
1.490 + * @param o object to be checked for containment in this queue
1.491 + * @return {@code true} if this queue contains the specified element
1.492 + */
1.493 + public boolean contains(Object o) {
1.494 + if (o == null) return false;
1.495 + final Object[] items = this.items;
1.496 + final ReentrantLock lock = this.lock;
1.497 + lock.lock();
1.498 + try {
1.499 + for (int i = takeIndex, k = count; k > 0; i = inc(i), k--)
1.500 + if (o.equals(items[i]))
1.501 + return true;
1.502 + return false;
1.503 + } finally {
1.504 + lock.unlock();
1.505 + }
1.506 + }
1.507 +
1.508 + /**
1.509 + * Returns an array containing all of the elements in this queue, in
1.510 + * proper sequence.
1.511 + *
1.512 + * <p>The returned array will be "safe" in that no references to it are
1.513 + * maintained by this queue. (In other words, this method must allocate
1.514 + * a new array). The caller is thus free to modify the returned array.
1.515 + *
1.516 + * <p>This method acts as bridge between array-based and collection-based
1.517 + * APIs.
1.518 + *
1.519 + * @return an array containing all of the elements in this queue
1.520 + */
1.521 + public Object[] toArray() {
1.522 + final Object[] items = this.items;
1.523 + final ReentrantLock lock = this.lock;
1.524 + lock.lock();
1.525 + try {
1.526 + final int count = this.count;
1.527 + Object[] a = new Object[count];
1.528 + for (int i = takeIndex, k = 0; k < count; i = inc(i), k++)
1.529 + a[k] = items[i];
1.530 + return a;
1.531 + } finally {
1.532 + lock.unlock();
1.533 + }
1.534 + }
1.535 +
1.536 + /**
1.537 + * Returns an array containing all of the elements in this queue, in
1.538 + * proper sequence; the runtime type of the returned array is that of
1.539 + * the specified array. If the queue fits in the specified array, it
1.540 + * is returned therein. Otherwise, a new array is allocated with the
1.541 + * runtime type of the specified array and the size of this queue.
1.542 + *
1.543 + * <p>If this queue fits in the specified array with room to spare
1.544 + * (i.e., the array has more elements than this queue), the element in
1.545 + * the array immediately following the end of the queue is set to
1.546 + * {@code null}.
1.547 + *
1.548 + * <p>Like the {@link #toArray()} method, this method acts as bridge between
1.549 + * array-based and collection-based APIs. Further, this method allows
1.550 + * precise control over the runtime type of the output array, and may,
1.551 + * under certain circumstances, be used to save allocation costs.
1.552 + *
1.553 + * <p>Suppose {@code x} is a queue known to contain only strings.
1.554 + * The following code can be used to dump the queue into a newly
1.555 + * allocated array of {@code String}:
1.556 + *
1.557 + * <pre>
1.558 + * String[] y = x.toArray(new String[0]);</pre>
1.559 + *
1.560 + * Note that {@code toArray(new Object[0])} is identical in function to
1.561 + * {@code toArray()}.
1.562 + *
1.563 + * @param a the array into which the elements of the queue are to
1.564 + * be stored, if it is big enough; otherwise, a new array of the
1.565 + * same runtime type is allocated for this purpose
1.566 + * @return an array containing all of the elements in this queue
1.567 + * @throws ArrayStoreException if the runtime type of the specified array
1.568 + * is not a supertype of the runtime type of every element in
1.569 + * this queue
1.570 + * @throws NullPointerException if the specified array is null
1.571 + */
1.572 + @SuppressWarnings("unchecked")
1.573 + public <T> T[] toArray(T[] a) {
1.574 + final Object[] items = this.items;
1.575 + final ReentrantLock lock = this.lock;
1.576 + lock.lock();
1.577 + try {
1.578 + final int count = this.count;
1.579 + final int len = a.length;
1.580 + if (len < count)
1.581 + a = (T[])java.lang.reflect.Array.newInstance(
1.582 + a.getClass().getComponentType(), count);
1.583 + for (int i = takeIndex, k = 0; k < count; i = inc(i), k++)
1.584 + a[k] = (T) items[i];
1.585 + if (len > count)
1.586 + a[count] = null;
1.587 + return a;
1.588 + } finally {
1.589 + lock.unlock();
1.590 + }
1.591 + }
1.592 +
1.593 + public String toString() {
1.594 + final ReentrantLock lock = this.lock;
1.595 + lock.lock();
1.596 + try {
1.597 + int k = count;
1.598 + if (k == 0)
1.599 + return "[]";
1.600 +
1.601 + StringBuilder sb = new StringBuilder();
1.602 + sb.append('[');
1.603 + for (int i = takeIndex; ; i = inc(i)) {
1.604 + Object e = items[i];
1.605 + sb.append(e == this ? "(this Collection)" : e);
1.606 + if (--k == 0)
1.607 + return sb.append(']').toString();
1.608 + sb.append(',').append(' ');
1.609 + }
1.610 + } finally {
1.611 + lock.unlock();
1.612 + }
1.613 + }
1.614 +
1.615 + /**
1.616 + * Atomically removes all of the elements from this queue.
1.617 + * The queue will be empty after this call returns.
1.618 + */
1.619 + public void clear() {
1.620 + final Object[] items = this.items;
1.621 + final ReentrantLock lock = this.lock;
1.622 + lock.lock();
1.623 + try {
1.624 + for (int i = takeIndex, k = count; k > 0; i = inc(i), k--)
1.625 + items[i] = null;
1.626 + count = 0;
1.627 + putIndex = 0;
1.628 + takeIndex = 0;
1.629 + notFull.signalAll();
1.630 + } finally {
1.631 + lock.unlock();
1.632 + }
1.633 + }
1.634 +
1.635 + /**
1.636 + * @throws UnsupportedOperationException {@inheritDoc}
1.637 + * @throws ClassCastException {@inheritDoc}
1.638 + * @throws NullPointerException {@inheritDoc}
1.639 + * @throws IllegalArgumentException {@inheritDoc}
1.640 + */
1.641 + public int drainTo(Collection<? super E> c) {
1.642 + checkNotNull(c);
1.643 + if (c == this)
1.644 + throw new IllegalArgumentException();
1.645 + final Object[] items = this.items;
1.646 + final ReentrantLock lock = this.lock;
1.647 + lock.lock();
1.648 + try {
1.649 + int i = takeIndex;
1.650 + int n = 0;
1.651 + int max = count;
1.652 + while (n < max) {
1.653 + c.add(this.<E>cast(items[i]));
1.654 + items[i] = null;
1.655 + i = inc(i);
1.656 + ++n;
1.657 + }
1.658 + if (n > 0) {
1.659 + count = 0;
1.660 + putIndex = 0;
1.661 + takeIndex = 0;
1.662 + notFull.signalAll();
1.663 + }
1.664 + return n;
1.665 + } finally {
1.666 + lock.unlock();
1.667 + }
1.668 + }
1.669 +
1.670 + /**
1.671 + * @throws UnsupportedOperationException {@inheritDoc}
1.672 + * @throws ClassCastException {@inheritDoc}
1.673 + * @throws NullPointerException {@inheritDoc}
1.674 + * @throws IllegalArgumentException {@inheritDoc}
1.675 + */
1.676 + public int drainTo(Collection<? super E> c, int maxElements) {
1.677 + checkNotNull(c);
1.678 + if (c == this)
1.679 + throw new IllegalArgumentException();
1.680 + if (maxElements <= 0)
1.681 + return 0;
1.682 + final Object[] items = this.items;
1.683 + final ReentrantLock lock = this.lock;
1.684 + lock.lock();
1.685 + try {
1.686 + int i = takeIndex;
1.687 + int n = 0;
1.688 + int max = (maxElements < count) ? maxElements : count;
1.689 + while (n < max) {
1.690 + c.add(this.<E>cast(items[i]));
1.691 + items[i] = null;
1.692 + i = inc(i);
1.693 + ++n;
1.694 + }
1.695 + if (n > 0) {
1.696 + count -= n;
1.697 + takeIndex = i;
1.698 + notFull.signalAll();
1.699 + }
1.700 + return n;
1.701 + } finally {
1.702 + lock.unlock();
1.703 + }
1.704 + }
1.705 +
1.706 + /**
1.707 + * Returns an iterator over the elements in this queue in proper sequence.
1.708 + * The elements will be returned in order from first (head) to last (tail).
1.709 + *
1.710 + * <p>The returned {@code Iterator} is a "weakly consistent" iterator that
1.711 + * will never throw {@link java.util.ConcurrentModificationException
1.712 + * ConcurrentModificationException},
1.713 + * and guarantees to traverse elements as they existed upon
1.714 + * construction of the iterator, and may (but is not guaranteed to)
1.715 + * reflect any modifications subsequent to construction.
1.716 + *
1.717 + * @return an iterator over the elements in this queue in proper sequence
1.718 + */
1.719 + public Iterator<E> iterator() {
1.720 + return new Itr();
1.721 + }
1.722 +
1.723 + /**
1.724 + * Iterator for ArrayBlockingQueue. To maintain weak consistency
1.725 + * with respect to puts and takes, we (1) read ahead one slot, so
1.726 + * as to not report hasNext true but then not have an element to
1.727 + * return -- however we later recheck this slot to use the most
1.728 + * current value; (2) ensure that each array slot is traversed at
1.729 + * most once (by tracking "remaining" elements); (3) skip over
1.730 + * null slots, which can occur if takes race ahead of iterators.
1.731 + * However, for circular array-based queues, we cannot rely on any
1.732 + * well established definition of what it means to be weakly
1.733 + * consistent with respect to interior removes since these may
1.734 + * require slot overwrites in the process of sliding elements to
1.735 + * cover gaps. So we settle for resiliency, operating on
1.736 + * established apparent nexts, which may miss some elements that
1.737 + * have moved between calls to next.
1.738 + */
1.739 + private class Itr implements Iterator<E> {
1.740 + private int remaining; // Number of elements yet to be returned
1.741 + private int nextIndex; // Index of element to be returned by next
1.742 + private E nextItem; // Element to be returned by next call to next
1.743 + private E lastItem; // Element returned by last call to next
1.744 + private int lastRet; // Index of last element returned, or -1 if none
1.745 +
1.746 + Itr() {
1.747 + final ReentrantLock lock = ArrayBlockingQueue.this.lock;
1.748 + lock.lock();
1.749 + try {
1.750 + lastRet = -1;
1.751 + if ((remaining = count) > 0)
1.752 + nextItem = itemAt(nextIndex = takeIndex);
1.753 + } finally {
1.754 + lock.unlock();
1.755 + }
1.756 + }
1.757 +
1.758 + public boolean hasNext() {
1.759 + return remaining > 0;
1.760 + }
1.761 +
1.762 + public E next() {
1.763 + final ReentrantLock lock = ArrayBlockingQueue.this.lock;
1.764 + lock.lock();
1.765 + try {
1.766 + if (remaining <= 0)
1.767 + throw new NoSuchElementException();
1.768 + lastRet = nextIndex;
1.769 + E x = itemAt(nextIndex); // check for fresher value
1.770 + if (x == null) {
1.771 + x = nextItem; // we are forced to report old value
1.772 + lastItem = null; // but ensure remove fails
1.773 + }
1.774 + else
1.775 + lastItem = x;
1.776 + while (--remaining > 0 && // skip over nulls
1.777 + (nextItem = itemAt(nextIndex = inc(nextIndex))) == null)
1.778 + ;
1.779 + return x;
1.780 + } finally {
1.781 + lock.unlock();
1.782 + }
1.783 + }
1.784 +
1.785 + public void remove() {
1.786 + final ReentrantLock lock = ArrayBlockingQueue.this.lock;
1.787 + lock.lock();
1.788 + try {
1.789 + int i = lastRet;
1.790 + if (i == -1)
1.791 + throw new IllegalStateException();
1.792 + lastRet = -1;
1.793 + E x = lastItem;
1.794 + lastItem = null;
1.795 + // only remove if item still at index
1.796 + if (x != null && x == items[i]) {
1.797 + boolean removingHead = (i == takeIndex);
1.798 + removeAt(i);
1.799 + if (!removingHead)
1.800 + nextIndex = dec(nextIndex);
1.801 + }
1.802 + } finally {
1.803 + lock.unlock();
1.804 + }
1.805 + }
1.806 + }
1.807 +
1.808 +}