7161229: PriorityBlockingQueue keeps hard reference to last removed element

Reviewed-by: dholmes, forax, alanb
Contributed-by: NDoug Lea <dl@cs.oswego.edu>

src/share/classes/java/util/concurrent/PriorityBlockingQueue.java

@@ -35,7 +35,8 @@
 
 package java.util.concurrent;
 
-import java.util.concurrent.locks.*;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.*;
 
 /**
@@ -111,7 +112,7 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
      * java.util.PriorityQueue operations within a lock, as was done
      * in a previous version of this class. To maintain
      * interoperability, a plain PriorityQueue is still used during
-     * serialization, which maintains compatibility at the espense of
+     * serialization, which maintains compatibility at the expense of
      * transiently doubling overhead.
      */
 
@@ -308,14 +309,13 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
     /**
      * Mechanics for poll().  Call only while holding lock.
      */
-    private E extract() {
-        E result;
+    private E dequeue() {
         int n = size - 1;
         if (n < 0)
-            result = null;
+            return null;
         else {
             Object[] array = queue;
-            result = (E) array[0];
+            E result = (E) array[0];
             E x = (E) array[n];
             array[n] = null;
             Comparator<? super E> cmp = comparator;
@@ -324,8 +324,8 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
             else
                 siftDownUsingComparator(0, x, array, n, cmp);
             size = n;
+            return result;
         }
-        return result;
     }
 
     /**
@@ -382,39 +382,43 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
      */
     private static <T> void siftDownComparable(int k, T x, Object[] array,
                                                int n) {
-        Comparable<? super T> key = (Comparable<? super T>)x;
-        int half = n >>> 1;           // loop while a non-leaf
-        while (k < half) {
-            int child = (k << 1) + 1; // assume left child is least
-            Object c = array[child];
-            int right = child + 1;
-            if (right < n &&
-                ((Comparable<? super T>) c).compareTo((T) array[right]) > 0)
-                c = array[child = right];
-            if (key.compareTo((T) c) <= 0)
-                break;
-            array[k] = c;
-            k = child;
+        if (n > 0) {
+            Comparable<? super T> key = (Comparable<? super T>)x;
+            int half = n >>> 1;           // loop while a non-leaf
+            while (k < half) {
+                int child = (k << 1) + 1; // assume left child is least
+                Object c = array[child];
+                int right = child + 1;
+                if (right < n &&
+                    ((Comparable<? super T>) c).compareTo((T) array[right]) > 0)
+                    c = array[child = right];
+                if (key.compareTo((T) c) <= 0)
+                    break;
+                array[k] = c;
+                k = child;
+            }
+            array[k] = key;
         }
-        array[k] = key;
     }
 
     private static <T> void siftDownUsingComparator(int k, T x, Object[] array,
                                                     int n,
                                                     Comparator<? super T> cmp) {
-        int half = n >>> 1;
-        while (k < half) {
-            int child = (k << 1) + 1;
-            Object c = array[child];
-            int right = child + 1;
-            if (right < n && cmp.compare((T) c, (T) array[right]) > 0)
-                c = array[child = right];
-            if (cmp.compare(x, (T) c) <= 0)
-                break;
-            array[k] = c;
-            k = child;
+        if (n > 0) {
+            int half = n >>> 1;
+            while (k < half) {
+                int child = (k << 1) + 1;
+                Object c = array[child];
+                int right = child + 1;
+                if (right < n && cmp.compare((T) c, (T) array[right]) > 0)
+                    c = array[child = right];
+                if (cmp.compare(x, (T) c) <= 0)
+                    break;
+                array[k] = c;
+                k = child;
+            }
+            array[k] = x;
         }
-        array[k] = x;
     }
 
     /**
@@ -520,13 +524,11 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
     public E poll() {
         final ReentrantLock lock = this.lock;
         lock.lock();
-        E result;
         try {
-            result = extract();
+            return dequeue();
         } finally {
             lock.unlock();
         }
-        return result;
     }
 
     public E take() throws InterruptedException {
@@ -534,7 +536,7 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
         lock.lockInterruptibly();
         E result;
         try {
-            while ( (result = extract()) == null)
+            while ( (result = dequeue()) == null)
                 notEmpty.await();
         } finally {
             lock.unlock();
@@ -548,7 +550,7 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
         lock.lockInterruptibly();
         E result;
         try {
-            while ( (result = extract()) == null && nanos > 0)
+            while ( (result = dequeue()) == null && nanos > 0)
                 nanos = notEmpty.awaitNanos(nanos);
         } finally {
             lock.unlock();
@@ -559,13 +561,11 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
     public E peek() {
         final ReentrantLock lock = this.lock;
         lock.lock();
-        E result;
         try {
-            result = size > 0 ? (E) queue[0] : null;
+            return (size == 0) ? null : (E) queue[0];
         } finally {
             lock.unlock();
         }
-        return result;
     }
 
     /**
@@ -649,32 +649,28 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
      * @return {@code true} if this queue changed as a result of the call
      */
     public boolean remove(Object o) {
-        boolean removed = false;
         final ReentrantLock lock = this.lock;
         lock.lock();
         try {
             int i = indexOf(o);
-            if (i != -1) {
-                removeAt(i);
-                removed = true;
-            }
+            if (i == -1)
+                return false;
+            removeAt(i);
+            return true;
         } finally {
             lock.unlock();
         }
-        return removed;
     }
 
-
     /**
      * Identity-based version for use in Itr.remove
      */
-    private void removeEQ(Object o) {
+    void removeEQ(Object o) {
         final ReentrantLock lock = this.lock;
         lock.lock();
         try {
             Object[] array = queue;
-            int n = size;
-            for (int i = 0; i < n; i++) {
+            for (int i = 0, n = size; i < n; i++) {
                 if (o == array[i]) {
                     removeAt(i);
                     break;
@@ -694,15 +690,13 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
      * @return {@code true} if this queue contains the specified element
      */
     public boolean contains(Object o) {
-        int index;
         final ReentrantLock lock = this.lock;
         lock.lock();
         try {
-            index = indexOf(o);
+            return indexOf(o) != -1;
         } finally {
             lock.unlock();
         }
-        return index != -1;
     }
 
     /**
@@ -728,7 +722,6 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
         }
     }
 
-
     public String toString() {
         final ReentrantLock lock = this.lock;
         lock.lock();
@@ -739,7 +732,7 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
             StringBuilder sb = new StringBuilder();
             sb.append('[');
             for (int i = 0; i < n; ++i) {
-                E e = (E)queue[i];
+                Object e = queue[i];
                 sb.append(e == this ? "(this Collection)" : e);
                 if (i != n - 1)
                     sb.append(',').append(' ');
@@ -757,23 +750,7 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
      * @throws IllegalArgumentException      {@inheritDoc}
      */
     public int drainTo(Collection<? super E> c) {
-        if (c == null)
-            throw new NullPointerException();
-        if (c == this)
-            throw new IllegalArgumentException();
-        final ReentrantLock lock = this.lock;
-        lock.lock();
-        try {
-            int n = 0;
-            E e;
-            while ( (e = extract()) != null) {
-                c.add(e);
-                ++n;
-            }
-            return n;
-        } finally {
-            lock.unlock();
-        }
+        return drainTo(c, Integer.MAX_VALUE);
     }
 
     /**
@@ -792,11 +769,10 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
         final ReentrantLock lock = this.lock;
         lock.lock();
         try {
-            int n = 0;
-            E e;
-            while (n < maxElements && (e = extract()) != null) {
-                c.add(e);
-                ++n;
+            int n = Math.min(size, maxElements);
+            for (int i = 0; i < n; i++) {
+                c.add((E) queue[0]); // In this order, in case add() throws.
+                dequeue();
             }
             return n;
         } finally {
@@ -844,8 +820,7 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
      * The following code can be used to dump the queue into a newly
      * allocated array of {@code String}:
      *
-     * <pre>
-     *     String[] y = x.toArray(new String[0]);</pre>
+     *  <pre> {@code String[] y = x.toArray(new String[0]);}</pre>
      *
      * Note that {@code toArray(new Object[0])} is identical in function to
      * {@code toArray()}.
@@ -898,7 +873,7 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
      */
     final class Itr implements Iterator<E> {
         final Object[] array; // Array of all elements
-        int cursor;           // index of next element to return;
+        int cursor;           // index of next element to return
         int lastRet;          // index of last element, or -1 if no such
 
         Itr(Object[] array) {
@@ -926,17 +901,18 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
     }
 
     /**
-     * Saves the state to a stream (that is, serializes it).  For
-     * compatibility with previous version of this class,
-     * elements are first copied to a java.util.PriorityQueue,
-     * which is then serialized.
+     * Saves this queue to a stream (that is, serializes it).
+     *
+     * For compatibility with previous version of this class, elements
+     * are first copied to a java.util.PriorityQueue, which is then
+     * serialized.
      */
     private void writeObject(java.io.ObjectOutputStream s)
         throws java.io.IOException {
         lock.lock();
         try {
-            int n = size; // avoid zero capacity argument
-            q = new PriorityQueue<E>(n == 0 ? 1 : n, comparator);
+            // avoid zero capacity argument
+            q = new PriorityQueue<E>(Math.max(size, 1), comparator);
             q.addAll(this);
             s.defaultWriteObject();
         } finally {
@@ -946,10 +922,7 @@ public class PriorityBlockingQueue<E> extends AbstractQueue<E>
     }
 
     /**
-     * Reconstitutes the {@code PriorityBlockingQueue} instance from a stream
-     * (that is, deserializes it).
-     *
-     * @param s the stream
+     * Reconstitutes this queue from a stream (that is, deserializes it).
      */
     private void readObject(java.io.ObjectInputStream s)
         throws java.io.IOException, ClassNotFoundException {

test/java/util/concurrent/BlockingQueue/LastElement.java

 /*
- * Copyright (c) 2005, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2005, 2012, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -23,7 +23,7 @@
 
 /*
  * @test
- * @bug 6215625
+ * @bug 6215625 7161229
  * @summary Check correct behavior when last element is removed.
  * @author Martin Buchholz
  */
@@ -38,9 +38,7 @@ public class LastElement {
         testQueue(new ArrayBlockingQueue<Integer>(10, true));
         testQueue(new ArrayBlockingQueue<Integer>(10, false));
         testQueue(new LinkedTransferQueue<Integer>());
-
-        System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
-        if (failed > 0) throw new Exception("Some tests failed");
+        testQueue(new PriorityBlockingQueue<Integer>());
     }
 
     void testQueue(BlockingQueue<Integer> q) throws Throwable {
@@ -59,6 +57,7 @@ public class LastElement {
         try {check(q.take() == three);}
         catch (Throwable t) {unexpected(t);}
         check(q.isEmpty() && q.size() == 0);
+        check(noRetention(q));
 
         // iterator().remove()
         q.clear();
@@ -77,6 +76,26 @@ public class LastElement {
         check(q.isEmpty() && q.size() == 0);
     }
 
+    boolean noRetention(BlockingQueue<?> q) {
+        if (q instanceof PriorityBlockingQueue) {
+            PriorityBlockingQueue<?> pbq = (PriorityBlockingQueue) q;
+            try {
+                java.lang.reflect.Field queue =
+                    PriorityBlockingQueue.class.getDeclaredField("queue");
+                queue.setAccessible(true);
+                Object[] a = (Object[]) queue.get(pbq);
+                return a[0] == null;
+            }
+            catch (NoSuchFieldException e) {
+                unexpected(e);
+            }
+            catch (IllegalAccessException e) {
+                // ignore - security manager must be installed
+            }
+        }
+        return true;
+    }
+
     //--------------------- Infrastructure ---------------------------
     volatile int passed = 0, failed = 0;
     void pass() {passed++;}