6785442: ConcurrentLinkedQueue.remove() and poll() can both remove the same element
6493942: ConcurrentLinkedQueue.remove sometimes very slow New algorithm for handling concurrent linked lists Reviewed-by: martin
This commit is contained in:
Doug Lea
parent
e364d187ea
commit
df5da31f55
@ -34,9 +34,13 @@
|
||||
*/
|
||||
|
||||
package java.util.concurrent;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.atomic.*;
|
||||
|
||||
import java.util.AbstractQueue;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Iterator;
|
||||
import java.util.NoSuchElementException;
|
||||
import java.util.Queue;
|
||||
|
||||
/**
|
||||
* An unbounded thread-safe {@linkplain Queue queue} based on linked nodes.
|
||||
@ -47,9 +51,9 @@ import java.util.concurrent.atomic.*;
|
||||
* queue the shortest time. New elements
|
||||
* are inserted at the tail of the queue, and the queue retrieval
|
||||
* operations obtain elements at the head of the queue.
|
||||
* A <tt>ConcurrentLinkedQueue</tt> is an appropriate choice when
|
||||
* A {@code ConcurrentLinkedQueue} is an appropriate choice when
|
||||
* many threads will share access to a common collection.
|
||||
* This queue does not permit <tt>null</tt> elements.
|
||||
* This queue does not permit {@code null} elements.
|
||||
*
|
||||
* <p>This implementation employs an efficient "wait-free"
|
||||
* algorithm based on one described in <a
|
||||
@ -57,7 +61,7 @@ import java.util.concurrent.atomic.*;
|
||||
* Fast, and Practical Non-Blocking and Blocking Concurrent Queue
|
||||
* Algorithms</a> by Maged M. Michael and Michael L. Scott.
|
||||
*
|
||||
* <p>Beware that, unlike in most collections, the <tt>size</tt> method
|
||||
* <p>Beware that, unlike in most collections, the {@code size} method
|
||||
* is <em>NOT</em> a constant-time operation. Because of the
|
||||
* asynchronous nature of these queues, determining the current number
|
||||
* of elements requires a traversal of the elements.
|
||||
@ -87,51 +91,102 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
private static final long serialVersionUID = 196745693267521676L;
|
||||
|
||||
/*
|
||||
* This is a straight adaptation of Michael & Scott algorithm.
|
||||
* For explanation, read the paper. The only (minor) algorithmic
|
||||
* difference is that this version supports lazy deletion of
|
||||
* internal nodes (method remove(Object)) -- remove CAS'es item
|
||||
* fields to null. The normal queue operations unlink but then
|
||||
* pass over nodes with null item fields. Similarly, iteration
|
||||
* methods ignore those with nulls.
|
||||
* This is a modification of the Michael & Scott algorithm,
|
||||
* adapted for a garbage-collected environment, with support for
|
||||
* interior node deletion (to support remove(Object)). For
|
||||
* explanation, read the paper.
|
||||
*
|
||||
* Also note that like most non-blocking algorithms in this
|
||||
* package, this implementation relies on the fact that in garbage
|
||||
* Note that like most non-blocking algorithms in this package,
|
||||
* this implementation relies on the fact that in garbage
|
||||
* collected systems, there is no possibility of ABA problems due
|
||||
* to recycled nodes, so there is no need to use "counted
|
||||
* pointers" or related techniques seen in versions used in
|
||||
* non-GC'ed settings.
|
||||
*
|
||||
* The fundamental invariants are:
|
||||
* - There is exactly one (last) Node with a null next reference,
|
||||
* which is CASed when enqueueing. This last Node can be
|
||||
* reached in O(1) time from tail, but tail is merely an
|
||||
* optimization - it can always be reached in O(N) time from
|
||||
* head as well.
|
||||
* - The elements contained in the queue are the non-null items in
|
||||
* Nodes that are reachable from head. CASing the item
|
||||
* reference of a Node to null atomically removes it from the
|
||||
* queue. Reachability of all elements from head must remain
|
||||
* true even in the case of concurrent modifications that cause
|
||||
* head to advance. A dequeued Node may remain in use
|
||||
* indefinitely due to creation of an Iterator or simply a
|
||||
* poll() that has lost its time slice.
|
||||
*
|
||||
* The above might appear to imply that all Nodes are GC-reachable
|
||||
* from a predecessor dequeued Node. That would cause two problems:
|
||||
* - allow a rogue Iterator to cause unbounded memory retention
|
||||
* - cause cross-generational linking of old Nodes to new Nodes if
|
||||
* a Node was tenured while live, which generational GCs have a
|
||||
* hard time dealing with, causing repeated major collections.
|
||||
* However, only non-deleted Nodes need to be reachable from
|
||||
* dequeued Nodes, and reachability does not necessarily have to
|
||||
* be of the kind understood by the GC. We use the trick of
|
||||
* linking a Node that has just been dequeued to itself. Such a
|
||||
* self-link implicitly means to advance to head.
|
||||
*
|
||||
* Both head and tail are permitted to lag. In fact, failing to
|
||||
* update them every time one could is a significant optimization
|
||||
* (fewer CASes). This is controlled by local "hops" variables
|
||||
* that only trigger helping-CASes after experiencing multiple
|
||||
* lags.
|
||||
*
|
||||
* Since head and tail are updated concurrently and independently,
|
||||
* it is possible for tail to lag behind head (why not)?
|
||||
*
|
||||
* CASing a Node's item reference to null atomically removes the
|
||||
* element from the queue. Iterators skip over Nodes with null
|
||||
* items. Prior implementations of this class had a race between
|
||||
* poll() and remove(Object) where the same element would appear
|
||||
* to be successfully removed by two concurrent operations. The
|
||||
* method remove(Object) also lazily unlinks deleted Nodes, but
|
||||
* this is merely an optimization.
|
||||
*
|
||||
* When constructing a Node (before enqueuing it) we avoid paying
|
||||
* for a volatile write to item by using lazySet instead of a
|
||||
* normal write. This allows the cost of enqueue to be
|
||||
* "one-and-a-half" CASes.
|
||||
*
|
||||
* Both head and tail may or may not point to a Node with a
|
||||
* non-null item. If the queue is empty, all items must of course
|
||||
* be null. Upon creation, both head and tail refer to a dummy
|
||||
* Node with null item. Both head and tail are only updated using
|
||||
* CAS, so they never regress, although again this is merely an
|
||||
* optimization.
|
||||
*/
|
||||
|
||||
private static class Node<E> {
|
||||
private volatile E item;
|
||||
private volatile Node<E> next;
|
||||
|
||||
private static final
|
||||
AtomicReferenceFieldUpdater<Node, Node>
|
||||
nextUpdater =
|
||||
AtomicReferenceFieldUpdater.newUpdater
|
||||
(Node.class, Node.class, "next");
|
||||
private static final
|
||||
AtomicReferenceFieldUpdater<Node, Object>
|
||||
itemUpdater =
|
||||
AtomicReferenceFieldUpdater.newUpdater
|
||||
(Node.class, Object.class, "item");
|
||||
|
||||
Node(E x) { item = x; }
|
||||
|
||||
Node(E x, Node<E> n) { item = x; next = n; }
|
||||
Node(E item) {
|
||||
// Piggyback on imminent casNext()
|
||||
lazySetItem(item);
|
||||
}
|
||||
|
||||
E getItem() {
|
||||
return item;
|
||||
}
|
||||
|
||||
boolean casItem(E cmp, E val) {
|
||||
return itemUpdater.compareAndSet(this, cmp, val);
|
||||
return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
|
||||
}
|
||||
|
||||
void setItem(E val) {
|
||||
itemUpdater.set(this, val);
|
||||
item = val;
|
||||
}
|
||||
|
||||
void lazySetItem(E val) {
|
||||
UNSAFE.putOrderedObject(this, itemOffset, val);
|
||||
}
|
||||
|
||||
void lazySetNext(Node<E> val) {
|
||||
UNSAFE.putOrderedObject(this, nextOffset, val);
|
||||
}
|
||||
|
||||
Node<E> getNext() {
|
||||
@ -139,52 +194,55 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
}
|
||||
|
||||
boolean casNext(Node<E> cmp, Node<E> val) {
|
||||
return nextUpdater.compareAndSet(this, cmp, val);
|
||||
return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
|
||||
}
|
||||
|
||||
void setNext(Node<E> val) {
|
||||
nextUpdater.set(this, val);
|
||||
}
|
||||
// Unsafe mechanics
|
||||
|
||||
private static final sun.misc.Unsafe UNSAFE =
|
||||
sun.misc.Unsafe.getUnsafe();
|
||||
private static final long nextOffset =
|
||||
objectFieldOffset(UNSAFE, "next", Node.class);
|
||||
private static final long itemOffset =
|
||||
objectFieldOffset(UNSAFE, "item", Node.class);
|
||||
}
|
||||
|
||||
private static final
|
||||
AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>
|
||||
tailUpdater =
|
||||
AtomicReferenceFieldUpdater.newUpdater
|
||||
(ConcurrentLinkedQueue.class, Node.class, "tail");
|
||||
private static final
|
||||
AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>
|
||||
headUpdater =
|
||||
AtomicReferenceFieldUpdater.newUpdater
|
||||
(ConcurrentLinkedQueue.class, Node.class, "head");
|
||||
|
||||
private boolean casTail(Node<E> cmp, Node<E> val) {
|
||||
return tailUpdater.compareAndSet(this, cmp, val);
|
||||
}
|
||||
|
||||
private boolean casHead(Node<E> cmp, Node<E> val) {
|
||||
return headUpdater.compareAndSet(this, cmp, val);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Pointer to header node, initialized to a dummy node. The first
|
||||
* actual node is at head.getNext().
|
||||
* A node from which the first live (non-deleted) node (if any)
|
||||
* can be reached in O(1) time.
|
||||
* Invariants:
|
||||
* - all live nodes are reachable from head via succ()
|
||||
* - head != null
|
||||
* - (tmp = head).next != tmp || tmp != head
|
||||
* Non-invariants:
|
||||
* - head.item may or may not be null.
|
||||
* - it is permitted for tail to lag behind head, that is, for tail
|
||||
* to not be reachable from head!
|
||||
*/
|
||||
private transient volatile Node<E> head = new Node<E>(null, null);
|
||||
private transient volatile Node<E> head = new Node<E>(null);
|
||||
|
||||
/** Pointer to last node on list **/
|
||||
/**
|
||||
* A node from which the last node on list (that is, the unique
|
||||
* node with node.next == null) can be reached in O(1) time.
|
||||
* Invariants:
|
||||
* - the last node is always reachable from tail via succ()
|
||||
* - tail != null
|
||||
* Non-invariants:
|
||||
* - tail.item may or may not be null.
|
||||
* - it is permitted for tail to lag behind head, that is, for tail
|
||||
* to not be reachable from head!
|
||||
* - tail.next may or may not be self-pointing to tail.
|
||||
*/
|
||||
private transient volatile Node<E> tail = head;
|
||||
|
||||
|
||||
/**
|
||||
* Creates a <tt>ConcurrentLinkedQueue</tt> that is initially empty.
|
||||
* Creates a {@code ConcurrentLinkedQueue} that is initially empty.
|
||||
*/
|
||||
public ConcurrentLinkedQueue() {}
|
||||
|
||||
/**
|
||||
* Creates a <tt>ConcurrentLinkedQueue</tt>
|
||||
* Creates a {@code ConcurrentLinkedQueue}
|
||||
* initially containing the elements of the given collection,
|
||||
* added in traversal order of the collection's iterator.
|
||||
* @param c the collection of elements to initially contain
|
||||
@ -201,115 +259,143 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
/**
|
||||
* Inserts the specified element at the tail of this queue.
|
||||
*
|
||||
* @return <tt>true</tt> (as specified by {@link Collection#add})
|
||||
* @return {@code true} (as specified by {@link Collection#add})
|
||||
* @throws NullPointerException if the specified element is null
|
||||
*/
|
||||
public boolean add(E e) {
|
||||
return offer(e);
|
||||
}
|
||||
|
||||
/**
|
||||
* We don't bother to update head or tail pointers if fewer than
|
||||
* HOPS links from "true" location. We assume that volatile
|
||||
* writes are significantly more expensive than volatile reads.
|
||||
*/
|
||||
private static final int HOPS = 1;
|
||||
|
||||
/**
|
||||
* Try to CAS head to p. If successful, repoint old head to itself
|
||||
* as sentinel for succ(), below.
|
||||
*/
|
||||
final void updateHead(Node<E> h, Node<E> p) {
|
||||
if (h != p && casHead(h, p))
|
||||
h.lazySetNext(h);
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the successor of p, or the head node if p.next has been
|
||||
* linked to self, which will only be true if traversing with a
|
||||
* stale pointer that is now off the list.
|
||||
*/
|
||||
final Node<E> succ(Node<E> p) {
|
||||
Node<E> next = p.getNext();
|
||||
return (p == next) ? head : next;
|
||||
}
|
||||
|
||||
/**
|
||||
* Inserts the specified element at the tail of this queue.
|
||||
*
|
||||
* @return <tt>true</tt> (as specified by {@link Queue#offer})
|
||||
* @return {@code true} (as specified by {@link Queue#offer})
|
||||
* @throws NullPointerException if the specified element is null
|
||||
*/
|
||||
public boolean offer(E e) {
|
||||
if (e == null) throw new NullPointerException();
|
||||
Node<E> n = new Node<E>(e, null);
|
||||
Node<E> n = new Node<E>(e);
|
||||
retry:
|
||||
for (;;) {
|
||||
Node<E> t = tail;
|
||||
Node<E> s = t.getNext();
|
||||
if (t == tail) {
|
||||
if (s == null) {
|
||||
if (t.casNext(s, n)) {
|
||||
casTail(t, n);
|
||||
return true;
|
||||
}
|
||||
Node<E> p = t;
|
||||
for (int hops = 0; ; hops++) {
|
||||
Node<E> next = succ(p);
|
||||
if (next != null) {
|
||||
if (hops > HOPS && t != tail)
|
||||
continue retry;
|
||||
p = next;
|
||||
} else if (p.casNext(null, n)) {
|
||||
if (hops >= HOPS)
|
||||
casTail(t, n); // Failure is OK.
|
||||
return true;
|
||||
} else {
|
||||
casTail(t, s);
|
||||
p = succ(p);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public E poll() {
|
||||
for (;;) {
|
||||
Node<E> h = head;
|
||||
Node<E> t = tail;
|
||||
Node<E> first = h.getNext();
|
||||
if (h == head) {
|
||||
if (h == t) {
|
||||
if (first == null)
|
||||
return null;
|
||||
else
|
||||
casTail(t, first);
|
||||
} else if (casHead(h, first)) {
|
||||
E item = first.getItem();
|
||||
if (item != null) {
|
||||
first.setItem(null);
|
||||
return item;
|
||||
}
|
||||
// else skip over deleted item, continue loop,
|
||||
Node<E> h = head;
|
||||
Node<E> p = h;
|
||||
for (int hops = 0; ; hops++) {
|
||||
E item = p.getItem();
|
||||
|
||||
if (item != null && p.casItem(item, null)) {
|
||||
if (hops >= HOPS) {
|
||||
Node<E> q = p.getNext();
|
||||
updateHead(h, (q != null) ? q : p);
|
||||
}
|
||||
return item;
|
||||
}
|
||||
Node<E> next = succ(p);
|
||||
if (next == null) {
|
||||
updateHead(h, p);
|
||||
break;
|
||||
}
|
||||
p = next;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
public E peek() { // same as poll except don't remove item
|
||||
public E peek() {
|
||||
Node<E> h = head;
|
||||
Node<E> p = h;
|
||||
E item;
|
||||
for (;;) {
|
||||
Node<E> h = head;
|
||||
Node<E> t = tail;
|
||||
Node<E> first = h.getNext();
|
||||
if (h == head) {
|
||||
if (h == t) {
|
||||
if (first == null)
|
||||
return null;
|
||||
else
|
||||
casTail(t, first);
|
||||
} else {
|
||||
E item = first.getItem();
|
||||
if (item != null)
|
||||
return item;
|
||||
else // remove deleted node and continue
|
||||
casHead(h, first);
|
||||
}
|
||||
item = p.getItem();
|
||||
if (item != null)
|
||||
break;
|
||||
Node<E> next = succ(p);
|
||||
if (next == null) {
|
||||
break;
|
||||
}
|
||||
p = next;
|
||||
}
|
||||
updateHead(h, p);
|
||||
return item;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the first actual (non-header) node on list. This is yet
|
||||
* another variant of poll/peek; here returning out the first
|
||||
* node, not element (so we cannot collapse with peek() without
|
||||
* introducing race.)
|
||||
* Returns the first live (non-deleted) node on list, or null if none.
|
||||
* This is yet another variant of poll/peek; here returning the
|
||||
* first node, not element. We could make peek() a wrapper around
|
||||
* first(), but that would cost an extra volatile read of item,
|
||||
* and the need to add a retry loop to deal with the possibility
|
||||
* of losing a race to a concurrent poll().
|
||||
*/
|
||||
Node<E> first() {
|
||||
Node<E> h = head;
|
||||
Node<E> p = h;
|
||||
Node<E> result;
|
||||
for (;;) {
|
||||
Node<E> h = head;
|
||||
Node<E> t = tail;
|
||||
Node<E> first = h.getNext();
|
||||
if (h == head) {
|
||||
if (h == t) {
|
||||
if (first == null)
|
||||
return null;
|
||||
else
|
||||
casTail(t, first);
|
||||
} else {
|
||||
if (first.getItem() != null)
|
||||
return first;
|
||||
else // remove deleted node and continue
|
||||
casHead(h, first);
|
||||
}
|
||||
E item = p.getItem();
|
||||
if (item != null) {
|
||||
result = p;
|
||||
break;
|
||||
}
|
||||
Node<E> next = succ(p);
|
||||
if (next == null) {
|
||||
result = null;
|
||||
break;
|
||||
}
|
||||
p = next;
|
||||
}
|
||||
updateHead(h, p);
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Returns <tt>true</tt> if this queue contains no elements.
|
||||
* Returns {@code true} if this queue contains no elements.
|
||||
*
|
||||
* @return <tt>true</tt> if this queue contains no elements
|
||||
* @return {@code true} if this queue contains no elements
|
||||
*/
|
||||
public boolean isEmpty() {
|
||||
return first() == null;
|
||||
@ -317,8 +403,8 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
|
||||
/**
|
||||
* Returns the number of elements in this queue. If this queue
|
||||
* contains more than <tt>Integer.MAX_VALUE</tt> elements, returns
|
||||
* <tt>Integer.MAX_VALUE</tt>.
|
||||
* contains more than {@code Integer.MAX_VALUE} elements, returns
|
||||
* {@code Integer.MAX_VALUE}.
|
||||
*
|
||||
* <p>Beware that, unlike in most collections, this method is
|
||||
* <em>NOT</em> a constant-time operation. Because of the
|
||||
@ -329,7 +415,7 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
*/
|
||||
public int size() {
|
||||
int count = 0;
|
||||
for (Node<E> p = first(); p != null; p = p.getNext()) {
|
||||
for (Node<E> p = first(); p != null; p = succ(p)) {
|
||||
if (p.getItem() != null) {
|
||||
// Collections.size() spec says to max out
|
||||
if (++count == Integer.MAX_VALUE)
|
||||
@ -340,16 +426,16 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns <tt>true</tt> if this queue contains the specified element.
|
||||
* More formally, returns <tt>true</tt> if and only if this queue contains
|
||||
* at least one element <tt>e</tt> such that <tt>o.equals(e)</tt>.
|
||||
* Returns {@code true} if this queue contains the specified element.
|
||||
* More formally, returns {@code true} if and only if this queue contains
|
||||
* at least one element {@code e} such that {@code o.equals(e)}.
|
||||
*
|
||||
* @param o object to be checked for containment in this queue
|
||||
* @return <tt>true</tt> if this queue contains the specified element
|
||||
* @return {@code true} if this queue contains the specified element
|
||||
*/
|
||||
public boolean contains(Object o) {
|
||||
if (o == null) return false;
|
||||
for (Node<E> p = first(); p != null; p = p.getNext()) {
|
||||
for (Node<E> p = first(); p != null; p = succ(p)) {
|
||||
E item = p.getItem();
|
||||
if (item != null &&
|
||||
o.equals(item))
|
||||
@ -360,23 +446,29 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
|
||||
/**
|
||||
* Removes a single instance of the specified element from this queue,
|
||||
* if it is present. More formally, removes an element <tt>e</tt> such
|
||||
* that <tt>o.equals(e)</tt>, if this queue contains one or more such
|
||||
* if it is present. More formally, removes an element {@code e} such
|
||||
* that {@code o.equals(e)}, if this queue contains one or more such
|
||||
* elements.
|
||||
* Returns <tt>true</tt> if this queue contained the specified element
|
||||
* Returns {@code true} if this queue contained the specified element
|
||||
* (or equivalently, if this queue changed as a result of the call).
|
||||
*
|
||||
* @param o element to be removed from this queue, if present
|
||||
* @return <tt>true</tt> if this queue changed as a result of the call
|
||||
* @return {@code true} if this queue changed as a result of the call
|
||||
*/
|
||||
public boolean remove(Object o) {
|
||||
if (o == null) return false;
|
||||
for (Node<E> p = first(); p != null; p = p.getNext()) {
|
||||
Node<E> pred = null;
|
||||
for (Node<E> p = first(); p != null; p = succ(p)) {
|
||||
E item = p.getItem();
|
||||
if (item != null &&
|
||||
o.equals(item) &&
|
||||
p.casItem(item, null))
|
||||
p.casItem(item, null)) {
|
||||
Node<E> next = succ(p);
|
||||
if (pred != null && next != null)
|
||||
pred.casNext(p, next);
|
||||
return true;
|
||||
}
|
||||
pred = p;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
@ -397,7 +489,7 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
public Object[] toArray() {
|
||||
// Use ArrayList to deal with resizing.
|
||||
ArrayList<E> al = new ArrayList<E>();
|
||||
for (Node<E> p = first(); p != null; p = p.getNext()) {
|
||||
for (Node<E> p = first(); p != null; p = succ(p)) {
|
||||
E item = p.getItem();
|
||||
if (item != null)
|
||||
al.add(item);
|
||||
@ -415,22 +507,22 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
* <p>If this queue fits in the specified array with room to spare
|
||||
* (i.e., the array has more elements than this queue), the element in
|
||||
* the array immediately following the end of the queue is set to
|
||||
* <tt>null</tt>.
|
||||
* {@code null}.
|
||||
*
|
||||
* <p>Like the {@link #toArray()} method, this method acts as bridge between
|
||||
* array-based and collection-based APIs. Further, this method allows
|
||||
* precise control over the runtime type of the output array, and may,
|
||||
* under certain circumstances, be used to save allocation costs.
|
||||
*
|
||||
* <p>Suppose <tt>x</tt> is a queue known to contain only strings.
|
||||
* <p>Suppose {@code x} is a queue known to contain only strings.
|
||||
* The following code can be used to dump the queue into a newly
|
||||
* allocated array of <tt>String</tt>:
|
||||
* allocated array of {@code String}:
|
||||
*
|
||||
* <pre>
|
||||
* String[] y = x.toArray(new String[0]);</pre>
|
||||
*
|
||||
* Note that <tt>toArray(new Object[0])</tt> is identical in function to
|
||||
* <tt>toArray()</tt>.
|
||||
* Note that {@code toArray(new Object[0])} is identical in function to
|
||||
* {@code toArray()}.
|
||||
*
|
||||
* @param a the array into which the elements of the queue are to
|
||||
* be stored, if it is big enough; otherwise, a new array of the
|
||||
@ -441,11 +533,12 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
* this queue
|
||||
* @throws NullPointerException if the specified array is null
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public <T> T[] toArray(T[] a) {
|
||||
// try to use sent-in array
|
||||
int k = 0;
|
||||
Node<E> p;
|
||||
for (p = first(); p != null && k < a.length; p = p.getNext()) {
|
||||
for (p = first(); p != null && k < a.length; p = succ(p)) {
|
||||
E item = p.getItem();
|
||||
if (item != null)
|
||||
a[k++] = (T)item;
|
||||
@ -458,7 +551,7 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
|
||||
// If won't fit, use ArrayList version
|
||||
ArrayList<E> al = new ArrayList<E>();
|
||||
for (Node<E> q = first(); q != null; q = q.getNext()) {
|
||||
for (Node<E> q = first(); q != null; q = succ(q)) {
|
||||
E item = q.getItem();
|
||||
if (item != null)
|
||||
al.add(item);
|
||||
@ -511,7 +604,15 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
lastRet = nextNode;
|
||||
E x = nextItem;
|
||||
|
||||
Node<E> p = (nextNode == null)? first() : nextNode.getNext();
|
||||
Node<E> pred, p;
|
||||
if (nextNode == null) {
|
||||
p = first();
|
||||
pred = null;
|
||||
} else {
|
||||
pred = nextNode;
|
||||
p = succ(nextNode);
|
||||
}
|
||||
|
||||
for (;;) {
|
||||
if (p == null) {
|
||||
nextNode = null;
|
||||
@ -523,8 +624,13 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
nextNode = p;
|
||||
nextItem = item;
|
||||
return x;
|
||||
} else // skip over nulls
|
||||
p = p.getNext();
|
||||
} else {
|
||||
// skip over nulls
|
||||
Node<E> next = succ(p);
|
||||
if (pred != null && next != null)
|
||||
pred.casNext(p, next);
|
||||
p = next;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -549,7 +655,7 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
/**
|
||||
* Save the state to a stream (that is, serialize it).
|
||||
*
|
||||
* @serialData All of the elements (each an <tt>E</tt>) in
|
||||
* @serialData All of the elements (each an {@code E}) in
|
||||
* the proper order, followed by a null
|
||||
* @param s the stream
|
||||
*/
|
||||
@ -560,7 +666,7 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
s.defaultWriteObject();
|
||||
|
||||
// Write out all elements in the proper order.
|
||||
for (Node<E> p = first(); p != null; p = p.getNext()) {
|
||||
for (Node<E> p = first(); p != null; p = succ(p)) {
|
||||
Object item = p.getItem();
|
||||
if (item != null)
|
||||
s.writeObject(item);
|
||||
@ -579,10 +685,11 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
throws java.io.IOException, ClassNotFoundException {
|
||||
// Read in capacity, and any hidden stuff
|
||||
s.defaultReadObject();
|
||||
head = new Node<E>(null, null);
|
||||
head = new Node<E>(null);
|
||||
tail = head;
|
||||
// Read in all elements and place in queue
|
||||
for (;;) {
|
||||
@SuppressWarnings("unchecked")
|
||||
E item = (E)s.readObject();
|
||||
if (item == null)
|
||||
break;
|
||||
@ -591,4 +698,35 @@ public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
|
||||
}
|
||||
}
|
||||
|
||||
// Unsafe mechanics
|
||||
|
||||
private static final sun.misc.Unsafe UNSAFE = sun.misc.Unsafe.getUnsafe();
|
||||
private static final long headOffset =
|
||||
objectFieldOffset(UNSAFE, "head", ConcurrentLinkedQueue.class);
|
||||
private static final long tailOffset =
|
||||
objectFieldOffset(UNSAFE, "tail", ConcurrentLinkedQueue.class);
|
||||
|
||||
private boolean casTail(Node<E> cmp, Node<E> val) {
|
||||
return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);
|
||||
}
|
||||
|
||||
private boolean casHead(Node<E> cmp, Node<E> val) {
|
||||
return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);
|
||||
}
|
||||
|
||||
private void lazySetHead(Node<E> val) {
|
||||
UNSAFE.putOrderedObject(this, headOffset, val);
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -33,9 +33,8 @@
|
||||
|
||||
/*
|
||||
* @test
|
||||
* @bug 4486658
|
||||
* @compile -source 1.5 ConcurrentQueueLoops.java
|
||||
* @run main/timeout=230 ConcurrentQueueLoops
|
||||
* @bug 4486658 6785442
|
||||
* @run main ConcurrentQueueLoops 8 123456
|
||||
* @summary Checks that a set of threads can repeatedly get and modify items
|
||||
*/
|
||||
|
||||
@ -44,34 +43,75 @@ import java.util.concurrent.*;
|
||||
import java.util.concurrent.atomic.*;
|
||||
|
||||
public class ConcurrentQueueLoops {
|
||||
static final ExecutorService pool = Executors.newCachedThreadPool();
|
||||
static AtomicInteger totalItems;
|
||||
static boolean print = false;
|
||||
ExecutorService pool;
|
||||
AtomicInteger totalItems;
|
||||
boolean print;
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
int maxStages = 8;
|
||||
int items = 100000;
|
||||
// Suitable for benchmarking. Overriden by args[0] for testing.
|
||||
int maxStages = 20;
|
||||
|
||||
// Suitable for benchmarking. Overriden by args[1] for testing.
|
||||
int items = 1024 * 1024;
|
||||
|
||||
Collection<Queue<Integer>> concurrentQueues() {
|
||||
List<Queue<Integer>> queues = new ArrayList<Queue<Integer>>();
|
||||
queues.add(new ConcurrentLinkedQueue<Integer>());
|
||||
queues.add(new ArrayBlockingQueue<Integer>(items, false));
|
||||
//queues.add(new ArrayBlockingQueue<Integer>(count, true));
|
||||
queues.add(new LinkedBlockingQueue<Integer>());
|
||||
queues.add(new LinkedBlockingDeque<Integer>());
|
||||
|
||||
try {
|
||||
queues.add((Queue<Integer>)
|
||||
Class.forName("java.util.concurrent.LinkedTransferQueue")
|
||||
.newInstance());
|
||||
} catch (IllegalAccessException e) {
|
||||
} catch (InstantiationException e) {
|
||||
} catch (ClassNotFoundException e) {
|
||||
// OK; not yet added to JDK
|
||||
}
|
||||
|
||||
// Following additional implementations are available from:
|
||||
// http://gee.cs.oswego.edu/dl/concurrency-interest/index.html
|
||||
// queues.add(new LinkedTransferQueue<Integer>());
|
||||
// queues.add(new SynchronizedLinkedListQueue<Integer>());
|
||||
|
||||
// Avoid "first fast, second slow" benchmark effect.
|
||||
Collections.shuffle(queues);
|
||||
return queues;
|
||||
}
|
||||
|
||||
void test(String[] args) throws Throwable {
|
||||
if (args.length > 0)
|
||||
maxStages = Integer.parseInt(args[0]);
|
||||
if (args.length > 1)
|
||||
items = Integer.parseInt(args[1]);
|
||||
|
||||
for (Queue<Integer> queue : concurrentQueues())
|
||||
test(queue);
|
||||
}
|
||||
|
||||
void test(final Queue<Integer> q) throws Throwable {
|
||||
System.out.println(q.getClass().getSimpleName());
|
||||
pool = Executors.newCachedThreadPool();
|
||||
print = false;
|
||||
|
||||
print = false;
|
||||
System.out.println("Warmup...");
|
||||
oneRun(1, items);
|
||||
Thread.sleep(100);
|
||||
oneRun(1, items);
|
||||
oneRun(1, items, q);
|
||||
//Thread.sleep(100);
|
||||
oneRun(3, items, q);
|
||||
Thread.sleep(100);
|
||||
print = true;
|
||||
|
||||
for (int i = 1; i <= maxStages; i += (i+1) >>> 1) {
|
||||
oneRun(i, items);
|
||||
oneRun(i, items, q);
|
||||
}
|
||||
pool.shutdown();
|
||||
if (! pool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS))
|
||||
throw new Error();
|
||||
check(pool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS));
|
||||
}
|
||||
|
||||
static class Stage implements Callable<Integer> {
|
||||
class Stage implements Callable<Integer> {
|
||||
final Queue<Integer> queue;
|
||||
final CyclicBarrier barrier;
|
||||
int items;
|
||||
@ -110,15 +150,11 @@ public class ConcurrentQueueLoops {
|
||||
}
|
||||
return new Integer(l);
|
||||
}
|
||||
catch (Exception ie) {
|
||||
ie.printStackTrace();
|
||||
throw new Error("Call loop failed");
|
||||
}
|
||||
catch (Throwable t) { unexpected(t); return null; }
|
||||
}
|
||||
}
|
||||
|
||||
static void oneRun(int n, int items) throws Exception {
|
||||
Queue<Integer> q = new ConcurrentLinkedQueue<Integer>();
|
||||
void oneRun(int n, int items, final Queue<Integer> q) throws Exception {
|
||||
LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
|
||||
CyclicBarrier barrier = new CyclicBarrier(n + 1, timer);
|
||||
totalItems = new AtomicInteger(n * items);
|
||||
@ -141,6 +177,22 @@ public class ConcurrentQueueLoops {
|
||||
System.out.println(LoopHelpers.rightJustify(time / (items * n)) + " ns per item");
|
||||
if (total == 0) // avoid overoptimization
|
||||
System.out.println("useless result: " + total);
|
||||
|
||||
}
|
||||
|
||||
//--------------------- Infrastructure ---------------------------
|
||||
volatile int passed = 0, failed = 0;
|
||||
void pass() {passed++;}
|
||||
void fail() {failed++; Thread.dumpStack();}
|
||||
void fail(String msg) {System.err.println(msg); fail();}
|
||||
void unexpected(Throwable t) {failed++; t.printStackTrace();}
|
||||
void check(boolean cond) {if (cond) pass(); else fail();}
|
||||
void equal(Object x, Object y) {
|
||||
if (x == null ? y == null : x.equals(y)) pass();
|
||||
else fail(x + " not equal to " + y);}
|
||||
public static void main(String[] args) throws Throwable {
|
||||
new ConcurrentQueueLoops().instanceMain(args);}
|
||||
public void instanceMain(String[] args) throws Throwable {
|
||||
try {test(args);} catch (Throwable t) {unexpected(t);}
|
||||
System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
|
||||
if (failed > 0) throw new AssertionError("Some tests failed");}
|
||||
}
|
||||
165
jdk/test/java/util/concurrent/ConcurrentQueues/GCRetention.java
Normal file
165
jdk/test/java/util/concurrent/ConcurrentQueues/GCRetention.java
Normal file
@ -0,0 +1,165 @@
|
||||
/*
|
||||
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
||||
*
|
||||
* This code is free software; you can redistribute it and/or modify it
|
||||
* under the terms of the GNU General Public License version 2 only, as
|
||||
* published by the Free Software Foundation.
|
||||
*
|
||||
* This code is distributed in the hope that it will be useful, but WITHOUT
|
||||
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
|
||||
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
|
||||
* version 2 for more details (a copy is included in the LICENSE file that
|
||||
* accompanied this code).
|
||||
*
|
||||
* You should have received a copy of the GNU General Public License version
|
||||
* 2 along with this work; if not, write to the Free Software Foundation,
|
||||
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
|
||||
*
|
||||
* Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
|
||||
* CA 95054 USA or visit www.sun.com if you need additional information or
|
||||
* have any questions.
|
||||
*/
|
||||
|
||||
/*
|
||||
* This file is available under and governed by the GNU General Public
|
||||
* License version 2 only, as published by the Free Software Foundation.
|
||||
* However, the following notice accompanied the original version of this
|
||||
* file:
|
||||
*
|
||||
* Written by Doug Lea with assistance from members of JCP JSR-166
|
||||
* Expert Group and released to the public domain, as explained at
|
||||
* http://creativecommons.org/licenses/publicdomain
|
||||
*/
|
||||
/*
|
||||
* @test
|
||||
* @bug 6785442
|
||||
* @summary Benchmark that tries to GC-tenure head, followed by
|
||||
* many add/remove operations.
|
||||
* @run main GCRetention 12345
|
||||
*/
|
||||
|
||||
import java.util.concurrent.ArrayBlockingQueue;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.LinkedBlockingDeque;
|
||||
import java.util.concurrent.PriorityBlockingQueue;
|
||||
import java.util.LinkedList;
|
||||
import java.util.PriorityQueue;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Queue;
|
||||
import java.util.Map;
|
||||
|
||||
public class GCRetention {
|
||||
// Suitable for benchmarking. Overriden by args[0] for testing.
|
||||
int count = 1024 * 1024;
|
||||
|
||||
final Map<String,String> results = new ConcurrentHashMap<String,String>();
|
||||
|
||||
Collection<Queue<Boolean>> queues() {
|
||||
List<Queue<Boolean>> queues = new ArrayList<Queue<Boolean>>();
|
||||
queues.add(new ConcurrentLinkedQueue<Boolean>());
|
||||
queues.add(new ArrayBlockingQueue<Boolean>(count, false));
|
||||
queues.add(new ArrayBlockingQueue<Boolean>(count, true));
|
||||
queues.add(new LinkedBlockingQueue<Boolean>());
|
||||
queues.add(new LinkedBlockingDeque<Boolean>());
|
||||
queues.add(new PriorityBlockingQueue<Boolean>());
|
||||
queues.add(new PriorityQueue<Boolean>());
|
||||
queues.add(new LinkedList<Boolean>());
|
||||
|
||||
try {
|
||||
queues.add((Queue<Boolean>)
|
||||
Class.forName("java.util.concurrent.LinkedTransferQueue")
|
||||
.newInstance());
|
||||
} catch (IllegalAccessException e) {
|
||||
} catch (InstantiationException e) {
|
||||
} catch (ClassNotFoundException e) {
|
||||
// OK; not yet added to JDK
|
||||
}
|
||||
|
||||
// Following additional implementations are available from:
|
||||
// http://gee.cs.oswego.edu/dl/concurrency-interest/index.html
|
||||
// queues.add(new LinkedTransferQueue<Boolean>());
|
||||
// queues.add(new SynchronizedLinkedListQueue<Boolean>());
|
||||
|
||||
// Avoid "first fast, second slow" benchmark effect.
|
||||
Collections.shuffle(queues);
|
||||
return queues;
|
||||
}
|
||||
|
||||
void prettyPrintResults() {
|
||||
List<String> classNames = new ArrayList<String>(results.keySet());
|
||||
Collections.sort(classNames);
|
||||
int maxClassNameLength = 0;
|
||||
int maxNanosLength = 0;
|
||||
for (String name : classNames) {
|
||||
if (maxClassNameLength < name.length())
|
||||
maxClassNameLength = name.length();
|
||||
if (maxNanosLength < results.get(name).length())
|
||||
maxNanosLength = results.get(name).length();
|
||||
}
|
||||
String format = String.format("%%%ds %%%ds nanos/item%%n",
|
||||
maxClassNameLength, maxNanosLength);
|
||||
for (String name : classNames)
|
||||
System.out.printf(format, name, results.get(name));
|
||||
}
|
||||
|
||||
void test(String[] args) {
|
||||
if (args.length > 0)
|
||||
count = Integer.valueOf(args[0]);
|
||||
// Warmup
|
||||
for (Queue<Boolean> queue : queues())
|
||||
test(queue);
|
||||
results.clear();
|
||||
for (Queue<Boolean> queue : queues())
|
||||
test(queue);
|
||||
|
||||
prettyPrintResults();
|
||||
}
|
||||
|
||||
void test(Queue<Boolean> q) {
|
||||
long t0 = System.nanoTime();
|
||||
for (int i = 0; i < count; i++)
|
||||
check(q.add(Boolean.TRUE));
|
||||
System.gc();
|
||||
System.gc();
|
||||
Boolean x;
|
||||
while ((x = q.poll()) != null)
|
||||
equal(x, Boolean.TRUE);
|
||||
check(q.isEmpty());
|
||||
|
||||
for (int i = 0; i < 10 * count; i++) {
|
||||
for (int k = 0; k < 3; k++)
|
||||
check(q.add(Boolean.TRUE));
|
||||
for (int k = 0; k < 3; k++)
|
||||
if (q.poll() != Boolean.TRUE)
|
||||
fail();
|
||||
}
|
||||
check(q.isEmpty());
|
||||
|
||||
String className = q.getClass().getSimpleName();
|
||||
long elapsed = System.nanoTime() - t0;
|
||||
int nanos = (int) ((double) elapsed / (10 * 3 * count));
|
||||
results.put(className, String.valueOf(nanos));
|
||||
}
|
||||
|
||||
//--------------------- Infrastructure ---------------------------
|
||||
volatile int passed = 0, failed = 0;
|
||||
void pass() {passed++;}
|
||||
void fail() {failed++; Thread.dumpStack();}
|
||||
void fail(String msg) {System.err.println(msg); fail();}
|
||||
void unexpected(Throwable t) {failed++; t.printStackTrace();}
|
||||
void check(boolean cond) {if (cond) pass(); else fail();}
|
||||
void equal(Object x, Object y) {
|
||||
if (x == null ? y == null : x.equals(y)) pass();
|
||||
else fail(x + " not equal to " + y);}
|
||||
public static void main(String[] args) throws Throwable {
|
||||
new GCRetention().instanceMain(args);}
|
||||
public void instanceMain(String[] args) throws Throwable {
|
||||
try {test(args);} catch (Throwable t) {unexpected(t);}
|
||||
System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
|
||||
if (failed > 0) throw new AssertionError("Some tests failed");}
|
||||
}
|
||||
@ -0,0 +1,230 @@
|
||||
/*
|
||||
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
|
||||
*
|
||||
* This code is free software; you can redistribute it and/or modify it
|
||||
* under the terms of the GNU General Public License version 2 only, as
|
||||
* published by the Free Software Foundation.
|
||||
*
|
||||
* This code is distributed in the hope that it will be useful, but WITHOUT
|
||||
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
|
||||
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
|
||||
* version 2 for more details (a copy is included in the LICENSE file that
|
||||
* accompanied this code).
|
||||
*
|
||||
* You should have received a copy of the GNU General Public License version
|
||||
* 2 along with this work; if not, write to the Free Software Foundation,
|
||||
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
|
||||
*
|
||||
* Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
|
||||
* CA 95054 USA or visit www.sun.com if you need additional information or
|
||||
* have any questions.
|
||||
*/
|
||||
|
||||
/*
|
||||
* This file is available under and governed by the GNU General Public
|
||||
* License version 2 only, as published by the Free Software Foundation.
|
||||
* However, the following notice accompanied the original version of this
|
||||
* file:
|
||||
*
|
||||
* Written by Doug Lea with assistance from members of JCP JSR-166
|
||||
* Expert Group and released to the public domain, as explained at
|
||||
* http://creativecommons.org/licenses/publicdomain
|
||||
*/
|
||||
|
||||
/*
|
||||
* @test
|
||||
* @bug 6785442
|
||||
* @summary Checks race between poll and remove(Object), while
|
||||
* occasionally moonlighting as a microbenchmark.
|
||||
* @run main RemovePollRace 12345
|
||||
*/
|
||||
|
||||
import java.util.concurrent.ArrayBlockingQueue;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentLinkedQueue;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.LinkedBlockingDeque;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Queue;
|
||||
import java.util.Map;
|
||||
|
||||
public class RemovePollRace {
|
||||
// Suitable for benchmarking. Overriden by args[0] for testing.
|
||||
int count = 1024 * 1024;
|
||||
|
||||
final Map<String,String> results = new ConcurrentHashMap<String,String>();
|
||||
|
||||
Collection<Queue<Boolean>> concurrentQueues() {
|
||||
List<Queue<Boolean>> queues = new ArrayList<Queue<Boolean>>();
|
||||
queues.add(new ConcurrentLinkedQueue<Boolean>());
|
||||
queues.add(new ArrayBlockingQueue<Boolean>(count, false));
|
||||
queues.add(new ArrayBlockingQueue<Boolean>(count, true));
|
||||
queues.add(new LinkedBlockingQueue<Boolean>());
|
||||
queues.add(new LinkedBlockingDeque<Boolean>());
|
||||
|
||||
try {
|
||||
queues.add((Queue<Boolean>)
|
||||
Class.forName("java.util.concurrent.LinkedTransferQueue")
|
||||
.newInstance());
|
||||
} catch (IllegalAccessException e) {
|
||||
} catch (InstantiationException e) {
|
||||
} catch (ClassNotFoundException e) {
|
||||
// OK; not yet added to JDK
|
||||
}
|
||||
|
||||
// Following additional implementations are available from:
|
||||
// http://gee.cs.oswego.edu/dl/concurrency-interest/index.html
|
||||
// queues.add(new LinkedTransferQueue<Boolean>());
|
||||
// queues.add(new SynchronizedLinkedListQueue<Boolean>());
|
||||
|
||||
// Avoid "first fast, second slow" benchmark effect.
|
||||
Collections.shuffle(queues);
|
||||
return queues;
|
||||
}
|
||||
|
||||
void prettyPrintResults() {
|
||||
List<String> classNames = new ArrayList<String>(results.keySet());
|
||||
Collections.sort(classNames);
|
||||
int maxClassNameLength = 0;
|
||||
int maxNanosLength = 0;
|
||||
for (String name : classNames) {
|
||||
if (maxClassNameLength < name.length())
|
||||
maxClassNameLength = name.length();
|
||||
if (maxNanosLength < results.get(name).length())
|
||||
maxNanosLength = results.get(name).length();
|
||||
}
|
||||
String format = String.format("%%%ds %%%ds nanos/item%%n",
|
||||
maxClassNameLength, maxNanosLength);
|
||||
for (String name : classNames)
|
||||
System.out.printf(format, name, results.get(name));
|
||||
}
|
||||
|
||||
void test(String[] args) throws Throwable {
|
||||
if (args.length > 0)
|
||||
count = Integer.valueOf(args[0]);
|
||||
// Warmup
|
||||
for (Queue<Boolean> queue : concurrentQueues())
|
||||
test(queue);
|
||||
results.clear();
|
||||
for (Queue<Boolean> queue : concurrentQueues())
|
||||
test(queue);
|
||||
|
||||
prettyPrintResults();
|
||||
}
|
||||
|
||||
void await(CountDownLatch latch) {
|
||||
try { latch.await(); }
|
||||
catch (InterruptedException e) { unexpected(e); }
|
||||
}
|
||||
|
||||
void test(final Queue<Boolean> q) throws Throwable {
|
||||
long t0 = System.nanoTime();
|
||||
final int SPINS = 5;
|
||||
final AtomicLong removes = new AtomicLong(0);
|
||||
final AtomicLong polls = new AtomicLong(0);
|
||||
final int adderCount =
|
||||
Math.max(1, Runtime.getRuntime().availableProcessors() / 4);
|
||||
final int removerCount =
|
||||
Math.max(1, Runtime.getRuntime().availableProcessors() / 4);
|
||||
final int pollerCount = removerCount;
|
||||
final int threadCount = adderCount + removerCount + pollerCount;
|
||||
final CountDownLatch startingGate = new CountDownLatch(1);
|
||||
final CountDownLatch addersDone = new CountDownLatch(adderCount);
|
||||
final Runnable remover = new Runnable() {
|
||||
public void run() {
|
||||
await(startingGate);
|
||||
int spins = 0;
|
||||
for (;;) {
|
||||
boolean quittingTime = (addersDone.getCount() == 0);
|
||||
if (q.remove(Boolean.TRUE))
|
||||
removes.getAndIncrement();
|
||||
else if (quittingTime)
|
||||
break;
|
||||
else if (++spins > SPINS) {
|
||||
Thread.yield();
|
||||
spins = 0;
|
||||
}}}};
|
||||
final Runnable poller = new Runnable() {
|
||||
public void run() {
|
||||
await(startingGate);
|
||||
int spins = 0;
|
||||
for (;;) {
|
||||
boolean quittingTime = (addersDone.getCount() == 0);
|
||||
if (q.poll() == Boolean.TRUE)
|
||||
polls.getAndIncrement();
|
||||
else if (quittingTime)
|
||||
break;
|
||||
else if (++spins > SPINS) {
|
||||
Thread.yield();
|
||||
spins = 0;
|
||||
}}}};
|
||||
final Runnable adder = new Runnable() {
|
||||
public void run() {
|
||||
await(startingGate);
|
||||
for (int i = 0; i < count; i++) {
|
||||
for (;;) {
|
||||
try { q.add(Boolean.TRUE); break; }
|
||||
catch (IllegalStateException e) { Thread.yield(); }
|
||||
}
|
||||
}
|
||||
addersDone.countDown();
|
||||
}};
|
||||
|
||||
final List<Thread> adders = new ArrayList<Thread>();
|
||||
final List<Thread> removers = new ArrayList<Thread>();
|
||||
final List<Thread> pollers = new ArrayList<Thread>();
|
||||
for (int i = 0; i < adderCount; i++)
|
||||
adders.add(checkedThread(adder));
|
||||
for (int i = 0; i < removerCount; i++)
|
||||
removers.add(checkedThread(remover));
|
||||
for (int i = 0; i < pollerCount; i++)
|
||||
pollers.add(checkedThread(poller));
|
||||
|
||||
final List<Thread> allThreads = new ArrayList<Thread>();
|
||||
allThreads.addAll(removers);
|
||||
allThreads.addAll(pollers);
|
||||
allThreads.addAll(adders);
|
||||
|
||||
for (Thread t : allThreads)
|
||||
t.start();
|
||||
startingGate.countDown();
|
||||
for (Thread t : allThreads)
|
||||
t.join();
|
||||
|
||||
String className = q.getClass().getSimpleName();
|
||||
long elapsed = System.nanoTime() - t0;
|
||||
int nanos = (int) ((double) elapsed / (adderCount * count));
|
||||
results.put(className, String.valueOf(nanos));
|
||||
if (removes.get() + polls.get() != adderCount * count) {
|
||||
String msg = String.format
|
||||
("class=%s removes=%s polls=%d count=%d",
|
||||
className, removes.get(), polls.get(), count);
|
||||
fail(msg);
|
||||
}
|
||||
}
|
||||
|
||||
//--------------------- Infrastructure ---------------------------
|
||||
volatile int passed = 0, failed = 0;
|
||||
void pass() {passed++;}
|
||||
void fail() {failed++; Thread.dumpStack();}
|
||||
void fail(String msg) {System.err.println(msg); fail();}
|
||||
void unexpected(Throwable t) {failed++; t.printStackTrace();}
|
||||
void check(boolean cond) {if (cond) pass(); else fail();}
|
||||
void equal(Object x, Object y) {
|
||||
if (x == null ? y == null : x.equals(y)) pass();
|
||||
else fail(x + " not equal to " + y);}
|
||||
public static void main(String[] args) throws Throwable {
|
||||
new RemovePollRace().instanceMain(args);}
|
||||
public void instanceMain(String[] args) throws Throwable {
|
||||
try {test(args);} catch (Throwable t) {unexpected(t);}
|
||||
System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
|
||||
if (failed > 0) throw new AssertionError("Some tests failed");}
|
||||
Thread checkedThread(final Runnable r) {
|
||||
return new Thread() {public void run() {
|
||||
try {r.run();} catch (Throwable t) {unexpected(t);}}};}
|
||||
}
|
||||
@ -28,62 +28,74 @@
|
||||
* @author Martin Buchholz
|
||||
*/
|
||||
|
||||
import java.util.*;
|
||||
import java.util.concurrent.*;
|
||||
|
||||
public class OfferRemoveLoops {
|
||||
private static void realMain(String[] args) throws Throwable {
|
||||
void test(String[] args) throws Throwable {
|
||||
testQueue(new LinkedBlockingQueue<String>(10));
|
||||
testQueue(new LinkedBlockingQueue<String>());
|
||||
testQueue(new LinkedBlockingDeque<String>(10));
|
||||
testQueue(new LinkedBlockingDeque<String>());
|
||||
testQueue(new ArrayBlockingQueue<String>(10));
|
||||
testQueue(new PriorityBlockingQueue<String>(10));
|
||||
testQueue(new ConcurrentLinkedQueue<String>());
|
||||
}
|
||||
|
||||
private abstract static class ControlledThread extends Thread {
|
||||
abstract class CheckedThread extends Thread {
|
||||
abstract protected void realRun();
|
||||
public void run() {
|
||||
try { realRun(); } catch (Throwable t) { unexpected(t); }
|
||||
}
|
||||
}
|
||||
|
||||
private static void testQueue(final BlockingQueue<String> q) throws Throwable {
|
||||
System.out.println(q.getClass());
|
||||
final int count = 10000;
|
||||
final long quittingTime = System.nanoTime() + 1L * 1000L * 1000L * 1000L;
|
||||
Thread t1 = new ControlledThread() {
|
||||
protected void realRun() {
|
||||
for (int i = 0, j = 0; i < count; i++)
|
||||
while (! q.remove(String.valueOf(i))
|
||||
&& System.nanoTime() - quittingTime < 0)
|
||||
Thread.yield();}};
|
||||
Thread t2 = new ControlledThread() {
|
||||
protected void realRun() {
|
||||
for (int i = 0, j = 0; i < count; i++)
|
||||
while (! q.offer(String.valueOf(i))
|
||||
&& System.nanoTime() - quittingTime < 0)
|
||||
Thread.yield();}};
|
||||
void testQueue(final Queue<String> q) throws Throwable {
|
||||
System.out.println(q.getClass().getSimpleName());
|
||||
final int count = 1000 * 1000;
|
||||
final long testDurationSeconds = 1L;
|
||||
final long testDurationMillis = testDurationSeconds * 1000L;
|
||||
final long quittingTimeNanos
|
||||
= System.nanoTime() + testDurationSeconds * 1000L * 1000L * 1000L;
|
||||
Thread t1 = new CheckedThread() {
|
||||
protected void realRun() {
|
||||
for (int i = 0; i < count; i++) {
|
||||
if ((i % 1024) == 0 &&
|
||||
System.nanoTime() - quittingTimeNanos > 0)
|
||||
return;
|
||||
while (! q.remove(String.valueOf(i)))
|
||||
Thread.yield();
|
||||
}}};
|
||||
Thread t2 = new CheckedThread() {
|
||||
protected void realRun() {
|
||||
for (int i = 0; i < count; i++) {
|
||||
if ((i % 1024) == 0 &&
|
||||
System.nanoTime() - quittingTimeNanos > 0)
|
||||
return;
|
||||
while (! q.offer(String.valueOf(i)))
|
||||
Thread.yield();
|
||||
}}};
|
||||
t1.setDaemon(true); t2.setDaemon(true);
|
||||
t1.start(); t2.start();
|
||||
t1.join(10000); t2.join(10000);
|
||||
t1.join(10 * testDurationMillis);
|
||||
t2.join(10 * testDurationMillis);
|
||||
check(! t1.isAlive());
|
||||
check(! t2.isAlive());
|
||||
}
|
||||
|
||||
//--------------------- Infrastructure ---------------------------
|
||||
static volatile int passed = 0, failed = 0;
|
||||
static void pass() { passed++; }
|
||||
static void fail() { failed++; Thread.dumpStack(); }
|
||||
static void unexpected(Throwable t) { failed++; t.printStackTrace(); }
|
||||
static void check(boolean cond) { if (cond) pass(); else fail(); }
|
||||
static void equal(Object x, Object y) {
|
||||
volatile int passed = 0, failed = 0;
|
||||
void pass() {passed++;}
|
||||
void fail() {failed++; Thread.dumpStack();}
|
||||
void fail(String msg) {System.err.println(msg); fail();}
|
||||
void unexpected(Throwable t) {failed++; t.printStackTrace();}
|
||||
void check(boolean cond) {if (cond) pass(); else fail();}
|
||||
void equal(Object x, Object y) {
|
||||
if (x == null ? y == null : x.equals(y)) pass();
|
||||
else {System.out.println(x + " not equal to " + y); fail(); }}
|
||||
|
||||
else fail(x + " not equal to " + y);}
|
||||
public static void main(String[] args) throws Throwable {
|
||||
try { realMain(args); } catch (Throwable t) { unexpected(t); }
|
||||
|
||||
new OfferRemoveLoops().instanceMain(args);}
|
||||
public void instanceMain(String[] args) throws Throwable {
|
||||
try {test(args);} catch (Throwable t) {unexpected(t);}
|
||||
System.out.printf("%nPassed = %d, failed = %d%n%n", passed, failed);
|
||||
if (failed > 0) throw new Exception("Some tests failed");
|
||||
}
|
||||
if (failed > 0) throw new AssertionError("Some tests failed");}
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user