32 * Expert Group and released to the public domain, as explained at 33 * http://creativecommons.org/licenses/publicdomain 34 */ 35 36 package java.util.concurrent; 37 import java.util.concurrent.locks.*; 38 import java.util.*; 39 40 /** 41 * A bounded {@linkplain BlockingQueue blocking queue} backed by an 42 * array. This queue orders elements FIFO (first-in-first-out). The 43 * <em>head</em> of the queue is that element that has been on the 44 * queue the longest time. The <em>tail</em> of the queue is that 45 * element that has been on the queue the shortest time. New elements 46 * are inserted at the tail of the queue, and the queue retrieval 47 * operations obtain elements at the head of the queue. 48 * 49 * <p>This is a classic "bounded buffer", in which a 50 * fixed-sized array holds elements inserted by producers and 51 * extracted by consumers. Once created, the capacity cannot be 52 * increased. Attempts to <tt>put</tt> an element into a full queue 53 * will result in the operation blocking; attempts to <tt>take</tt> an 54 * element from an empty queue will similarly block. 55 * 56 * <p> This class supports an optional fairness policy for ordering 57 * waiting producer and consumer threads. By default, this ordering 58 * is not guaranteed. However, a queue constructed with fairness set 59 * to <tt>true</tt> grants threads access in FIFO order. Fairness 60 * generally decreases throughput but reduces variability and avoids 61 * starvation. 62 * 63 * <p>This class and its iterator implement all of the 64 * <em>optional</em> methods of the {@link Collection} and {@link 65 * Iterator} interfaces. 66 * 67 * <p>This class is a member of the 68 * <a href="{@docRoot}/../technotes/guides/collections/index.html"> 69 * Java Collections Framework</a>. 70 * 71 * @since 1.5 72 * @author Doug Lea 73 * @param <E> the type of elements held in this collection 74 */ 75 public class ArrayBlockingQueue<E> extends AbstractQueue<E> 76 implements BlockingQueue<E>, java.io.Serializable { 77 78 /** 79 * Serialization ID. This class relies on default serialization 80 * even for the items array, which is default-serialized, even if 81 * it is empty. Otherwise it could not be declared final, which is 82 * necessary here. 83 */ 84 private static final long serialVersionUID = -817911632652898426L; 85 86 /** The queued items */ 87 private final E[] items; 88 /** items index for next take, poll or remove */ 89 private int takeIndex; 90 /** items index for next put, offer, or add. */ 91 private int putIndex; 92 /** Number of items in the queue */ 93 private int count; 94 95 /* 96 * Concurrency control uses the classic two-condition algorithm 97 * found in any textbook. 98 */ 99 100 /** Main lock guarding all access */ 101 private final ReentrantLock lock; 102 /** Condition for waiting takes */ 103 private final Condition notEmpty; 104 /** Condition for waiting puts */ 105 private final Condition notFull; 106 107 // Internal helper methods 108 109 /** 110 * Circularly increment i. 111 */ 112 final int inc(int i) { 113 return (++i == items.length)? 0 : i; 114 } 115 116 /** 117 * Inserts element at current put position, advances, and signals. 118 * Call only when holding lock. 119 */ 120 private void insert(E x) { 121 items[putIndex] = x; 122 putIndex = inc(putIndex); 123 ++count; 124 notEmpty.signal(); 125 } 126 127 /** 128 * Extracts element at current take position, advances, and signals. 129 * Call only when holding lock. 130 */ 131 private E extract() { 132 final E[] items = this.items; 133 E x = items[takeIndex]; 134 items[takeIndex] = null; 135 takeIndex = inc(takeIndex); 136 --count; 137 notFull.signal(); 138 return x; 139 } 140 141 /** 142 * Utility for remove and iterator.remove: Delete item at position i. 143 * Call only when holding lock. 144 */ 145 void removeAt(int i) { 146 final E[] items = this.items; 147 // if removing front item, just advance 148 if (i == takeIndex) { 149 items[takeIndex] = null; 150 takeIndex = inc(takeIndex); 151 } else { 152 // slide over all others up through putIndex. 153 for (;;) { 154 int nexti = inc(i); 155 if (nexti != putIndex) { 156 items[i] = items[nexti]; 157 i = nexti; 158 } else { 159 items[i] = null; 160 putIndex = i; 161 break; 162 } 163 } 164 } 165 --count; 166 notFull.signal(); 167 } 168 169 /** 170 * Creates an <tt>ArrayBlockingQueue</tt> with the given (fixed) 171 * capacity and default access policy. 172 * 173 * @param capacity the capacity of this queue 174 * @throws IllegalArgumentException if <tt>capacity</tt> is less than 1 175 */ 176 public ArrayBlockingQueue(int capacity) { 177 this(capacity, false); 178 } 179 180 /** 181 * Creates an <tt>ArrayBlockingQueue</tt> with the given (fixed) 182 * capacity and the specified access policy. 183 * 184 * @param capacity the capacity of this queue 185 * @param fair if <tt>true</tt> then queue accesses for threads blocked 186 * on insertion or removal, are processed in FIFO order; 187 * if <tt>false</tt> the access order is unspecified. 188 * @throws IllegalArgumentException if <tt>capacity</tt> is less than 1 189 */ 190 public ArrayBlockingQueue(int capacity, boolean fair) { 191 if (capacity <= 0) 192 throw new IllegalArgumentException(); 193 this.items = (E[]) new Object[capacity]; 194 lock = new ReentrantLock(fair); 195 notEmpty = lock.newCondition(); 196 notFull = lock.newCondition(); 197 } 198 199 /** 200 * Creates an <tt>ArrayBlockingQueue</tt> with the given (fixed) 201 * capacity, the specified access policy and initially containing the 202 * elements of the given collection, 203 * added in traversal order of the collection's iterator. 204 * 205 * @param capacity the capacity of this queue 206 * @param fair if <tt>true</tt> then queue accesses for threads blocked 207 * on insertion or removal, are processed in FIFO order; 208 * if <tt>false</tt> the access order is unspecified. 209 * @param c the collection of elements to initially contain 210 * @throws IllegalArgumentException if <tt>capacity</tt> is less than 211 * <tt>c.size()</tt>, or less than 1. 212 * @throws NullPointerException if the specified collection or any 213 * of its elements are null 214 */ 215 public ArrayBlockingQueue(int capacity, boolean fair, 216 Collection<? extends E> c) { 217 this(capacity, fair); 218 if (capacity < c.size()) 219 throw new IllegalArgumentException(); 220 221 for (E e : c) 222 add(e); 223 } 224 225 /** 226 * Inserts the specified element at the tail of this queue if it is 227 * possible to do so immediately without exceeding the queue's capacity, 228 * returning <tt>true</tt> upon success and throwing an 229 * <tt>IllegalStateException</tt> if this queue is full. 230 * 231 * @param e the element to add 232 * @return <tt>true</tt> (as specified by {@link Collection#add}) 233 * @throws IllegalStateException if this queue is full 234 * @throws NullPointerException if the specified element is null 235 */ 236 public boolean add(E e) { 237 return super.add(e); 238 } 239 240 /** 241 * Inserts the specified element at the tail of this queue if it is 242 * possible to do so immediately without exceeding the queue's capacity, 243 * returning <tt>true</tt> upon success and <tt>false</tt> if this queue 244 * is full. This method is generally preferable to method {@link #add}, 245 * which can fail to insert an element only by throwing an exception. 246 * 247 * @throws NullPointerException if the specified element is null 248 */ 249 public boolean offer(E e) { 250 if (e == null) throw new NullPointerException(); 251 final ReentrantLock lock = this.lock; 252 lock.lock(); 253 try { 254 if (count == items.length) 255 return false; 256 else { 257 insert(e); 258 return true; 259 } 260 } finally { 261 lock.unlock(); 262 } 263 } 264 265 /** 266 * Inserts the specified element at the tail of this queue, waiting 267 * for space to become available if the queue is full. 268 * 269 * @throws InterruptedException {@inheritDoc} 270 * @throws NullPointerException {@inheritDoc} 271 */ 272 public void put(E e) throws InterruptedException { 273 if (e == null) throw new NullPointerException(); 274 final E[] items = this.items; 275 final ReentrantLock lock = this.lock; 276 lock.lockInterruptibly(); 277 try { 278 try { 279 while (count == items.length) 280 notFull.await(); 281 } catch (InterruptedException ie) { 282 notFull.signal(); // propagate to non-interrupted thread 283 throw ie; 284 } 285 insert(e); 286 } finally { 287 lock.unlock(); 288 } 289 } 290 291 /** 292 * Inserts the specified element at the tail of this queue, waiting 293 * up to the specified wait time for space to become available if 294 * the queue is full. 295 * 296 * @throws InterruptedException {@inheritDoc} 297 * @throws NullPointerException {@inheritDoc} 298 */ 299 public boolean offer(E e, long timeout, TimeUnit unit) 300 throws InterruptedException { 301 302 if (e == null) throw new NullPointerException(); 303 long nanos = unit.toNanos(timeout); 304 final ReentrantLock lock = this.lock; 305 lock.lockInterruptibly(); 306 try { 307 for (;;) { 308 if (count != items.length) { 309 insert(e); 310 return true; 311 } 312 if (nanos <= 0) 313 return false; 314 try { 315 nanos = notFull.awaitNanos(nanos); 316 } catch (InterruptedException ie) { 317 notFull.signal(); // propagate to non-interrupted thread 318 throw ie; 319 } 320 } 321 } finally { 322 lock.unlock(); 323 } 324 } 325 326 public E poll() { 327 final ReentrantLock lock = this.lock; 328 lock.lock(); 329 try { 330 if (count == 0) 331 return null; 332 E x = extract(); 333 return x; 334 } finally { 335 lock.unlock(); 336 } 337 } 338 339 public E take() throws InterruptedException { 340 final ReentrantLock lock = this.lock; 341 lock.lockInterruptibly(); 342 try { 343 try { 344 while (count == 0) 345 notEmpty.await(); 346 } catch (InterruptedException ie) { 347 notEmpty.signal(); // propagate to non-interrupted thread 348 throw ie; 349 } 350 E x = extract(); 351 return x; 352 } finally { 353 lock.unlock(); 354 } 355 } 356 357 public E poll(long timeout, TimeUnit unit) throws InterruptedException { 358 long nanos = unit.toNanos(timeout); 359 final ReentrantLock lock = this.lock; 360 lock.lockInterruptibly(); 361 try { 362 for (;;) { 363 if (count != 0) { 364 E x = extract(); 365 return x; 366 } 367 if (nanos <= 0) 368 return null; 369 try { 370 nanos = notEmpty.awaitNanos(nanos); 371 } catch (InterruptedException ie) { 372 notEmpty.signal(); // propagate to non-interrupted thread 373 throw ie; 374 } 375 376 } 377 } finally { 378 lock.unlock(); 379 } 380 } 381 382 public E peek() { 383 final ReentrantLock lock = this.lock; 384 lock.lock(); 385 try { 386 return (count == 0) ? null : items[takeIndex]; 387 } finally { 388 lock.unlock(); 389 } 390 } 391 392 // this doc comment is overridden to remove the reference to collections 393 // greater in size than Integer.MAX_VALUE 394 /** 395 * Returns the number of elements in this queue. 396 * 397 * @return the number of elements in this queue 398 */ 399 public int size() { 400 final ReentrantLock lock = this.lock; 401 lock.lock(); 402 try { 403 return count; 404 } finally { 405 lock.unlock(); 406 } 407 } 408 409 // this doc comment is a modified copy of the inherited doc comment, 410 // without the reference to unlimited queues. 411 /** 412 * Returns the number of additional elements that this queue can ideally 413 * (in the absence of memory or resource constraints) accept without 414 * blocking. This is always equal to the initial capacity of this queue 415 * less the current <tt>size</tt> of this queue. 416 * 417 * <p>Note that you <em>cannot</em> always tell if an attempt to insert 418 * an element will succeed by inspecting <tt>remainingCapacity</tt> 419 * because it may be the case that another thread is about to 420 * insert or remove an element. 421 */ 422 public int remainingCapacity() { 423 final ReentrantLock lock = this.lock; 424 lock.lock(); 425 try { 426 return items.length - count; 427 } finally { 428 lock.unlock(); 429 } 430 } 431 432 /** 433 * Removes a single instance of the specified element from this queue, 434 * if it is present. More formally, removes an element <tt>e</tt> such 435 * that <tt>o.equals(e)</tt>, if this queue contains one or more such 436 * elements. 437 * Returns <tt>true</tt> if this queue contained the specified element 438 * (or equivalently, if this queue changed as a result of the call). 439 * 440 * @param o element to be removed from this queue, if present 441 * @return <tt>true</tt> if this queue changed as a result of the call 442 */ 443 public boolean remove(Object o) { 444 if (o == null) return false; 445 final E[] items = this.items; 446 final ReentrantLock lock = this.lock; 447 lock.lock(); 448 try { 449 int i = takeIndex; 450 int k = 0; 451 for (;;) { 452 if (k++ >= count) 453 return false; 454 if (o.equals(items[i])) { 455 removeAt(i); 456 return true; 457 } 458 i = inc(i); 459 } 460 461 } finally { 462 lock.unlock(); 463 } 464 } 465 466 /** 467 * Returns <tt>true</tt> if this queue contains the specified element. 468 * More formally, returns <tt>true</tt> if and only if this queue contains 469 * at least one element <tt>e</tt> such that <tt>o.equals(e)</tt>. 470 * 471 * @param o object to be checked for containment in this queue 472 * @return <tt>true</tt> if this queue contains the specified element 473 */ 474 public boolean contains(Object o) { 475 if (o == null) return false; 476 final E[] items = this.items; 477 final ReentrantLock lock = this.lock; 478 lock.lock(); 479 try { 480 int i = takeIndex; 481 int k = 0; 482 while (k++ < count) { 483 if (o.equals(items[i])) 484 return true; 485 i = inc(i); 486 } 487 return false; 488 } finally { 489 lock.unlock(); 490 } 491 } 492 493 /** 494 * Returns an array containing all of the elements in this queue, in 495 * proper sequence. 496 * 497 * <p>The returned array will be "safe" in that no references to it are 498 * maintained by this queue. (In other words, this method must allocate 499 * a new array). The caller is thus free to modify the returned array. 500 * 501 * <p>This method acts as bridge between array-based and collection-based 502 * APIs. 503 * 504 * @return an array containing all of the elements in this queue 505 */ 506 public Object[] toArray() { 507 final E[] items = this.items; 508 final ReentrantLock lock = this.lock; 509 lock.lock(); 510 try { 511 Object[] a = new Object[count]; 512 int k = 0; 513 int i = takeIndex; 514 while (k < count) { 515 a[k++] = items[i]; 516 i = inc(i); 517 } 518 return a; 519 } finally { 520 lock.unlock(); 521 } 522 } 523 524 /** 525 * Returns an array containing all of the elements in this queue, in 526 * proper sequence; the runtime type of the returned array is that of 527 * the specified array. If the queue fits in the specified array, it 528 * is returned therein. Otherwise, a new array is allocated with the 529 * runtime type of the specified array and the size of this queue. 530 * 531 * <p>If this queue fits in the specified array with room to spare 532 * (i.e., the array has more elements than this queue), the element in 533 * the array immediately following the end of the queue is set to 534 * <tt>null</tt>. 535 * 536 * <p>Like the {@link #toArray()} method, this method acts as bridge between 537 * array-based and collection-based APIs. Further, this method allows 538 * precise control over the runtime type of the output array, and may, 539 * under certain circumstances, be used to save allocation costs. 540 * 541 * <p>Suppose <tt>x</tt> is a queue known to contain only strings. 542 * The following code can be used to dump the queue into a newly 543 * allocated array of <tt>String</tt>: 544 * 545 * <pre> 546 * String[] y = x.toArray(new String[0]);</pre> 547 * 548 * Note that <tt>toArray(new Object[0])</tt> is identical in function to 549 * <tt>toArray()</tt>. 550 * 551 * @param a the array into which the elements of the queue are to 552 * be stored, if it is big enough; otherwise, a new array of the 553 * same runtime type is allocated for this purpose 554 * @return an array containing all of the elements in this queue 555 * @throws ArrayStoreException if the runtime type of the specified array 556 * is not a supertype of the runtime type of every element in 557 * this queue 558 * @throws NullPointerException if the specified array is null 559 */ 560 public <T> T[] toArray(T[] a) { 561 final E[] items = this.items; 562 final ReentrantLock lock = this.lock; 563 lock.lock(); 564 try { 565 if (a.length < count) 566 a = (T[])java.lang.reflect.Array.newInstance( 567 a.getClass().getComponentType(), 568 count 569 ); 570 571 int k = 0; 572 int i = takeIndex; 573 while (k < count) { 574 a[k++] = (T)items[i]; 575 i = inc(i); 576 } 577 if (a.length > count) 578 a[count] = null; 579 return a; 580 } finally { 581 lock.unlock(); 582 } 583 } 584 585 public String toString() { 586 final ReentrantLock lock = this.lock; 587 lock.lock(); 588 try { 589 return super.toString(); 590 } finally { 591 lock.unlock(); 592 } 593 } 594 595 /** 596 * Atomically removes all of the elements from this queue. 597 * The queue will be empty after this call returns. 598 */ 599 public void clear() { 600 final E[] items = this.items; 601 final ReentrantLock lock = this.lock; 602 lock.lock(); 603 try { 604 int i = takeIndex; 605 int k = count; 606 while (k-- > 0) { 607 items[i] = null; 608 i = inc(i); 609 } 610 count = 0; 611 putIndex = 0; 612 takeIndex = 0; 613 notFull.signalAll(); 614 } finally { 615 lock.unlock(); 616 } 617 } 618 619 /** 620 * @throws UnsupportedOperationException {@inheritDoc} 621 * @throws ClassCastException {@inheritDoc} 622 * @throws NullPointerException {@inheritDoc} 623 * @throws IllegalArgumentException {@inheritDoc} 624 */ 625 public int drainTo(Collection<? super E> c) { 626 if (c == null) 627 throw new NullPointerException(); 628 if (c == this) 629 throw new IllegalArgumentException(); 630 final E[] items = this.items; 631 final ReentrantLock lock = this.lock; 632 lock.lock(); 633 try { 634 int i = takeIndex; 635 int n = 0; 636 int max = count; 637 while (n < max) { 638 c.add(items[i]); 639 items[i] = null; 640 i = inc(i); 641 ++n; 642 } 643 if (n > 0) { 644 count = 0; 645 putIndex = 0; 646 takeIndex = 0; 647 notFull.signalAll(); 648 } 649 return n; 650 } finally { 651 lock.unlock(); 652 } 653 } 654 655 /** 656 * @throws UnsupportedOperationException {@inheritDoc} 657 * @throws ClassCastException {@inheritDoc} 658 * @throws NullPointerException {@inheritDoc} 659 * @throws IllegalArgumentException {@inheritDoc} 660 */ 661 public int drainTo(Collection<? super E> c, int maxElements) { 662 if (c == null) 663 throw new NullPointerException(); 664 if (c == this) 665 throw new IllegalArgumentException(); 666 if (maxElements <= 0) 667 return 0; 668 final E[] items = this.items; 669 final ReentrantLock lock = this.lock; 670 lock.lock(); 671 try { 672 int i = takeIndex; 673 int n = 0; 674 int sz = count; 675 int max = (maxElements < count)? maxElements : count; 676 while (n < max) { 677 c.add(items[i]); 678 items[i] = null; 679 i = inc(i); 680 ++n; 681 } 682 if (n > 0) { 683 count -= n; 684 takeIndex = i; 685 notFull.signalAll(); 686 } 687 return n; 688 } finally { 689 lock.unlock(); 690 } 691 } 692 693 694 /** 695 * Returns an iterator over the elements in this queue in proper sequence. 696 * The returned <tt>Iterator</tt> is a "weakly consistent" iterator that 697 * will never throw {@link ConcurrentModificationException}, 698 * and guarantees to traverse elements as they existed upon 699 * construction of the iterator, and may (but is not guaranteed to) 700 * reflect any modifications subsequent to construction. 701 * 702 * @return an iterator over the elements in this queue in proper sequence 703 */ 704 public Iterator<E> iterator() { 705 final ReentrantLock lock = this.lock; 706 lock.lock(); 707 try { 708 return new Itr(); 709 } finally { 710 lock.unlock(); 711 } 712 } 713 714 /** 715 * Iterator for ArrayBlockingQueue 716 */ 717 private class Itr implements Iterator<E> { 718 /** 719 * Index of element to be returned by next, 720 * or a negative number if no such. 721 */ 722 private int nextIndex; 723 724 /** 725 * nextItem holds on to item fields because once we claim 726 * that an element exists in hasNext(), we must return it in 727 * the following next() call even if it was in the process of 728 * being removed when hasNext() was called. 729 */ 730 private E nextItem; 731 732 /** 733 * Index of element returned by most recent call to next. 734 * Reset to -1 if this element is deleted by a call to remove. 735 */ 736 private int lastRet; 737 738 Itr() { 739 lastRet = -1; 740 if (count == 0) 741 nextIndex = -1; 742 else { 743 nextIndex = takeIndex; 744 nextItem = items[takeIndex]; 745 } 746 } 747 748 public boolean hasNext() { 749 /* 750 * No sync. We can return true by mistake here 751 * only if this iterator passed across threads, 752 * which we don't support anyway. 753 */ 754 return nextIndex >= 0; 755 } 756 757 /** 758 * Checks whether nextIndex is valid; if so setting nextItem. 759 * Stops iterator when either hits putIndex or sees null item. 760 */ 761 private void checkNext() { 762 if (nextIndex == putIndex) { 763 nextIndex = -1; 764 nextItem = null; 765 } else { 766 nextItem = items[nextIndex]; 767 if (nextItem == null) 768 nextIndex = -1; 769 } 770 } 771 772 public E next() { 773 final ReentrantLock lock = ArrayBlockingQueue.this.lock; 774 lock.lock(); 775 try { 776 if (nextIndex < 0) 777 throw new NoSuchElementException(); 778 lastRet = nextIndex; 779 E x = nextItem; 780 nextIndex = inc(nextIndex); 781 checkNext(); 782 return x; 783 } finally { 784 lock.unlock(); 785 } 786 } 787 788 public void remove() { 789 final ReentrantLock lock = ArrayBlockingQueue.this.lock; 790 lock.lock(); 791 try { 792 int i = lastRet; 793 if (i == -1) 794 throw new IllegalStateException(); 795 lastRet = -1; 796 797 int ti = takeIndex; 798 removeAt(i); 799 // back up cursor (reset to front if was first element) 800 nextIndex = (i == ti) ? takeIndex : i; 801 checkNext(); 802 } finally { 803 lock.unlock(); 804 } 805 } 806 } 807 } | 32 * Expert Group and released to the public domain, as explained at 33 * http://creativecommons.org/licenses/publicdomain 34 */ 35 36 package java.util.concurrent; 37 import java.util.concurrent.locks.*; 38 import java.util.*; 39 40 /** 41 * A bounded {@linkplain BlockingQueue blocking queue} backed by an 42 * array. This queue orders elements FIFO (first-in-first-out). The 43 * <em>head</em> of the queue is that element that has been on the 44 * queue the longest time. The <em>tail</em> of the queue is that 45 * element that has been on the queue the shortest time. New elements 46 * are inserted at the tail of the queue, and the queue retrieval 47 * operations obtain elements at the head of the queue. 48 * 49 * <p>This is a classic "bounded buffer", in which a 50 * fixed-sized array holds elements inserted by producers and 51 * extracted by consumers. Once created, the capacity cannot be 52 * changed. Attempts to {@code put} an element into a full queue 53 * will result in the operation blocking; attempts to {@code take} an 54 * element from an empty queue will similarly block. 55 * 56 * <p>This class supports an optional fairness policy for ordering 57 * waiting producer and consumer threads. By default, this ordering 58 * is not guaranteed. However, a queue constructed with fairness set 59 * to {@code true} grants threads access in FIFO order. Fairness 60 * generally decreases throughput but reduces variability and avoids 61 * starvation. 62 * 63 * <p>This class and its iterator implement all of the 64 * <em>optional</em> methods of the {@link Collection} and {@link 65 * Iterator} interfaces. 66 * 67 * <p>This class is a member of the 68 * <a href="{@docRoot}/../technotes/guides/collections/index.html"> 69 * Java Collections Framework</a>. 70 * 71 * @since 1.5 72 * @author Doug Lea 73 * @param <E> the type of elements held in this collection 74 */ 75 public class ArrayBlockingQueue<E> extends AbstractQueue<E> 76 implements BlockingQueue<E>, java.io.Serializable { 77 78 /** 79 * Serialization ID. This class relies on default serialization 80 * even for the items array, which is default-serialized, even if 81 * it is empty. Otherwise it could not be declared final, which is 82 * necessary here. 83 */ 84 private static final long serialVersionUID = -817911632652898426L; 85 86 /** The queued items */ 87 final Object[] items; 88 89 /** items index for next take, poll, peek or remove */ 90 int takeIndex; 91 92 /** items index for next put, offer, or add */ 93 int putIndex; 94 95 /** Number of elements in the queue */ 96 int count; 97 98 /* 99 * Concurrency control uses the classic two-condition algorithm 100 * found in any textbook. 101 */ 102 103 /** Main lock guarding all access */ 104 final ReentrantLock lock; 105 /** Condition for waiting takes */ 106 private final Condition notEmpty; 107 /** Condition for waiting puts */ 108 private final Condition notFull; 109 110 // Internal helper methods 111 112 /** 113 * Circularly increment i. 114 */ 115 final int inc(int i) { 116 return (++i == items.length) ? 0 : i; 117 } 118 119 /** 120 * Circularly decrement i. 121 */ 122 final int dec(int i) { 123 return ((i == 0) ? items.length : i) - 1; 124 } 125 126 @SuppressWarnings("unchecked") 127 static <E> E cast(Object item) { 128 return (E) item; 129 } 130 131 /** 132 * Returns item at index i. 133 */ 134 final E itemAt(int i) { 135 return this.<E>cast(items[i]); 136 } 137 138 /** 139 * Throws NullPointerException if argument is null. 140 * 141 * @param v the element 142 */ 143 private static void checkNotNull(Object v) { 144 if (v == null) 145 throw new NullPointerException(); 146 } 147 148 /** 149 * Inserts element at current put position, advances, and signals. 150 * Call only when holding lock. 151 */ 152 private void insert(E x) { 153 items[putIndex] = x; 154 putIndex = inc(putIndex); 155 ++count; 156 notEmpty.signal(); 157 } 158 159 /** 160 * Extracts element at current take position, advances, and signals. 161 * Call only when holding lock. 162 */ 163 private E extract() { 164 final Object[] items = this.items; 165 E x = this.<E>cast(items[takeIndex]); 166 items[takeIndex] = null; 167 takeIndex = inc(takeIndex); 168 --count; 169 notFull.signal(); 170 return x; 171 } 172 173 /** 174 * Deletes item at position i. 175 * Utility for remove and iterator.remove. 176 * Call only when holding lock. 177 */ 178 void removeAt(int i) { 179 final Object[] items = this.items; 180 // if removing front item, just advance 181 if (i == takeIndex) { 182 items[takeIndex] = null; 183 takeIndex = inc(takeIndex); 184 } else { 185 // slide over all others up through putIndex. 186 for (;;) { 187 int nexti = inc(i); 188 if (nexti != putIndex) { 189 items[i] = items[nexti]; 190 i = nexti; 191 } else { 192 items[i] = null; 193 putIndex = i; 194 break; 195 } 196 } 197 } 198 --count; 199 notFull.signal(); 200 } 201 202 /** 203 * Creates an {@code ArrayBlockingQueue} with the given (fixed) 204 * capacity and default access policy. 205 * 206 * @param capacity the capacity of this queue 207 * @throws IllegalArgumentException if {@code capacity < 1} 208 */ 209 public ArrayBlockingQueue(int capacity) { 210 this(capacity, false); 211 } 212 213 /** 214 * Creates an {@code ArrayBlockingQueue} with the given (fixed) 215 * capacity and the specified access policy. 216 * 217 * @param capacity the capacity of this queue 218 * @param fair if {@code true} then queue accesses for threads blocked 219 * on insertion or removal, are processed in FIFO order; 220 * if {@code false} the access order is unspecified. 221 * @throws IllegalArgumentException if {@code capacity < 1} 222 */ 223 public ArrayBlockingQueue(int capacity, boolean fair) { 224 if (capacity <= 0) 225 throw new IllegalArgumentException(); 226 this.items = new Object[capacity]; 227 lock = new ReentrantLock(fair); 228 notEmpty = lock.newCondition(); 229 notFull = lock.newCondition(); 230 } 231 232 /** 233 * Creates an {@code ArrayBlockingQueue} with the given (fixed) 234 * capacity, the specified access policy and initially containing the 235 * elements of the given collection, 236 * added in traversal order of the collection's iterator. 237 * 238 * @param capacity the capacity of this queue 239 * @param fair if {@code true} then queue accesses for threads blocked 240 * on insertion or removal, are processed in FIFO order; 241 * if {@code false} the access order is unspecified. 242 * @param c the collection of elements to initially contain 243 * @throws IllegalArgumentException if {@code capacity} is less than 244 * {@code c.size()}, or less than 1. 245 * @throws NullPointerException if the specified collection or any 246 * of its elements are null 247 */ 248 public ArrayBlockingQueue(int capacity, boolean fair, 249 Collection<? extends E> c) { 250 this(capacity, fair); 251 252 final ReentrantLock lock = this.lock; 253 lock.lock(); // Lock only for visibility, not mutual exclusion 254 try { 255 int i = 0; 256 try { 257 for (E e : c) { 258 checkNotNull(e); 259 items[i++] = e; 260 } 261 } catch (ArrayIndexOutOfBoundsException ex) { 262 throw new IllegalArgumentException(); 263 } 264 count = i; 265 putIndex = (i == capacity) ? 0 : i; 266 } finally { 267 lock.unlock(); 268 } 269 } 270 271 /** 272 * Inserts the specified element at the tail of this queue if it is 273 * possible to do so immediately without exceeding the queue's capacity, 274 * returning {@code true} upon success and throwing an 275 * {@code IllegalStateException} if this queue is full. 276 * 277 * @param e the element to add 278 * @return {@code true} (as specified by {@link Collection#add}) 279 * @throws IllegalStateException if this queue is full 280 * @throws NullPointerException if the specified element is null 281 */ 282 public boolean add(E e) { 283 return super.add(e); 284 } 285 286 /** 287 * Inserts the specified element at the tail of this queue if it is 288 * possible to do so immediately without exceeding the queue's capacity, 289 * returning {@code true} upon success and {@code false} if this queue 290 * is full. This method is generally preferable to method {@link #add}, 291 * which can fail to insert an element only by throwing an exception. 292 * 293 * @throws NullPointerException if the specified element is null 294 */ 295 public boolean offer(E e) { 296 checkNotNull(e); 297 final ReentrantLock lock = this.lock; 298 lock.lock(); 299 try { 300 if (count == items.length) 301 return false; 302 else { 303 insert(e); 304 return true; 305 } 306 } finally { 307 lock.unlock(); 308 } 309 } 310 311 /** 312 * Inserts the specified element at the tail of this queue, waiting 313 * for space to become available if the queue is full. 314 * 315 * @throws InterruptedException {@inheritDoc} 316 * @throws NullPointerException {@inheritDoc} 317 */ 318 public void put(E e) throws InterruptedException { 319 checkNotNull(e); 320 final ReentrantLock lock = this.lock; 321 lock.lockInterruptibly(); 322 try { 323 while (count == items.length) 324 notFull.await(); 325 insert(e); 326 } finally { 327 lock.unlock(); 328 } 329 } 330 331 /** 332 * Inserts the specified element at the tail of this queue, waiting 333 * up to the specified wait time for space to become available if 334 * the queue is full. 335 * 336 * @throws InterruptedException {@inheritDoc} 337 * @throws NullPointerException {@inheritDoc} 338 */ 339 public boolean offer(E e, long timeout, TimeUnit unit) 340 throws InterruptedException { 341 342 checkNotNull(e); 343 long nanos = unit.toNanos(timeout); 344 final ReentrantLock lock = this.lock; 345 lock.lockInterruptibly(); 346 try { 347 while (count == items.length) { 348 if (nanos <= 0) 349 return false; 350 nanos = notFull.awaitNanos(nanos); 351 } 352 insert(e); 353 return true; 354 } finally { 355 lock.unlock(); 356 } 357 } 358 359 public E poll() { 360 final ReentrantLock lock = this.lock; 361 lock.lock(); 362 try { 363 return (count == 0) ? null : extract(); 364 } finally { 365 lock.unlock(); 366 } 367 } 368 369 public E take() throws InterruptedException { 370 final ReentrantLock lock = this.lock; 371 lock.lockInterruptibly(); 372 try { 373 while (count == 0) 374 notEmpty.await(); 375 return extract(); 376 } finally { 377 lock.unlock(); 378 } 379 } 380 381 public E poll(long timeout, TimeUnit unit) throws InterruptedException { 382 long nanos = unit.toNanos(timeout); 383 final ReentrantLock lock = this.lock; 384 lock.lockInterruptibly(); 385 try { 386 while (count == 0) { 387 if (nanos <= 0) 388 return null; 389 nanos = notEmpty.awaitNanos(nanos); 390 } 391 return extract(); 392 } finally { 393 lock.unlock(); 394 } 395 } 396 397 public E peek() { 398 final ReentrantLock lock = this.lock; 399 lock.lock(); 400 try { 401 return (count == 0) ? null : itemAt(takeIndex); 402 } finally { 403 lock.unlock(); 404 } 405 } 406 407 // this doc comment is overridden to remove the reference to collections 408 // greater in size than Integer.MAX_VALUE 409 /** 410 * Returns the number of elements in this queue. 411 * 412 * @return the number of elements in this queue 413 */ 414 public int size() { 415 final ReentrantLock lock = this.lock; 416 lock.lock(); 417 try { 418 return count; 419 } finally { 420 lock.unlock(); 421 } 422 } 423 424 // this doc comment is a modified copy of the inherited doc comment, 425 // without the reference to unlimited queues. 426 /** 427 * Returns the number of additional elements that this queue can ideally 428 * (in the absence of memory or resource constraints) accept without 429 * blocking. This is always equal to the initial capacity of this queue 430 * less the current {@code size} of this queue. 431 * 432 * <p>Note that you <em>cannot</em> always tell if an attempt to insert 433 * an element will succeed by inspecting {@code remainingCapacity} 434 * because it may be the case that another thread is about to 435 * insert or remove an element. 436 */ 437 public int remainingCapacity() { 438 final ReentrantLock lock = this.lock; 439 lock.lock(); 440 try { 441 return items.length - count; 442 } finally { 443 lock.unlock(); 444 } 445 } 446 447 /** 448 * Removes a single instance of the specified element from this queue, 449 * if it is present. More formally, removes an element {@code e} such 450 * that {@code o.equals(e)}, if this queue contains one or more such 451 * elements. 452 * Returns {@code true} if this queue contained the specified element 453 * (or equivalently, if this queue changed as a result of the call). 454 * 455 * <p>Removal of interior elements in circular array based queues 456 * is an intrinsically slow and disruptive operation, so should 457 * be undertaken only in exceptional circumstances, ideally 458 * only when the queue is known not to be accessible by other 459 * threads. 460 * 461 * @param o element to be removed from this queue, if present 462 * @return {@code true} if this queue changed as a result of the call 463 */ 464 public boolean remove(Object o) { 465 if (o == null) return false; 466 final Object[] items = this.items; 467 final ReentrantLock lock = this.lock; 468 lock.lock(); 469 try { 470 for (int i = takeIndex, k = count; k > 0; i = inc(i), k--) { 471 if (o.equals(items[i])) { 472 removeAt(i); 473 return true; 474 } 475 } 476 return false; 477 } finally { 478 lock.unlock(); 479 } 480 } 481 482 /** 483 * Returns {@code true} if this queue contains the specified element. 484 * More formally, returns {@code true} if and only if this queue contains 485 * at least one element {@code e} such that {@code o.equals(e)}. 486 * 487 * @param o object to be checked for containment in this queue 488 * @return {@code true} if this queue contains the specified element 489 */ 490 public boolean contains(Object o) { 491 if (o == null) return false; 492 final Object[] items = this.items; 493 final ReentrantLock lock = this.lock; 494 lock.lock(); 495 try { 496 for (int i = takeIndex, k = count; k > 0; i = inc(i), k--) 497 if (o.equals(items[i])) 498 return true; 499 return false; 500 } finally { 501 lock.unlock(); 502 } 503 } 504 505 /** 506 * Returns an array containing all of the elements in this queue, in 507 * proper sequence. 508 * 509 * <p>The returned array will be "safe" in that no references to it are 510 * maintained by this queue. (In other words, this method must allocate 511 * a new array). The caller is thus free to modify the returned array. 512 * 513 * <p>This method acts as bridge between array-based and collection-based 514 * APIs. 515 * 516 * @return an array containing all of the elements in this queue 517 */ 518 public Object[] toArray() { 519 final Object[] items = this.items; 520 final ReentrantLock lock = this.lock; 521 lock.lock(); 522 try { 523 final int count = this.count; 524 Object[] a = new Object[count]; 525 for (int i = takeIndex, k = 0; k < count; i = inc(i), k++) 526 a[k] = items[i]; 527 return a; 528 } finally { 529 lock.unlock(); 530 } 531 } 532 533 /** 534 * Returns an array containing all of the elements in this queue, in 535 * proper sequence; the runtime type of the returned array is that of 536 * the specified array. If the queue fits in the specified array, it 537 * is returned therein. Otherwise, a new array is allocated with the 538 * runtime type of the specified array and the size of this queue. 539 * 540 * <p>If this queue fits in the specified array with room to spare 541 * (i.e., the array has more elements than this queue), the element in 542 * the array immediately following the end of the queue is set to 543 * {@code null}. 544 * 545 * <p>Like the {@link #toArray()} method, this method acts as bridge between 546 * array-based and collection-based APIs. Further, this method allows 547 * precise control over the runtime type of the output array, and may, 548 * under certain circumstances, be used to save allocation costs. 549 * 550 * <p>Suppose {@code x} is a queue known to contain only strings. 551 * The following code can be used to dump the queue into a newly 552 * allocated array of {@code String}: 553 * 554 * <pre> 555 * String[] y = x.toArray(new String[0]);</pre> 556 * 557 * Note that {@code toArray(new Object[0])} is identical in function to 558 * {@code toArray()}. 559 * 560 * @param a the array into which the elements of the queue are to 561 * be stored, if it is big enough; otherwise, a new array of the 562 * same runtime type is allocated for this purpose 563 * @return an array containing all of the elements in this queue 564 * @throws ArrayStoreException if the runtime type of the specified array 565 * is not a supertype of the runtime type of every element in 566 * this queue 567 * @throws NullPointerException if the specified array is null 568 */ 569 @SuppressWarnings("unchecked") 570 public <T> T[] toArray(T[] a) { 571 final Object[] items = this.items; 572 final ReentrantLock lock = this.lock; 573 lock.lock(); 574 try { 575 final int count = this.count; 576 final int len = a.length; 577 if (len < count) 578 a = (T[])java.lang.reflect.Array.newInstance( 579 a.getClass().getComponentType(), count); 580 for (int i = takeIndex, k = 0; k < count; i = inc(i), k++) 581 a[k] = (T) items[i]; 582 if (len > count) 583 a[count] = null; 584 return a; 585 } finally { 586 lock.unlock(); 587 } 588 } 589 590 public String toString() { 591 final ReentrantLock lock = this.lock; 592 lock.lock(); 593 try { 594 int k = count; 595 if (k == 0) 596 return "[]"; 597 598 StringBuilder sb = new StringBuilder(); 599 sb.append('['); 600 for (int i = takeIndex; ; i = inc(i)) { 601 Object e = items[i]; 602 sb.append(e == this ? "(this Collection)" : e); 603 if (--k == 0) 604 return sb.append(']').toString(); 605 sb.append(',').append(' '); 606 } 607 } finally { 608 lock.unlock(); 609 } 610 } 611 612 /** 613 * Atomically removes all of the elements from this queue. 614 * The queue will be empty after this call returns. 615 */ 616 public void clear() { 617 final Object[] items = this.items; 618 final ReentrantLock lock = this.lock; 619 lock.lock(); 620 try { 621 for (int i = takeIndex, k = count; k > 0; i = inc(i), k--) 622 items[i] = null; 623 count = 0; 624 putIndex = 0; 625 takeIndex = 0; 626 notFull.signalAll(); 627 } finally { 628 lock.unlock(); 629 } 630 } 631 632 /** 633 * @throws UnsupportedOperationException {@inheritDoc} 634 * @throws ClassCastException {@inheritDoc} 635 * @throws NullPointerException {@inheritDoc} 636 * @throws IllegalArgumentException {@inheritDoc} 637 */ 638 public int drainTo(Collection<? super E> c) { 639 checkNotNull(c); 640 if (c == this) 641 throw new IllegalArgumentException(); 642 final Object[] items = this.items; 643 final ReentrantLock lock = this.lock; 644 lock.lock(); 645 try { 646 int i = takeIndex; 647 int n = 0; 648 int max = count; 649 while (n < max) { 650 c.add(this.<E>cast(items[i])); 651 items[i] = null; 652 i = inc(i); 653 ++n; 654 } 655 if (n > 0) { 656 count = 0; 657 putIndex = 0; 658 takeIndex = 0; 659 notFull.signalAll(); 660 } 661 return n; 662 } finally { 663 lock.unlock(); 664 } 665 } 666 667 /** 668 * @throws UnsupportedOperationException {@inheritDoc} 669 * @throws ClassCastException {@inheritDoc} 670 * @throws NullPointerException {@inheritDoc} 671 * @throws IllegalArgumentException {@inheritDoc} 672 */ 673 public int drainTo(Collection<? super E> c, int maxElements) { 674 checkNotNull(c); 675 if (c == this) 676 throw new IllegalArgumentException(); 677 if (maxElements <= 0) 678 return 0; 679 final Object[] items = this.items; 680 final ReentrantLock lock = this.lock; 681 lock.lock(); 682 try { 683 int i = takeIndex; 684 int n = 0; 685 int max = (maxElements < count) ? maxElements : count; 686 while (n < max) { 687 c.add(this.<E>cast(items[i])); 688 items[i] = null; 689 i = inc(i); 690 ++n; 691 } 692 if (n > 0) { 693 count -= n; 694 takeIndex = i; 695 notFull.signalAll(); 696 } 697 return n; 698 } finally { 699 lock.unlock(); 700 } 701 } 702 703 /** 704 * Returns an iterator over the elements in this queue in proper sequence. 705 * The elements will be returned in order from first (head) to last (tail). 706 * 707 * <p>The returned {@code Iterator} is a "weakly consistent" iterator that 708 * will never throw {@link java.util.ConcurrentModificationException 709 * ConcurrentModificationException}, 710 * and guarantees to traverse elements as they existed upon 711 * construction of the iterator, and may (but is not guaranteed to) 712 * reflect any modifications subsequent to construction. 713 * 714 * @return an iterator over the elements in this queue in proper sequence 715 */ 716 public Iterator<E> iterator() { 717 return new Itr(); 718 } 719 720 /** 721 * Iterator for ArrayBlockingQueue. To maintain weak consistency 722 * with respect to puts and takes, we (1) read ahead one slot, so 723 * as to not report hasNext true but then not have an element to 724 * return -- however we later recheck this slot to use the most 725 * current value; (2) ensure that each array slot is traversed at 726 * most once (by tracking "remaining" elements); (3) skip over 727 * null slots, which can occur if takes race ahead of iterators. 728 * However, for circular array-based queues, we cannot rely on any 729 * well established definition of what it means to be weakly 730 * consistent with respect to interior removes since these may 731 * require slot overwrites in the process of sliding elements to 732 * cover gaps. So we settle for resiliency, operating on 733 * established apparent nexts, which may miss some elements that 734 * have moved between calls to next. 735 */ 736 private class Itr implements Iterator<E> { 737 private int remaining; // Number of elements yet to be returned 738 private int nextIndex; // Index of element to be returned by next 739 private E nextItem; // Element to be returned by next call to next 740 private E lastItem; // Element returned by last call to next 741 private int lastRet; // Index of last element returned, or -1 if none 742 743 Itr() { 744 final ReentrantLock lock = ArrayBlockingQueue.this.lock; 745 lock.lock(); 746 try { 747 lastRet = -1; 748 if ((remaining = count) > 0) 749 nextItem = itemAt(nextIndex = takeIndex); 750 } finally { 751 lock.unlock(); 752 } 753 } 754 755 public boolean hasNext() { 756 return remaining > 0; 757 } 758 759 public E next() { 760 final ReentrantLock lock = ArrayBlockingQueue.this.lock; 761 lock.lock(); 762 try { 763 if (remaining <= 0) 764 throw new NoSuchElementException(); 765 lastRet = nextIndex; 766 E x = itemAt(nextIndex); // check for fresher value 767 if (x == null) { 768 x = nextItem; // we are forced to report old value 769 lastItem = null; // but ensure remove fails 770 } 771 else 772 lastItem = x; 773 while (--remaining > 0 && // skip over nulls 774 (nextItem = itemAt(nextIndex = inc(nextIndex))) == null) 775 ; 776 return x; 777 } finally { 778 lock.unlock(); 779 } 780 } 781 782 public void remove() { 783 final ReentrantLock lock = ArrayBlockingQueue.this.lock; 784 lock.lock(); 785 try { 786 int i = lastRet; 787 if (i == -1) 788 throw new IllegalStateException(); 789 lastRet = -1; 790 E x = lastItem; 791 lastItem = null; 792 // only remove if item still at index 793 if (x != null && x == items[i]) { 794 boolean removingHead = (i == takeIndex); 795 removeAt(i); 796 if (!removingHead) 797 nextIndex = dec(nextIndex); 798 } 799 } finally { 800 lock.unlock(); 801 } 802 } 803 } 804 805 } |