You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@harmony.apache.org by hi...@apache.org on 2009/07/28 11:30:48 UTC
svn commit: r798469 [8/28] - in /harmony/enhanced/classlib/branches/java6:
./ depends/build/platform/ depends/files/ depends/jars/
depends/manifests/icu4j_4.0/ depends/manifests/icu4j_4.2.1/
depends/manifests/icu4j_4.2.1/META-INF/ make/ modules/accessi...
Modified: harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/CopyOnWriteArraySet.java
URL: http://svn.apache.org/viewvc/harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/CopyOnWriteArraySet.java?rev=798469&r1=798468&r2=798469&view=diff
==============================================================================
--- harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/CopyOnWriteArraySet.java (original)
+++ harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/CopyOnWriteArraySet.java Tue Jul 28 09:30:33 2009
@@ -1,42 +1,41 @@
/*
* Written by Doug Lea with assistance from members of JCP JSR-166
- * Expert Group and released to the public domain. Use, modify, and
- * redistribute this code in any way without acknowledgement.
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/licenses/publicdomain
*/
package java.util.concurrent;
import java.util.*;
/**
- * A {@link java.util.Set} that uses {@link
- * java.util.concurrent.CopyOnWriteArrayList} for all of its
- * operations. Thus, it shares the same basic properties:
+ * A {@link java.util.Set} that uses an internal {@link CopyOnWriteArrayList}
+ * for all of its operations. Thus, it shares the same basic properties:
* <ul>
* <li>It is best suited for applications in which set sizes generally
* stay small, read-only operations
* vastly outnumber mutative operations, and you need
* to prevent interference among threads during traversal.
- * <li>Mutative operations(add, set, remove, etc) are expensive
- * since they usually entail copying the entire underlying array.
- * <li>Iterators do not support the mutative remove operation
- * <li>Traversal via iterators is very fast and cannot ever encounter
+ * <li>It is thread-safe.
+ * <li>Mutative operations (<tt>add</tt>, <tt>set</tt>, <tt>remove</tt>, etc.)
+ * are expensive since they usually entail copying the entire underlying
+ * array.
+ * <li>Iterators do not support the mutative <tt>remove</tt> operation.
+ * <li>Traversal via iterators is fast and cannot encounter
* interference from other threads. Iterators rely on
* unchanging snapshots of the array at the time the iterators were
- * constructed.
+ * constructed.
* </ul>
- * <p>
- * <b>Sample Usage.</b> Probably the main application
- * of copy-on-write sets are classes that maintain
- * sets of Handler objects
- * that must be multicasted to upon an update command. This
- * is a classic case where you do not want to be holding a
- * lock while sending a message, and where traversals normally
- * vastly overwhelm additions.
+ *
+ * <p> <b>Sample Usage.</b> The following code sketch uses a
+ * copy-on-write set to maintain a set of Handler objects that
+ * perform some action upon state updates.
+ *
* <pre>
* class Handler { void handle(); ... }
*
* class X {
- * private final CopyOnWriteArraySet<Handler> handlers = new CopyOnWriteArraySet<Handler>();
+ * private final CopyOnWriteArraySet<Handler> handlers
+ * = new CopyOnWriteArraySet<Handler>();
* public void addHandler(Handler h) { handlers.add(h); }
*
* private long internalState;
@@ -44,18 +43,17 @@
*
* public void update() {
* changeState();
- * Iterator it = handlers.iterator();
- * while (it.hasNext())
- * it.next().handle();
+ * for (Handler handler : handlers)
+ * handler.handle();
* }
* }
* </pre>
- * @see CopyOnWriteArrayList
*
* <p>This class is a member of the
- * <a href="{@docRoot}/../guide/collections/index.html">
+ * <a href="{@docRoot}/../technotes/guides/collections/index.html">
* Java Collections Framework</a>.
*
+ * @see CopyOnWriteArrayList
* @since 1.5
* @author Doug Lea
* @param <E> the type of elements held in this collection
@@ -75,27 +73,292 @@
/**
* Creates a set containing all of the elements of the specified
- * Collection.
- * @param c the collection
+ * collection.
+ *
+ * @param c the collection of elements to initially contain
+ * @throws NullPointerException if the specified collection is null
*/
public CopyOnWriteArraySet(Collection<? extends E> c) {
al = new CopyOnWriteArrayList<E>();
al.addAllAbsent(c);
}
+ /**
+ * Returns the number of elements in this set.
+ *
+ * @return the number of elements in this set
+ */
+ public int size() {
+ return al.size();
+ }
+
+ /**
+ * Returns <tt>true</tt> if this set contains no elements.
+ *
+ * @return <tt>true</tt> if this set contains no elements
+ */
+ public boolean isEmpty() {
+ return al.isEmpty();
+ }
+
+ /**
+ * Returns <tt>true</tt> if this set contains the specified element.
+ * More formally, returns <tt>true</tt> if and only if this set
+ * contains an element <tt>e</tt> such that
+ * <tt>(o==null ? e==null : o.equals(e))</tt>.
+ *
+ * @param o element whose presence in this set is to be tested
+ * @return <tt>true</tt> if this set contains the specified element
+ */
+ public boolean contains(Object o) {
+ return al.contains(o);
+ }
+
+ /**
+ * Returns an array containing all of the elements in this set.
+ * If this set makes any guarantees as to what order its elements
+ * are returned by its iterator, this method must return the
+ * elements in the same order.
+ *
+ * <p>The returned array will be "safe" in that no references to it
+ * are maintained by this set. (In other words, this method must
+ * allocate a new array even if this set is backed by an array).
+ * The caller is thus free to modify the returned array.
+ *
+ * <p>This method acts as bridge between array-based and collection-based
+ * APIs.
+ *
+ * @return an array containing all the elements in this set
+ */
+ public Object[] toArray() {
+ return al.toArray();
+ }
+
+ /**
+ * Returns an array containing all of the elements in this set; the
+ * runtime type of the returned array is that of the specified array.
+ * If the set fits in the specified array, it is returned therein.
+ * Otherwise, a new array is allocated with the runtime type of the
+ * specified array and the size of this set.
+ *
+ * <p>If this set fits in the specified array with room to spare
+ * (i.e., the array has more elements than this set), the element in
+ * the array immediately following the end of the set is set to
+ * <tt>null</tt>. (This is useful in determining the length of this
+ * set <i>only</i> if the caller knows that this set does not contain
+ * any null elements.)
+ *
+ * <p>If this set makes any guarantees as to what order its elements
+ * are returned by its iterator, this method must return the elements
+ * in the same order.
+ *
+ * <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 set known to contain only strings.
+ * The following code can be used to dump the set into a newly allocated
+ * array of <tt>String</tt>:
+ *
+ * <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>.
+ *
+ * @param a the array into which the elements of this set are to be
+ * stored, if it is big enough; otherwise, a new array of the same
+ * runtime type is allocated for this purpose.
+ * @return an array containing all the elements in this set
+ * @throws ArrayStoreException if the runtime type of the specified array
+ * is not a supertype of the runtime type of every element in this
+ * set
+ * @throws NullPointerException if the specified array is null
+ */
+ public <T> T[] toArray(T[] a) {
+ return al.toArray(a);
+ }
+
+ /**
+ * Removes all of the elements from this set.
+ * The set will be empty after this call returns.
+ */
+ public void clear() {
+ al.clear();
+ }
- public int size() { return al.size(); }
- public boolean isEmpty() { return al.isEmpty(); }
- public boolean contains(Object o) { return al.contains(o); }
- public Object[] toArray() { return al.toArray(); }
- public <T> T[] toArray(T[] a) { return al.toArray(a); }
- public void clear() { al.clear(); }
- public Iterator<E> iterator() { return al.iterator(); }
- public boolean remove(Object o) { return al.remove(o); }
- public boolean add(E o) { return al.addIfAbsent(o); }
- public boolean containsAll(Collection<?> c) { return al.containsAll(c); }
- public boolean addAll(Collection<? extends E> c) { return al.addAllAbsent(c) > 0; }
- public boolean removeAll(Collection<?> c) { return al.removeAll(c); }
- public boolean retainAll(Collection<?> c) { return al.retainAll(c); }
+ /**
+ * Removes the specified element from this set if it is present.
+ * More formally, removes an element <tt>e</tt> such that
+ * <tt>(o==null ? e==null : o.equals(e))</tt>,
+ * if this set contains such an element. Returns <tt>true</tt> if
+ * this set contained the element (or equivalently, if this set
+ * changed as a result of the call). (This set will not contain the
+ * element once the call returns.)
+ *
+ * @param o object to be removed from this set, if present
+ * @return <tt>true</tt> if this set contained the specified element
+ */
+ public boolean remove(Object o) {
+ return al.remove(o);
+ }
+ /**
+ * Adds the specified element to this set if it is not already present.
+ * More formally, adds the specified element <tt>e</tt> to this set if
+ * the set contains no element <tt>e2</tt> such that
+ * <tt>(e==null ? e2==null : e.equals(e2))</tt>.
+ * If this set already contains the element, the call leaves the set
+ * unchanged and returns <tt>false</tt>.
+ *
+ * @param e element to be added to this set
+ * @return <tt>true</tt> if this set did not already contain the specified
+ * element
+ */
+ public boolean add(E e) {
+ return al.addIfAbsent(e);
+ }
+
+ /**
+ * Returns <tt>true</tt> if this set contains all of the elements of the
+ * specified collection. If the specified collection is also a set, this
+ * method returns <tt>true</tt> if it is a <i>subset</i> of this set.
+ *
+ * @param c collection to be checked for containment in this set
+ * @return <tt>true</tt> if this set contains all of the elements of the
+ * specified collection
+ * @throws NullPointerException if the specified collection is null
+ * @see #contains(Object)
+ */
+ public boolean containsAll(Collection<?> c) {
+ return al.containsAll(c);
+ }
+
+ /**
+ * Adds all of the elements in the specified collection to this set if
+ * they're not already present. If the specified collection is also a
+ * set, the <tt>addAll</tt> operation effectively modifies this set so
+ * that its value is the <i>union</i> of the two sets. The behavior of
+ * this operation is undefined if the specified collection is modified
+ * while the operation is in progress.
+ *
+ * @param c collection containing elements to be added to this set
+ * @return <tt>true</tt> if this set changed as a result of the call
+ * @throws NullPointerException if the specified collection is null
+ * @see #add(Object)
+ */
+ public boolean addAll(Collection<? extends E> c) {
+ return al.addAllAbsent(c) > 0;
+ }
+
+ /**
+ * Removes from this set all of its elements that are contained in the
+ * specified collection. If the specified collection is also a set,
+ * this operation effectively modifies this set so that its value is the
+ * <i>asymmetric set difference</i> of the two sets.
+ *
+ * @param c collection containing elements to be removed from this set
+ * @return <tt>true</tt> if this set changed as a result of the call
+ * @throws ClassCastException if the class of an element of this set
+ * is incompatible with the specified collection (optional)
+ * @throws NullPointerException if this set contains a null element and the
+ * specified collection does not permit null elements (optional),
+ * or if the specified collection is null
+ * @see #remove(Object)
+ */
+ public boolean removeAll(Collection<?> c) {
+ return al.removeAll(c);
+ }
+
+ /**
+ * Retains only the elements in this set that are contained in the
+ * specified collection. In other words, removes from this set all of
+ * its elements that are not contained in the specified collection. If
+ * the specified collection is also a set, this operation effectively
+ * modifies this set so that its value is the <i>intersection</i> of the
+ * two sets.
+ *
+ * @param c collection containing elements to be retained in this set
+ * @return <tt>true</tt> if this set changed as a result of the call
+ * @throws ClassCastException if the class of an element of this set
+ * is incompatible with the specified collection (optional)
+ * @throws NullPointerException if this set contains a null element and the
+ * specified collection does not permit null elements (optional),
+ * or if the specified collection is null
+ * @see #remove(Object)
+ */
+ public boolean retainAll(Collection<?> c) {
+ return al.retainAll(c);
+ }
+
+ /**
+ * Returns an iterator over the elements contained in this set
+ * in the order in which these elements were added.
+ *
+ * <p>The returned iterator provides a snapshot of the state of the set
+ * when the iterator was constructed. No synchronization is needed while
+ * traversing the iterator. The iterator does <em>NOT</em> support the
+ * <tt>remove</tt> method.
+ *
+ * @return an iterator over the elements in this set
+ */
+ public Iterator<E> iterator() {
+ return al.iterator();
+ }
+
+ /**
+ * Compares the specified object with this set for equality.
+ * Returns {@code true} if the specified object is the same object
+ * as this object, or if it is also a {@link Set} and the elements
+ * returned by an {@linkplain List#iterator() iterator} over the
+ * specified set are the same as the elements returned by an
+ * iterator over this set. More formally, the two iterators are
+ * considered to return the same elements if they return the same
+ * number of elements and for every element {@code e1} returned by
+ * the iterator over the specified set, there is an element
+ * {@code e2} returned by the iterator over this set such that
+ * {@code (e1==null ? e2==null : e1.equals(e2))}.
+ *
+ * @param o object to be compared for equality with this set
+ * @return {@code true} if the specified object is equal to this set
+ */
+ public boolean equals(Object o) {
+ if (o == this)
+ return true;
+ if (!(o instanceof Set))
+ return false;
+ Set<?> set = (Set<?>)(o);
+ Iterator<?> it = set.iterator();
+
+ // Uses O(n^2) algorithm that is only appropriate
+ // for small sets, which CopyOnWriteArraySets should be.
+
+ // Use a single snapshot of underlying array
+ Object[] elements = al.getArray();
+ int len = elements.length;
+ // Mark matched elements to avoid re-checking
+ boolean[] matched = new boolean[len];
+ int k = 0;
+ outer: while (it.hasNext()) {
+ if (++k > len)
+ return false;
+ Object x = it.next();
+ for (int i = 0; i < len; ++i) {
+ if (!matched[i] && eq(x, elements[i])) {
+ matched[i] = true;
+ continue outer;
+ }
+ }
+ return false;
+ }
+ return k == len;
+ }
+
+ /**
+ * Test for equality, coping with nulls.
+ */
+ private static boolean eq(Object o1, Object o2) {
+ return (o1 == null ? o2 == null : o1.equals(o2));
+ }
}
Modified: harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/CountDownLatch.java
URL: http://svn.apache.org/viewvc/harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/CountDownLatch.java?rev=798469&r1=798468&r2=798469&view=diff
==============================================================================
--- harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/CountDownLatch.java (original)
+++ harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/CountDownLatch.java Tue Jul 28 09:30:33 2009
@@ -12,25 +12,25 @@
* A synchronization aid that allows one or more threads to wait until
* a set of operations being performed in other threads completes.
*
- * <p>A <tt>CountDownLatch</tt> is initialized with a given
- * <em>count</em>. The {@link #await await} methods block until the current
- * {@link #getCount count} reaches zero due to invocations of the
- * {@link #countDown} method, after which all waiting threads are
- * released and any subsequent invocations of {@link #await await} return
- * immediately. This is a one-shot phenomenon -- the count cannot be
- * reset. If you need a version that resets the count, consider using
- * a {@link CyclicBarrier}.
+ * <p>A {@code CountDownLatch} is initialized with a given <em>count</em>.
+ * The {@link #await await} methods block until the current count reaches
+ * zero due to invocations of the {@link #countDown} method, after which
+ * all waiting threads are released and any subsequent invocations of
+ * {@link #await await} return immediately. This is a one-shot phenomenon
+ * -- the count cannot be reset. If you need a version that resets the
+ * count, consider using a {@link CyclicBarrier}.
*
- * <p>A <tt>CountDownLatch</tt> is a versatile synchronization tool
+ * <p>A {@code CountDownLatch} is a versatile synchronization tool
* and can be used for a number of purposes. A
- * <tt>CountDownLatch</tt> initialized with a count of one serves as a
+ * {@code CountDownLatch} initialized with a count of one serves as a
* simple on/off latch, or gate: all threads invoking {@link #await await}
* wait at the gate until it is opened by a thread invoking {@link
- * #countDown}. A <tt>CountDownLatch</tt> initialized to <em>N</em>
+ * #countDown}. A {@code CountDownLatch} initialized to <em>N</em>
* can be used to make one thread wait until <em>N</em> threads have
* completed some action, or some action has been completed N times.
- * <p>A useful property of a <tt>CountDownLatch</tt> is that it
- * doesn't require that threads calling <tt>countDown</tt> wait for
+ *
+ * <p>A useful property of a {@code CountDownLatch} is that it
+ * doesn't require that threads calling {@code countDown} wait for
* the count to reach zero before proceeding, it simply prevents any
* thread from proceeding past an {@link #await await} until all
* threads could pass.
@@ -119,6 +119,13 @@
*
* </pre>
*
+ * <p>Memory consistency effects: Until the count reaches
+ * zero, actions in a thread prior to calling
+ * {@code countDown()}
+ * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
+ * actions following a successful return from a corresponding
+ * {@code await()} in another thread.
+ *
* @since 1.5
* @author Doug Lea
*/
@@ -128,118 +135,120 @@
* Uses AQS state to represent count.
*/
private static final class Sync extends AbstractQueuedSynchronizer {
+ private static final long serialVersionUID = 4982264981922014374L;
+
Sync(int count) {
- setState(count);
+ setState(count);
}
-
+
int getCount() {
return getState();
}
- public int tryAcquireShared(int acquires) {
+ protected int tryAcquireShared(int acquires) {
return getState() == 0? 1 : -1;
}
-
- public boolean tryReleaseShared(int releases) {
+
+ protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
- if (compareAndSetState(c, nextc))
+ if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}
private final Sync sync;
+
/**
- * Constructs a <tt>CountDownLatch</tt> initialized with the given
- * count.
- *
- * @param count the number of times {@link #countDown} must be invoked
- * before threads can pass through {@link #await}.
+ * Constructs a {@code CountDownLatch} initialized with the given count.
*
- * @throws IllegalArgumentException if <tt>count</tt> is less than zero.
+ * @param count the number of times {@link #countDown} must be invoked
+ * before threads can pass through {@link #await}
+ * @throws IllegalArgumentException if {@code count} is negative
*/
- public CountDownLatch(int count) {
+ public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
/**
- * Causes the current thread to wait until the latch has counted down to
- * zero, unless the thread is {@link Thread#interrupt interrupted}.
+ * Causes the current thread to wait until the latch has counted down to
+ * zero, unless the thread is {@linkplain Thread#interrupt interrupted}.
*
- * <p>If the current {@link #getCount count} is zero then this method
- * returns immediately.
- * <p>If the current {@link #getCount count} is greater than zero then
- * the current thread becomes disabled for thread scheduling
- * purposes and lies dormant until one of two things happen:
+ * <p>If the current count is zero then this method returns immediately.
+ *
+ * <p>If the current count is greater than zero then the current
+ * thread becomes disabled for thread scheduling purposes and lies
+ * dormant until one of two things happen:
* <ul>
* <li>The count reaches zero due to invocations of the
* {@link #countDown} method; or
- * <li>Some other thread {@link Thread#interrupt interrupts} the current
- * thread.
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts}
+ * the current thread.
* </ul>
+ *
* <p>If the current thread:
* <ul>
- * <li>has its interrupted status set on entry to this method; or
- * <li>is {@link Thread#interrupt interrupted} while waiting,
+ * <li>has its interrupted status set on entry to this method; or
+ * <li>is {@linkplain Thread#interrupt interrupted} while waiting,
* </ul>
- * then {@link InterruptedException} is thrown and the current thread's
- * interrupted status is cleared.
+ * then {@link InterruptedException} is thrown and the current thread's
+ * interrupted status is cleared.
*
* @throws InterruptedException if the current thread is interrupted
- * while waiting.
+ * while waiting
*/
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
/**
- * Causes the current thread to wait until the latch has counted down to
- * zero, unless the thread is {@link Thread#interrupt interrupted},
+ * Causes the current thread to wait until the latch has counted down to
+ * zero, unless the thread is {@linkplain Thread#interrupt interrupted},
* or the specified waiting time elapses.
*
- * <p>If the current {@link #getCount count} is zero then this method
- * returns immediately with the value <tt>true</tt>.
+ * <p>If the current count is zero then this method returns immediately
+ * with the value {@code true}.
*
- * <p>If the current {@link #getCount count} is greater than zero then
- * the current thread becomes disabled for thread scheduling
- * purposes and lies dormant until one of three things happen:
+ * <p>If the current count is greater than zero then the current
+ * thread becomes disabled for thread scheduling purposes and lies
+ * dormant until one of three things happen:
* <ul>
* <li>The count reaches zero due to invocations of the
* {@link #countDown} method; or
- * <li>Some other thread {@link Thread#interrupt interrupts} the current
- * thread; or
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts}
+ * the current thread; or
* <li>The specified waiting time elapses.
* </ul>
+ *
* <p>If the count reaches zero then the method returns with the
- * value <tt>true</tt>.
+ * value {@code true}.
+ *
* <p>If the current thread:
* <ul>
- * <li>has its interrupted status set on entry to this method; or
- * <li>is {@link Thread#interrupt interrupted} while waiting,
+ * <li>has its interrupted status set on entry to this method; or
+ * <li>is {@linkplain Thread#interrupt interrupted} while waiting,
* </ul>
- * then {@link InterruptedException} is thrown and the current thread's
- * interrupted status is cleared.
+ * then {@link InterruptedException} is thrown and the current thread's
+ * interrupted status is cleared.
*
- * <p>If the specified waiting time elapses then the value <tt>false</tt>
- * is returned.
- * If the time is
- * less than or equal to zero, the method will not wait at all.
+ * <p>If the specified waiting time elapses then the value {@code false}
+ * is returned. If the time is less than or equal to zero, the method
+ * will not wait at all.
*
* @param timeout the maximum time to wait
- * @param unit the time unit of the <tt>timeout</tt> argument.
- * @return <tt>true</tt> if the count reached zero and <tt>false</tt>
- * if the waiting time elapsed before the count reached zero.
- *
+ * @param unit the time unit of the {@code timeout} argument
+ * @return {@code true} if the count reached zero and {@code false}
+ * if the waiting time elapsed before the count reached zero
* @throws InterruptedException if the current thread is interrupted
- * while waiting.
+ * while waiting
*/
- public boolean await(long timeout, TimeUnit unit)
+ public boolean await(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
@@ -247,11 +256,12 @@
/**
* Decrements the count of the latch, releasing all waiting threads if
* the count reaches zero.
- * <p>If the current {@link #getCount count} is greater than zero then
- * it is decremented. If the new count is zero then all waiting threads
- * are re-enabled for thread scheduling purposes.
- * <p>If the current {@link #getCount count} equals zero then nothing
- * happens.
+ *
+ * <p>If the current count is greater than zero then it is decremented.
+ * If the new count is zero then all waiting threads are re-enabled for
+ * thread scheduling purposes.
+ *
+ * <p>If the current count equals zero then nothing happens.
*/
public void countDown() {
sync.releaseShared(1);
@@ -259,8 +269,10 @@
/**
* Returns the current count.
+ *
* <p>This method is typically used for debugging and testing purposes.
- * @return the current count.
+ *
+ * @return the current count
*/
public long getCount() {
return sync.getCount();
@@ -268,13 +280,12 @@
/**
* Returns a string identifying this latch, as well as its state.
- * The state, in brackets, includes the String
- * "Count =" followed by the current count.
- * @return a string identifying this latch, as well as its
- * state
+ * The state, in brackets, includes the String {@code "Count ="}
+ * followed by the current count.
+ *
+ * @return a string identifying this latch, as well as its state
*/
public String toString() {
return super.toString() + "[Count = " + sync.getCount() + "]";
}
-
}
Modified: harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/CyclicBarrier.java
URL: http://svn.apache.org/viewvc/harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/CyclicBarrier.java?rev=798469&r1=798468&r2=798469&view=diff
==============================================================================
--- harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/CyclicBarrier.java (original)
+++ harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/CyclicBarrier.java Tue Jul 28 09:30:33 2009
@@ -17,10 +17,10 @@
*
* <p>A <tt>CyclicBarrier</tt> supports an optional {@link Runnable} command
* that is run once per barrier point, after the last thread in the party
- * arrives, but before any threads are released.
+ * arrives, but before any threads are released.
* This <em>barrier action</em> is useful
* for updating shared-state before any of the parties continue.
- *
+ *
* <p><b>Sample usage:</b> Here is an example of
* using a barrier in a parallel decomposition design:
* <pre>
@@ -28,7 +28,7 @@
* final int N;
* final float[][] data;
* final CyclicBarrier barrier;
- *
+ *
* class Worker implements Runnable {
* int myRow;
* Worker(int row) { myRow = row; }
@@ -37,11 +37,11 @@
* processRow(myRow);
*
* try {
- * barrier.await();
- * } catch (InterruptedException ex) {
- * return;
- * } catch (BrokenBarrierException ex) {
- * return;
+ * barrier.await();
+ * } catch (InterruptedException ex) {
+ * return;
+ * } catch (BrokenBarrierException ex) {
+ * return;
* }
* }
* }
@@ -50,22 +50,22 @@
* public Solver(float[][] matrix) {
* data = matrix;
* N = matrix.length;
- * barrier = new CyclicBarrier(N,
+ * barrier = new CyclicBarrier(N,
* new Runnable() {
- * public void run() {
- * mergeRows(...);
+ * public void run() {
+ * mergeRows(...);
* }
* });
- * for (int i = 0; i < N; ++i)
+ * for (int i = 0; i < N; ++i)
* new Thread(new Worker(i)).start();
*
* waitUntilDone();
* }
* }
* </pre>
- * Here, each worker thread processes a row of the matrix then waits at the
+ * Here, each worker thread processes a row of the matrix then waits at the
* barrier until all rows have been processed. When all rows are processed
- * the supplied {@link Runnable} barrier action is executed and merges the
+ * the supplied {@link Runnable} barrier action is executed and merges the
* rows. If the merger
* determines that a solution has been found then <tt>done()</tt> will return
* <tt>true</tt> and each worker will terminate.
@@ -74,19 +74,26 @@
* it is executed, then any of the threads in the party could execute that
* action when it is released. To facilitate this, each invocation of
* {@link #await} returns the arrival index of that thread at the barrier.
- * You can then choose which thread should execute the barrier action, for
+ * You can then choose which thread should execute the barrier action, for
* example:
* <pre> if (barrier.await() == 0) {
* // log the completion of this iteration
* }</pre>
*
- * <p>The <tt>CyclicBarrier</tt> uses a fast-fail all-or-none breakage
- * model for failed synchronization attempts: If a thread leaves a
- * barrier point prematurely because of interruption, failure, or
- * timeout, all other threads, even those that have not yet resumed
- * from a previous {@link #await}. will also leave abnormally via
- * {@link BrokenBarrierException} (or <tt>InterruptedException</tt> if
- * they too were interrupted at about the same time).
+ * <p>The <tt>CyclicBarrier</tt> uses an all-or-none breakage model
+ * for failed synchronization attempts: If a thread leaves a barrier
+ * point prematurely because of interruption, failure, or timeout, all
+ * other threads waiting at that barrier point will also leave
+ * abnormally via {@link BrokenBarrierException} (or
+ * {@link InterruptedException} if they too were interrupted at about
+ * the same time).
+ *
+ * <p>Memory consistency effects: Actions in a thread prior to calling
+ * {@code await()}
+ * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
+ * actions that are part of the barrier action, which in turn
+ * <i>happen-before</i> actions following a successful return from the
+ * corresponding {@code await()} in other threads.
*
* @since 1.5
* @see CountDownLatch
@@ -94,6 +101,21 @@
* @author Doug Lea
*/
public class CyclicBarrier {
+ /**
+ * Each use of the barrier is represented as a generation instance.
+ * The generation changes whenever the barrier is tripped, or
+ * is reset. There can be many generations associated with threads
+ * using the barrier - due to the non-deterministic way the lock
+ * may be allocated to waiting threads - but only one of these
+ * can be active at a time (the one to which <tt>count</tt> applies)
+ * and all the rest are either broken or tripped.
+ * There need not be an active generation if there has been a break
+ * but no subsequent reset.
+ */
+ private static class Generation {
+ boolean broken = false;
+ }
+
/** The lock for guarding barrier entry */
private final ReentrantLock lock = new ReentrantLock();
/** Condition to wait on until tripped */
@@ -102,53 +124,50 @@
private final int parties;
/* The command to run when tripped */
private final Runnable barrierCommand;
-
- /**
- * The generation number. Incremented upon barrier trip.
- * Retracted upon reset.
- */
- private long generation;
-
- /**
- * Breakage indicator.
- */
- private boolean broken;
+ /** The current generation */
+ private Generation generation = new Generation();
/**
* Number of parties still waiting. Counts down from parties to 0
- * on each cycle.
+ * on each generation. It is reset to parties on each new
+ * generation or when broken.
*/
- private int count;
+ private int count;
/**
- * Updates state on barrier trip and wake up everyone.
- */
+ * Updates state on barrier trip and wakes up everyone.
+ * Called only while holding lock.
+ */
private void nextGeneration() {
- count = parties;
- ++generation;
+ // signal completion of last generation
trip.signalAll();
+ // set up next generation
+ count = parties;
+ generation = new Generation();
}
/**
- * Sets barrier as broken and wake up everyone
+ * Sets current barrier generation as broken and wakes up everyone.
+ * Called only while holding lock.
*/
private void breakBarrier() {
- broken = true;
+ generation.broken = true;
+ count = parties;
trip.signalAll();
}
/**
* Main barrier code, covering the various policies.
*/
- private int dowait(boolean timed, long nanos)
- throws InterruptedException, BrokenBarrierException, TimeoutException {
+ private int dowait(boolean timed, long nanos)
+ throws InterruptedException, BrokenBarrierException,
+ TimeoutException {
final ReentrantLock lock = this.lock;
lock.lock();
try {
- int index = --count;
- long g = generation;
+ final Generation g = generation;
- if (broken)
+ if (g.broken)
throw new BrokenBarrierException();
if (Thread.interrupted()) {
@@ -156,37 +175,45 @@
throw new InterruptedException();
}
- if (index == 0) { // tripped
- nextGeneration();
- boolean ranAction = false;
- try {
- Runnable command = barrierCommand;
- if (command != null)
- command.run();
- ranAction = true;
- return 0;
- } finally {
- if (!ranAction)
- breakBarrier();
- }
- }
+ int index = --count;
+ if (index == 0) { // tripped
+ boolean ranAction = false;
+ try {
+ final Runnable command = barrierCommand;
+ if (command != null)
+ command.run();
+ ranAction = true;
+ nextGeneration();
+ return 0;
+ } finally {
+ if (!ranAction)
+ breakBarrier();
+ }
+ }
+ // loop until tripped, broken, interrupted, or timed out
for (;;) {
try {
- if (!timed)
+ if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
- breakBarrier();
- throw ie;
+ if (g == generation && ! g.broken) {
+ breakBarrier();
+ throw ie;
+ } else {
+ // We're about to finish waiting even if we had not
+ // been interrupted, so this interrupt is deemed to
+ // "belong" to subsequent execution.
+ Thread.currentThread().interrupt();
+ }
}
-
- if (broken ||
- g > generation) // true if a reset occurred while waiting
+
+ if (g.broken)
throw new BrokenBarrierException();
- if (g < generation)
+ if (g != generation)
return index;
if (timed && nanos <= 0L) {
@@ -194,7 +221,6 @@
throw new TimeoutException();
}
}
-
} finally {
lock.unlock();
}
@@ -207,15 +233,14 @@
* performed by the last thread entering the barrier.
*
* @param parties the number of threads that must invoke {@link #await}
- * before the barrier is tripped.
+ * before the barrier is tripped
* @param barrierAction the command to execute when the barrier is
- * tripped, or <tt>null</tt> if there is no action.
- *
- * @throws IllegalArgumentException if <tt>parties</tt> is less than 1.
+ * tripped, or {@code null} if there is no action
+ * @throws IllegalArgumentException if {@code parties} is less than 1
*/
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
- this.parties = parties;
+ this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
@@ -223,12 +248,11 @@
/**
* Creates a new <tt>CyclicBarrier</tt> that will trip when the
* given number of parties (threads) are waiting upon it, and
- * does not perform a predefined action upon each barrier.
+ * does not perform a predefined action when the barrier is tripped.
*
* @param parties the number of threads that must invoke {@link #await}
- * before the barrier is tripped.
- *
- * @throws IllegalArgumentException if <tt>parties</tt> is less than 1.
+ * before the barrier is tripped
+ * @throws IllegalArgumentException if {@code parties} is less than 1
*/
public CyclicBarrier(int parties) {
this(parties, null);
@@ -236,64 +260,66 @@
/**
* Returns the number of parties required to trip this barrier.
- * @return the number of parties required to trip this barrier.
- **/
+ *
+ * @return the number of parties required to trip this barrier
+ */
public int getParties() {
return parties;
}
/**
- * Waits until all {@link #getParties parties} have invoked <tt>await</tt>
- * on this barrier.
+ * Waits until all {@linkplain #getParties parties} have invoked
+ * <tt>await</tt> on this barrier.
*
* <p>If the current thread is not the last to arrive then it is
* disabled for thread scheduling purposes and lies dormant until
- * one of following things happens:
+ * one of the following things happens:
* <ul>
* <li>The last thread arrives; or
- * <li>Some other thread {@link Thread#interrupt interrupts} the current
- * thread; or
- * <li>Some other thread {@link Thread#interrupt interrupts} one of the
- * other waiting threads; or
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts}
+ * the current thread; or
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts}
+ * one of the other waiting threads; or
* <li>Some other thread times out while waiting for barrier; or
* <li>Some other thread invokes {@link #reset} on this barrier.
* </ul>
+ *
* <p>If the current thread:
* <ul>
* <li>has its interrupted status set on entry to this method; or
- * <li>is {@link Thread#interrupt interrupted} while waiting
+ * <li>is {@linkplain Thread#interrupt interrupted} while waiting
* </ul>
* then {@link InterruptedException} is thrown and the current thread's
* interrupted status is cleared.
*
- * <p>If the barrier is {@link #reset} while any thread is waiting, or if
- * the barrier {@link #isBroken is broken} when <tt>await</tt> is invoked,
- * or while any thread is waiting,
- * then {@link BrokenBarrierException} is thrown.
+ * <p>If the barrier is {@link #reset} while any thread is waiting,
+ * or if the barrier {@linkplain #isBroken is broken} when
+ * <tt>await</tt> is invoked, or while any thread is waiting, then
+ * {@link BrokenBarrierException} is thrown.
*
- * <p>If any thread is {@link Thread#interrupt interrupted} while waiting,
- * then all other waiting threads will throw
+ * <p>If any thread is {@linkplain Thread#interrupt interrupted} while waiting,
+ * then all other waiting threads will throw
* {@link BrokenBarrierException} and the barrier is placed in the broken
* state.
*
* <p>If the current thread is the last thread to arrive, and a
* non-null barrier action was supplied in the constructor, then the
- * current thread runs the action before allowing the other threads to
+ * current thread runs the action before allowing the other threads to
* continue.
* If an exception occurs during the barrier action then that exception
* will be propagated in the current thread and the barrier is placed in
* the broken state.
*
* @return the arrival index of the current thread, where index
- * <tt>{@link #getParties()} - 1</tt> indicates the first to arrive and
- * zero indicates the last to arrive.
- *
- * @throws InterruptedException if the current thread was interrupted
- * while waiting
+ * <tt>{@link #getParties()} - 1</tt> indicates the first
+ * to arrive and zero indicates the last to arrive
+ * @throws InterruptedException if the current thread was interrupted
+ * while waiting
* @throws BrokenBarrierException if <em>another</em> thread was
- * interrupted while the current thread was waiting, or the barrier was
- * reset, or the barrier was broken when <tt>await</tt> was called,
- * or the barrier action (if present) failed due an exception.
+ * interrupted or timed out while the current thread was
+ * waiting, or the barrier was reset, or the barrier was
+ * broken when {@code await} was called, or the barrier
+ * action (if present) failed due an exception.
*/
public int await() throws InterruptedException, BrokenBarrierException {
try {
@@ -304,8 +330,8 @@
}
/**
- * Waits until all {@link #getParties parties} have invoked <tt>await</tt>
- * on this barrier.
+ * Waits until all {@linkplain #getParties parties} have invoked
+ * <tt>await</tt> on this barrier, or the specified waiting time elapses.
*
* <p>If the current thread is not the last to arrive then it is
* disabled for thread scheduling purposes and lies dormant until
@@ -313,34 +339,39 @@
* <ul>
* <li>The last thread arrives; or
* <li>The specified timeout elapses; or
- * <li>Some other thread {@link Thread#interrupt interrupts} the current
- * thread; or
- * <li>Some other thread {@link Thread#interrupt interrupts} one of the
- * other waiting threads; or
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts}
+ * the current thread; or
+ * <li>Some other thread {@linkplain Thread#interrupt interrupts}
+ * one of the other waiting threads; or
* <li>Some other thread times out while waiting for barrier; or
* <li>Some other thread invokes {@link #reset} on this barrier.
* </ul>
+ *
* <p>If the current thread:
* <ul>
* <li>has its interrupted status set on entry to this method; or
- * <li>is {@link Thread#interrupt interrupted} while waiting
+ * <li>is {@linkplain Thread#interrupt interrupted} while waiting
* </ul>
* then {@link InterruptedException} is thrown and the current thread's
* interrupted status is cleared.
*
- * <p>If the barrier is {@link #reset} while any thread is waiting, or if
- * the barrier {@link #isBroken is broken} when <tt>await</tt> is invoked,
- * or while any thread is waiting,
- * then {@link BrokenBarrierException} is thrown.
- *
- * <p>If any thread is {@link Thread#interrupt interrupted} while waiting,
- * then all other waiting threads will throw
- * {@link BrokenBarrierException} and the barrier is placed in the broken
+ * <p>If the specified waiting time elapses then {@link TimeoutException}
+ * is thrown. If the time is less than or equal to zero, the
+ * method will not wait at all.
+ *
+ * <p>If the barrier is {@link #reset} while any thread is waiting,
+ * or if the barrier {@linkplain #isBroken is broken} when
+ * <tt>await</tt> is invoked, or while any thread is waiting, then
+ * {@link BrokenBarrierException} is thrown.
+ *
+ * <p>If any thread is {@linkplain Thread#interrupt interrupted} while
+ * waiting, then all other waiting threads will throw {@link
+ * BrokenBarrierException} and the barrier is placed in the broken
* state.
*
* <p>If the current thread is the last thread to arrive, and a
* non-null barrier action was supplied in the constructor, then the
- * current thread runs the action before allowing the other threads to
+ * current thread runs the action before allowing the other threads to
* continue.
* If an exception occurs during the barrier action then that exception
* will be propagated in the current thread and the barrier is placed in
@@ -349,36 +380,37 @@
* @param timeout the time to wait for the barrier
* @param unit the time unit of the timeout parameter
* @return the arrival index of the current thread, where index
- * <tt>{@link #getParties()} - 1</tt> indicates the first to arrive and
- * zero indicates the last to arrive.
- *
- * @throws InterruptedException if the current thread was interrupted
- * while waiting
- * @throws TimeoutException if the specified timeout elapses.
+ * <tt>{@link #getParties()} - 1</tt> indicates the first
+ * to arrive and zero indicates the last to arrive
+ * @throws InterruptedException if the current thread was interrupted
+ * while waiting
+ * @throws TimeoutException if the specified timeout elapses
* @throws BrokenBarrierException if <em>another</em> thread was
- * interrupted while the current thread was waiting, or the barrier was
- * reset, or the barrier was broken when <tt>await</tt> was called,
- * or the barrier action (if present) failed due an exception.
- */
- public int await(long timeout, TimeUnit unit)
- throws InterruptedException,
- BrokenBarrierException,
- TimeoutException {
+ * interrupted or timed out while the current thread was
+ * waiting, or the barrier was reset, or the barrier was broken
+ * when {@code await} was called, or the barrier action (if
+ * present) failed due an exception
+ */
+ public int await(long timeout, TimeUnit unit)
+ throws InterruptedException,
+ BrokenBarrierException,
+ TimeoutException {
return dowait(true, unit.toNanos(timeout));
}
/**
* Queries if this barrier is in a broken state.
- * @return <tt>true</tt> if one or more parties broke out of this
- * barrier due to interruption or timeout since construction or
- * the last reset, or a barrier action failed due to an exception;
- * and <tt>false</tt> otherwise.
+ *
+ * @return {@code true} if one or more parties broke out of this
+ * barrier due to interruption or timeout since
+ * construction or the last reset, or a barrier action
+ * failed due to an exception; {@code false} otherwise.
*/
public boolean isBroken() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
- return broken;
+ return generation.broken;
} finally {
lock.unlock();
}
@@ -397,15 +429,8 @@
final ReentrantLock lock = this.lock;
lock.lock();
try {
- /*
- * Retract generation number enough to cover threads
- * currently waiting on current and still resuming from
- * previous generation, plus similarly accommodating spans
- * after the reset.
- */
- generation -= 4;
- broken = false;
- trip.signalAll();
+ breakBarrier(); // break the current generation
+ nextGeneration(); // start a new generation
} finally {
lock.unlock();
}
@@ -416,7 +441,7 @@
* This method is primarily useful for debugging and assertions.
*
* @return the number of parties currently blocked in {@link #await}
- **/
+ */
public int getNumberWaiting() {
final ReentrantLock lock = this.lock;
lock.lock();
@@ -426,5 +451,4 @@
lock.unlock();
}
}
-
}
Modified: harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/DelayQueue.java
URL: http://svn.apache.org/viewvc/harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/DelayQueue.java?rev=798469&r1=798468&r2=798469&view=diff
==============================================================================
--- harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/DelayQueue.java (original)
+++ harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/DelayQueue.java Tue Jul 28 09:30:33 2009
@@ -10,17 +10,25 @@
import java.util.*;
/**
- * An unbounded {@linkplain BlockingQueue blocking queue} of <tt>Delayed</tt>
- * elements, in which an element can only be taken when its delay has expired.
- * The <em>head</em> of the queue is that <tt>Delayed</tt> element whose delay
- * expired furthest in the past - if no delay has expired there is no head and
- * <tt>poll</tt> will return <tt>null</tt>.
- * This queue does not permit <tt>null</tt> elements.
- * <p>This class implements all of the <em>optional</em> methods
- * of the {@link Collection} and {@link Iterator} interfaces.
+ * An unbounded {@linkplain BlockingQueue blocking queue} of
+ * <tt>Delayed</tt> elements, in which an element can only be taken
+ * when its delay has expired. The <em>head</em> of the queue is that
+ * <tt>Delayed</tt> element whose delay expired furthest in the
+ * past. If no delay has expired there is no head and <tt>poll</tt>
+ * will return <tt>null</tt>. Expiration occurs when an element's
+ * <tt>getDelay(TimeUnit.NANOSECONDS)</tt> method returns a value less
+ * than or equal to zero. Even though unexpired elements cannot be
+ * removed using <tt>take</tt> or <tt>poll</tt>, they are otherwise
+ * treated as normal elements. For example, the <tt>size</tt> method
+ * returns the count of both expired and unexpired elements.
+ * This queue does not permit null elements.
+ *
+ * <p>This class and its iterator implement all of the
+ * <em>optional</em> methods of the {@link Collection} and {@link
+ * Iterator} interfaces.
*
* <p>This class is a member of the
- * <a href="{@docRoot}/../guide/collections/index.html">
+ * <a href="{@docRoot}/../technotes/guides/collections/index.html">
* Java Collections Framework</a>.
*
* @since 1.5
@@ -32,10 +40,34 @@
implements BlockingQueue<E> {
private transient final ReentrantLock lock = new ReentrantLock();
- private transient final Condition available = lock.newCondition();
private final PriorityQueue<E> q = new PriorityQueue<E>();
/**
+ * Thread designated to wait for the element at the head of
+ * the queue. This variant of the Leader-Follower pattern
+ * (http://www.cs.wustl.edu/~schmidt/POSA/POSA2/) serves to
+ * minimize unnecessary timed waiting. When a thread becomes
+ * the leader, it waits only for the next delay to elapse, but
+ * other threads await indefinitely. The leader thread must
+ * signal some other thread before returning from take() or
+ * poll(...), unless some other thread becomes leader in the
+ * interim. Whenever the head of the queue is replaced with
+ * an element with an earlier expiration time, the leader
+ * field is invalidated by being reset to null, and some
+ * waiting thread, but not necessarily the current leader, is
+ * signalled. So waiting threads must be prepared to acquire
+ * and lose leadership while waiting.
+ */
+ private Thread leader = null;
+
+ /**
+ * Condition signalled when a newer element becomes available
+ * at the head of the queue or a new thread may need to
+ * become leader.
+ */
+ private final Condition available = lock.newCondition();
+
+ /**
* Creates a new <tt>DelayQueue</tt> that is initially empty.
*/
public DelayQueue() {}
@@ -44,10 +76,9 @@
* Creates a <tt>DelayQueue</tt> initially containing the elements of the
* given collection of {@link Delayed} instances.
*
- * @param c the collection
- * @throws NullPointerException if <tt>c</tt> or any element within it
- * is <tt>null</tt>
- *
+ * @param c the collection of elements to initially contain
+ * @throws NullPointerException if the specified collection or any
+ * of its elements are null
*/
public DelayQueue(Collection<? extends E> c) {
this.addAll(c);
@@ -56,91 +87,136 @@
/**
* Inserts the specified element into this delay queue.
*
- * @param o the element to add
+ * @param e the element to add
+ * @return <tt>true</tt> (as specified by {@link Collection#add})
+ * @throws NullPointerException if the specified element is null
+ */
+ public boolean add(E e) {
+ return offer(e);
+ }
+
+ /**
+ * Inserts the specified element into this delay queue.
+ *
+ * @param e the element to add
* @return <tt>true</tt>
- * @throws NullPointerException if the specified element is <tt>null</tt>.
+ * @throws NullPointerException if the specified element is null
*/
- public boolean offer(E o) {
+ public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
- E first = q.peek();
- q.offer(o);
- if (first == null || o.compareTo(first) < 0)
- available.signalAll();
+ q.offer(e);
+ if (q.peek() == e) {
+ leader = null;
+ available.signal();
+ }
return true;
} finally {
lock.unlock();
}
}
-
/**
- * Adds the specified element to this delay queue. As the queue is
+ * Inserts the specified element into this delay queue. As the queue is
* unbounded this method will never block.
- * @param o the element to add
- * @throws NullPointerException if the specified element is <tt>null</tt>.
+ *
+ * @param e the element to add
+ * @throws NullPointerException {@inheritDoc}
*/
- public void put(E o) {
- offer(o);
+ public void put(E e) {
+ offer(e);
}
/**
* Inserts the specified element into this delay queue. As the queue is
* unbounded this method will never block.
- * @param o the element to add
+ *
+ * @param e the element to add
* @param timeout This parameter is ignored as the method never blocks
* @param unit This parameter is ignored as the method never blocks
* @return <tt>true</tt>
- * @throws NullPointerException if the specified element is <tt>null</tt>.
+ * @throws NullPointerException {@inheritDoc}
*/
- public boolean offer(E o, long timeout, TimeUnit unit) {
- return offer(o);
+ public boolean offer(E e, long timeout, TimeUnit unit) {
+ return offer(e);
}
/**
- * Adds the specified element to this queue.
- * @param o the element to add
- * @return <tt>true</tt> (as per the general contract of
- * <tt>Collection.add</tt>).
+ * Retrieves and removes the head of this queue, or returns <tt>null</tt>
+ * if this queue has no elements with an expired delay.
*
- * @throws NullPointerException if the specified element is <tt>null</tt>.
+ * @return the head of this queue, or <tt>null</tt> if this
+ * queue has no elements with an expired delay
*/
- public boolean add(E o) {
- return offer(o);
+ public E poll() {
+ final ReentrantLock lock = this.lock;
+ lock.lock();
+ try {
+ E first = q.peek();
+ if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)
+ return null;
+ else
+ return q.poll();
+ } finally {
+ lock.unlock();
+ }
}
+ /**
+ * Retrieves and removes the head of this queue, waiting if necessary
+ * until an element with an expired delay is available on this queue.
+ *
+ * @return the head of this queue
+ * @throws InterruptedException {@inheritDoc}
+ */
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
E first = q.peek();
- if (first == null) {
+ if (first == null)
available.await();
- } else {
- long delay = first.getDelay(TimeUnit.NANOSECONDS);
- if (delay > 0) {
- long tl = available.awaitNanos(delay);
- } else {
- E x = q.poll();
- assert x != null;
- if (q.size() != 0)
- available.signalAll(); // wake up other takers
- return x;
-
+ else {
+ long delay = first.getDelay(TimeUnit.NANOSECONDS);
+ if (delay <= 0)
+ return q.poll();
+ else if (leader != null)
+ available.await();
+ else {
+ Thread thisThread = Thread.currentThread();
+ leader = thisThread;
+ try {
+ available.awaitNanos(delay);
+ } finally {
+ if (leader == thisThread)
+ leader = null;
+ }
}
}
}
} finally {
+ if (leader == null && q.peek() != null)
+ available.signal();
lock.unlock();
}
}
- public E poll(long time, TimeUnit unit) throws InterruptedException {
+ /**
+ * Retrieves and removes the head of this queue, waiting if necessary
+ * until an element with an expired delay is available on this queue,
+ * or the specified wait time expires.
+ *
+ * @return the head of this queue, or <tt>null</tt> if the
+ * specified waiting time elapses before an element with
+ * an expired delay becomes available
+ * @throws InterruptedException {@inheritDoc}
+ */
+ public E poll(long timeout, TimeUnit unit) throws InterruptedException {
+ long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
- long nanos = unit.toNanos(time);
try {
for (;;) {
E first = q.peek();
@@ -150,46 +226,43 @@
else
nanos = available.awaitNanos(nanos);
} else {
- long delay = first.getDelay(TimeUnit.NANOSECONDS);
- if (delay > 0) {
- if (delay > nanos)
- delay = nanos;
- long timeLeft = available.awaitNanos(delay);
- nanos -= delay - timeLeft;
- } else {
- E x = q.poll();
- assert x != null;
- if (q.size() != 0)
- available.signalAll();
- return x;
+ long delay = first.getDelay(TimeUnit.NANOSECONDS);
+ if (delay <= 0)
+ return q.poll();
+ if (nanos <= 0)
+ return null;
+ if (nanos < delay || leader != null)
+ nanos = available.awaitNanos(nanos);
+ else {
+ Thread thisThread = Thread.currentThread();
+ leader = thisThread;
+ try {
+ long timeLeft = available.awaitNanos(delay);
+ nanos -= delay - timeLeft;
+ } finally {
+ if (leader == thisThread)
+ leader = null;
+ }
}
}
}
} finally {
+ if (leader == null && q.peek() != null)
+ available.signal();
lock.unlock();
}
}
-
- public E poll() {
- final ReentrantLock lock = this.lock;
- lock.lock();
- try {
- E first = q.peek();
- if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)
- return null;
- else {
- E x = q.poll();
- assert x != null;
- if (q.size() != 0)
- available.signalAll();
- return x;
- }
- } finally {
- lock.unlock();
- }
- }
-
+ /**
+ * Retrieves, but does not remove, the head of this queue, or
+ * returns <tt>null</tt> if this queue is empty. Unlike
+ * <tt>poll</tt>, if no expired elements are available in the queue,
+ * this method returns the element that will expire next,
+ * if one exists.
+ *
+ * @return the head of this queue, or <tt>null</tt> if this
+ * queue is empty.
+ */
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
@@ -210,6 +283,12 @@
}
}
+ /**
+ * @throws UnsupportedOperationException {@inheritDoc}
+ * @throws ClassCastException {@inheritDoc}
+ * @throws NullPointerException {@inheritDoc}
+ * @throws IllegalArgumentException {@inheritDoc}
+ */
public int drainTo(Collection<? super E> c) {
if (c == null)
throw new NullPointerException();
@@ -226,14 +305,18 @@
c.add(q.poll());
++n;
}
- if (n > 0)
- available.signalAll();
return n;
} finally {
lock.unlock();
}
}
+ /**
+ * @throws UnsupportedOperationException {@inheritDoc}
+ * @throws ClassCastException {@inheritDoc}
+ * @throws NullPointerException {@inheritDoc}
+ * @throws IllegalArgumentException {@inheritDoc}
+ */
public int drainTo(Collection<? super E> c, int maxElements) {
if (c == null)
throw new NullPointerException();
@@ -252,8 +335,6 @@
c.add(q.poll());
++n;
}
- if (n > 0)
- available.signalAll();
return n;
} finally {
lock.unlock();
@@ -263,6 +344,8 @@
/**
* Atomically removes all of the elements from this delay queue.
* The queue will be empty after this call returns.
+ * Elements with an unexpired delay are not waited for; they are
+ * simply discarded from the queue.
*/
public void clear() {
final ReentrantLock lock = this.lock;
@@ -277,12 +360,26 @@
/**
* Always returns <tt>Integer.MAX_VALUE</tt> because
* a <tt>DelayQueue</tt> is not capacity constrained.
+ *
* @return <tt>Integer.MAX_VALUE</tt>
*/
public int remainingCapacity() {
return Integer.MAX_VALUE;
}
+ /**
+ * Returns an array containing all of the elements in this queue.
+ * The returned array elements are in no particular order.
+ *
+ * <p>The returned array will be "safe" in that no references to it are
+ * maintained by this queue. (In other words, this method must allocate
+ * a new array). The caller is thus free to modify the returned array.
+ *
+ * <p>This method acts as bridge between array-based and collection-based
+ * APIs.
+ *
+ * @return an array containing all of the elements in this queue
+ */
public Object[] toArray() {
final ReentrantLock lock = this.lock;
lock.lock();
@@ -293,16 +390,56 @@
}
}
- public <T> T[] toArray(T[] array) {
+ /**
+ * Returns an array containing all of the elements in this queue; the
+ * runtime type of the returned array is that of the specified array.
+ * The returned array elements are in no particular order.
+ * If the queue fits in the specified array, it is returned therein.
+ * Otherwise, a new array is allocated with the runtime type of the
+ * specified array and the size of this queue.
+ *
+ * <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>.
+ *
+ * <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>The following code can be used to dump a delay queue into a newly
+ * allocated array of <tt>Delayed</tt>:
+ *
+ * <pre>
+ * Delayed[] a = q.toArray(new Delayed[0]);</pre>
+ *
+ * Note that <tt>toArray(new Object[0])</tt> is identical in function to
+ * <tt>toArray()</tt>.
+ *
+ * @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
+ * same runtime type is allocated for this purpose
+ * @return an array containing all of the elements in this queue
+ * @throws ArrayStoreException if the runtime type of the specified array
+ * is not a supertype of the runtime type of every element in
+ * this queue
+ * @throws NullPointerException if the specified array is null
+ */
+ public <T> T[] toArray(T[] a) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
- return q.toArray(array);
+ return q.toArray(a);
} finally {
lock.unlock();
}
}
+ /**
+ * Removes a single instance of the specified element from this
+ * queue, if it is present, whether or not it has expired.
+ */
public boolean remove(Object o) {
final ReentrantLock lock = this.lock;
lock.lock();
@@ -314,49 +451,61 @@
}
/**
- * Returns an iterator over the elements in this queue. The iterator
- * does not return the elements in any particular order. The
- * returned iterator is a thread-safe "fast-fail" iterator that will
- * throw {@link java.util.ConcurrentModificationException}
- * upon detected interference.
+ * Returns an iterator over all the elements (both expired and
+ * unexpired) in this queue. The iterator does not return the
+ * elements in any particular order. The returned
+ * <tt>Iterator</tt> is a "weakly consistent" iterator that will
+ * never throw {@link ConcurrentModificationException}, and
+ * guarantees to traverse elements as they existed upon
+ * construction of the iterator, and may (but is not guaranteed
+ * to) reflect any modifications subsequent to construction.
*
- * @return an iterator over the elements in this queue.
+ * @return an iterator over the elements in this queue
*/
public Iterator<E> iterator() {
- final ReentrantLock lock = this.lock;
- lock.lock();
- try {
- return new Itr(q.iterator());
- } finally {
- lock.unlock();
- }
+ return new Itr(toArray());
}
- private class Itr<E> implements Iterator<E> {
- private final Iterator<E> iter;
- Itr(Iterator<E> i) {
- iter = i;
+ /**
+ * Snapshot iterator that works off copy of underlying q array.
+ */
+ private class Itr implements Iterator<E> {
+ final Object[] array; // Array of all elements
+ int cursor; // index of next element to return;
+ int lastRet; // index of last element, or -1 if no such
+
+ Itr(Object[] array) {
+ lastRet = -1;
+ this.array = array;
}
public boolean hasNext() {
- return iter.hasNext();
+ return cursor < array.length;
}
+ @SuppressWarnings("unchecked")
public E next() {
- final ReentrantLock lock = DelayQueue.this.lock;
- lock.lock();
- try {
- return iter.next();
- } finally {
- lock.unlock();
- }
+ if (cursor >= array.length)
+ throw new NoSuchElementException();
+ lastRet = cursor;
+ return (E)array[cursor++];
}
public void remove() {
- final ReentrantLock lock = DelayQueue.this.lock;
+ if (lastRet < 0)
+ throw new IllegalStateException();
+ Object x = array[lastRet];
+ lastRet = -1;
+ // Traverse underlying queue to find == element,
+ // not just a .equals element.
lock.lock();
try {
- iter.remove();
+ for (Iterator it = q.iterator(); it.hasNext(); ) {
+ if (it.next() == x) {
+ it.remove();
+ return;
+ }
+ }
} finally {
lock.unlock();
}
Modified: harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/Delayed.java
URL: http://svn.apache.org/viewvc/harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/Delayed.java?rev=798469&r1=798468&r2=798469&view=diff
==============================================================================
--- harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/Delayed.java (original)
+++ harmony/enhanced/classlib/branches/java6/modules/concurrent/src/main/java/java/util/concurrent/Delayed.java Tue Jul 28 09:30:33 2009
@@ -27,11 +27,12 @@
public interface Delayed extends Comparable<Delayed> {
/**
- * Returns the delay associated with this object, in the given time unit.
+ * Returns the remaining delay associated with this object, in the
+ * given time unit.
*
* @param unit the time unit
- * @return the delay; zero or negative values indicate that the
- * delay has already elapsed
+ * @return the remaining delay; zero or negative values indicate
+ * that the delay has already elapsed
*/
long getDelay(TimeUnit unit);
}