Extension example. Here is a sketch of a class + *
Extension example. Here is a sketch of a class * that customizes {@link ThreadPoolExecutor} to use * a {@code CustomTask} class instead of the default {@code FutureTask}: *
{@code
* public class CustomThreadPoolExecutor extends ThreadPoolExecutor {
*
- * static class CustomTask implements RunnableFuture {...}
+ * static class CustomTask implements RunnableFuture { ... }
*
* protected RunnableFuture newTaskFor(Callable c) {
* return new CustomTask(c);
diff --git a/src/java.base/share/classes/java/util/concurrent/CopyOnWriteArraySet.java b/src/java.base/share/classes/java/util/concurrent/CopyOnWriteArraySet.java
--- a/src/java.base/share/classes/java/util/concurrent/CopyOnWriteArraySet.java
+++ b/src/java.base/share/classes/java/util/concurrent/CopyOnWriteArraySet.java
@@ -69,7 +69,7 @@
* perform some action upon state updates.
*
* {@code
- * class Handler { void handle(); ... }
+ * class Handler { void handle() { ... } }
*
* class X {
* private final CopyOnWriteArraySet handlers
diff --git a/src/java.base/share/classes/java/util/concurrent/CountDownLatch.java b/src/java.base/share/classes/java/util/concurrent/CountDownLatch.java
--- a/src/java.base/share/classes/java/util/concurrent/CountDownLatch.java
+++ b/src/java.base/share/classes/java/util/concurrent/CountDownLatch.java
@@ -118,7 +118,7 @@
* class Driver2 { // ...
* void main() throws InterruptedException {
* CountDownLatch doneSignal = new CountDownLatch(N);
- * Executor e = ...
+ * Executor e = ...;
*
* for (int i = 0; i < N; ++i) // create and start threads
* e.execute(new WorkerRunnable(doneSignal, i));
@@ -135,10 +135,8 @@
* this.i = i;
* }
* public void run() {
- * try {
- * doWork(i);
- * doneSignal.countDown();
- * } catch (InterruptedException ex) {} // return;
+ * doWork();
+ * doneSignal.countDown();
* }
*
* void doWork() { ... }
diff --git a/src/java.base/share/classes/java/util/concurrent/CyclicBarrier.java b/src/java.base/share/classes/java/util/concurrent/CyclicBarrier.java
--- a/src/java.base/share/classes/java/util/concurrent/CyclicBarrier.java
+++ b/src/java.base/share/classes/java/util/concurrent/CyclicBarrier.java
@@ -94,7 +94,9 @@
*
* // wait until done
* for (Thread thread : threads)
- * thread.join();
+ * try {
+ * thread.join();
+ * } catch (InterruptedException ex) { }
* }
* }}
*
diff --git a/src/java.base/share/classes/java/util/concurrent/Exchanger.java b/src/java.base/share/classes/java/util/concurrent/Exchanger.java
--- a/src/java.base/share/classes/java/util/concurrent/Exchanger.java
+++ b/src/java.base/share/classes/java/util/concurrent/Exchanger.java
@@ -57,8 +57,8 @@
* {@code
* class FillAndEmpty {
* Exchanger exchanger = new Exchanger<>();
- * DataBuffer initialEmptyBuffer = ... a made-up type
- * DataBuffer initialFullBuffer = ...
+ * DataBuffer initialEmptyBuffer = ...; // a made-up type
+ * DataBuffer initialFullBuffer = ...;
*
* class FillingLoop implements Runnable {
* public void run() {
@@ -69,7 +69,7 @@
* if (currentBuffer.isFull())
* currentBuffer = exchanger.exchange(currentBuffer);
* }
- * } catch (InterruptedException ex) { ... handle ... }
+ * } catch (InterruptedException ex) { ... handle ...}
* }
* }
*
diff --git a/src/java.base/share/classes/java/util/concurrent/ExecutorService.java b/src/java.base/share/classes/java/util/concurrent/ExecutorService.java
--- a/src/java.base/share/classes/java/util/concurrent/ExecutorService.java
+++ b/src/java.base/share/classes/java/util/concurrent/ExecutorService.java
@@ -117,7 +117,7 @@
* if (!pool.awaitTermination(60, TimeUnit.SECONDS))
* System.err.println("Pool did not terminate");
* }
- * } catch (InterruptedException ie) {
+ * } catch (InterruptedException ex) {
* // (Re-)Cancel if current thread also interrupted
* pool.shutdownNow();
* // Preserve interrupt status
diff --git a/src/java.base/share/classes/java/util/concurrent/Future.java b/src/java.base/share/classes/java/util/concurrent/Future.java
--- a/src/java.base/share/classes/java/util/concurrent/Future.java
+++ b/src/java.base/share/classes/java/util/concurrent/Future.java
@@ -56,8 +56,8 @@
* {@code
* interface ArchiveSearcher { String search(String target); }
* class App {
- * ExecutorService executor = ...
- * ArchiveSearcher searcher = ...
+ * ExecutorService executor = ...;
+ * ArchiveSearcher searcher = ...;
* void showSearch(String target) throws InterruptedException {
* Callable task = () -> searcher.search(target);
* Future future = executor.submit(task);
@@ -88,25 +88,28 @@
public interface Future {
/**
- * Attempts to cancel execution of this task. This attempt will
- * fail if the task has already completed, has already been cancelled,
- * or could not be cancelled for some other reason. If successful,
- * and this task has not started when {@code cancel} is called,
- * this task should never run. If the task has already started,
- * then the {@code mayInterruptIfRunning} parameter determines
- * whether the thread executing this task should be interrupted in
- * an attempt to stop the task.
+ * Attempts to cancel execution of this task. This method has no
+ * effect if the task is already completed or cancelled, or could
+ * not be cancelled for some other reason. Otherwise, if this
+ * task has not started when {@code cancel} is called, this task
+ * should never run. If the task has already started, then the
+ * {@code mayInterruptIfRunning} parameter determines whether the
+ * thread executing this task (when known by the implementation)
+ * is interrupted in an attempt to stop the task.
*
- * After this method returns, subsequent calls to {@link #isDone} will
- * always return {@code true}. Subsequent calls to {@link #isCancelled}
- * will always return {@code true} if this method returned {@code true}.
+ *
The return value from this method does not necessarily
+ * indicate whether the task is now cancelled; use {@link
+ * #isCancelled}.
*
- * @param mayInterruptIfRunning {@code true} if the thread executing this
- * task should be interrupted; otherwise, in-progress tasks are allowed
- * to complete
+ * @param mayInterruptIfRunning {@code true} if the thread
+ * executing this task should be interrupted (if the thread is
+ * known to the implementation); otherwise, in-progress tasks are
+ * allowed to complete
* @return {@code false} if the task could not be cancelled,
- * typically because it has already completed normally;
- * {@code true} otherwise
+ * typically because it has already completed; {@code true}
+ * otherwise. If two or more threads cause a task to be cancelled,
+ * then at least one of them returns {@code true}. Implementations
+ * may provide stronger guarantees.
*/
boolean cancel(boolean mayInterruptIfRunning);
diff --git a/src/java.base/share/classes/java/util/concurrent/Phaser.java b/src/java.base/share/classes/java/util/concurrent/Phaser.java
--- a/src/java.base/share/classes/java/util/concurrent/Phaser.java
+++ b/src/java.base/share/classes/java/util/concurrent/Phaser.java
@@ -245,7 +245,7 @@
* be appropriate for extremely small per-phase task bodies (thus
* high rates), or up to hundreds for extremely large ones.
*
- *
Implementation notes: This implementation restricts the
+ *
Implementation notes: This implementation restricts the
* maximum number of parties to 65535. Attempts to register additional
* parties result in {@code IllegalStateException}. However, you can and
* should create tiered phasers to accommodate arbitrarily large sets
@@ -919,7 +919,7 @@
*
{@code
* Phaser phaser = new Phaser() {
* protected boolean onAdvance(int phase, int parties) { return false; }
- * }}
+ * };}
*
* @param phase the current phase number on entry to this method,
* before this phaser is advanced
diff --git a/src/java.base/share/classes/java/util/concurrent/PriorityBlockingQueue.java b/src/java.base/share/classes/java/util/concurrent/PriorityBlockingQueue.java
--- a/src/java.base/share/classes/java/util/concurrent/PriorityBlockingQueue.java
+++ b/src/java.base/share/classes/java/util/concurrent/PriorityBlockingQueue.java
@@ -87,7 +87,7 @@
* {@code
* class FIFOEntry>
* implements Comparable> {
- * static final AtomicLong seq = new AtomicLong(0);
+ * static final AtomicLong seq = new AtomicLong();
* final long seqNum;
* final E entry;
* public FIFOEntry(E entry) {
@@ -226,7 +226,7 @@
/**
* Creates a {@code PriorityBlockingQueue} containing the elements
* in the specified collection. If the specified collection is a
- * {@link SortedSet} or a {@link PriorityQueue}, this
+ * {@link SortedSet} or a {@link PriorityBlockingQueue}, this
* priority queue will be ordered according to the same ordering.
* Otherwise, this priority queue will be ordered according to the
* {@linkplain Comparable natural ordering} of its elements.
@@ -290,7 +290,9 @@
if (allocationSpinLock == 0 &&
ALLOCATIONSPINLOCK.compareAndSet(this, 0, 1)) {
try {
- int growth = oldCap < 64 ? oldCap + 2 : oldCap >> 1;
+ int growth = (oldCap < 64)
+ ? (oldCap + 2) // grow faster if small
+ : (oldCap >> 1);
int newCap = ArraysSupport.newLength(oldCap, 1, growth);
if (queue == array)
newArray = new Object[newCap];
diff --git a/src/java.base/share/classes/java/util/concurrent/Semaphore.java b/src/java.base/share/classes/java/util/concurrent/Semaphore.java
--- a/src/java.base/share/classes/java/util/concurrent/Semaphore.java
+++ b/src/java.base/share/classes/java/util/concurrent/Semaphore.java
@@ -66,7 +66,7 @@
*
* // Not a particularly efficient data structure; just for demo
*
- * protected Object[] items = ... whatever kinds of items being managed
+ * protected Object[] items = ...; // whatever kinds of items being managed
* protected boolean[] used = new boolean[MAX_AVAILABLE];
*
* protected synchronized Object getNextAvailableItem() {
diff --git a/src/java.base/share/classes/java/util/concurrent/SubmissionPublisher.java b/src/java.base/share/classes/java/util/concurrent/SubmissionPublisher.java
--- a/src/java.base/share/classes/java/util/concurrent/SubmissionPublisher.java
+++ b/src/java.base/share/classes/java/util/concurrent/SubmissionPublisher.java
@@ -41,6 +41,7 @@
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.locks.LockSupport;
+import java.util.concurrent.locks.ReentrantLock;
import java.util.function.BiConsumer;
import java.util.function.BiPredicate;
import java.util.function.Consumer;
@@ -175,11 +176,11 @@
/*
* Most mechanics are handled by BufferedSubscription. This class
* mainly tracks subscribers and ensures sequentiality, by using
- * built-in synchronization locks across public methods. Using
- * built-in locks works well in the most typical case in which
- * only one thread submits items. We extend this idea in
- * submission methods by detecting single-ownership to reduce
- * producer-consumer synchronization strength.
+ * locks across public methods, to ensure thread-safety in the
+ * presence of multiple sources and maintain acquire-release
+ * ordering around user operations. However, we also track whether
+ * there is only a single source, and if so streamline some buffer
+ * operations by avoiding some atomics.
*/
/** The largest possible power of two array size. */
@@ -234,6 +235,8 @@
*/
BufferedSubscription clients;
+ /** Lock for exclusion across multiple sources */
+ final ReentrantLock lock;
/** Run status, updated only within locks */
volatile boolean closed;
/** Set true on first call to subscribe, to initialize possible owner */
@@ -274,6 +277,7 @@
throw new NullPointerException();
if (maxBufferCapacity <= 0)
throw new IllegalArgumentException("capacity must be positive");
+ this.lock = new ReentrantLock();
this.executor = executor;
this.onNextHandler = handler;
this.maxBufferCapacity = roundCapacity(maxBufferCapacity);
@@ -337,13 +341,15 @@
*/
public void subscribe(Subscriber subscriber) {
if (subscriber == null) throw new NullPointerException();
+ ReentrantLock lock = this.lock;
int max = maxBufferCapacity; // allocate initial array
Object[] array = new Object[max < INITIAL_CAPACITY ?
max : INITIAL_CAPACITY];
BufferedSubscription subscription =
new BufferedSubscription(subscriber, executor, onNextHandler,
array, max);
- synchronized (this) {
+ lock.lock();
+ try {
if (!subscribed) {
subscribed = true;
owner = Thread.currentThread();
@@ -378,6 +384,8 @@
pred = b;
b = next;
}
+ } finally {
+ lock.unlock();
}
}
@@ -390,7 +398,9 @@
if (item == null) throw new NullPointerException();
int lag = 0;
boolean complete, unowned;
- synchronized (this) {
+ ReentrantLock lock = this.lock;
+ lock.lock();
+ try {
Thread t = Thread.currentThread(), o;
BufferedSubscription b = clients;
if ((unowned = ((o = owner) != t)) && o != null)
@@ -421,6 +431,8 @@
if (retries != null || cleanMe)
lag = retryOffer(item, nanos, onDrop, retries, lag, cleanMe);
}
+ } finally {
+ lock.unlock();
}
if (complete)
throw new IllegalStateException("Closed");
@@ -609,14 +621,18 @@
* subscribers have yet completed.
*/
public void close() {
+ ReentrantLock lock = this.lock;
if (!closed) {
BufferedSubscription b;
- synchronized (this) {
+ lock.lock();
+ try {
// no need to re-check closed here
b = clients;
clients = null;
owner = null;
closed = true;
+ } finally {
+ lock.unlock();
}
while (b != null) {
BufferedSubscription next = b.next;
@@ -641,9 +657,11 @@
public void closeExceptionally(Throwable error) {
if (error == null)
throw new NullPointerException();
+ ReentrantLock lock = this.lock;
if (!closed) {
BufferedSubscription b;
- synchronized (this) {
+ lock.lock();
+ try {
b = clients;
if (!closed) { // don't clobber racing close
closedException = error;
@@ -651,6 +669,8 @@
owner = null;
closed = true;
}
+ } finally {
+ lock.unlock();
}
while (b != null) {
BufferedSubscription next = b.next;
@@ -688,7 +708,9 @@
*/
public boolean hasSubscribers() {
boolean nonEmpty = false;
- synchronized (this) {
+ ReentrantLock lock = this.lock;
+ lock.lock();
+ try {
for (BufferedSubscription b = clients; b != null;) {
BufferedSubscription next = b.next;
if (b.isClosed()) {
@@ -700,6 +722,8 @@
break;
}
}
+ } finally {
+ lock.unlock();
}
return nonEmpty;
}
@@ -710,9 +734,15 @@
* @return the number of current subscribers
*/
public int getNumberOfSubscribers() {
- synchronized (this) {
- return cleanAndCount();
+ int n;
+ ReentrantLock lock = this.lock;
+ lock.lock();
+ try {
+ n = cleanAndCount();
+ } finally {
+ lock.unlock();
}
+ return n;
}
/**
@@ -742,7 +772,9 @@
*/
public List> getSubscribers() {
ArrayList> subs = new ArrayList<>();
- synchronized (this) {
+ ReentrantLock lock = this.lock;
+ lock.lock();
+ try {
BufferedSubscription pred = null, next;
for (BufferedSubscription b = clients; b != null; b = next) {
next = b.next;
@@ -758,6 +790,8 @@
pred = b;
}
}
+ } finally {
+ lock.unlock();
}
return subs;
}
@@ -771,8 +805,11 @@
*/
public boolean isSubscribed(Subscriber subscriber) {
if (subscriber == null) throw new NullPointerException();
+ boolean subscribed = false;
+ ReentrantLock lock = this.lock;
if (!closed) {
- synchronized (this) {
+ lock.lock();
+ try {
BufferedSubscription pred = null, next;
for (BufferedSubscription b = clients; b != null; b = next) {
next = b.next;
@@ -783,14 +820,16 @@
else
pred.next = next;
}
- else if (subscriber.equals(b.subscriber))
- return true;
+ else if (subscribed = subscriber.equals(b.subscriber))
+ break;
else
pred = b;
}
+ } finally {
+ lock.unlock();
}
}
- return false;
+ return subscribed;
}
/**
@@ -803,7 +842,9 @@
public long estimateMinimumDemand() {
long min = Long.MAX_VALUE;
boolean nonEmpty = false;
- synchronized (this) {
+ ReentrantLock lock = this.lock;
+ lock.lock();
+ try {
BufferedSubscription pred = null, next;
for (BufferedSubscription b = clients; b != null; b = next) {
int n; long d;
@@ -822,6 +863,8 @@
pred = b;
}
}
+ } finally {
+ lock.unlock();
}
return nonEmpty ? min : 0;
}
@@ -834,7 +877,9 @@
*/
public int estimateMaximumLag() {
int max = 0;
- synchronized (this) {
+ ReentrantLock lock = this.lock;
+ lock.lock();
+ try {
BufferedSubscription pred = null, next;
for (BufferedSubscription b = clients; b != null; b = next) {
int n;
@@ -852,6 +897,8 @@
pred = b;
}
}
+ } finally {
+ lock.unlock();
}
return max;
}
diff --git a/src/java.base/share/classes/java/util/concurrent/ThreadPoolExecutor.java b/src/java.base/share/classes/java/util/concurrent/ThreadPoolExecutor.java
--- a/src/java.base/share/classes/java/util/concurrent/ThreadPoolExecutor.java
+++ b/src/java.base/share/classes/java/util/concurrent/ThreadPoolExecutor.java
@@ -221,13 +221,18 @@
* simple feedback control mechanism that will slow down the rate that
* new tasks are submitted.
*
- * In {@link ThreadPoolExecutor.DiscardPolicy}, a task that
- * cannot be executed is simply dropped.
+ * In {@link ThreadPoolExecutor.DiscardPolicy}, a task that cannot
+ * be executed is simply dropped. This policy is designed only for
+ * those rare cases in which task completion is never relied upon.
*
* In {@link ThreadPoolExecutor.DiscardOldestPolicy}, if the
* executor is not shut down, the task at the head of the work queue
* is dropped, and then execution is retried (which can fail again,
- * causing this to be repeated.)
+ * causing this to be repeated.) This policy is rarely acceptable. In
+ * nearly all cases, you should also cancel the task to cause an
+ * exception in any component waiting for its completion, and/or log
+ * the failure, as illustrated in {@link
+ * ThreadPoolExecutor.DiscardOldestPolicy} documentation.
*
*
*
@@ -272,7 +277,7 @@
*
*
*
- * Extension example. Most extensions of this class
+ *
Extension example. Most extensions of this class
* override one or more of the protected hook methods. For example,
* here is a subclass that adds a simple pause/resume feature:
*
@@ -1149,8 +1154,10 @@
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
- * parameters, the default thread factory and the default rejected
- * execution handler.
+ * parameters, the
+ * {@linkplain Executors#defaultThreadFactory default thread factory}
+ * and the {@linkplain ThreadPoolExecutor.AbortPolicy
+ * default rejected execution handler}.
*
*
It may be more convenient to use one of the {@link Executors}
* factory methods instead of this general purpose constructor.
@@ -1184,7 +1191,7 @@
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
- * parameters and {@linkplain ThreadPoolExecutor.AbortPolicy
+ * parameters and the {@linkplain ThreadPoolExecutor.AbortPolicy
* default rejected execution handler}.
*
* @param corePoolSize the number of threads to keep in the pool, even
@@ -1220,7 +1227,7 @@
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial
- * parameters and
+ * parameters and the
* {@linkplain Executors#defaultThreadFactory default thread factory}.
*
* @param corePoolSize the number of threads to keep in the pool, even
@@ -2081,7 +2088,20 @@
/**
* A handler for rejected tasks that discards the oldest unhandled
* request and then retries {@code execute}, unless the executor
- * is shut down, in which case the task is discarded.
+ * is shut down, in which case the task is discarded. This policy is
+ * rarely useful in cases where other threads may be waiting for
+ * tasks to terminate, or failures must be recorded. Instead consider
+ * using a handler of the form:
+ *
{@code
+ * new RejectedExecutionHandler() {
+ * public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
+ * Runnable dropped = e.getQueue().poll();
+ * if (dropped instanceof Future) {
+ * ((Future)dropped).cancel(false);
+ * // also consider logging the failure
+ * }
+ * e.execute(r); // retry
+ * }}}
*/
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
/**
diff --git a/src/java.base/share/classes/java/util/concurrent/atomic/AtomicBoolean.java b/src/java.base/share/classes/java/util/concurrent/atomic/AtomicBoolean.java
--- a/src/java.base/share/classes/java/util/concurrent/atomic/AtomicBoolean.java
+++ b/src/java.base/share/classes/java/util/concurrent/atomic/AtomicBoolean.java
@@ -68,9 +68,8 @@
* @param initialValue the initial value
*/
public AtomicBoolean(boolean initialValue) {
- if (initialValue) {
+ if (initialValue)
value = 1;
- }
}
/**
diff --git a/src/java.base/share/classes/java/util/concurrent/atomic/AtomicReferenceFieldUpdater.java b/src/java.base/share/classes/java/util/concurrent/atomic/AtomicReferenceFieldUpdater.java
--- a/src/java.base/share/classes/java/util/concurrent/atomic/AtomicReferenceFieldUpdater.java
+++ b/src/java.base/share/classes/java/util/concurrent/atomic/AtomicReferenceFieldUpdater.java
@@ -62,7 +62,7 @@
*
* private static final AtomicReferenceFieldUpdater leftUpdater =
* AtomicReferenceFieldUpdater.newUpdater(Node.class, Node.class, "left");
- * private static AtomicReferenceFieldUpdater rightUpdater =
+ * private static final AtomicReferenceFieldUpdater rightUpdater =
* AtomicReferenceFieldUpdater.newUpdater(Node.class, Node.class, "right");
*
* Node getLeft() { return left; }
diff --git a/src/java.base/share/classes/java/util/concurrent/atomic/DoubleAccumulator.java b/src/java.base/share/classes/java/util/concurrent/atomic/DoubleAccumulator.java
--- a/src/java.base/share/classes/java/util/concurrent/atomic/DoubleAccumulator.java
+++ b/src/java.base/share/classes/java/util/concurrent/atomic/DoubleAccumulator.java
@@ -111,16 +111,17 @@
|| ((r = doubleToRawLongBits
(function.applyAsDouble(longBitsToDouble(b = base), x))) != b
&& !casBase(b, r))) {
+ int index = getProbe();
boolean uncontended = true;
if (cs == null
|| (m = cs.length - 1) < 0
- || (c = cs[getProbe() & m]) == null
+ || (c = cs[index & m]) == null
|| !(uncontended =
((r = doubleToRawLongBits
(function.applyAsDouble
(longBitsToDouble(v = c.value), x))) == v)
|| c.cas(v, r)))
- doubleAccumulate(x, function, uncontended);
+ doubleAccumulate(x, function, uncontended, index);
}
}
diff --git a/src/java.base/share/classes/java/util/concurrent/atomic/DoubleAdder.java b/src/java.base/share/classes/java/util/concurrent/atomic/DoubleAdder.java
--- a/src/java.base/share/classes/java/util/concurrent/atomic/DoubleAdder.java
+++ b/src/java.base/share/classes/java/util/concurrent/atomic/DoubleAdder.java
@@ -92,13 +92,14 @@
!casBase(b = base,
Double.doubleToRawLongBits
(Double.longBitsToDouble(b) + x))) {
+ int index = getProbe();
boolean uncontended = true;
if (cs == null || (m = cs.length - 1) < 0 ||
- (c = cs[getProbe() & m]) == null ||
+ (c = cs[index & m]) == null ||
!(uncontended = c.cas(v = c.value,
Double.doubleToRawLongBits
(Double.longBitsToDouble(v) + x))))
- doubleAccumulate(x, null, uncontended);
+ doubleAccumulate(x, null, uncontended, index);
}
}
diff --git a/src/java.base/share/classes/java/util/concurrent/atomic/LongAccumulator.java b/src/java.base/share/classes/java/util/concurrent/atomic/LongAccumulator.java
--- a/src/java.base/share/classes/java/util/concurrent/atomic/LongAccumulator.java
+++ b/src/java.base/share/classes/java/util/concurrent/atomic/LongAccumulator.java
@@ -108,14 +108,15 @@
if ((cs = cells) != null
|| ((r = function.applyAsLong(b = base, x)) != b
&& !casBase(b, r))) {
+ int index = getProbe();
boolean uncontended = true;
if (cs == null
|| (m = cs.length - 1) < 0
- || (c = cs[getProbe() & m]) == null
+ || (c = cs[index & m]) == null
|| !(uncontended =
(r = function.applyAsLong(v = c.value, x)) == v
|| c.cas(v, r)))
- longAccumulate(x, function, uncontended);
+ longAccumulate(x, function, uncontended, index);
}
}
diff --git a/src/java.base/share/classes/java/util/concurrent/atomic/LongAdder.java b/src/java.base/share/classes/java/util/concurrent/atomic/LongAdder.java
--- a/src/java.base/share/classes/java/util/concurrent/atomic/LongAdder.java
+++ b/src/java.base/share/classes/java/util/concurrent/atomic/LongAdder.java
@@ -85,11 +85,12 @@
public void add(long x) {
Cell[] cs; long b, v; int m; Cell c;
if ((cs = cells) != null || !casBase(b = base, b + x)) {
+ int index = getProbe();
boolean uncontended = true;
if (cs == null || (m = cs.length - 1) < 0 ||
- (c = cs[getProbe() & m]) == null ||
+ (c = cs[index & m]) == null ||
!(uncontended = c.cas(v = c.value, v + x)))
- longAccumulate(x, null, uncontended);
+ longAccumulate(x, null, uncontended, index);
}
}
diff --git a/src/java.base/share/classes/java/util/concurrent/atomic/Striped64.java b/src/java.base/share/classes/java/util/concurrent/atomic/Striped64.java
--- a/src/java.base/share/classes/java/util/concurrent/atomic/Striped64.java
+++ b/src/java.base/share/classes/java/util/concurrent/atomic/Striped64.java
@@ -125,7 +125,7 @@
volatile long value;
Cell(long x) { value = x; }
final boolean cas(long cmp, long val) {
- return VALUE.compareAndSet(this, cmp, val);
+ return VALUE.weakCompareAndSetRelease(this, cmp, val);
}
final void reset() {
VALUE.setVolatile(this, 0L);
@@ -178,7 +178,7 @@
* CASes the base field.
*/
final boolean casBase(long cmp, long val) {
- return BASE.compareAndSet(this, cmp, val);
+ return BASE.weakCompareAndSetRelease(this, cmp, val);
}
final long getAndSetBase(long val) {
@@ -224,20 +224,19 @@
* @param fn the update function, or null for add (this convention
* avoids the need for an extra field or function in LongAdder).
* @param wasUncontended false if CAS failed before call
+ * @param index thread index from getProbe
*/
final void longAccumulate(long x, LongBinaryOperator fn,
- boolean wasUncontended) {
- int h;
- if ((h = getProbe()) == 0) {
+ boolean wasUncontended, int index) {
+ if (index == 0) {
ThreadLocalRandom.current(); // force initialization
- h = getProbe();
+ index = getProbe();
wasUncontended = true;
}
- boolean collide = false; // True if last slot nonempty
- done: for (;;) {
+ for (boolean collide = false;;) { // True if last slot nonempty
Cell[] cs; Cell c; int n; long v;
if ((cs = cells) != null && (n = cs.length) > 0) {
- if ((c = cs[(n - 1) & h]) == null) {
+ if ((c = cs[(n - 1) & index]) == null) {
if (cellsBusy == 0) { // Try to attach new Cell
Cell r = new Cell(x); // Optimistically create
if (cellsBusy == 0 && casCellsBusy()) {
@@ -245,9 +244,9 @@
Cell[] rs; int m, j;
if ((rs = cells) != null &&
(m = rs.length) > 0 &&
- rs[j = (m - 1) & h] == null) {
+ rs[j = (m - 1) & index] == null) {
rs[j] = r;
- break done;
+ break;
}
} finally {
cellsBusy = 0;
@@ -276,15 +275,15 @@
collide = false;
continue; // Retry with expanded table
}
- h = advanceProbe(h);
+ index = advanceProbe(index);
}
else if (cellsBusy == 0 && cells == cs && casCellsBusy()) {
try { // Initialize table
if (cells == cs) {
Cell[] rs = new Cell[2];
- rs[h & 1] = new Cell(x);
+ rs[index & 1] = new Cell(x);
cells = rs;
- break done;
+ break;
}
} finally {
cellsBusy = 0;
@@ -293,7 +292,7 @@
// Fall back on using base
else if (casBase(v = base,
(fn == null) ? v + x : fn.applyAsLong(v, x)))
- break done;
+ break;
}
}
@@ -310,18 +309,16 @@
* maintained by copy/paste/adapt.
*/
final void doubleAccumulate(double x, DoubleBinaryOperator fn,
- boolean wasUncontended) {
- int h;
- if ((h = getProbe()) == 0) {
+ boolean wasUncontended, int index) {
+ if (index == 0) {
ThreadLocalRandom.current(); // force initialization
- h = getProbe();
+ index = getProbe();
wasUncontended = true;
}
- boolean collide = false; // True if last slot nonempty
- done: for (;;) {
+ for (boolean collide = false;;) { // True if last slot nonempty
Cell[] cs; Cell c; int n; long v;
if ((cs = cells) != null && (n = cs.length) > 0) {
- if ((c = cs[(n - 1) & h]) == null) {
+ if ((c = cs[(n - 1) & index]) == null) {
if (cellsBusy == 0) { // Try to attach new Cell
Cell r = new Cell(Double.doubleToRawLongBits(x));
if (cellsBusy == 0 && casCellsBusy()) {
@@ -329,9 +326,9 @@
Cell[] rs; int m, j;
if ((rs = cells) != null &&
(m = rs.length) > 0 &&
- rs[j = (m - 1) & h] == null) {
+ rs[j = (m - 1) & index] == null) {
rs[j] = r;
- break done;
+ break;
}
} finally {
cellsBusy = 0;
@@ -359,15 +356,15 @@
collide = false;
continue; // Retry with expanded table
}
- h = advanceProbe(h);
+ index = advanceProbe(index);
}
else if (cellsBusy == 0 && cells == cs && casCellsBusy()) {
try { // Initialize table
if (cells == cs) {
Cell[] rs = new Cell[2];
- rs[h & 1] = new Cell(Double.doubleToRawLongBits(x));
+ rs[index & 1] = new Cell(Double.doubleToRawLongBits(x));
cells = rs;
- break done;
+ break;
}
} finally {
cellsBusy = 0;
@@ -375,7 +372,7 @@
}
// Fall back on using base
else if (casBase(v = base, apply(fn, v, x)))
- break done;
+ break;
}
}
diff --git a/src/java.base/share/classes/java/util/concurrent/atomic/package-info.java b/src/java.base/share/classes/java/util/concurrent/atomic/package-info.java
--- a/src/java.base/share/classes/java/util/concurrent/atomic/package-info.java
+++ b/src/java.base/share/classes/java/util/concurrent/atomic/package-info.java
@@ -54,7 +54,7 @@
* {@code
* class Sequencer {
* private final AtomicLong sequenceNumber
- * = new AtomicLong(0);
+ * = new AtomicLong(17);
* public long next() {
* return sequenceNumber.getAndIncrement();
* }
diff --git a/src/java.base/share/classes/java/util/concurrent/locks/AbstractQueuedLongSynchronizer.java b/src/java.base/share/classes/java/util/concurrent/locks/AbstractQueuedLongSynchronizer.java
--- a/src/java.base/share/classes/java/util/concurrent/locks/AbstractQueuedLongSynchronizer.java
+++ b/src/java.base/share/classes/java/util/concurrent/locks/AbstractQueuedLongSynchronizer.java
@@ -40,6 +40,7 @@
import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.RejectedExecutionException;
import jdk.internal.misc.Unsafe;
/**
@@ -1197,13 +1198,18 @@
ConditionNode node = new ConditionNode();
long savedState = enableWait(node);
LockSupport.setCurrentBlocker(this); // for back-compatibility
- boolean interrupted = false;
+ boolean interrupted = false, rejected = false;
while (!canReacquire(node)) {
if (Thread.interrupted())
interrupted = true;
else if ((node.status & COND) != 0) {
try {
- ForkJoinPool.managedBlock(node);
+ if (rejected)
+ node.block();
+ else
+ ForkJoinPool.managedBlock(node);
+ } catch (RejectedExecutionException ex) {
+ rejected = true;
} catch (InterruptedException ie) {
interrupted = true;
}
@@ -1236,14 +1242,19 @@
ConditionNode node = new ConditionNode();
long savedState = enableWait(node);
LockSupport.setCurrentBlocker(this); // for back-compatibility
- boolean interrupted = false, cancelled = false;
+ boolean interrupted = false, cancelled = false, rejected = false;
while (!canReacquire(node)) {
if (interrupted |= Thread.interrupted()) {
if (cancelled = (node.getAndUnsetStatus(COND) & COND) != 0)
break; // else interrupted after signal
} else if ((node.status & COND) != 0) {
try {
- ForkJoinPool.managedBlock(node);
+ if (rejected)
+ node.block();
+ else
+ ForkJoinPool.managedBlock(node);
+ } catch (RejectedExecutionException ex) {
+ rejected = true;
} catch (InterruptedException ie) {
interrupted = true;
}
diff --git a/src/java.base/share/classes/java/util/concurrent/locks/AbstractQueuedSynchronizer.java b/src/java.base/share/classes/java/util/concurrent/locks/AbstractQueuedSynchronizer.java
--- a/src/java.base/share/classes/java/util/concurrent/locks/AbstractQueuedSynchronizer.java
+++ b/src/java.base/share/classes/java/util/concurrent/locks/AbstractQueuedSynchronizer.java
@@ -40,6 +40,7 @@
import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.RejectedExecutionException;
import jdk.internal.misc.Unsafe;
/**
@@ -137,13 +138,13 @@
* of exclusive synchronization takes the form:
*
*
- * Acquire:
+ * Acquire:
* while (!tryAcquire(arg)) {
* enqueue thread if it is not already queued;
* possibly block current thread;
* }
*
- * Release:
+ * Release:
* if (tryRelease(arg))
* unblock the first queued thread;
*
@@ -1565,13 +1566,18 @@
ConditionNode node = new ConditionNode();
int savedState = enableWait(node);
LockSupport.setCurrentBlocker(this); // for back-compatibility
- boolean interrupted = false;
+ boolean interrupted = false, rejected = false;
while (!canReacquire(node)) {
if (Thread.interrupted())
interrupted = true;
else if ((node.status & COND) != 0) {
try {
- ForkJoinPool.managedBlock(node);
+ if (rejected)
+ node.block();
+ else
+ ForkJoinPool.managedBlock(node);
+ } catch (RejectedExecutionException ex) {
+ rejected = true;
} catch (InterruptedException ie) {
interrupted = true;
}
@@ -1604,14 +1610,19 @@
ConditionNode node = new ConditionNode();
int savedState = enableWait(node);
LockSupport.setCurrentBlocker(this); // for back-compatibility
- boolean interrupted = false, cancelled = false;
+ boolean interrupted = false, cancelled = false, rejected = false;
while (!canReacquire(node)) {
if (interrupted |= Thread.interrupted()) {
if (cancelled = (node.getAndUnsetStatus(COND) & COND) != 0)
break; // else interrupted after signal
} else if ((node.status & COND) != 0) {
try {
- ForkJoinPool.managedBlock(node);
+ if (rejected)
+ node.block();
+ else
+ ForkJoinPool.managedBlock(node);
+ } catch (RejectedExecutionException ex) {
+ rejected = true;
} catch (InterruptedException ie) {
interrupted = true;
}
diff --git a/src/java.base/share/classes/java/util/concurrent/locks/ReentrantLock.java b/src/java.base/share/classes/java/util/concurrent/locks/ReentrantLock.java
--- a/src/java.base/share/classes/java/util/concurrent/locks/ReentrantLock.java
+++ b/src/java.base/share/classes/java/util/concurrent/locks/ReentrantLock.java
@@ -550,7 +550,7 @@
*
* {@code
* class X {
- * ReentrantLock lock = new ReentrantLock();
+ * final ReentrantLock lock = new ReentrantLock();
* // ...
* public void m() {
* assert lock.getHoldCount() == 0;
@@ -580,7 +580,7 @@
*
* {@code
* class X {
- * ReentrantLock lock = new ReentrantLock();
+ * final ReentrantLock lock = new ReentrantLock();
* // ...
*
* public void m() {
@@ -594,7 +594,7 @@
*
* {@code
* class X {
- * ReentrantLock lock = new ReentrantLock();
+ * final ReentrantLock lock = new ReentrantLock();
* // ...
*
* public void m() {
diff --git a/src/java.base/share/classes/java/util/concurrent/locks/ReentrantReadWriteLock.java b/src/java.base/share/classes/java/util/concurrent/locks/ReentrantReadWriteLock.java
--- a/src/java.base/share/classes/java/util/concurrent/locks/ReentrantReadWriteLock.java
+++ b/src/java.base/share/classes/java/util/concurrent/locks/ReentrantReadWriteLock.java
@@ -128,7 +128,7 @@
* locks: a deserialized lock is in the unlocked state, regardless of
* its state when serialized.
*
- * Sample usages. Here is a code sketch showing how to perform
+ *
Sample usages. Here is a code sketch showing how to perform
* lock downgrading after updating a cache (exception handling is
* particularly tricky when handling multiple locks in a non-nested
* fashion):
@@ -149,7 +149,7 @@
* // Recheck state because another thread might have
* // acquired write lock and changed state before we did.
* if (!cacheValid) {
- * data = ...
+ * data = ...;
* cacheValid = true;
* }
* // Downgrade by acquiring read lock before releasing write lock
diff --git a/src/java.base/share/classes/java/util/concurrent/locks/StampedLock.java b/src/java.base/share/classes/java/util/concurrent/locks/StampedLock.java
--- a/src/java.base/share/classes/java/util/concurrent/locks/StampedLock.java
+++ b/src/java.base/share/classes/java/util/concurrent/locks/StampedLock.java
@@ -212,8 +212,8 @@
* }
* }
*
- * // Upgrade read lock to write lock
- * void moveIfAtOrigin(double newX, double newY) {
+ * // upgrade read lock to write lock
+ * void moveIfAtOrigin2(double newX, double newY) {
* long stamp = sl.readLock();
* try {
* while (x == 0.0 && y == 0.0) {
diff --git a/test/jdk/java/util/Collection/IteratorMicroBenchmark.java b/test/jdk/java/util/Collection/IteratorMicroBenchmark.java
--- a/test/jdk/java/util/Collection/IteratorMicroBenchmark.java
+++ b/test/jdk/java/util/Collection/IteratorMicroBenchmark.java
@@ -360,7 +360,6 @@
}
Stream jobs(Collection x) {
- final String klazz = goodClassName(x);
return concatStreams(
collectionJobs(x),
diff --git a/test/jdk/java/util/concurrent/ExecutorService/Invoke.java b/test/jdk/java/util/concurrent/ExecutorService/Invoke.java
--- a/test/jdk/java/util/concurrent/ExecutorService/Invoke.java
+++ b/test/jdk/java/util/concurrent/ExecutorService/Invoke.java
@@ -84,9 +84,16 @@
public static void main(String[] args) {
try {
- testInvokeAll();
- testInvokeAny();
- testInvokeAny_cancellationInterrupt();
+ for (int nThreads = 1; nThreads <= 6; ++nThreads) {
+ // untimed
+ testInvokeAll(nThreads, false);
+ testInvokeAny(nThreads, false);
+ testInvokeAny_cancellationInterrupt(nThreads, false);
+ // timed
+ testInvokeAll(nThreads, true);
+ testInvokeAny(nThreads, true);
+ testInvokeAny_cancellationInterrupt(nThreads, true);
+ }
} catch (Throwable t) { unexpected(t); }
if (failed > 0)
@@ -96,10 +103,8 @@
static final long timeoutSeconds = 10L;
- static void testInvokeAll() throws Throwable {
+ static void testInvokeAll(int nThreads, boolean timed) throws Throwable {
final ThreadLocalRandom rnd = ThreadLocalRandom.current();
- final int nThreads = rnd.nextInt(2, 7);
- final boolean timed = rnd.nextBoolean();
final ExecutorService pool = Executors.newFixedThreadPool(nThreads);
final AtomicLong count = new AtomicLong(0);
class Task implements Callable {
@@ -136,17 +141,15 @@
}
}
- static void testInvokeAny() throws Throwable {
+ static void testInvokeAny(int nThreads, boolean timed) throws Throwable {
final ThreadLocalRandom rnd = ThreadLocalRandom.current();
- final boolean timed = rnd.nextBoolean();
- final ExecutorService pool = Executors.newSingleThreadExecutor();
+ final ExecutorService pool = Executors.newFixedThreadPool(nThreads);
final AtomicLong count = new AtomicLong(0);
final CountDownLatch invokeAnyDone = new CountDownLatch(1);
class Task implements Callable {
public Long call() throws Exception {
long x = count.incrementAndGet();
- check(x <= 2);
- if (x == 2) {
+ if (x > 1) {
// wait for main thread to interrupt us ...
awaitInterrupt(timeoutSeconds);
// ... and then for invokeAny to return
@@ -173,12 +176,12 @@
check(val == 1);
invokeAnyDone.countDown();
- // inherent race between main thread interrupt and
- // start of second task
- check(count.get() == 1 || count.get() == 2);
-
pool.shutdown();
check(pool.awaitTermination(timeoutSeconds, SECONDS));
+
+ long c = count.get();
+ check(c >= 1 && c <= tasks.size());
+
} finally {
pool.shutdownNow();
}
@@ -187,18 +190,16 @@
/**
* Every remaining running task is sent an interrupt for cancellation.
*/
- static void testInvokeAny_cancellationInterrupt() throws Throwable {
+ static void testInvokeAny_cancellationInterrupt(int nThreads, boolean timed) throws Throwable {
final ThreadLocalRandom rnd = ThreadLocalRandom.current();
- final int nThreads = rnd.nextInt(2, 7);
- final boolean timed = rnd.nextBoolean();
final ExecutorService pool = Executors.newFixedThreadPool(nThreads);
final AtomicLong count = new AtomicLong(0);
final AtomicLong interruptedCount = new AtomicLong(0);
final CyclicBarrier allStarted = new CyclicBarrier(nThreads);
class Task implements Callable {
public Long call() throws Exception {
+ long x = count.incrementAndGet();
allStarted.await();
- long x = count.incrementAndGet();
if (x > 1)
// main thread will interrupt us
awaitInterrupt(timeoutSeconds);
diff --git a/test/jdk/java/util/concurrent/tck/ArrayBlockingQueueTest.java b/test/jdk/java/util/concurrent/tck/ArrayBlockingQueueTest.java
--- a/test/jdk/java/util/concurrent/tck/ArrayBlockingQueueTest.java
+++ b/test/jdk/java/util/concurrent/tck/ArrayBlockingQueueTest.java
@@ -715,7 +715,7 @@
Integer x = (Integer) it.next();
assertEquals(s + i, (int) x);
for (Object[] a : as)
- assertSame(a1[i], x);
+ assertSame(a[i], x);
}
}
diff --git a/test/jdk/java/util/concurrent/tck/ArrayDequeTest.java b/test/jdk/java/util/concurrent/tck/ArrayDequeTest.java
--- a/test/jdk/java/util/concurrent/tck/ArrayDequeTest.java
+++ b/test/jdk/java/util/concurrent/tck/ArrayDequeTest.java
@@ -738,7 +738,7 @@
Integer x = (Integer) it.next();
assertEquals(s + i, (int) x);
for (Object[] a : as)
- assertSame(a1[i], x);
+ assertSame(a[i], x);
}
}
diff --git a/test/jdk/java/util/concurrent/tck/ArrayListTest.java b/test/jdk/java/util/concurrent/tck/ArrayListTest.java
--- a/test/jdk/java/util/concurrent/tck/ArrayListTest.java
+++ b/test/jdk/java/util/concurrent/tck/ArrayListTest.java
@@ -57,7 +57,7 @@
}
}
return newTestSuite(
- // ArrayListTest.class,
+ ArrayListTest.class,
CollectionTest.testSuite(new Implementation()),
CollectionTest.testSuite(new SubListImplementation()));
}
diff --git a/test/jdk/java/util/concurrent/tck/Collection8Test.java b/test/jdk/java/util/concurrent/tck/Collection8Test.java
--- a/test/jdk/java/util/concurrent/tck/Collection8Test.java
+++ b/test/jdk/java/util/concurrent/tck/Collection8Test.java
@@ -440,7 +440,6 @@
if (rnd.nextBoolean()) assertTrue(it.hasNext());
it.next();
}
- Consumer alwaysThrows = e -> { throw new AssertionError(); };
// TODO: many more removal methods
if (rnd.nextBoolean()) {
for (Iterator z = c.iterator(); z.hasNext(); ) {
@@ -661,7 +660,6 @@
*/
public void testStreamForEach() throws Throwable {
final Collection c = impl.emptyCollection();
- final AtomicLong count = new AtomicLong(0L);
final Object x = impl.makeElement(1);
final Object y = impl.makeElement(2);
final ArrayList found = new ArrayList();
@@ -719,7 +717,6 @@
*/
public void testForEach() throws Throwable {
final Collection c = impl.emptyCollection();
- final AtomicLong count = new AtomicLong(0L);
final Object x = impl.makeElement(1);
final Object y = impl.makeElement(2);
final ArrayList found = new ArrayList();
diff --git a/test/jdk/java/util/concurrent/tck/CompletableFutureTest.java b/test/jdk/java/util/concurrent/tck/CompletableFutureTest.java
--- a/test/jdk/java/util/concurrent/tck/CompletableFutureTest.java
+++ b/test/jdk/java/util/concurrent/tck/CompletableFutureTest.java
@@ -3779,16 +3779,6 @@
}
/**
- * failedFuture(null) throws NPE
- */
- public void testFailedFuture_null() {
- try {
- CompletableFuture f = CompletableFuture.failedFuture(null);
- shouldThrow();
- } catch (NullPointerException success) {}
- }
-
- /**
* copy returns a CompletableFuture that is completed normally,
* with the same value, when source is.
*/
@@ -4217,12 +4207,9 @@
// Manufacture boxed primitives for primitive params
for (int i = 0; i < args.length; i++) {
Class type = parameterTypes[i];
- if (parameterTypes[i] == boolean.class)
- args[i] = false;
- else if (parameterTypes[i] == int.class)
- args[i] = 0;
- else if (parameterTypes[i] == long.class)
- args[i] = 0L;
+ if (type == boolean.class) args[i] = false;
+ else if (type == int.class) args[i] = 0;
+ else if (type == long.class) args[i] = 0L;
}
for (CompletionStage stage : stages) {
try {
diff --git a/test/jdk/java/util/concurrent/tck/ConcurrentLinkedDequeTest.java b/test/jdk/java/util/concurrent/tck/ConcurrentLinkedDequeTest.java
--- a/test/jdk/java/util/concurrent/tck/ConcurrentLinkedDequeTest.java
+++ b/test/jdk/java/util/concurrent/tck/ConcurrentLinkedDequeTest.java
@@ -1010,7 +1010,6 @@
* ant -Djsr166.expensiveTests=true -Djsr166.tckTestClass=ConcurrentLinkedDequeTest -Djsr166.methodFilter=testBug8189387 tck
*/
public void testBug8189387() {
- final ThreadLocalRandom rnd = ThreadLocalRandom.current();
Object x = new Object();
for (int n = expensiveTests ? 100_000 : 10; n--> 0; ) {
ConcurrentLinkedDeque