archive/jdk
1
0
Fork 0
mirror of https://github.com/openjdk/jdk.git synced 2025-08-27 14:54:52 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

8195590: Miscellaneous changes imported from jsr166 CVS 2018-02

Browse source 
Reviewed-by: martin, psandoz, dholmes
This commit is contained in:
Doug Lea 2018-02-10 09:23:41 -08:00
parent b6c2b234ef
commit f9b19eb874
16 changed files with 187 additions and 186 deletions
Download patch file Download diff file Expand all files Collapse all files

62
src/java.base/share/classes/java/util/concurrent/ArrayBlockingQueue.java
View file

@ -1584,34 +1584,40 @@ public class ArrayBlockingQueue<E> extends AbstractQueue<E>
void checkInvariants() {
// meta-assertions
// assert lock.isHeldByCurrentThread();
try {
// Unlike ArrayDeque, we have a count field but no spare slot.
// We prefer ArrayDeque's strategy (and the names of its fields!),
// but our field layout is baked into the serial form, and so is
// too annoying to change.
//
// putIndex == takeIndex must be disambiguated by checking count.
int capacity = items.length;
// assert capacity > 0;
// assert takeIndex >= 0 && takeIndex < capacity;
// assert putIndex >= 0 && putIndex < capacity;
// assert count <= capacity;
// assert takeIndex == putIndex || items[takeIndex] != null;
// assert count == capacity || items[putIndex] == null;
// assert takeIndex == putIndex || items[dec(putIndex, capacity)] != null;
} catch (Throwable t) {
System.err.printf("takeIndex=%d putIndex=%d count=%d capacity=%d%n",
takeIndex, putIndex, count, items.length);
System.err.printf("items=%s%n",
Arrays.toString(items));
throw t;
if (!invariantsSatisfied()) {
String detail = String.format(
"takeIndex=%d putIndex=%d count=%d capacity=%d items=%s",
takeIndex, putIndex, count, items.length,
Arrays.toString(items));
System.err.println(detail);
throw new AssertionError(detail);
}
}
private boolean invariantsSatisfied() {
// Unlike ArrayDeque, we have a count field but no spare slot.
// We prefer ArrayDeque's strategy (and the names of its fields!),
// but our field layout is baked into the serial form, and so is
// too annoying to change.
//
// putIndex == takeIndex must be disambiguated by checking count.
int capacity = items.length;
return capacity > 0
&& items.getClass() == Object[].class
&& (takeIndex | putIndex | count) >= 0
&& takeIndex < capacity
&& putIndex < capacity
&& count <= capacity
&& (putIndex - takeIndex - count) % capacity == 0
&& (count == 0 || items[takeIndex] != null)
&& (count == capacity || items[putIndex] == null)
&& (count == 0 || items[dec(putIndex, capacity)] != null);
}
/**
* Deserializes this queue and then checks some invariants.
* Reconstitutes this queue from a stream (that is, deserializes it).
*
* @param s the input stream
* @param s the stream
* @throws ClassNotFoundException if the class of a serialized object
* could not be found
* @throws java.io.InvalidObjectException if invariants are violated
@ -1623,15 +1629,7 @@ public class ArrayBlockingQueue<E> extends AbstractQueue<E>
// Read in items array and various fields
s.defaultReadObject();
// Check invariants over count and index fields. Note that
// if putIndex==takeIndex, count can be either 0 or items.length.
if (items.length == 0 ||
takeIndex < 0 || takeIndex >= items.length ||
putIndex < 0 || putIndex >= items.length ||
count < 0 || count > items.length ||
Math.floorMod(putIndex - takeIndex, items.length) !=
Math.floorMod(count, items.length)) {
if (!invariantsSatisfied())
throw new java.io.InvalidObjectException("invariants violated");
}
}
}

75
src/java.base/share/classes/java/util/concurrent/Executors.java
View file

@ -35,6 +35,7 @@
package java.util.concurrent;
import static java.lang.ref.Reference.reachabilityFence;
import java.security.AccessControlContext;
import java.security.AccessControlException;
import java.security.AccessController;
@ -185,9 +186,7 @@ public class Executors {
* returned executor is guaranteed not to be reconfigurable to use
* additional threads.
*
* @param threadFactory the factory to use when creating new
* threads
*
* @param threadFactory the factory to use when creating new threads
* @return the newly created single-threaded Executor
* @throws NullPointerException if threadFactory is null
*/
@ -226,6 +225,7 @@ public class Executors {
* will reuse previously constructed threads when they are
* available, and uses the provided
* ThreadFactory to create new threads when needed.
*
* @param threadFactory the factory to use when creating new threads
* @return the newly created thread pool
* @throws NullPointerException if threadFactory is null
@ -248,6 +248,7 @@ public class Executors {
* given time. Unlike the otherwise equivalent
* {@code newScheduledThreadPool(1)} the returned executor is
* guaranteed not to be reconfigurable to use additional threads.
*
* @return the newly created scheduled executor
*/
public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
@ -266,9 +267,9 @@ public class Executors {
* equivalent {@code newScheduledThreadPool(1, threadFactory)}
* the returned executor is guaranteed not to be reconfigurable to
* use additional threads.
* @param threadFactory the factory to use when creating new
* threads
* @return a newly created scheduled executor
*
* @param threadFactory the factory to use when creating new threads
* @return the newly created scheduled executor
* @throws NullPointerException if threadFactory is null
*/
public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
@ -281,7 +282,7 @@ public class Executors {
* given delay, or to execute periodically.
* @param corePoolSize the number of threads to keep in the pool,
* even if they are idle
* @return a newly created scheduled thread pool
* @return the newly created scheduled thread pool
* @throws IllegalArgumentException if {@code corePoolSize < 0}
*/
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
@ -295,7 +296,7 @@ public class Executors {
* even if they are idle
* @param threadFactory the factory to use when the executor
* creates a new thread
* @return a newly created scheduled thread pool
* @return the newly created scheduled thread pool
* @throws IllegalArgumentException if {@code corePoolSize < 0}
* @throws NullPointerException if threadFactory is null
*/
@ -678,44 +679,76 @@ public class Executors {
* of an ExecutorService implementation.
*/
private static class DelegatedExecutorService
extends AbstractExecutorService {
implements ExecutorService {
private final ExecutorService e;
DelegatedExecutorService(ExecutorService executor) { e = executor; }
public void execute(Runnable command) { e.execute(command); }
public void execute(Runnable command) {
try {
e.execute(command);
} finally { reachabilityFence(this); }
}
public void shutdown() { e.shutdown(); }
public List<Runnable> shutdownNow() { return e.shutdownNow(); }
public boolean isShutdown() { return e.isShutdown(); }
public boolean isTerminated() { return e.isTerminated(); }
public List<Runnable> shutdownNow() {
try {
return e.shutdownNow();
} finally { reachabilityFence(this); }
}
public boolean isShutdown() {
try {
return e.isShutdown();
} finally { reachabilityFence(this); }
}
public boolean isTerminated() {
try {
return e.isTerminated();
} finally { reachabilityFence(this); }
}
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
return e.awaitTermination(timeout, unit);
try {
return e.awaitTermination(timeout, unit);
} finally { reachabilityFence(this); }
}
public Future<?> submit(Runnable task) {
return e.submit(task);
try {
return e.submit(task);
} finally { reachabilityFence(this); }
}
public <T> Future<T> submit(Callable<T> task) {
return e.submit(task);
try {
return e.submit(task);
} finally { reachabilityFence(this); }
}
public <T> Future<T> submit(Runnable task, T result) {
return e.submit(task, result);
try {
return e.submit(task, result);
} finally { reachabilityFence(this); }
}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
return e.invokeAll(tasks);
try {
return e.invokeAll(tasks);
} finally { reachabilityFence(this); }
}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {
return e.invokeAll(tasks, timeout, unit);
try {
return e.invokeAll(tasks, timeout, unit);
} finally { reachabilityFence(this); }
}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
return e.invokeAny(tasks);
try {
return e.invokeAny(tasks);
} finally { reachabilityFence(this); }
}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return e.invokeAny(tasks, timeout, unit);
try {
return e.invokeAny(tasks, timeout, unit);
} finally { reachabilityFence(this); }
}
}

51
src/java.base/share/classes/java/util/concurrent/LinkedBlockingQueue.java
View file

@ -323,10 +323,8 @@ public class LinkedBlockingQueue<E> extends AbstractQueue<E>
*/
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
// Note: convention in all put/take/etc is to preset local var
// holding count negative to indicate failure unless set.
int c = -1;
Node<E> node = new Node<E>(e);
final int c;
final Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
@ -367,7 +365,7 @@ public class LinkedBlockingQueue<E> extends AbstractQueue<E>
if (e == null) throw new NullPointerException();
long nanos = unit.toNanos(timeout);
int c = -1;
final int c;
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
@ -405,28 +403,28 @@ public class LinkedBlockingQueue<E> extends AbstractQueue<E>
final AtomicInteger count = this.count;
if (count.get() == capacity)
return false;
int c = -1;
Node<E> node = new Node<E>(e);
final int c;
final Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
if (count.get() < capacity) {
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
}
if (count.get() == capacity)
return false;
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
return c >= 0;
return true;
}
public E take() throws InterruptedException {
E x;
int c = -1;
final E x;
final int c;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
@ -447,8 +445,8 @@ public class LinkedBlockingQueue<E> extends AbstractQueue<E>
}
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E x = null;
int c = -1;
final E x;
final int c;
long nanos = unit.toNanos(timeout);
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
@ -475,17 +473,17 @@ public class LinkedBlockingQueue<E> extends AbstractQueue<E>
final AtomicInteger count = this.count;
if (count.get() == 0)
return null;
E x = null;
int c = -1;
final E x;
final int c;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
if (count.get() > 0) {
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
}
if (count.get() == 0)
return null;
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
@ -495,6 +493,7 @@ public class LinkedBlockingQueue<E> extends AbstractQueue<E>
}
public E peek() {
final AtomicInteger count = this.count;
if (count.get() == 0)
return null;
final ReentrantLock takeLock = this.takeLock;