--- jsr166/src/jsr166y/LinkedTransferQueue.java 2009/07/24 23:48:26 1.25 +++ jsr166/src/jsr166y/LinkedTransferQueue.java 2009/07/30 13:30:19 1.33 @@ -5,11 +5,15 @@ */ package jsr166y; + import java.util.concurrent.*; -import java.util.concurrent.locks.*; -import java.util.concurrent.atomic.*; -import java.util.*; -import java.io.*; + +import java.util.AbstractQueue; +import java.util.Collection; +import java.util.Iterator; +import java.util.NoSuchElementException; +import java.util.concurrent.locks.LockSupport; +import java.util.concurrent.atomic.AtomicReference; /** * An unbounded {@linkplain TransferQueue} based on linked nodes. @@ -110,17 +114,46 @@ public class LinkedTransferQueue exte this.isData = isData; } - @SuppressWarnings("rawtypes") - static final AtomicReferenceFieldUpdater - nextUpdater = AtomicReferenceFieldUpdater.newUpdater - (Node.class, Node.class, "next"); + // Unsafe mechanics + + private static final sun.misc.Unsafe UNSAFE = getUnsafe(); + private static final long nextOffset = + objectFieldOffset(UNSAFE, "next", Node.class); final boolean casNext(Node cmp, Node val) { - return nextUpdater.compareAndSet(this, cmp, val); + return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val); } final void clearNext() { - nextUpdater.lazySet(this, this); + UNSAFE.putOrderedObject(this, nextOffset, this); + } + + /** + * Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package. + * Replace with a simple call to Unsafe.getUnsafe when integrating + * into a jdk. + * + * @return a sun.misc.Unsafe + */ + private static sun.misc.Unsafe getUnsafe() { + try { + return sun.misc.Unsafe.getUnsafe(); + } catch (SecurityException se) { + try { + return java.security.AccessController.doPrivileged + (new java.security + .PrivilegedExceptionAction() { + public sun.misc.Unsafe run() throws Exception { + java.lang.reflect.Field f = sun.misc + .Unsafe.class.getDeclaredField("theUnsafe"); + f.setAccessible(true); + return (sun.misc.Unsafe) f.get(null); + }}); + } catch (java.security.PrivilegedActionException e) { + throw new RuntimeException("Could not initialize intrinsics", + e.getCause()); + } + } } private static final long serialVersionUID = -3375979862319811754L; @@ -437,12 +470,20 @@ public class LinkedTransferQueue exte addAll(c); } + /** + * @throws InterruptedException {@inheritDoc} + * @throws NullPointerException {@inheritDoc} + */ public void put(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); if (Thread.interrupted()) throw new InterruptedException(); xfer(e, NOWAIT, 0); } + /** + * @throws InterruptedException {@inheritDoc} + * @throws NullPointerException {@inheritDoc} + */ public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException { if (e == null) throw new NullPointerException(); @@ -451,18 +492,28 @@ public class LinkedTransferQueue exte return true; } + /** + * @throws NullPointerException {@inheritDoc} + */ public boolean offer(E e) { if (e == null) throw new NullPointerException(); xfer(e, NOWAIT, 0); return true; } + /** + * @throws NullPointerException {@inheritDoc} + */ public boolean add(E e) { if (e == null) throw new NullPointerException(); xfer(e, NOWAIT, 0); return true; } + /** + * @throws InterruptedException {@inheritDoc} + * @throws NullPointerException {@inheritDoc} + */ public void transfer(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); if (xfer(e, WAIT, 0) == null) { @@ -471,6 +522,10 @@ public class LinkedTransferQueue exte } } + /** + * @throws InterruptedException {@inheritDoc} + * @throws NullPointerException {@inheritDoc} + */ public boolean tryTransfer(E e, long timeout, TimeUnit unit) throws InterruptedException { if (e == null) throw new NullPointerException(); @@ -481,11 +536,17 @@ public class LinkedTransferQueue exte throw new InterruptedException(); } + /** + * @throws NullPointerException {@inheritDoc} + */ public boolean tryTransfer(E e) { if (e == null) throw new NullPointerException(); return fulfill(e) != null; } + /** + * @throws InterruptedException {@inheritDoc} + */ public E take() throws InterruptedException { Object e = xfer(null, WAIT, 0); if (e != null) @@ -494,6 +555,9 @@ public class LinkedTransferQueue exte throw new InterruptedException(); } + /** + * @throws InterruptedException {@inheritDoc} + */ public E poll(long timeout, TimeUnit unit) throws InterruptedException { Object e = xfer(null, TIMEOUT, unit.toNanos(timeout)); if (e != null || !Thread.interrupted()) @@ -505,6 +569,10 @@ public class LinkedTransferQueue exte return fulfill(null); } + /** + * @throws NullPointerException {@inheritDoc} + * @throws IllegalArgumentException {@inheritDoc} + */ public int drainTo(Collection c) { if (c == null) throw new NullPointerException(); @@ -519,6 +587,10 @@ public class LinkedTransferQueue exte return n; } + /** + * @throws NullPointerException {@inheritDoc} + * @throws IllegalArgumentException {@inheritDoc} + */ public int drainTo(Collection c, int maxElements) { if (c == null) throw new NullPointerException(); @@ -563,7 +635,6 @@ public class LinkedTransferQueue exte } } - public Iterator iterator() { return new Itr(); } @@ -578,42 +649,39 @@ public class LinkedTransferQueue exte class Itr implements Iterator { Node next; // node to return next Node pnext; // predecessor of next - Node snext; // successor of next Node curr; // last returned node, for remove() Node pcurr; // predecessor of curr, for remove() - E nextItem; // Cache of next item, once committed to in next + E nextItem; // Cache of next item, once committed to in next Itr() { - findNext(); + advance(); } /** - * Ensures next points to next valid node, or null if none. + * Moves to next valid node and returns item to return for + * next(), or null if no such. */ - void findNext() { + private E advance() { + pcurr = pnext; + curr = next; + E item = nextItem; + for (;;) { - Node pred = pnext; - Node q = next; - if (pred == null || pred == q) { - pred = traversalHead(); - q = pred.next; - } - if (q == null || !q.isData) { + pnext = next == null ? traversalHead() : next; + next = pnext.next; + if (next == pnext) { next = null; - return; + continue; // restart } - Object x = q.get(); - Node s = q.next; - if (x != null && q != x && q != s) { + if (next == null) + break; + Object x = next.get(); + if (x != null && x != next) { nextItem = (E) x; - snext = s; - pnext = pred; - next = q; - return; + break; } - pnext = q; - next = s; } + return item; } public boolean hasNext() { @@ -621,14 +689,9 @@ public class LinkedTransferQueue exte } public E next() { - if (next == null) throw new NoSuchElementException(); - pcurr = pnext; - curr = next; - pnext = next; - next = snext; - E x = nextItem; - findNext(); - return x; + if (next == null) + throw new NoSuchElementException(); + return advance(); } public void remove() { @@ -698,32 +761,44 @@ public class LinkedTransferQueue exte * @return the number of elements in this queue */ public int size() { - int count = 0; - Node h = traversalHead(); - for (Node p = h.next; p != null && p.isData; p = p.next) { - Object x = p.get(); - if (x != null && x != p) { - if (++count == Integer.MAX_VALUE) // saturated + for (;;) { + int count = 0; + Node pred = traversalHead(); + for (;;) { + Node q = pred.next; + if (q == pred) // restart break; + if (q == null || !q.isData) + return count; + Object x = q.get(); + if (x != null && x != q) { + if (++count == Integer.MAX_VALUE) // saturated + return count; + } + pred = q; } } - return count; } public int getWaitingConsumerCount() { - int count = 0; - Node h = traversalHead(); - for (Node p = h.next; p != null && !p.isData; p = p.next) { - if (p.get() == null) { - if (++count == Integer.MAX_VALUE) + // converse of size -- count valid non-data nodes + for (;;) { + int count = 0; + Node pred = traversalHead(); + for (;;) { + Node q = pred.next; + if (q == pred) // restart break; + if (q == null || q.isData) + return count; + Object x = q.get(); + if (x == null) { + if (++count == Integer.MAX_VALUE) // saturated + return count; + } + pred = q; } } - return count; - } - - public int remainingCapacity() { - return Integer.MAX_VALUE; } public boolean remove(Object o) { @@ -733,10 +808,10 @@ public class LinkedTransferQueue exte Node pred = traversalHead(); for (;;) { Node q = pred.next; - if (q == null || !q.isData) - return false; if (q == pred) // restart break; + if (q == null || !q.isData) + return false; Object x = q.get(); if (x != null && x != q && o.equals(x) && q.compareAndSet(x, q)) { @@ -748,6 +823,10 @@ public class LinkedTransferQueue exte } } + public int remainingCapacity() { + return Integer.MAX_VALUE; + } + /** * Save the state to a stream (that is, serialize it). * @@ -794,16 +873,49 @@ public class LinkedTransferQueue exte new PaddedAtomicReference>(null)); } - // Unsafe mechanics for jsr166y 3rd party package. + // Unsafe mechanics + + private static final sun.misc.Unsafe UNSAFE = getUnsafe(); + private static final long headOffset = + objectFieldOffset(UNSAFE, "head", LinkedTransferQueue.class); + private static final long tailOffset = + objectFieldOffset(UNSAFE, "tail", LinkedTransferQueue.class); + private static final long cleanMeOffset = + objectFieldOffset(UNSAFE, "cleanMe", LinkedTransferQueue.class); + + + static long objectFieldOffset(sun.misc.Unsafe UNSAFE, + String field, Class klazz) { + try { + return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field)); + } catch (NoSuchFieldException e) { + // Convert Exception to corresponding Error + NoSuchFieldError error = new NoSuchFieldError(field); + error.initCause(e); + throw error; + } + } + + /** + * Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package. + * Replace with a simple call to Unsafe.getUnsafe when integrating + * into a jdk. + * + * @return a sun.misc.Unsafe + */ private static sun.misc.Unsafe getUnsafe() { try { return sun.misc.Unsafe.getUnsafe(); } catch (SecurityException se) { try { return java.security.AccessController.doPrivileged - (new java.security.PrivilegedExceptionAction() { + (new java.security + .PrivilegedExceptionAction() { public sun.misc.Unsafe run() throws Exception { - return getUnsafeByReflection(); + java.lang.reflect.Field f = sun.misc + .Unsafe.class.getDeclaredField("theUnsafe"); + f.setAccessible(true); + return (sun.misc.Unsafe) f.get(null); }}); } catch (java.security.PrivilegedActionException e) { throw new RuntimeException("Could not initialize intrinsics", @@ -811,32 +923,4 @@ public class LinkedTransferQueue exte } } } - - private static sun.misc.Unsafe getUnsafeByReflection() - throws NoSuchFieldException, IllegalAccessException { - java.lang.reflect.Field f = - sun.misc.Unsafe.class.getDeclaredField("theUnsafe"); - f.setAccessible(true); - return (sun.misc.Unsafe) f.get(null); - } - - private static long fieldOffset(String fieldName, Class klazz) { - try { - return UNSAFE.objectFieldOffset(klazz.getDeclaredField(fieldName)); - } catch (NoSuchFieldException e) { - // Convert Exception to Error - NoSuchFieldError error = new NoSuchFieldError(fieldName); - error.initCause(e); - throw error; - } - } - - private static final sun.misc.Unsafe UNSAFE = getUnsafe(); - static final long headOffset = - fieldOffset("head", LinkedTransferQueue.class); - static final long tailOffset = - fieldOffset("tail", LinkedTransferQueue.class); - static final long cleanMeOffset = - fieldOffset("cleanMe", LinkedTransferQueue.class); - }