libcore/luni/src/main/java/java/util/concurrent/LinkedBlockingQueue.java

   1 /*
   2  * Written by Doug Lea with assistance from members of JCP JSR-166
   3  * Expert Group and released to the public domain, as explained at
   4  * http://creativecommons.org/licenses/publicdomain
   5  */
   6
   7 package java.util.concurrent;
   8
   9 // BEGIN android-note
  10 // removed link to collections framework docs
  11 // END android-note
  12
  13 import java.util.AbstractQueue;
  14 import java.util.Collection;
  15 import java.util.Iterator;
  16 import java.util.NoSuchElementException;
  17 import java.util.concurrent.atomic.AtomicInteger;
  18 import java.util.concurrent.locks.Condition;
  19 import java.util.concurrent.locks.ReentrantLock;
  20
  21 /**
  22  * An optionally-bounded {@linkplain BlockingQueue blocking queue} based on
  23  * linked nodes.
  24  * This queue orders elements FIFO (first-in-first-out).
  25  * The <em>head</em> of the queue is that element that has been on the
  26  * queue the longest time.
  27  * The <em>tail</em> of the queue is that element that has been on the
  28  * queue the shortest time. New elements
  29  * are inserted at the tail of the queue, and the queue retrieval
  30  * operations obtain elements at the head of the queue.
  31  * Linked queues typically have higher throughput than array-based queues but
  32  * less predictable performance in most concurrent applications.
  33  *
  34  * <p> The optional capacity bound constructor argument serves as a
  35  * way to prevent excessive queue expansion. The capacity, if unspecified,
  36  * is equal to {@link Integer#MAX_VALUE}.  Linked nodes are
  37  * dynamically created upon each insertion unless this would bring the
  38  * queue above capacity.
  39  *
  40  * <p>This class and its iterator implement all of the
  41  * <em>optional</em> methods of the {@link Collection} and {@link
  42  * Iterator} interfaces.
  43  *
  44  * @since 1.5
  45  * @author Doug Lea
  46  * @param <E> the type of elements held in this collection
  47  *
  48  */
  49 public class LinkedBlockingQueue<E> extends AbstractQueue<E>
  50         implements BlockingQueue<E>, java.io.Serializable {
  51     private static final long serialVersionUID = -6903933977591709194L;
  52
  53     /*
  54      * A variant of the "two lock queue" algorithm.  The putLock gates
  55      * entry to put (and offer), and has an associated condition for
  56      * waiting puts.  Similarly for the takeLock.  The "count" field
  57      * that they both rely on is maintained as an atomic to avoid
  58      * needing to get both locks in most cases. Also, to minimize need
  59      * for puts to get takeLock and vice-versa, cascading notifies are
  60      * used. When a put notices that it has enabled at least one take,
  61      * it signals taker. That taker in turn signals others if more
  62      * items have been entered since the signal. And symmetrically for
  63      * takes signalling puts. Operations such as remove(Object) and
  64      * iterators acquire both locks.
  65      *
  66      * Visibility between writers and readers is provided as follows:
  67      *
  68      * Whenever an element is enqueued, the putLock is acquired and
  69      * count updated.  A subsequent reader guarantees visibility to the
  70      * enqueued Node by either acquiring the putLock (via fullyLock)
  71      * or by acquiring the takeLock, and then reading n = count.get();
  72      * this gives visibility to the first n items.
  73      *
  74      * To implement weakly consistent iterators, it appears we need to
  75      * keep all Nodes GC-reachable from a predecessor dequeued Node.
  76      * That would cause two problems:
  77      * - allow a rogue Iterator to cause unbounded memory retention
  78      * - cause cross-generational linking of old Nodes to new Nodes if
  79      *   a Node was tenured while live, which generational GCs have a
  80      *   hard time dealing with, causing repeated major collections.
  81      * However, only non-deleted Nodes need to be reachable from
  82      * dequeued Nodes, and reachability does not necessarily have to
  83      * be of the kind understood by the GC.  We use the trick of
  84      * linking a Node that has just been dequeued to itself.  Such a
  85      * self-link implicitly means to advance to head.next.
  86      */
  87
  88     /**
  89      * Linked list node class
  90      */
  91     static class Node<E> {
  92         E item;
  93
  94         /**
  95          * One of:
  96          * - the real successor Node
  97          * - this Node, meaning the successor is head.next
  98          * - null, meaning there is no successor (this is the last node)
  99          */
 100         Node<E> next;
 101
 102         Node(E x) { item = x; }
 103     }
 104
 105     /** The capacity bound, or Integer.MAX_VALUE if none */
 106     private final int capacity;
 107
 108     /** Current number of elements */
 109     private final AtomicInteger count = new AtomicInteger(0);
 110
 111     /**
 112      * Head of linked list.
 113      * Invariant: head.item == null
 114      */
 115     private transient Node<E> head;
 116
 117     /**
 118      * Tail of linked list.
 119      * Invariant: last.next == null
 120      */
 121     private transient Node<E> last;
 122
 123     /** Lock held by take, poll, etc */
 124     private final ReentrantLock takeLock = new ReentrantLock();
 125
 126     /** Wait queue for waiting takes */
 127     private final Condition notEmpty = takeLock.newCondition();
 128
 129     /** Lock held by put, offer, etc */
 130     private final ReentrantLock putLock = new ReentrantLock();
 131
 132     /** Wait queue for waiting puts */
 133     private final Condition notFull = putLock.newCondition();
 134
 135     /**
 136      * Signals a waiting take. Called only from put/offer (which do not
 137      * otherwise ordinarily lock takeLock.)
 138      */
 139     private void signalNotEmpty() {
 140         final ReentrantLock takeLock = this.takeLock;
 141         takeLock.lock();
 142         try {
 143             notEmpty.signal();
 144         } finally {
 145             takeLock.unlock();
 146         }
 147     }
 148
 149     /**
 150      * Signals a waiting put. Called only from take/poll.
 151      */
 152     private void signalNotFull() {
 153         final ReentrantLock putLock = this.putLock;
 154         putLock.lock();
 155         try {
 156             notFull.signal();
 157         } finally {
 158             putLock.unlock();
 159         }
 160     }
 161
 162     /**
 163      * Creates a node and links it at end of queue.
 164      *
 165      * @param x the item
 166      */
 167     private void enqueue(E x) {
 168         // assert putLock.isHeldByCurrentThread();
 169         // assert last.next == null;
 170         last = last.next = new Node<E>(x);
 171     }
 172
 173     /**
 174      * Removes a node from head of queue.
 175      *
 176      * @return the node
 177      */
 178     private E dequeue() {
 179         // assert takeLock.isHeldByCurrentThread();
 180         // assert head.item == null;
 181         Node<E> h = head;
 182         Node<E> first = h.next;
 183         h.next = h; // help GC
 184         head = first;
 185         E x = first.item;
 186         first.item = null;
 187         return x;
 188     }
 189
 190     /**
 191      * Lock to prevent both puts and takes.
 192      */
 193     void fullyLock() {
 194         putLock.lock();
 195         takeLock.lock();
 196     }
 197
 198     /**
 199      * Unlock to allow both puts and takes.
 200      */
 201     void fullyUnlock() {
 202         takeLock.unlock();
 203         putLock.unlock();
 204     }
 205
 206 //     /**
 207 //      * Tells whether both locks are held by current thread.
 208 //      */
 209 //     boolean isFullyLocked() {
 210 //         return (putLock.isHeldByCurrentThread() &&
 211 //                 takeLock.isHeldByCurrentThread());
 212 //     }
 213
 214     /**
 215      * Creates a {@code LinkedBlockingQueue} with a capacity of
 216      * {@link Integer#MAX_VALUE}.
 217      */
 218     public LinkedBlockingQueue() {
 219         this(Integer.MAX_VALUE);
 220     }
 221
 222     /**
 223      * Creates a {@code LinkedBlockingQueue} with the given (fixed) capacity.
 224      *
 225      * @param capacity the capacity of this queue
 226      * @throws IllegalArgumentException if {@code capacity} is not greater
 227      *         than zero
 228      */
 229     public LinkedBlockingQueue(int capacity) {
 230         if (capacity <= 0) throw new IllegalArgumentException();
 231         this.capacity = capacity;
 232         last = head = new Node<E>(null);
 233     }
 234
 235     /**
 236      * Creates a {@code LinkedBlockingQueue} with a capacity of
 237      * {@link Integer#MAX_VALUE}, initially containing the elements of the
 238      * given collection,
 239      * added in traversal order of the collection's iterator.
 240      *
 241      * @param c the collection of elements to initially contain
 242      * @throws NullPointerException if the specified collection or any
 243      *         of its elements are null
 244      */
 245     public LinkedBlockingQueue(Collection<? extends E> c) {
 246         this(Integer.MAX_VALUE);
 247         final ReentrantLock putLock = this.putLock;
 248         putLock.lock(); // Never contended, but necessary for visibility
 249         try {
 250             int n = 0;
 251             for (E e : c) {
 252                 if (e == null)
 253                     throw new NullPointerException();
 254                 if (n == capacity)
 255                     throw new IllegalStateException("Queue full");
 256                 enqueue(e);
 257                 ++n;
 258             }
 259             count.set(n);
 260         } finally {
 261             putLock.unlock();
 262         }
 263     }
 264
 265
 266     // this doc comment is overridden to remove the reference to collections
 267     // greater in size than Integer.MAX_VALUE
 268     /**
 269      * Returns the number of elements in this queue.
 270      *
 271      * @return the number of elements in this queue
 272      */
 273     public int size() {
 274         return count.get();
 275     }
 276
 277     // this doc comment is a modified copy of the inherited doc comment,
 278     // without the reference to unlimited queues.
 279     /**
 280      * Returns the number of additional elements that this queue can ideally
 281      * (in the absence of memory or resource constraints) accept without
 282      * blocking. This is always equal to the initial capacity of this queue
 283      * less the current {@code size} of this queue.
 284      *
 285      * <p>Note that you <em>cannot</em> always tell if an attempt to insert
 286      * an element will succeed by inspecting {@code remainingCapacity}
 287      * because it may be the case that another thread is about to
 288      * insert or remove an element.
 289      */
 290     public int remainingCapacity() {
 291         return capacity - count.get();
 292     }
 293
 294     /**
 295      * Inserts the specified element at the tail of this queue, waiting if
 296      * necessary for space to become available.
 297      *
 298      * @throws InterruptedException {@inheritDoc}
 299      * @throws NullPointerException {@inheritDoc}
 300      */
 301     public void put(E e) throws InterruptedException {
 302         if (e == null) throw new NullPointerException();
 303         // Note: convention in all put/take/etc is to preset local var
 304         // holding count negative to indicate failure unless set.
 305         int c = -1;
 306         final ReentrantLock putLock = this.putLock;
 307         final AtomicInteger count = this.count;
 308         putLock.lockInterruptibly();
 309         try {
 310             /*
 311              * Note that count is used in wait guard even though it is
 312              * not protected by lock. This works because count can
 313              * only decrease at this point (all other puts are shut
 314              * out by lock), and we (or some other waiting put) are
 315              * signalled if it ever changes from capacity. Similarly
 316              * for all other uses of count in other wait guards.
 317              */
 318             while (count.get() == capacity) {
 319                 notFull.await();
 320             }
 321             enqueue(e);
 322             c = count.getAndIncrement();
 323             if (c + 1 < capacity)
 324                 notFull.signal();
 325         } finally {
 326             putLock.unlock();
 327         }
 328         if (c == 0)
 329             signalNotEmpty();
 330     }
 331
 332     /**
 333      * Inserts the specified element at the tail of this queue, waiting if
 334      * necessary up to the specified wait time for space to become available.
 335      *
 336      * @return {@code true} if successful, or {@code false} if
 337      *         the specified waiting time elapses before space is available.
 338      * @throws InterruptedException {@inheritDoc}
 339      * @throws NullPointerException {@inheritDoc}
 340      */
 341     public boolean offer(E e, long timeout, TimeUnit unit)
 342         throws InterruptedException {
 343
 344         if (e == null) throw new NullPointerException();
 345         long nanos = unit.toNanos(timeout);
 346         int c = -1;
 347         final ReentrantLock putLock = this.putLock;
 348         final AtomicInteger count = this.count;
 349         putLock.lockInterruptibly();
 350         try {
 351             while (count.get() == capacity) {
 352                 if (nanos <= 0)
 353                     return false;
 354                 nanos = notFull.awaitNanos(nanos);
 355             }
 356             enqueue(e);
 357             c = count.getAndIncrement();
 358             if (c + 1 < capacity)
 359                 notFull.signal();
 360         } finally {
 361             putLock.unlock();
 362         }
 363         if (c == 0)
 364             signalNotEmpty();
 365         return true;
 366     }
 367
 368     /**
 369      * Inserts the specified element at the tail of this queue if it is
 370      * possible to do so immediately without exceeding the queue's capacity,
 371      * returning {@code true} upon success and {@code false} if this queue
 372      * is full.
 373      * When using a capacity-restricted queue, this method is generally
 374      * preferable to method {@link BlockingQueue#add add}, which can fail to
 375      * insert an element only by throwing an exception.
 376      *
 377      * @throws NullPointerException if the specified element is null
 378      */
 379     public boolean offer(E e) {
 380         if (e == null) throw new NullPointerException();
 381         final AtomicInteger count = this.count;
 382         if (count.get() == capacity)
 383             return false;
 384         int c = -1;
 385         final ReentrantLock putLock = this.putLock;
 386         putLock.lock();
 387         try {
 388             if (count.get() < capacity) {
 389                 enqueue(e);
 390                 c = count.getAndIncrement();
 391                 if (c + 1 < capacity)
 392                     notFull.signal();
 393             }
 394         } finally {
 395             putLock.unlock();
 396         }
 397         if (c == 0)
 398             signalNotEmpty();
 399         return c >= 0;
 400     }
 401
 402
 403     public E take() throws InterruptedException {
 404         E x;
 405         int c = -1;
 406         final AtomicInteger count = this.count;
 407         final ReentrantLock takeLock = this.takeLock;
 408         takeLock.lockInterruptibly();
 409         try {
 410             while (count.get() == 0) {
 411                 notEmpty.await();
 412             }
 413             x = dequeue();
 414             c = count.getAndDecrement();
 415             if (c > 1)
 416                 notEmpty.signal();
 417         } finally {
 418             takeLock.unlock();
 419         }
 420         if (c == capacity)
 421             signalNotFull();
 422         return x;
 423     }
 424
 425     public E poll(long timeout, TimeUnit unit) throws InterruptedException {
 426         E x = null;
 427         int c = -1;
 428         long nanos = unit.toNanos(timeout);
 429         final AtomicInteger count = this.count;
 430         final ReentrantLock takeLock = this.takeLock;
 431         takeLock.lockInterruptibly();
 432         try {
 433             while (count.get() == 0) {
 434                 if (nanos <= 0)
 435                     return null;
 436                 nanos = notEmpty.awaitNanos(nanos);
 437             }
 438             x = dequeue();
 439             c = count.getAndDecrement();
 440             if (c > 1)
 441                 notEmpty.signal();
 442         } finally {
 443             takeLock.unlock();
 444         }
 445         if (c == capacity)
 446             signalNotFull();
 447         return x;
 448     }
 449
 450     public E poll() {
 451         final AtomicInteger count = this.count;
 452         if (count.get() == 0)
 453             return null;
 454         E x = null;
 455         int c = -1;
 456         final ReentrantLock takeLock = this.takeLock;
 457         takeLock.lock();
 458         try {
 459             if (count.get() > 0) {
 460                 x = dequeue();
 461                 c = count.getAndDecrement();
 462                 if (c > 1)
 463                     notEmpty.signal();
 464             }
 465         } finally {
 466             takeLock.unlock();
 467         }
 468         if (c == capacity)
 469             signalNotFull();
 470         return x;
 471     }
 472
 473     public E peek() {
 474         if (count.get() == 0)
 475             return null;
 476         final ReentrantLock takeLock = this.takeLock;
 477         takeLock.lock();
 478         try {
 479             Node<E> first = head.next;
 480             if (first == null)
 481                 return null;
 482             else
 483                 return first.item;
 484         } finally {
 485             takeLock.unlock();
 486         }
 487     }
 488
 489     /**
 490      * Unlinks interior Node p with predecessor trail.
 491      */
 492     void unlink(Node<E> p, Node<E> trail) {
 493         // assert isFullyLocked();
 494         // p.next is not changed, to allow iterators that are
 495         // traversing p to maintain their weak-consistency guarantee.
 496         p.item = null;
 497         trail.next = p.next;
 498         if (last == p)
 499             last = trail;
 500         if (count.getAndDecrement() == capacity)
 501             notFull.signal();
 502     }
 503
 504     /**
 505      * Removes a single instance of the specified element from this queue,
 506      * if it is present.  More formally, removes an element {@code e} such
 507      * that {@code o.equals(e)}, if this queue contains one or more such
 508      * elements.
 509      * Returns {@code true} if this queue contained the specified element
 510      * (or equivalently, if this queue changed as a result of the call).
 511      *
 512      * @param o element to be removed from this queue, if present
 513      * @return {@code true} if this queue changed as a result of the call
 514      */
 515     public boolean remove(Object o) {
 516         if (o == null) return false;
 517         fullyLock();
 518         try {
 519             for (Node<E> trail = head, p = trail.next;
 520                  p != null;
 521                  trail = p, p = p.next) {
 522                 if (o.equals(p.item)) {
 523                     unlink(p, trail);
 524                     return true;
 525                 }
 526             }
 527             return false;
 528         } finally {
 529             fullyUnlock();
 530         }
 531     }
 532
 533     /**
 534      * Returns an array containing all of the elements in this queue, in
 535      * proper sequence.
 536      *
 537      * <p>The returned array will be "safe" in that no references to it are
 538      * maintained by this queue.  (In other words, this method must allocate
 539      * a new array).  The caller is thus free to modify the returned array.
 540      *
 541      * <p>This method acts as bridge between array-based and collection-based
 542      * APIs.
 543      *
 544      * @return an array containing all of the elements in this queue
 545      */
 546     public Object[] toArray() {
 547         fullyLock();
 548         try {
 549             int size = count.get();
 550             Object[] a = new Object[size];
 551             int k = 0;
 552             for (Node<E> p = head.next; p != null; p = p.next)
 553                 a[k++] = p.item;
 554             return a;
 555         } finally {
 556             fullyUnlock();
 557         }
 558     }
 559
 560     /**
 561      * Returns an array containing all of the elements in this queue, in
 562      * proper sequence; the runtime type of the returned array is that of
 563      * the specified array.  If the queue fits in the specified array, it
 564      * is returned therein.  Otherwise, a new array is allocated with the
 565      * runtime type of the specified array and the size of this queue.
 566      *
 567      * <p>If this queue fits in the specified array with room to spare
 568      * (i.e., the array has more elements than this queue), the element in
 569      * the array immediately following the end of the queue is set to
 570      * {@code null}.
 571      *
 572      * <p>Like the {@link #toArray()} method, this method acts as bridge between
 573      * array-based and collection-based APIs.  Further, this method allows
 574      * precise control over the runtime type of the output array, and may,
 575      * under certain circumstances, be used to save allocation costs.
 576      *
 577      * <p>Suppose {@code x} is a queue known to contain only strings.
 578      * The following code can be used to dump the queue into a newly
 579      * allocated array of {@code String}:
 580      *
 581      * <pre>
 582      *     String[] y = x.toArray(new String[0]);</pre>
 583      *
 584      * Note that {@code toArray(new Object[0])} is identical in function to
 585      * {@code toArray()}.
 586      *
 587      * @param a the array into which the elements of the queue are to
 588      *          be stored, if it is big enough; otherwise, a new array of the
 589      *          same runtime type is allocated for this purpose
 590      * @return an array containing all of the elements in this queue
 591      * @throws ArrayStoreException if the runtime type of the specified array
 592      *         is not a supertype of the runtime type of every element in
 593      *         this queue
 594      * @throws NullPointerException if the specified array is null
 595      */
 596     @SuppressWarnings("unchecked")
 597     public <T> T[] toArray(T[] a) {
 598         fullyLock();
 599         try {
 600             int size = count.get();
 601             if (a.length < size)
 602                 a = (T[])java.lang.reflect.Array.newInstance
 603                     (a.getClass().getComponentType(), size);
 604
 605             int k = 0;
 606             for (Node<E> p = head.next; p != null; p = p.next)
 607                 a[k++] = (T)p.item;
 608             if (a.length > k)
 609                 a[k] = null;
 610             return a;
 611         } finally {
 612             fullyUnlock();
 613         }
 614     }
 615
 616     public String toString() {
 617         fullyLock();
 618         try {
 619             return super.toString();
 620         } finally {
 621             fullyUnlock();
 622         }
 623     }
 624
 625     /**
 626      * Atomically removes all of the elements from this queue.
 627      * The queue will be empty after this call returns.
 628      */
 629     public void clear() {
 630         fullyLock();
 631         try {
 632             for (Node<E> p, h = head; (p = h.next) != null; h = p) {
 633                 h.next = h;
 634                 p.item = null;
 635             }
 636             head = last;
 637             // assert head.item == null && head.next == null;
 638             if (count.getAndSet(0) == capacity)
 639                 notFull.signal();
 640         } finally {
 641             fullyUnlock();
 642         }
 643     }
 644
 645     /**
 646      * @throws UnsupportedOperationException {@inheritDoc}
 647      * @throws ClassCastException            {@inheritDoc}
 648      * @throws NullPointerException          {@inheritDoc}
 649      * @throws IllegalArgumentException      {@inheritDoc}
 650      */
 651     public int drainTo(Collection<? super E> c) {
 652         return drainTo(c, Integer.MAX_VALUE);
 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         boolean signalNotFull = false;
 667         final ReentrantLock takeLock = this.takeLock;
 668         takeLock.lock();
 669         try {
 670             int n = Math.min(maxElements, count.get());
 671             // count.get provides visibility to first n Nodes
 672             Node<E> h = head;
 673             int i = 0;
 674             try {
 675                 while (i < n) {
 676                     Node<E> p = h.next;
 677                     c.add(p.item);
 678                     p.item = null;
 679                     h.next = h;
 680                     h = p;
 681                     ++i;
 682                 }
 683                 return n;
 684             } finally {
 685                 // Restore invariants even if c.add() threw
 686                 if (i > 0) {
 687                     // assert h.item == null;
 688                     head = h;
 689                     signalNotFull = (count.getAndAdd(-i) == capacity);
 690                 }
 691             }
 692         } finally {
 693             takeLock.unlock();
 694             if (signalNotFull)
 695                 signalNotFull();
 696         }
 697     }
 698
 699     /**
 700      * Returns an iterator over the elements in this queue in proper sequence.
 701      * The returned {@code Iterator} is a "weakly consistent" iterator that
 702      * will never throw {@link java.util.ConcurrentModificationException
 703      * ConcurrentModificationException},
 704      * and guarantees to traverse elements as they existed upon
 705      * construction of the iterator, and may (but is not guaranteed to)
 706      * reflect any modifications subsequent to construction.
 707      *
 708      * @return an iterator over the elements in this queue in proper sequence
 709      */
 710     public Iterator<E> iterator() {
 711       return new Itr();
 712     }
 713
 714     private class Itr implements Iterator<E> {
 715         /*
 716          * Basic weakly-consistent iterator.  At all times hold the next
 717          * item to hand out so that if hasNext() reports true, we will
 718          * still have it to return even if lost race with a take etc.
 719          */
 720         private Node<E> current;
 721         private Node<E> lastRet;
 722         private E currentElement;
 723
 724         Itr() {
 725             fullyLock();
 726             try {
 727                 current = head.next;
 728                 if (current != null)
 729                     currentElement = current.item;
 730             } finally {
 731                 fullyUnlock();
 732             }
 733         }
 734
 735         public boolean hasNext() {
 736             return current != null;
 737         }
 738
 739         /**
 740          * Returns the next live successor of p, or null if no such.
 741          *
 742          * Unlike other traversal methods, iterators need to handle both:
 743          * - dequeued nodes (p.next == p)
 744          * - (possibly multiple) interior removed nodes (p.item == null)
 745          */
 746         private Node<E> nextNode(Node<E> p) {
 747             for (;;) {
 748                 Node<E> s = p.next;
 749                 if (s == p)
 750                     return head.next;
 751                 if (s == null || s.item != null)
 752                     return s;
 753                 p = s;
 754             }
 755         }
 756
 757         public E next() {
 758             fullyLock();
 759             try {
 760                 if (current == null)
 761                     throw new NoSuchElementException();
 762                 E x = currentElement;
 763                 lastRet = current;
 764                 current = nextNode(current);
 765                 currentElement = (current == null) ? null : current.item;
 766                 return x;
 767             } finally {
 768                 fullyUnlock();
 769             }
 770         }
 771
 772         public void remove() {
 773             if (lastRet == null)
 774                 throw new IllegalStateException();
 775             fullyLock();
 776             try {
 777                 Node<E> node = lastRet;
 778                 lastRet = null;
 779                 for (Node<E> trail = head, p = trail.next;
 780                      p != null;
 781                      trail = p, p = p.next) {
 782                     if (p == node) {
 783                         unlink(p, trail);
 784                         break;
 785                     }
 786                 }
 787             } finally {
 788                 fullyUnlock();
 789             }
 790         }
 791     }
 792
 793     /**
 794      * Save the state to a stream (that is, serialize it).
 795      *
 796      * @serialData The capacity is emitted (int), followed by all of
 797      * its elements (each an {@code Object}) in the proper order,
 798      * followed by a null
 799      * @param s the stream
 800      */
 801     private void writeObject(java.io.ObjectOutputStream s)
 802         throws java.io.IOException {
 803
 804         fullyLock();
 805         try {
 806             // Write out any hidden stuff, plus capacity
 807             s.defaultWriteObject();
 808
 809             // Write out all elements in the proper order.
 810             for (Node<E> p = head.next; p != null; p = p.next)
 811                 s.writeObject(p.item);
 812
 813             // Use trailing null as sentinel
 814             s.writeObject(null);
 815         } finally {
 816             fullyUnlock();
 817         }
 818     }
 819
 820     /**
 821      * Reconstitute this queue instance from a stream (that is,
 822      * deserialize it).
 823      *
 824      * @param s the stream
 825      */
 826     private void readObject(java.io.ObjectInputStream s)
 827         throws java.io.IOException, ClassNotFoundException {
 828         // Read in capacity, and any hidden stuff
 829         s.defaultReadObject();
 830
 831         count.set(0);
 832         last = head = new Node<E>(null);
 833
 834         // Read in all elements and place in queue
 835         for (;;) {
 836             @SuppressWarnings("unchecked")
 837             E item = (E)s.readObject();
 838             if (item == null)
 839                 break;
 840             add(item);
 841         }
 842     }
 843 }