/** The value is used for character storage. */privatefinalchar value[];
/** Cache the hash code for the string */privateint hash; // Default to 0/** use serialVersionUID from JDK 1.0.2 for interoperability */privatestaticfinallongserialVersionUID= -6849794470754667710L;
/**
* Allocates a new {@code String} so that it represents the sequence of
* characters currently contained in the character array argument. The
* contents of the character array are copied; subsequent modification of
* the character array does not affect the newly created string.
*
* @param value
* The initial value of the string
*/publicString(char value[]) {
this.value = Arrays.copyOf(value, value.length);
}
/**
* Returns a hash code for this string. The hash code for a
* {@code String} object is computed as
* <blockquote><pre>
* s[0]*31^(n-1) + s[1]*31^(n-2) + ... + s[n-1]
* </pre></blockquote>
* using {@code int} arithmetic, where {@code s[i]} is the
* <i>i</i>th character of the string, {@code n} is the length of
* the string, and {@code ^} indicates exponentiation.
* (The hash value of the empty string is zero.)
*
* @return a hash code value for this object.
*/publicinthashCode() {
inth= hash;
if (h == 0 && value.length > 0) {
char val[] = value;
for (inti=0; i < value.length; i++) {
h = 31 * h + val[i];
}
hash = h;
}
return h;
}
/**
* Compares this string to the specified object. The result is {@code
* true} if and only if the argument is not {@code null} and is a {@code
* String} object that represents the same sequence of characters as this
* object.
*
* @param anObject
* The object to compare this {@code String} against
*
* @return {@code true} if the given object represents a {@code String}
* equivalent to this string, {@code false} otherwise
*
* @see #compareTo(String)
* @see #equalsIgnoreCase(String)
*/publicbooleanequals(Object anObject) {
if (this == anObject) {
returntrue;
}
if (anObject instanceof String) {
StringanotherString= (String)anObject;
intn= value.length;
if (n == anotherString.value.length) {
char v1[] = value;
char v2[] = anotherString.value;
inti=0;
while (n-- != 0) {
if (v1[i] != v2[i])
returnfalse;
i++;
}
returntrue;
}
}
returnfalse;
}
/**
* For positive values of {@code minimumCapacity}, this method
* behaves like {@code ensureCapacity}, however it is never
* synchronized.
* If {@code minimumCapacity} is non positive due to numeric
* overflow, this method throws {@code OutOfMemoryError}.
*/privatevoidensureCapacityInternal(int minimumCapacity) {
// overflow-conscious codeif (minimumCapacity - value.length > 0) {
value = Arrays.copyOf(value,
newCapacity(minimumCapacity));
}
}
/**
* Returns a capacity at least as large as the given minimum capacity.
* Returns the current capacity increased by the same amount + 2 if
* that suffices.
* Will not return a capacity greater than {@code MAX_ARRAY_SIZE}
* unless the given minimum capacity is greater than that.
*
* @param minCapacity the desired minimum capacity
* @throws OutOfMemoryError if minCapacity is less than zero or
* greater than Integer.MAX_VALUE
*/privateintnewCapacity(int minCapacity) {
// overflow-conscious codeintnewCapacity= (value.length << 1) + 2;
if (newCapacity - minCapacity < 0) {
newCapacity = minCapacity;
}
return (newCapacity <= 0 || MAX_ARRAY_SIZE - newCapacity < 0)
? hugeCapacity(minCapacity)
: newCapacity;
}
/**
* Appends the specified string to this character sequence.
* <p>
* The characters of the {@code String} argument are appended, in
* order, increasing the length of this sequence by the length of the
* argument. If {@code str} is {@code null}, then the four
* characters {@code "null"} are appended.
* <p>
* Let <i>n</i> be the length of this character sequence just prior to
* execution of the {@code append} method. Then the character at
* index <i>k</i> in the new character sequence is equal to the character
* at index <i>k</i> in the old character sequence, if <i>k</i> is less
* than <i>n</i>; otherwise, it is equal to the character at index
* <i>k-n</i> in the argument {@code str}.
*
* @param str a string.
* @return a reference to this object.
*/public AbstractStringBuilder append(String str) {
if (str == null)
return appendNull();
intlen= str.length();
ensureCapacityInternal(count + len);
str.getChars(0, len, value, count);
count += len;
returnthis;
}
/**
* Allocates a new string that contains the sequence of characters
* currently contained in the string builder argument. The contents of the
* string builder are copied; subsequent modification of the string builder
* does not affect the newly created string.
*
* <p> This constructor is provided to ease migration to {@code
* StringBuilder}. Obtaining a string from a string builder via the {@code
* toString} method is likely to run faster and is generally preferred.
*
* @param builder
* A {@code StringBuilder}
*
* @since 1.5
*/publicString(StringBuilder builder) {
this.value = Arrays.copyOf(builder.getValue(), builder.length());
}
/**
* Cache to support the object identity semantics of autoboxing for values between
* -128 and 127 (inclusive) as required by JLS.
*
* The cache is initialized on first usage. The size of the cache
* may be controlled by the {@code -XX:AutoBoxCacheMax=<size>} option.
* During VM initialization, java.lang.Integer.IntegerCache.high property
* may be set and saved in the private system properties in the
* sun.misc.VM class.
*/privatestaticclassIntegerCache {
staticfinalintlow= -128;
staticfinalint high;
staticfinal Integer cache[];
static {
// high value may be configured by propertyinth=127;
StringintegerCacheHighPropValue=
sun.misc.VM.getSavedProperty("java.lang.Integer.IntegerCache.high");
if (integerCacheHighPropValue != null) {
try {
inti= parseInt(integerCacheHighPropValue);
i = Math.max(i, 127);
// Maximum array size is Integer.MAX_VALUE
h = Math.min(i, Integer.MAX_VALUE - (-low) -1);
} catch( NumberFormatException nfe) {
// If the property cannot be parsed into an int, ignore it.
}
}
high = h;
cache = newInteger[(high - low) + 1];
intj= low;
/**
* Returns an {@code Integer} instance representing the specified
* {@code int} value. If a new {@code Integer} instance is not
* required, this method should generally be used in preference to
* the constructor {@link #Integer(int)}, as this method is likely
* to yield significantly better space and time performance by
* caching frequently requested values.
*
* This method will always cache values in the range -128 to 127,
* inclusive, and may cache other values outside of this range.
*
* @param i an {@code int} value.
* @return an {@code Integer} instance representing {@code i}.
* @since 1.5
*/publicstatic Integer valueOf(int i) {
if (i >= IntegerCache.low && i <= IntegerCache.high)
return IntegerCache.cache[i + (-IntegerCache.low)];
returnnewInteger(i);
}
/**
* Compares this object to the specified object. The result is
* {@code true} if and only if the argument is not
* {@code null} and is an {@code Integer} object that
* contains the same {@code int} value as this object.
*
* @param obj the object to compare with.
* @return {@code true} if the objects are the same;
* {@code false} otherwise.
*/publicbooleanequals(Object obj) {
if (obj instanceof Integer) {
return value == ((Integer)obj).intValue();
}
returnfalse;
}
/**
* Appends the specified element to the end of this list.
*
* @param e element to be appended to this list
* @return <tt>true</tt> (as specified by {@link Collection#add})
*/publicbooleanadd(E e) {
ensureCapacityInternal(size + 1); // Increments modCount!!
elementData[size++] = e;
returntrue;
}
/**
* Increases the capacity to ensure that it can hold at least the
* number of elements specified by the minimum capacity argument.
*
* @param minCapacity the desired minimum capacity
*/privatevoidgrow(int minCapacity) {
// overflow-conscious codeintoldCapacity= elementData.length;
intnewCapacity= oldCapacity + (oldCapacity >> 1);
if (newCapacity - minCapacity < 0)
newCapacity = minCapacity;
if (newCapacity - MAX_ARRAY_SIZE > 0)
newCapacity = hugeCapacity(minCapacity);
// minCapacity is usually close to size, so this is a win:
elementData = Arrays.copyOf(elementData, newCapacity);
}
/**
* Removes the element at the specified position in this list.
* Shifts any subsequent elements to the left (subtracts one from their
* indices).
*
* @param index the index of the element to be removed
* @return the element that was removed from the list
* @throws IndexOutOfBoundsException {@inheritDoc}
*/public E remove(int index) {
rangeCheck(index);
modCount++;
EoldValue= elementData(index);
intnumMoved= size - index - 1;
if (numMoved > 0)
System.arraycopy(elementData, index+1, elementData, index,
numMoved);
elementData[--size] = null; // clear to let GC do its workreturn oldValue;
}
/**
* Removes all of the elements from this list. The list will
* be empty after this call returns.
*/publicvoidclear() {
modCount++;
// clear to let GC do its workfor (inti=0; i < size; i++)
elementData[i] = null;
size = 0;
}
/**
* Links e as last element.
*/voidlinkLast(E e) {
final Node<E> l = last;
final Node<E> newNode = newNode<>(l, e, null);
last = newNode;
if (l == null)
first = newNode;
else
l.next = newNode;
size++;
modCount++;
}
/**
* Retrieves and removes the first element of this list,
* or returns {@code null} if this list is empty.
*
* @return the first element of this list, or {@code null} if
* this list is empty
* @since 1.6
*/public E pollFirst() {
final Node<E> f = first;
return (f == null) ? null : unlinkFirst(f);
}
/**
* Returns a power of two size for the given target capacity.
*/staticfinalinttableSizeFor(int cap) {
intn= cap - 1;
n |= n >>> 1;
n |= n >>> 2;
n |= n >>> 4;
n |= n >>> 8;
n |= n >>> 16;
return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}
/* ---------------- Static utilities -------------- *//**
* Computes key.hashCode() and spreads (XORs) higher bits of hash
* to lower. Because the table uses power-of-two masking, sets of
* hashes that vary only in bits above the current mask will
* always collide. (Among known examples are sets of Float keys
* holding consecutive whole numbers in small tables.) So we
* apply a transform that spreads the impact of higher bits
* downward. There is a tradeoff between speed, utility, and
* quality of bit-spreading. Because many common sets of hashes
* are already reasonably distributed (so don't benefit from
* spreading), and because we use trees to handle large sets of
* collisions in bins, we just XOR some shifted bits in the
* cheapest possible way to reduce systematic lossage, as well as
* to incorporate impact of the highest bits that would otherwise
* never be used in index calculations because of table bounds.
*/staticfinalinthash(Object key) {
int h;
return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
}
/**
* Replaces all linked nodes in bin at index for given hash unless
* table is too small, in which case resizes instead.
*/finalvoidtreeifyBin(Node<K,V>[] tab, int hash) {
int n, index; Node<K,V> e;
if (tab == null || (n = tab.length) < MIN_TREEIFY_CAPACITY)
resize();
elseif ((e = tab[index = (n - 1) & hash]) != null) {
TreeNode<K,V> hd = null, tl = null;
do {
TreeNode<K,V> p = replacementTreeNode(e, null);
if (tl == null)
hd = p;
else {
p.prev = tl;
tl.next = p;
}
tl = p;
} while ((e = e.next) != null);
if ((tab[index] = hd) != null)
hd.treeify(tab);
}
}
/**
* Splits nodes in a tree bin into lower and upper tree bins,
* or untreeifies if now too small. Called only from resize;
* see above discussion about split bits and indices.
*
* @param map the map
* @param tab the table for recording bin heads
* @param index the index of the table being split
* @param bit the bit of hash to split on
*/finalvoidsplit(HashMap<K,V> map, Node<K,V>[] tab, int index, int bit) {
TreeNode<K,V> b = this;
// Relink into lo and hi lists, preserving order
TreeNode<K,V> loHead = null, loTail = null;
TreeNode<K,V> hiHead = null, hiTail = null;
intlc=0, hc = 0;
for (TreeNode<K,V> e = b, next; e != null; e = next) {
next = (TreeNode<K,V>)e.next;
e.next = null;
if ((e.hash & bit) == 0) {
if ((e.prev = loTail) == null)
loHead = e;
else
loTail.next = e;
loTail = e;
++lc;
}
else {
if ((e.prev = hiTail) == null)
hiHead = e;
else
hiTail.next = e;
hiTail = e;
++hc;
}
}
if (loHead != null) {
if (lc <= UNTREEIFY_THRESHOLD)
tab[index] = loHead.untreeify(map);
else {
tab[index] = loHead;
if (hiHead != null) // (else is already treeified)
loHead.treeify(tab);
}
}
if (hiHead != null) {
if (hc <= UNTREEIFY_THRESHOLD)
tab[index + bit] = hiHead.untreeify(map);
else {
tab[index + bit] = hiHead;
if (loHead != null)
hiHead.treeify(tab);
}
}
}
/**
* Tie-breaking utility for ordering insertions when equal
* hashCodes and non-comparable. We don't require a total
* order, just a consistent insertion rule to maintain
* equivalence across rebalancings. Tie-breaking further than
* necessary simplifies testing a bit.
*/staticinttieBreakOrder(Object a, Object b) {
int d;
if (a == null || b == null ||
(d = a.getClass().getName().
compareTo(b.getClass().getName())) == 0)
d = (System.identityHashCode(a) <= System.identityHashCode(b) ?
-1 : 1);
return d;
}
/**
* 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
* this 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
*/publicbooleanadd(E e) {
return map.put(e, PRESENT)==null;
}
/**
* 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 the set contained the specified element
*/publicbooleanremove(Object o) {
return map.remove(o)==PRESENT;
}
/**
* Associates the specified value with the specified key in this map.
* If the map previously contained a mapping for the key, the old
* value is replaced.
*
* @param key key with which the specified value is to be associated
* @param value value to be associated with the specified key
*
* @return the previous value associated with {@code key}, or
* {@code null} if there was no mapping for {@code key}.
* (A {@code null} return can also indicate that the map
* previously associated {@code null} with {@code key}.)
* @throws ClassCastException if the specified key cannot be compared
* with the keys currently in the map
* @throws NullPointerException if the specified key is null
* and this map uses natural ordering, or its comparator
* does not permit null keys
*/public V put(K key, V value) {
Entry<K,V> t = root;
if (t == null) {
compare(key, key); // type (and possibly null) check
root = newEntry<>(key, value, null);
size = 1;
modCount++;
returnnull;
}
int cmp;
Entry<K,V> parent;
// split comparator and comparable paths
Comparator<? super K> cpr = comparator;
if (cpr != null) {
do {
parent = t;
cmp = cpr.compare(key, t.key);
if (cmp < 0)
t = t.left;
elseif (cmp > 0)
t = t.right;
elsereturn t.setValue(value);
} while (t != null);
}
else {
if (key == null)
thrownewNullPointerException();
@SuppressWarnings("unchecked")
Comparable<? super K> k = (Comparable<? super K>) key;
do {
parent = t;
cmp = k.compareTo(t.key);
if (cmp < 0)
t = t.left;
elseif (cmp > 0)
t = t.right;
elsereturn t.setValue(value);
} while (t != null);
}
Entry<K,V> e = newEntry<>(key, value, parent);
if (cmp < 0)
parent.left = e;
else
parent.right = e;
fixAfterInsertion(e);
size++;
modCount++;
returnnull;
}
/*
* Encodings for Node hash fields. See above for explanation.
*/staticfinalintMOVED= -1; // hash for forwarding nodesstaticfinalintTREEBIN= -2; // hash for roots of treesstaticfinalintRESERVED= -3; // hash for transient reservationsstaticfinalintHASH_BITS=0x7fffffff; // usable bits of normal node hash
/* ---------------- Table element access -------------- *//*
* Volatile access methods are used for table elements as well as
* elements of in-progress next table while resizing. All uses of
* the tab arguments must be null checked by callers. All callers
* also paranoically precheck that tab's length is not zero (or an
* equivalent check), thus ensuring that any index argument taking
* the form of a hash value anded with (length - 1) is a valid
* index. Note that, to be correct wrt arbitrary concurrency
* errors by users, these checks must operate on local variables,
* which accounts for some odd-looking inline assignments below.
* Note that calls to setTabAt always occur within locked regions,
* and so in principle require only release ordering, not
* full volatile semantics, but are currently coded as volatile
* writes to be conservative.
*/@SuppressWarnings("unchecked")staticfinal <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}
/**
* Returns the value to which the specified key is mapped,
* or {@code null} if this map contains no mapping for the key.
*
* <p>More formally, if this map contains a mapping from a key
* {@code k} to a value {@code v} such that {@code key.equals(k)},
* then this method returns {@code v}; otherwise it returns
* {@code null}. (There can be at most one such mapping.)
*
* @throws NullPointerException if the specified key is null
*/public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
inth= spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
elseif (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
returnnull;
}
/**
* If the specified key is not already associated with a value,
* attempts to compute its value using the given mapping function
* and enters it into this map unless {@code null}. The entire
* method invocation is performed atomically, so the function is
* applied at most once per key. Some attempted update operations
* on this map by other threads may be blocked while computation
* is in progress, so the computation should be short and simple,
* and must not attempt to update any other mappings of this map.
*
* @param key key with which the specified value is to be associated
* @param mappingFunction the function to compute a value
* @return the current (existing or computed) value associated with
* the specified key, or null if the computed value is null
* @throws NullPointerException if the specified key or mappingFunction
* is null
* @throws IllegalStateException if the computation detectably
* attempts a recursive update to this map that would
* otherwise never complete
* @throws RuntimeException or Error if the mappingFunction does so,
* in which case the mapping is left unestablished
*/public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {
if (key == null || mappingFunction == null)
thrownewNullPointerException();
inth= spread(key.hashCode());
Vval=null;
intbinCount=0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
tab = initTable();
elseif ((f = tabAt(tab, i = (n - 1) & h)) == null) {
Node<K,V> r = newReservationNode<K,V>();
synchronized (r) {
if (casTabAt(tab, i, null, r)) {
binCount = 1;
Node<K,V> node = null;
try {
if ((val = mappingFunction.apply(key)) != null)
node = newNode<K,V>(h, key, val, null);
} finally {
setTabAt(tab, i, node);
}
}
}
if (binCount != 0)
break;
}
elseif ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
booleanadded=false;
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek; V ev;
if (e.hash == h &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
val = e.val;
break;
}
Node<K,V> pred = e;
if ((e = e.next) == null) {
if ((val = mappingFunction.apply(key)) != null) {
added = true;
pred.next = newNode<K,V>(h, key, val, null);
}
break;
}
}
}
elseif (f instanceof TreeBin) {
binCount = 2;
TreeBin<K,V> t = (TreeBin<K,V>)f;
TreeNode<K,V> r, p;
if ((r = t.root) != null &&
(p = r.findTreeNode(h, key, null)) != null)
val = p.val;
elseif ((val = mappingFunction.apply(key)) != null) {
added = true;
t.putTreeVal(h, key, val);
}
}
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (!added)
return val;
break;
}
}
}
if (val != null)
addCount(1L, binCount);
return val;
}
/**
* Adds to count, and if table is too small and not already
* resizing, initiates transfer. If already resizing, helps
* perform transfer if work is available. Rechecks occupancy
* after a transfer to see if another resize is already needed
* because resizings are lagging additions.
*
* @param x the count to add
* @param check if <0, don't check resize, if <= 1 only check if uncontended
*/privatefinalvoidaddCount(long x, int check) {
CounterCell[] as; long b, s;
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell a; long v; int m;
booleanuncontended=true;
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
s = sumCount();
}
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
intrs= resizeStamp(n) << RESIZE_STAMP_SHIFT;
if (sc < 0) {
if (sc == rs + MAX_RESIZERS || sc == rs + 1 ||
(nt = nextTable) == null || transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
elseif (U.compareAndSwapInt(this, SIZECTL, sc, rs + 2))
transfer(tab, null);
s = sumCount();
}
}
}
/**
* Appends the specified element to the end of this list.
*
* @param e element to be appended to this list
* @return {@code true} (as specified by {@link Collection#add})
*/publicbooleanadd(E e) {
finalReentrantLocklock=this.lock;
lock.lock();
try {
Object[] elements = getArray();
intlen= elements.length;
Object[] newElements = Arrays.copyOf(elements, len + 1);
newElements[len] = e;
setArray(newElements);
returntrue;
} finally {
lock.unlock();
}
}
/**
* Returns an iterator over the elements in this list in proper sequence.
*
* <p>The returned iterator provides a snapshot of the state of the list
* when the iterator was constructed. No synchronization is needed while
* traversing the iterator. The iterator does <em>NOT</em> support the
* {@code remove} method.
*
* @return an iterator over the elements in this list in proper sequence
*/public Iterator<E> iterator() {
returnnewCOWIterator<E>(getArray(), 0);
}
staticfinalclassCOWIterator<E> implementsListIterator<E> {
/** Snapshot of the array */privatefinal Object[] snapshot;
/** Index of element to be returned by subsequent call to next. */privateint cursor;
privateCOWIterator(Object[] elements, int initialCursor) {
cursor = initialCursor;
snapshot = elements;
}
publicbooleanhasNext() {
return cursor < snapshot.length;
}
publicbooleanhasPrevious() {
return cursor > 0;
}
@SuppressWarnings("unchecked")public E next() {
if (! hasNext())
thrownewNoSuchElementException();
return (E) snapshot[cursor++];
}
/**
* Appends the element, if not present.
*
* @param e element to be added to this list, if absent
* @return {@code true} if the element was added
*/publicbooleanaddIfAbsent(E e) {
Object[] snapshot = getArray();
return indexOf(e, snapshot, 0, snapshot.length) >= 0 ? false :
addIfAbsent(e, snapshot);
}
/**
* Inserts the specified element at the tail of this queue.
* As the queue is unbounded, this method will never return {@code false}.
*
* @return {@code true} (as specified by {@link Queue#offer})
* @throws NullPointerException if the specified element is null
*/publicbooleanoffer(E e) {
checkNotNull(e);
final Node<E> newNode = newNode<E>(e);
for (Node<E> t = tail, p = t;;) {
Node<E> q = p.next;
if (q == null) {
// p is last nodeif (p.casNext(null, newNode)) {
// Successful CAS is the linearization point// for e to become an element of this queue,// and for newNode to become "live".if (p != t) // hop two nodes at a time
casTail(t, newNode); // Failure is OK.returntrue;
}
// Lost CAS race to another thread; re-read next
}
elseif (p == q)
// We have fallen off list. If tail is unchanged, it// will also be off-list, in which case we need to// jump to head, from which all live nodes are always// reachable. Else the new tail is a better bet.
p = (t != (t = tail)) ? t : head;
else// Check for tail updates after two hops.
p = (p != t && t != (t = tail)) ? t : q;
}
}
public E poll() {
restartFromHead:
for (;;) {
for (Node<E> h = head, p = h, q;;) {
Eitem= p.item;
if (item != null && p.casItem(item, null)) {
// Successful CAS is the linearization point// for item to be removed from this queue.if (p != h) // hop two nodes at a time
updateHead(h, ((q = p.next) != null) ? q : p);
return item;
}
elseif ((q = p.next) == null) {
updateHead(h, p);
returnnull;
}
elseif (p == q)
continue restartFromHead;
else
p = q;
}
}
}
/**
* Tries to CAS head to p. If successful, repoint old head to itself
* as sentinel for succ(), below.
*/finalvoidupdateHead(Node<E> h, Node<E> p) {
if (h != p && casHead(h, p))
h.lazySetNext(h);
}
/**
* Returns the value in the current thread's copy of this
* thread-local variable. If the variable has no value for the
* current thread, it is first initialized to the value returned
* by an invocation of the {@link #initialValue} method.
*
* @return the current thread's value of this thread-local
*/public T get() {
Threadt= Thread.currentThread();
ThreadLocalMapmap= getMap(t);
if (map != null) {
ThreadLocalMap.Entrye= map.getEntry(this);
if (e != null) {
@SuppressWarnings("unchecked")Tresult= (T)e.value;
return result;
}
}
return setInitialValue();
}
/**
* The entries in this hash map extend WeakReference, using
* its main ref field as the key (which is always a
* ThreadLocal object). Note that null keys (i.e. entry.get()
* == null) mean that the key is no longer referenced, so the
* entry can be expunged from table. Such entries are referred to
* as "stale entries" in the code that follows.
*/staticclassEntryextendsWeakReference<ThreadLocal<?>> {
/** The value associated with this ThreadLocal. */
Object value;
Entry(ThreadLocal<?> k, Object v) {
super(k);
value = v;
}
}
/**
* Removes the current thread's value for this thread-local
* variable. If this thread-local variable is subsequently
* {@linkplain #get read} by the current thread, its value will be
* reinitialized by invoking its {@link #initialValue} method,
* unless its value is {@linkplain #set set} by the current thread
* in the interim. This may result in multiple invocations of the
* {@code initialValue} method in the current thread.
*
* @since 1.5
*/publicvoidremove() {
ThreadLocalMapm= getMap(Thread.currentThread());
if (m != null) {
m.remove(this);
}
}
/**
* Variant of set() to establish initialValue. Used instead
* of set() in case user has overridden the set() method.
*
* @return the initial value
*/private T setInitialValue() {
Tvalue= initialValue();
Threadt= Thread.currentThread();
ThreadLocalMapmap= getMap(t);
if (map != null) {
map.set(this, value);
} else {
createMap(t, value);
}
if (thisinstanceof TerminatingThreadLocal) {
TerminatingThreadLocal.register((TerminatingThreadLocal<?>) this);
}
return value;
}
/**
* Get the map associated with a ThreadLocal. Overridden in
* InheritableThreadLocal.
*
* @param t the current thread
* @return the map
*/
ThreadLocalMap getMap(Thread t) {
return t.threadLocals;
}
/**
* Get the entry associated with key. This method
* itself handles only the fast path: a direct hit of existing
* key. It otherwise relays to getEntryAfterMiss. This is
* designed to maximize performance for direct hits, in part
* by making this method readily inlinable.
*
* @param key the thread local object
* @return the entry associated with key, or null if no such
*/private Entry getEntry(ThreadLocal<?> key) {
inti= key.threadLocalHashCode & (table.length - 1);
Entrye= table[i];
if (e != null && e.get() == key)
return e;
elsereturn getEntryAfterMiss(key, i, e);
}
先检查理想槽，未直接命中则走getEntryAfterMiss。常见无碰撞读取路径很短。
源码2：private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e)
/**
* Version of getEntry method for use when key is not found in
* its direct hash slot.
*
* @param key the thread local object
* @param i the table index for key's hash code
* @param e the entry at table[i]
* @return the entry associated with key, or null if no such
*/private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e) {
Entry[] tab = table;
intlen= tab.length;
while (e != null) {
ThreadLocal<?> k = e.get();
if (k == key)
return e;
if (k == null)
expungeStaleEntry(i);
else
i = nextIndex(i, len);
e = tab[i];
}
returnnull;
}
/**
* Expunge a stale entry by rehashing any possibly colliding entries
* lying between staleSlot and the next null slot. This also expunges
* any other stale entries encountered before the trailing null. See
* Knuth, Section 6.4
*
* @param staleSlot index of slot known to have null key
* @return the index of the next null slot after staleSlot
* (all between staleSlot and this slot will have been checked
* for expunging).
*/privateintexpungeStaleEntry(int staleSlot) {
Entry[] tab = table;
intlen= tab.length;
// expunge entry at staleSlot
tab[staleSlot].value = null;
tab[staleSlot] = null;
size--;
// Rehash until we encounter null
Entry e;
int i;
for (i = nextIndex(staleSlot, len);
(e = tab[i]) != null;
i = nextIndex(i, len)) {
ThreadLocal<?> k = e.get();
if (k == null) {
e.value = null;
tab[i] = null;
size--;
} else {
inth= k.threadLocalHashCode & (len - 1);
if (h != i) {
tab[i] = null;
// Unlike Knuth 6.4 Algorithm R, we must scan until// null because multiple entries could have been stale.while (tab[h] != null)
h = nextIndex(h, len);
tab[h] = e;
}
}
}
return i;
}
/**
* Heuristically scan some cells looking for stale entries.
* This is invoked when either a new element is added, or
* another stale one has been expunged. It performs a
* logarithmic number of scans, as a balance between no
* scanning (fast but retains garbage) and a number of scans
* proportional to number of elements, that would find all
* garbage but would cause some insertions to take O(n) time.
*
* @param i a position known NOT to hold a stale entry. The
* scan starts at the element after i.
*
* @param n scan control: {@code log2(n)} cells are scanned,
* unless a stale entry is found, in which case
* {@code log2(table.length)-1} additional cells are scanned.
* When called from insertions, this parameter is the number
* of elements, but when from replaceStaleEntry, it is the
* table length. (Note: all this could be changed to be either
* more or less aggressive by weighting n instead of just
* using straight log n. But this version is simple, fast, and
* seems to work well.)
*
* @return true if any stale entries have been removed.
*/privatebooleancleanSomeSlots(int i, int n) {
booleanremoved=false;
Entry[] tab = table;
intlen= tab.length;
do {
i = nextIndex(i, len);
Entrye= tab[i];
if (e != null && e.get() == null) {
n = len;
removed = true;
i = expungeStaleEntry(i);
}
} while ( (n >>>= 1) != 0);
return removed;
}
/**
* Set the value associated with key.
*
* @param key the thread local object
* @param value the value to be set
*/privatevoidset(ThreadLocal<?> key, Object value) {
// We don't use a fast path as with get() because it is at// least as common to use set() to create new entries as// it is to replace existing ones, in which case, a fast// path would fail more often than not.
Entry[] tab = table;
intlen= tab.length;
inti= key.threadLocalHashCode & (len-1);
for (Entrye= tab[i];
e != null;
e = tab[i = nextIndex(i, len)]) {
ThreadLocal<?> k = e.get();
if (k == key) {
e.value = value;
return;
}
if (k == null) {
replaceStaleEntry(key, value, i);
return;
}
}
tab[i] = newEntry(key, value);
intsz= ++size;
if (!cleanSomeSlots(i, sz) && sz >= threshold)
rehash();
}
/**
* Replace a stale entry encountered during a set operation
* with an entry for the specified key. The value passed in
* the value parameter is stored in the entry, whether or not
* an entry already exists for the specified key.
*
* As a side effect, this method expunges all stale entries in the
* "run" containing the stale entry. (A run is a sequence of entries
* between two null slots.)
*
* @param key the key
* @param value the value to be associated with key
* @param staleSlot index of the first stale entry encountered while
* searching for key.
*/privatevoidreplaceStaleEntry(ThreadLocal<?> key, Object value,
int staleSlot) {
Entry[] tab = table;
intlen= tab.length;
Entry e;
// Back up to check for prior stale entry in current run.// We clean out whole runs at a time to avoid continual// incremental rehashing due to garbage collector freeing// up refs in bunches (i.e., whenever the collector runs).intslotToExpunge= staleSlot;
for (inti= prevIndex(staleSlot, len);
(e = tab[i]) != null;
i = prevIndex(i, len))
if (e.get() == null)
slotToExpunge = i;
// Find either the key or trailing null slot of run, whichever// occurs firstfor (inti= nextIndex(staleSlot, len);
(e = tab[i]) != null;
i = nextIndex(i, len)) {
ThreadLocal<?> k = e.get();
// If we find key, then we need to swap it// with the stale entry to maintain hash table order.// The newly stale slot, or any other stale slot// encountered above it, can then be sent to expungeStaleEntry// to remove or rehash all of the other entries in run.if (k == key) {
e.value = value;
tab[i] = tab[staleSlot];
tab[staleSlot] = e;
// Start expunge at preceding stale entry if it existsif (slotToExpunge == staleSlot)
slotToExpunge = i;
cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
return;
}
// If we didn't find stale entry on backward scan, the// first stale entry seen while scanning for key is the// first still present in the run.if (k == null && slotToExpunge == staleSlot)
slotToExpunge = i;
}
// If key not found, put new entry in stale slot
tab[staleSlot].value = null;
tab[staleSlot] = newEntry(key, value);
// If there are any other stale entries in run, expunge themif (slotToExpunge != staleSlot)
cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
}
/**
* Atomically increments by one the current value.
*
* @return the updated value
*/publicfinalintincrementAndGet() {
return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
}
/**
* Atomically updates the current value with the results of
* applying the given function, returning the updated value. The
* function should be side-effect-free, since it may be re-applied
* when attempted updates fail due to contention among threads.
*
* @param updateFunction a side-effect-free function
* @return the updated value
* @since 1.8
*/publicfinalintupdateAndGet(IntUnaryOperator updateFunction) {
int prev, next;
do {
prev = get();
next = updateFunction.applyAsInt(prev);
} while (!compareAndSet(prev, next));
return next;
}
循环中函数可能被重复调用，因此应无副作用；CAS失败后必须基于新prev重算next。
源码3：public final int getAndAddInt(Object o, long offset, int delta)
// The following contain CAS-based Java implementations used on// platforms not supporting native instructions/**
* Atomically adds the given value to the current value of a field
* or array element within the given object <code>o</code>
* at the given <code>offset</code>.
*
* @param o object/array to update the field/element in
* @param offset field/element offset
* @param delta the value to add
* @return the previous value
* @since 1.8
*/publicfinalintgetAndAddInt(Object o, long offset, int delta) {
int v;
do {
v = getIntVolatile(o, offset);
} while (!compareAndSwapInt(o, offset, v, v + delta));
return v;
}
/**
* Atomically sets the value to the given updated value
* if the current value {@code ==} the expected value.
*
* @param expect the expected value
* @param update the new value
* @return {@code true} if successful. False return indicates that
* the actual value was not equal to the expected value.
*/publicfinalbooleancompareAndSet(int expect, int update) {
return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}
/**
* Returns the current sum. The returned value is <em>NOT</em> an
* atomic snapshot; invocation in the absence of concurrent
* updates returns an accurate result, but concurrent updates that
* occur while the sum is being calculated might not be
* incorporated.
*
* @return the sum
*/publiclongsum() {
Cell[] as = cells; Cell a;
longsum= base;
if (as != null) {
for (inti=0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}
/**
* Makes available the permit for the given thread, if it
* was not already available. If the thread was blocked on
* {@code park} then it will unblock. Otherwise, its next call
* to {@code park} is guaranteed not to block. This operation
* is not guaranteed to have any effect at all if the given
* thread has not been started.
*
* @param thread the thread to unpark, or {@code null}, in which case
* this operation has no effect
*/publicstaticvoidunpark(Thread thread) {
if (thread != null)
UNSAFE.unpark(thread);
}
/**
* Disables the current thread for thread scheduling purposes unless the
* permit is available.
*
* <p>If the permit is available then it is consumed and the call returns
* immediately; otherwise
* the current thread becomes disabled for thread scheduling
* purposes and lies dormant until one of three things happens:
*
* <ul>
* <li>Some other thread invokes {@link #unpark unpark} with the
* current thread as the target; or
*
* <li>Some other thread {@linkplain Thread#interrupt interrupts}
* the current thread; or
*
* <li>The call spuriously (that is, for no reason) returns.
* </ul>
*
* <p>This method does <em>not</em> report which of these caused the
* method to return. Callers should re-check the conditions which caused
* the thread to park in the first place. Callers may also determine,
* for example, the interrupt status of the thread upon return.
*
* @param blocker the synchronization object responsible for this
* thread parking
* @since 1.6
*/publicstaticvoidpark(Object blocker) {
Threadt= Thread.currentThread();
setBlocker(t, blocker);
UNSAFE.park(false, 0L);
setBlocker(t, null);
}
/**
* Disables the current thread for thread scheduling purposes, for up to
* the specified waiting time, unless the permit is available.
*
* <p>If the permit is available then it is consumed and the call
* returns immediately; otherwise the current thread becomes disabled
* for thread scheduling purposes and lies dormant until one of four
* things happens:
*
* <ul>
* <li>Some other thread invokes {@link #unpark unpark} with the
* current thread as the target; or
*
* <li>Some other thread {@linkplain Thread#interrupt interrupts}
* the current thread; or
*
* <li>The specified waiting time elapses; or
*
* <li>The call spuriously (that is, for no reason) returns.
* </ul>
*
* <p>This method does <em>not</em> report which of these caused the
* method to return. Callers should re-check the conditions which caused
* the thread to park in the first place. Callers may also determine,
* for example, the interrupt status of the thread, or the elapsed time
* upon return.
*
* @param blocker the synchronization object responsible for this
* thread parking
* @param nanos the maximum number of nanoseconds to wait
* @since 1.6
*/publicstaticvoidparkNanos(Object blocker, long nanos) {
if (nanos > 0) {
Threadt= Thread.currentThread();
setBlocker(t, blocker);
UNSAFE.park(false, nanos);
setBlocker(t, null);
}
}
/**
* Acquires in exclusive mode, ignoring interrupts. Implemented
* by invoking at least once {@link #tryAcquire},
* returning on success. Otherwise the thread is queued, possibly
* repeatedly blocking and unblocking, invoking {@link
* #tryAcquire} until success. This method can be used
* to implement method {@link Lock#lock}.
*
* @param arg the acquire argument. This value is conveyed to
* {@link #tryAcquire} but is otherwise uninterpreted and
* can represent anything you like.
*/publicfinalvoidacquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
/*
* Various flavors of acquire, varying in exclusive/shared and
* control modes. Each is mostly the same, but annoyingly
* different. Only a little bit of factoring is possible due to
* interactions of exception mechanics (including ensuring that we
* cancel if tryAcquire throws exception) and other control, at
* least not without hurting performance too much.
*//**
* Acquires in exclusive uninterruptible mode for thread already in
* queue. Used by condition wait methods as well as acquire.
*
* @param node the node
* @param arg the acquire argument
* @return {@code true} if interrupted while waiting
*/finalbooleanacquireQueued(final Node node, int arg) {
booleanfailed=true;
try {
booleaninterrupted=false;
for (;;) {
finalNodep= node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
/**
* Checks and updates status for a node that failed to acquire.
* Returns true if thread should block. This is the main signal
* control in all acquire loops. Requires that pred == node.prev.
*
* @param pred node's predecessor holding status
* @param node the node
* @return {@code true} if thread should block
*/privatestaticbooleanshouldParkAfterFailedAcquire(Node pred, Node node) {
intws= pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/returntrue;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* waitStatus must be 0 or PROPAGATE. Indicate that we
* need a signal, but don't park yet. Caller will need to
* retry to make sure it cannot acquire before parking.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
returnfalse;
}
/**
* Wakes up node's successor, if one exists.
*
* @param node the node
*/privatevoidunparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling. It is OK if this
* fails or if status is changed by waiting thread.
*/intws= node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
/*
* Thread to unpark is held in successor, which is normally
* just the next node. But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/Nodes= node.next;
if (s == null || s.waitStatus > 0) {
s = null;
for (Nodet= tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
if (s != null)
LockSupport.unpark(s.thread);
}
/**
* Release action for shared mode -- signals successor and ensures
* propagation. (Note: For exclusive mode, release just amounts
* to calling unparkSuccessor of head if it needs signal.)
*/privatevoiddoReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases. This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/for (;;) {
Nodeh= head;
if (h != null && h != tail) {
intws= h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
elseif (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changedbreak;
}
}
/**
* Implements interruptible condition wait.
* <ol>
* <li> If current thread is interrupted, throw InterruptedException.
* <li> Save lock state returned by {@link #getState}.
* <li> Invoke {@link #release} with saved state as argument,
* throwing IllegalMonitorStateException if it fails.
* <li> Block until signalled or interrupted.
* <li> Reacquire by invoking specialized version of
* {@link #acquire} with saved state as argument.
* <li> If interrupted while blocked in step 4, throw InterruptedException.
* </ol>
*/publicfinalvoidawait()throws InterruptedException {
if (Thread.interrupted())
thrownewInterruptedException();
Nodenode= addConditionWaiter();
intsavedState= fullyRelease(node);
intinterruptMode=0;
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
/**
* Transfers a node from a condition queue onto sync queue.
* Returns true if successful.
* @param node the node
* @return true if successfully transferred (else the node was
* cancelled before signal)
*/finalbooleantransferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
returnfalse;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/Nodep= enq(node);
intws= p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
returntrue;
}
// public methods/**
* Moves the longest-waiting thread, if one exists, from the
* wait queue for this condition to the wait queue for the
* owning lock.
*
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code false}
*/publicfinalvoidsignal() {
if (!isHeldExclusively())
thrownewIllegalMonitorStateException();
Nodefirst= firstWaiter;
if (first != null)
doSignal(first);
}
/**
* Acquires the lock only if it is not held by another thread at the time
* of invocation.
*
* <p>Acquires the lock if it is not held by another thread and
* returns immediately with the value {@code true}, setting the
* lock hold count to one. Even when this lock has been set to use a
* fair ordering policy, a call to {@code tryLock()} <em>will</em>
* immediately acquire the lock if it is available, whether or not
* other threads are currently waiting for the lock.
* This "barging" behavior can be useful in certain
* circumstances, even though it breaks fairness. If you want to honor
* the fairness setting for this lock, then use
* {@link #tryLock(long, TimeUnit) tryLock(0, TimeUnit.SECONDS) }
* which is almost equivalent (it also detects interruption).
*
* <p>If the current thread already holds this lock then the hold
* count is incremented by one and the method returns {@code true}.
*
* <p>If the lock is held by another thread then this method will return
* immediately with the value {@code false}.
*
* @return {@code true} if the lock was free and was acquired by the
* current thread, or the lock was already held by the current
* thread; and {@code false} otherwise
*/publicbooleantryLock() {
return sync.nonfairTryAcquire(1);
}
/*
* Read vs write count extraction constants and functions.
* Lock state is logically divided into two unsigned shorts:
* The lower one representing the exclusive (writer) lock hold count,
* and the upper the shared (reader) hold count.
*/staticfinalintSHARED_SHIFT=16;
staticfinalintSHARED_UNIT= (1 << SHARED_SHIFT);
staticfinalintMAX_COUNT= (1 << SHARED_SHIFT) - 1;
staticfinalintEXCLUSIVE_MASK= (1 << SHARED_SHIFT) - 1;
/** Returns the number of shared holds represented in count */staticintsharedCount(int c) { return c >>> SHARED_SHIFT; }
/** Returns the number of exclusive holds represented in count */staticintexclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
protectedfinalbooleantryAcquire(int acquires) {
/*
* Walkthrough:
* 1. If read count nonzero or write count nonzero
* and owner is a different thread, fail.
* 2. If count would saturate, fail. (This can only
* happen if count is already nonzero.)
* 3. Otherwise, this thread is eligible for lock if
* it is either a reentrant acquire or
* queue policy allows it. If so, update state
* and set owner.
*/Threadcurrent= Thread.currentThread();
intc= getState();
intw= exclusiveCount(c);
if (c != 0) {
// (Note: if c != 0 and w == 0 then shared count != 0)if (w == 0 || current != getExclusiveOwnerThread())
returnfalse;
if (w + exclusiveCount(acquires) > MAX_COUNT)
thrownewError("Maximum lock count exceeded");
// Reentrant acquire
setState(c + acquires);
returntrue;
}
if (writerShouldBlock() ||
!compareAndSetState(c, c + acquires))
returnfalse;
setExclusiveOwnerThread(current);
returntrue;
}
protectedfinalinttryAcquireShared(int unused) {
/*
* Walkthrough:
* 1. If write lock held by another thread, fail.
* 2. Otherwise, this thread is eligible for
* lock wrt state, so ask if it should block
* because of queue policy. If not, try
* to grant by CASing state and updating count.
* Note that step does not check for reentrant
* acquires, which is postponed to full version
* to avoid having to check hold count in
* the more typical non-reentrant case.
* 3. If step 2 fails either because thread
* apparently not eligible or CAS fails or count
* saturated, chain to version with full retry loop.
*/Threadcurrent= Thread.currentThread();
intc= getState();
if (exclusiveCount(c) != 0 &&
getExclusiveOwnerThread() != current)
return -1;
intr= sharedCount(c);
if (!readerShouldBlock() &&
r < MAX_COUNT &&
compareAndSetState(c, c + SHARED_UNIT)) {
if (r == 0) {
firstReader = current;
firstReaderHoldCount = 1;
} elseif (firstReader == current) {
firstReaderHoldCount++;
} else {
HoldCounterrh= cachedHoldCounter;
if (rh == null || rh.tid != getThreadId(current))
cachedHoldCounter = rh = readHolds.get();
elseif (rh.count == 0)
readHolds.set(rh);
rh.count++;
}
return1;
}
return fullTryAcquireShared(current);
}
/**
* Inserts the specified element at the tail of this queue, waiting
* for space to become available if the queue is full.
*
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/publicvoidput(E e)throws InterruptedException {
checkNotNull(e);
finalReentrantLocklock=this.lock;
lock.lockInterruptibly();
try {
while (count == items.length)
notFull.await();
enqueue(e);
} finally {
lock.unlock();
}
}
/**
* Inserts the specified element at the tail of this queue, waiting if
* necessary for space to become available.
*
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/publicvoidput(E e)throws InterruptedException {
if (e == null) thrownewNullPointerException();
// Note: convention in all put/take/etc is to preset local var// holding count negative to indicate failure unless set.intc= -1;
Node<E> node = newNode<E>(e);
finalReentrantLockputLock=this.putLock;
finalAtomicIntegercount=this.count;
putLock.lockInterruptibly();
try {
/*
* Note that count is used in wait guard even though it is
* not protected by lock. This works because count can
* only decrease at this point (all other puts are shut
* out by lock), and we (or some other waiting put) are
* signalled if it ever changes from capacity. Similarly
* for all other uses of count in other wait guards.
*/while (count.get() == capacity) {
notFull.await();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}
/**
* Removes a node from head of queue.
*
* @return the node
*/private E dequeue() {
// assert takeLock.isHeldByCurrentThread();// assert head.item == null;
Node<E> h = head;
Node<E> first = h.next;
h.next = h; // help GC
head = first;
Ex= first.item;
first.item = null;
return x;
}
/**
* Creates a {@code SynchronousQueue} with the specified fairness policy.
*
* @param fair if true, waiting threads contend in FIFO order for
* access; otherwise the order is unspecified.
*/publicSynchronousQueue(boolean fair) {
transferer = fair ? newTransferQueue<E>() : newTransferStack<E>();
}
/**
* Adds the specified element to this queue, waiting if necessary for
* another thread to receive it.
*
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/publicvoidput(E e)throws InterruptedException {
if (e == null) thrownewNullPointerException();
if (transferer.transfer(e, false, 0) == null) {
Thread.interrupted();
thrownewInterruptedException();
}
}
/**
* Inserts the specified element into this queue, if another thread is
* waiting to receive it.
*
* @param e the element to add
* @return {@code true} if the element was added to this queue, else
* {@code false}
* @throws NullPointerException if the specified element is null
*/publicbooleanoffer(E e) {
if (e == null) thrownewNullPointerException();
return transferer.transfer(e, true, 0) != null;
}
/**
* 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 {
finalReentrantLocklock=this.lock;
lock.lockInterruptibly();
try {
for (;;) {
Efirst= q.peek();
if (first == null)
available.await();
else {
longdelay= first.getDelay(NANOSECONDS);
if (delay <= 0)
return q.poll();
first = null; // don't retain ref while waitingif (leader != null)
available.await();
else {
ThreadthisThread= 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();
}
}
/**
* Inserts the specified element into this delay queue.
*
* @param e the element to add
* @return {@code true}
* @throws NullPointerException if the specified element is null
*/publicbooleanoffer(E e) {
finalReentrantLocklock=this.lock;
lock.lock();
try {
q.offer(e);
if (q.peek() == e) {
leader = null;
available.signal();
}
returntrue;
} finally {
lock.unlock();
}
}
/**
* Executes the given task sometime in the future. The task
* may execute in a new thread or in an existing pooled thread.
*
* If the task cannot be submitted for execution, either because this
* executor has been shutdown or because its capacity has been reached,
* the task is handled by the current {@code RejectedExecutionHandler}.
*
* @param command the task to execute
* @throws RejectedExecutionException at discretion of
* {@code RejectedExecutionHandler}, if the task
* cannot be accepted for execution
* @throws NullPointerException if {@code command} is null
*/publicvoidexecute(Runnable command) {
if (command == null)
thrownewNullPointerException();
/*
* Proceed in 3 steps:
*
* 1. If fewer than corePoolSize threads are running, try to
* start a new thread with the given command as its first
* task. The call to addWorker atomically checks runState and
* workerCount, and so prevents false alarms that would add
* threads when it shouldn't, by returning false.
*
* 2. If a task can be successfully queued, then we still need
* to double-check whether we should have added a thread
* (because existing ones died since last checking) or that
* the pool shut down since entry into this method. So we
* recheck state and if necessary roll back the enqueuing if
* stopped, or start a new thread if there are none.
*
* 3. If we cannot queue task, then we try to add a new
* thread. If it fails, we know we are shut down or saturated
* and so reject the task.
*/intc= ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
intrecheck= ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
elseif (workerCountOf(recheck) == 0)
addWorker(null, false);
}
elseif (!addWorker(command, false))
reject(command);
}
/*
* Methods for creating, running and cleaning up after workers
*//**
* Checks if a new worker can be added with respect to current
* pool state and the given bound (either core or maximum). If so,
* the worker count is adjusted accordingly, and, if possible, a
* new worker is created and started, running firstTask as its
* first task. This method returns false if the pool is stopped or
* eligible to shut down. It also returns false if the thread
* factory fails to create a thread when asked. If the thread
* creation fails, either due to the thread factory returning
* null, or due to an exception (typically OutOfMemoryError in
* Thread.start()), we roll back cleanly.
*
* @param firstTask the task the new thread should run first (or
* null if none). Workers are created with an initial first task
* (in method execute()) to bypass queuing when there are fewer
* than corePoolSize threads (in which case we always start one),
* or when the queue is full (in which case we must bypass queue).
* Initially idle threads are usually created via
* prestartCoreThread or to replace other dying workers.
*
* @param core if true use corePoolSize as bound, else
* maximumPoolSize. (A boolean indicator is used here rather than a
* value to ensure reads of fresh values after checking other pool
* state).
* @return true if successful
*/privatebooleanaddWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
intc= ctl.get();
intrs= runStateOf(c);
// Check if queue empty only if necessary.if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
returnfalse;
for (;;) {
intwc= workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
returnfalse;
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctlif (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
booleanworkerStarted=false;
booleanworkerAdded=false;
Workerw=null;
try {
w = newWorker(firstTask);
finalThreadt= w.thread;
if (t != null) {
finalReentrantLockmainLock=this.mainLock;
mainLock.lock();
try {
// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.intrs= runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startablethrownewIllegalThreadStateException();
workers.add(w);
ints= workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
/*
* Misc utilities, most of which are also exported to
* ScheduledThreadPoolExecutor
*//**
* Invokes the rejected execution handler for the given command.
* Package-protected for use by ScheduledThreadPoolExecutor.
*/finalvoidreject(Runnable command) {
handler.rejectedExecution(command, this);
}
/**
* The main pool control state, ctl, is an atomic integer packing
* two conceptual fields
* workerCount, indicating the effective number of threads
* runState, indicating whether running, shutting down etc
*
* In order to pack them into one int, we limit workerCount to
* (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
* billion) otherwise representable. If this is ever an issue in
* the future, the variable can be changed to be an AtomicLong,
* and the shift/mask constants below adjusted. But until the need
* arises, this code is a bit faster and simpler using an int.
*
* The workerCount is the number of workers that have been
* permitted to start and not permitted to stop. The value may be
* transiently different from the actual number of live threads,
* for example when a ThreadFactory fails to create a thread when
* asked, and when exiting threads are still performing
* bookkeeping before terminating. The user-visible pool size is
* reported as the current size of the workers set.
*
* The runState provides the main lifecycle control, taking on values:
*
* RUNNING: Accept new tasks and process queued tasks
* SHUTDOWN: Don't accept new tasks, but process queued tasks
* STOP: Don't accept new tasks, don't process queued tasks,
* and interrupt in-progress tasks
* TIDYING: All tasks have terminated, workerCount is zero,
* the thread transitioning to state TIDYING
* will run the terminated() hook method
* TERMINATED: terminated() has completed
*
* The numerical order among these values matters, to allow
* ordered comparisons. The runState monotonically increases over
* time, but need not hit each state. The transitions are:
*
* RUNNING -> SHUTDOWN
* On invocation of shutdown(), perhaps implicitly in finalize()
* (RUNNING or SHUTDOWN) -> STOP
* On invocation of shutdownNow()
* SHUTDOWN -> TIDYING
* When both queue and pool are empty
* STOP -> TIDYING
* When pool is empty
* TIDYING -> TERMINATED
* When the terminated() hook method has completed
*
* Threads waiting in awaitTermination() will return when the
* state reaches TERMINATED.
*
* Detecting the transition from SHUTDOWN to TIDYING is less
* straightforward than you'd like because the queue may become
* empty after non-empty and vice versa during SHUTDOWN state, but
* we can only terminate if, after seeing that it is empty, we see
* that workerCount is 0 (which sometimes entails a recheck -- see
* below).
*/privatefinalAtomicIntegerctl=newAtomicInteger(ctlOf(RUNNING, 0));
privatestaticfinalintCOUNT_BITS= Integer.SIZE - 3;
privatestaticfinalintCAPACITY= (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bitsprivatestaticfinalintRUNNING= -1 << COUNT_BITS;
privatestaticfinalintSHUTDOWN=0 << COUNT_BITS;
privatestaticfinalintSTOP=1 << COUNT_BITS;
privatestaticfinalintTIDYING=2 << COUNT_BITS;
privatestaticfinalintTERMINATED=3 << COUNT_BITS;
// Packing and unpacking ctlprivatestaticintrunStateOf(int c) { return c & ~CAPACITY; }
privatestaticintworkerCountOf(int c) { return c & CAPACITY; }
privatestaticintctlOf(int rs, int wc) { return rs | wc; }
/*
* Bit field accessors that don't require unpacking ctl.
* These depend on the bit layout and on workerCount being never negative.
*/privatestaticbooleanrunStateLessThan(int c, int s) {
return c < s;
}
privatestaticbooleanrunStateAtLeast(int c, int s) {
return c >= s;
}
/**
* Main worker run loop. Repeatedly gets tasks from queue and
* executes them, while coping with a number of issues:
*
* 1. We may start out with an initial task, in which case we
* don't need to get the first one. Otherwise, as long as pool is
* running, we get tasks from getTask. If it returns null then the
* worker exits due to changed pool state or configuration
* parameters. Other exits result from exception throws in
* external code, in which case completedAbruptly holds, which
* usually leads processWorkerExit to replace this thread.
*
* 2. Before running any task, the lock is acquired to prevent
* other pool interrupts while the task is executing, and then we
* ensure that unless pool is stopping, this thread does not have
* its interrupt set.
*
* 3. Each task run is preceded by a call to beforeExecute, which
* might throw an exception, in which case we cause thread to die
* (breaking loop with completedAbruptly true) without processing
* the task.
*
* 4. Assuming beforeExecute completes normally, we run the task,
* gathering any of its thrown exceptions to send to afterExecute.
* We separately handle RuntimeException, Error (both of which the
* specs guarantee that we trap) and arbitrary Throwables.
* Because we cannot rethrow Throwables within Runnable.run, we
* wrap them within Errors on the way out (to the thread's
* UncaughtExceptionHandler). Any thrown exception also
* conservatively causes thread to die.
*
* 5. After task.run completes, we call afterExecute, which may
* also throw an exception, which will also cause thread to
* die. According to JLS Sec 14.20, this exception is the one that
* will be in effect even if task.run throws.
*
* The net effect of the exception mechanics is that afterExecute
* and the thread's UncaughtExceptionHandler have as accurate
* information as we can provide about any problems encountered by
* user code.
*
* @param w the worker
*/finalvoidrunWorker(Worker w) {
Threadwt= Thread.currentThread();
Runnabletask= w.firstTask;
w.firstTask = null;
w.unlock(); // allow interruptsbooleancompletedAbruptly=true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;// if not, ensure thread is not interrupted. This// requires a recheck in second case to deal with// shutdownNow race while clearing interruptif ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwablethrown=null;
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; thrownewError(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
/**
* Performs blocking or timed wait for a task, depending on
* current configuration settings, or returns null if this worker
* must exit because of any of:
* 1. There are more than maximumPoolSize workers (due to
* a call to setMaximumPoolSize).
* 2. The pool is stopped.
* 3. The pool is shutdown and the queue is empty.
* 4. This worker timed out waiting for a task, and timed-out
* workers are subject to termination (that is,
* {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
* both before and after the timed wait, and if the queue is
* non-empty, this worker is not the last thread in the pool.
*
* @return task, or null if the worker must exit, in which case
* workerCount is decremented
*/private Runnable getTask() {
booleantimedOut=false; // Did the last poll() time out?for (;;) {
intc= ctl.get();
intrs= runStateOf(c);
// Check if queue empty only if necessary.if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
returnnull;
}
intwc= workerCountOf(c);
// Are workers subject to culling?booleantimed= allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
returnnull;
continue;
}
try {
Runnabler= timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
/**
* Rolls back the worker thread creation.
* - removes worker from workers, if present
* - decrements worker count
* - rechecks for termination, in case the existence of this
* worker was holding up termination
*/privatevoidaddWorkerFailed(Worker w) {
finalReentrantLockmainLock=this.mainLock;
mainLock.lock();
try {
if (w != null)
workers.remove(w);
decrementWorkerCount();
tryTerminate();
} finally {
mainLock.unlock();
}
}
/**
* Performs cleanup and bookkeeping for a dying worker. Called
* only from worker threads. Unless completedAbruptly is set,
* assumes that workerCount has already been adjusted to account
* for exit. This method removes thread from worker set, and
* possibly terminates the pool or replaces the worker if either
* it exited due to user task exception or if fewer than
* corePoolSize workers are running or queue is non-empty but
* there are no workers.
*
* @param w the worker
* @param completedAbruptly if the worker died due to user exception
*/privatevoidprocessWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
finalReentrantLockmainLock=this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
intc= ctl.get();
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
intmin= allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
addWorker(null, false);
}
}
/**
* Initiates an orderly shutdown in which previously submitted
* tasks are executed, but no new tasks will be accepted.
* Invocation has no additional effect if already shut down.
*
* <p>This method does not wait for previously submitted tasks to
* complete execution. Use {@link #awaitTermination awaitTermination}
* to do that.
*
* @throws SecurityException {@inheritDoc}
*/publicvoidshutdown() {
finalReentrantLockmainLock=this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
/**
* Attempts to stop all actively executing tasks, halts the
* processing of waiting tasks, and returns a list of the tasks
* that were awaiting execution. These tasks are drained (removed)
* from the task queue upon return from this method.
*
* <p>This method does not wait for actively executing tasks to
* terminate. Use {@link #awaitTermination awaitTermination} to
* do that.
*
* <p>There are no guarantees beyond best-effort attempts to stop
* processing actively executing tasks. This implementation
* cancels tasks via {@link Thread#interrupt}, so any task that
* fails to respond to interrupts may never terminate.
*
* @throws SecurityException {@inheritDoc}
*/public List<Runnable> shutdownNow() {
List<Runnable> tasks;
finalReentrantLockmainLock=this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);
interruptWorkers();
tasks = drainQueue();
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
}
/* Predefined RejectedExecutionHandlers *//**
* A handler for rejected tasks that runs the rejected task
* directly in the calling thread of the {@code execute} method,
* unless the executor has been shut down, in which case the task
* is discarded.
*/publicstaticclassCallerRunsPolicyimplementsRejectedExecutionHandler {
/**
* Creates a {@code CallerRunsPolicy}.
*/publicCallerRunsPolicy() { }
/**
* Executes task r in the caller's thread, unless the executor
* has been shut down, in which case the task is discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/publicvoidrejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
/**
* 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.
*/publicstaticclassDiscardOldestPolicyimplementsRejectedExecutionHandler {
/**
* Creates a {@code DiscardOldestPolicy} for the given executor.
*/publicDiscardOldestPolicy() { }
/**
* Obtains and ignores the next task that the executor
* would otherwise execute, if one is immediately available,
* and then retries execution of task r, unless the executor
* is shut down, in which case task r is instead discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/publicvoidrejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
}
/*
* Revision notes: This differs from previous versions of this
* class that relied on AbstractQueuedSynchronizer, mainly to
* avoid surprising users about retaining interrupt status during
* cancellation races. Sync control in the current design relies
* on a "state" field updated via CAS to track completion, along
* with a simple Treiber stack to hold waiting threads.
*
* Style note: As usual, we bypass overhead of using
* AtomicXFieldUpdaters and instead directly use Unsafe intrinsics.
*//**
* The run state of this task, initially NEW. The run state
* transitions to a terminal state only in methods set,
* setException, and cancel. During completion, state may take on
* transient values of COMPLETING (while outcome is being set) or
* INTERRUPTING (only while interrupting the runner to satisfy a
* cancel(true)). Transitions from these intermediate to final
* states use cheaper ordered/lazy writes because values are unique
* and cannot be further modified.
*
* Possible state transitions:
* NEW -> COMPLETING -> NORMAL
* NEW -> COMPLETING -> EXCEPTIONAL
* NEW -> CANCELLED
* NEW -> INTERRUPTING -> INTERRUPTED
*/privatevolatileint state;
privatestaticfinalintNEW=0;
privatestaticfinalintCOMPLETING=1;
privatestaticfinalintNORMAL=2;
privatestaticfinalintEXCEPTIONAL=3;
privatestaticfinalintCANCELLED=4;
privatestaticfinalintINTERRUPTING=5;
privatestaticfinalintINTERRUPTED=6;
/** The underlying callable; nulled out after running */
publicvoidrun() {
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
set(result);
}
} finally {
// runner must be non-null until state is settled to// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent// leaked interruptsints= state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
/**
* Sets the result of this future to the given value unless
* this future has already been set or has been cancelled.
*
* <p>This method is invoked internally by the {@link #run} method
* upon successful completion of the computation.
*
* @param v the value
*/protectedvoidset(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion();
}
}
/**
* Pops and tries to trigger all reachable dependents. Call only
* when known to be done.
*/finalvoidpostComplete() {
/*
* On each step, variable f holds current dependents to pop
* and run. It is extended along only one path at a time,
* pushing others to avoid unbounded recursion.
*/
CompletableFuture<?> f = this; Completion h;
while ((h = f.stack) != null ||
(f != this && (h = (f = this).stack) != null)) {
CompletableFuture<?> d; Completion t;
if (f.casStack(h, t = h.next)) {
if (t != null) {
if (f != this) {
pushStack(h);
continue;
}
h.next = null; // detach
}
f = (d = h.tryFire(NESTED)) == null ? this : d;
}
}
}
/* ------------- Control and status methods -------------- *//**
* If not already completed, completes this CompletableFuture with
* a {@link CancellationException}. Dependent CompletableFutures
* that have not already completed will also complete
* exceptionally, with a {@link CompletionException} caused by
* this {@code CancellationException}.
*
* @param mayInterruptIfRunning this value has no effect in this
* implementation because interrupts are not used to control
* processing.
*
* @return {@code true} if this task is now cancelled
*/publicbooleancancel(boolean mayInterruptIfRunning) {
booleancancelled= (result == null) &&
internalComplete(newAltResult(newCancellationException()));
postComplete();
return cancelled || isCancelled();
}
/**
* Sets the next time to run for a periodic task.
*/privatevoidsetNextRunTime() {
longp= period;
if (p > 0)
time += p;
else
time = triggerTime(-p);
}
/**
* Overrides FutureTask version so as to reset/requeue if periodic.
*/publicvoidrun() {
booleanperiodic= isPeriodic();
if (!canRunInCurrentRunState(periodic))
cancel(false);
elseif (!periodic)
ScheduledFutureTask.super.run();
elseif (ScheduledFutureTask.super.runAndReset()) {
setNextRunTime();
reExecutePeriodic(outerTask);
}
}
/**
* Main execution method for delayed or periodic tasks. If pool
* is shut down, rejects the task. Otherwise adds task to queue
* and starts a thread, if necessary, to run it. (We cannot
* prestart the thread to run the task because the task (probably)
* shouldn't be run yet.) If the pool is shut down while the task
* is being added, cancel and remove it if required by state and
* run-after-shutdown parameters.
*
* @param task the task
*/privatevoiddelayedExecute(RunnableScheduledFuture<?> task) {
if (isShutdown())
reject(task);
else {
super.getQueue().add(task);
if (isShutdown() &&
!canRunInCurrentRunState(task.isPeriodic()) &&
remove(task))
task.cancel(false);
else
ensurePrestart();
}
}
/**
* Causes this thread to begin execution; the Java Virtual Machine
* calls the <code>run</code> method of this thread.
* <p>
* The result is that two threads are running concurrently: the
* current thread (which returns from the call to the
* <code>start</code> method) and the other thread (which executes its
* <code>run</code> method).
* <p>
* It is never legal to start a thread more than once.
* In particular, a thread may not be restarted once it has completed
* execution.
*
* @exception IllegalThreadStateException if the thread was already
* started.
* @see #run()
* @see #stop()
*/publicsynchronizedvoidstart() {
/**
* This method is not invoked for the main method thread or "system"
* group threads created/set up by the VM. Any new functionality added
* to this method in the future may have to also be added to the VM.
*
* A zero status value corresponds to state "NEW".
*/if (threadStatus != 0)
thrownewIllegalThreadStateException();
/* Notify the group that this thread is about to be started
* so that it can be added to the group's list of threads
* and the group's unstarted count can be decremented. */
group.add(this);
booleanstarted=false;
try {
start0();
started = true;
} finally {
try {
if (!started) {
group.threadStartFailed(this);
}
} catch (Throwable ignore) {
/* do nothing. If start0 threw a Throwable then
it will be passed up the call stack */
}
}
}
/**
* If this thread was constructed using a separate
* <code>Runnable</code> run object, then that
* <code>Runnable</code> object's <code>run</code> method is called;
* otherwise, this method does nothing and returns.
* <p>
* Subclasses of <code>Thread</code> should override this method.
*
* @see #start()
* @see #stop()
* @see #Thread(ThreadGroup, Runnable, String)
*/@Overridepublicvoidrun() {
if (target != null) {
target.run();
}
}
/**
* Interrupts this thread.
*
* <p> Unless the current thread is interrupting itself, which is
* always permitted, the {@link #checkAccess() checkAccess} method
* of this thread is invoked, which may cause a {@link
* SecurityException} to be thrown.
*
* <p> If this thread is blocked in an invocation of the {@link
* Object#wait() wait()}, {@link Object#wait(long) wait(long)}, or {@link
* Object#wait(long, int) wait(long, int)} methods of the {@link Object}
* class, or of the {@link #join()}, {@link #join(long)}, {@link
* #join(long, int)}, {@link #sleep(long)}, or {@link #sleep(long, int)},
* methods of this class, then its interrupt status will be cleared and it
* will receive an {@link InterruptedException}.
*
* <p> If this thread is blocked in an I/O operation upon an {@link
* java.nio.channels.InterruptibleChannel InterruptibleChannel}
* then the channel will be closed, the thread's interrupt
* status will be set, and the thread will receive a {@link
* java.nio.channels.ClosedByInterruptException}.
*
* <p> If this thread is blocked in a {@link java.nio.channels.Selector}
* then the thread's interrupt status will be set and it will return
* immediately from the selection operation, possibly with a non-zero
* value, just as if the selector's {@link
* java.nio.channels.Selector#wakeup wakeup} method were invoked.
*
* <p> If none of the previous conditions hold then this thread's interrupt
* status will be set. </p>
*
* <p> Interrupting a thread that is not alive need not have any effect.
*
* @throws SecurityException
* if the current thread cannot modify this thread
*
* @revised 6.0
* @spec JSR-51
*/publicvoidinterrupt() {
if (this != Thread.currentThread())
checkAccess();
synchronized (blockerLock) {
Interruptibleb= blocker;
if (b != null) {
interrupt0(); // Just to set the interrupt flag
b.interrupt(this);
return;
}
}
interrupt0();
}
/**
* Tests whether the current thread has been interrupted. The
* <i>interrupted status</i> of the thread is cleared by this method. In
* other words, if this method were to be called twice in succession, the
* second call would return false (unless the current thread were
* interrupted again, after the first call had cleared its interrupted
* status and before the second call had examined it).
*
* <p>A thread interruption ignored because a thread was not alive
* at the time of the interrupt will be reflected by this method
* returning false.
*
* @return <code>true</code> if the current thread has been interrupted;
* <code>false</code> otherwise.
* @see #isInterrupted()
* @revised 6.0
*/publicstaticbooleaninterrupted() {
return currentThread().isInterrupted(true);
}
/**
* Tests whether this thread has been interrupted. The <i>interrupted
* status</i> of the thread is unaffected by this method.
*
* <p>A thread interruption ignored because a thread was not alive
* at the time of the interrupt will be reflected by this method
* returning false.
*
* @return <code>true</code> if this thread has been interrupted;
* <code>false</code> otherwise.
* @see #interrupted()
* @revised 6.0
*/publicbooleanisInterrupted() {
return isInterrupted(false);
}
/**
* Clears this buffer. The position is set to zero, the limit is set to
* the capacity, and the mark is discarded.
*
* <p> Invoke this method before using a sequence of channel-read or
* <i>put</i> operations to fill this buffer. For example:
*
* <blockquote><pre>
* buf.clear(); // Prepare buffer for reading
* in.read(buf); // Read data</pre></blockquote>
*
* <p> This method does not actually erase the data in the buffer, but it
* is named as if it did because it will most often be used in situations
* in which that might as well be the case. </p>
*
* @return This buffer
*/publicfinal Buffer clear() {
position = 0;
limit = capacity;
mark = -1;
returnthis;
}
/**
* Flips this buffer. The limit is set to the current position and then
* the position is set to zero. If the mark is defined then it is
* discarded.
*
* <p> After a sequence of channel-read or <i>put</i> operations, invoke
* this method to prepare for a sequence of channel-write or relative
* <i>get</i> operations. For example:
*
* <blockquote><pre>
* buf.put(magic); // Prepend header
* in.read(buf); // Read data into rest of buffer
* buf.flip(); // Flip buffer
* out.write(buf); // Write header + data to channel</pre></blockquote>
*
* <p> This method is often used in conjunction with the {@link
* java.nio.ByteBuffer#compact compact} method when transferring data from
* one place to another. </p>
*
* @return This buffer
*/publicfinal Buffer flip() {
limit = position;
position = 0;
mark = -1;
returnthis;
}
/**
* Rewinds this buffer. The position is set to zero and the mark is
* discarded.
*
* <p> Invoke this method before a sequence of channel-write or <i>get</i>
* operations, assuming that the limit has already been set
* appropriately. For example:
*
* <blockquote><pre>
* out.write(buf); // Write remaining data
* buf.rewind(); // Rewind buffer
* buf.get(array); // Copy data into array</pre></blockquote>
*
* @return This buffer
*/publicfinal Buffer rewind() {
position = 0;
mark = -1;
returnthis;
}
privatestaticbooleantryReserveMemory(long size, int cap) {
// -XX:MaxDirectMemorySize limits the total capacity rather than the// actual memory usage, which will differ when buffers are page// aligned.long totalCap;
while (cap <= maxMemory - (totalCap = totalCapacity.get())) {
if (totalCapacity.compareAndSet(totalCap, totalCap + cap)) {
reservedMemory.addAndGet(size);
count.incrementAndGet();
returntrue;
}
}
returnfalse;
}
// These methods should be called whenever direct memory is allocated or// freed. They allow the user to control the amount of direct memory// which a process may access. All sizes are specified in bytes.staticvoidreserveMemory(long size, int cap) {
if (!memoryLimitSet && VM.isBooted()) {
maxMemory = VM.maxDirectMemory();
memoryLimitSet = true;
}
// optimist!if (tryReserveMemory(size, cap)) {
return;
}
finalJavaLangRefAccessjlra= SharedSecrets.getJavaLangRefAccess();
// retry while helping enqueue pending Reference objects// which includes executing pending Cleaner(s) which includes// Cleaner(s) that free direct buffer memorywhile (jlra.tryHandlePendingReference()) {
if (tryReserveMemory(size, cap)) {
return;
}
}
// trigger VM's Reference processing
System.gc();
// a retry loop with exponential back-off delays// (this gives VM some time to do it's job)booleaninterrupted=false;
try {
longsleepTime=1;
intsleeps=0;
while (true) {
if (tryReserveMemory(size, cap)) {
return;
}
if (sleeps >= MAX_SLEEPS) {
break;
}
if (!jlra.tryHandlePendingReference()) {
try {
Thread.sleep(sleepTime);
sleepTime <<= 1;
sleeps++;
} catch (InterruptedException e) {
interrupted = true;
}
}
}
// no luckthrownewOutOfMemoryError("Direct buffer memory");
} finally {
if (interrupted) {
// don't swallow interrupts
Thread.currentThread().interrupt();
}
}
}
// Terminal evaluation methods/**
* Evaluate the pipeline with a terminal operation to produce a result.
*
* @param <R> the type of result
* @param terminalOp the terminal operation to be applied to the pipeline.
* @return the result
*/final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
assertgetOutputShape() == terminalOp.inputShape();
if (linkedOrConsumed)
thrownewIllegalStateException(MSG_STREAM_LINKED);
linkedOrConsumed = true;
return isParallel()
? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
: terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
}
/**
* Returns an instance of a proxy class for the specified interfaces
* that dispatches method invocations to the specified invocation
* handler.
*
* <p>{@code Proxy.newProxyInstance} throws
* {@code IllegalArgumentException} for the same reasons that
* {@code Proxy.getProxyClass} does.
*
* @param loader the class loader to define the proxy class
* @param interfaces the list of interfaces for the proxy class
* to implement
* @param h the invocation handler to dispatch method invocations to
* @return a proxy instance with the specified invocation handler of a
* proxy class that is defined by the specified class loader
* and that implements the specified interfaces
* @throws IllegalArgumentException if any of the restrictions on the
* parameters that may be passed to {@code getProxyClass}
* are violated
* @throws SecurityException if a security manager, <em>s</em>, is present
* and any of the following conditions is met:
* <ul>
* <li> the given {@code loader} is {@code null} and
* the caller's class loader is not {@code null} and the
* invocation of {@link SecurityManager#checkPermission
* s.checkPermission} with
* {@code RuntimePermission("getClassLoader")} permission
* denies access;</li>
* <li> for each proxy interface, {@code intf},
* the caller's class loader is not the same as or an
* ancestor of the class loader for {@code intf} and
* invocation of {@link SecurityManager#checkPackageAccess
* s.checkPackageAccess()} denies access to {@code intf};</li>
* <li> any of the given proxy interfaces is non-public and the
* caller class is not in the same {@linkplain Package runtime package}
* as the non-public interface and the invocation of
* {@link SecurityManager#checkPermission s.checkPermission} with
* {@code ReflectPermission("newProxyInPackage.{package name}")}
* permission denies access.</li>
* </ul>
* @throws NullPointerException if the {@code interfaces} array
* argument or any of its elements are {@code null}, or
* if the invocation handler, {@code h}, is
* {@code null}
*/@CallerSensitivepublicstatic Object newProxyInstance(ClassLoader loader,
Class<?>[] interfaces,
InvocationHandler h)throws IllegalArgumentException
{
Objects.requireNonNull(h);
final Class<?>[] intfs = interfaces.clone();
finalSecurityManagersm= System.getSecurityManager();
if (sm != null) {
checkProxyAccess(Reflection.getCallerClass(), loader, intfs);
}
/*
* Look up or generate the designated proxy class.
*/
Class<?> cl = getProxyClass0(loader, intfs);
/*
* Invoke its constructor with the designated invocation handler.
*/try {
if (sm != null) {
checkNewProxyPermission(Reflection.getCallerClass(), cl);
}
final Constructor<?> cons = cl.getConstructor(constructorParams);
finalInvocationHandlerih= h;
if (!Modifier.isPublic(cl.getModifiers())) {
AccessController.doPrivileged(newPrivilegedAction<Void>() {
public Void run() {
cons.setAccessible(true);
returnnull;
}
/**
* A factory function that generates, defines and returns the proxy class given
* the ClassLoader and array of interfaces.
*/privatestaticfinalclassProxyClassFactoryimplementsBiFunction<ClassLoader, Class<?>[], Class<?>>
{
// prefix for all proxy class namesprivatestaticfinalStringproxyClassNamePrefix="$Proxy";
// next number to use for generation of unique proxy class namesprivatestaticfinalAtomicLongnextUniqueNumber=newAtomicLong();
@Overridepublic Class<?> apply(ClassLoader loader, Class<?>[] interfaces) {
Map<Class<?>, Boolean> interfaceSet = newIdentityHashMap<>(interfaces.length);
for (Class<?> intf : interfaces) {
/*
* Verify that the class loader resolves the name of this
* interface to the same Class object.
*/
Class<?> interfaceClass = null;
try {
interfaceClass = Class.forName(intf.getName(), false, loader);
} catch (ClassNotFoundException e) {
}
if (interfaceClass != intf) {
thrownewIllegalArgumentException(
intf + " is not visible from class loader: " + loader);
}
/*
* Verify that the Class object actually represents an
* interface.
*/if (!interfaceClass.isInterface()) {
thrownewIllegalArgumentException(
interfaceClass.getName() + " is not an interface");
}
/*
* Verify that this interface is not a duplicate.
*/if (interfaceSet.put(interfaceClass, Boolean.TRUE) != null) {
thrownewIllegalArgumentException(
"repeated interface: " + interfaceClass.getName());
}
}
StringproxyPkg=null; // package to define proxy class inintaccessFlags= Modifier.PUBLIC | Modifier.FINAL;
/*
* Record the package of a non-public proxy interface so that the
* proxy class will be defined in the same package. Verify that
* all non-public proxy interfaces are in the same package.
*/for (Class<?> intf : interfaces) {
intflags= intf.getModifiers();
if (!Modifier.isPublic(flags)) {
accessFlags = Modifier.FINAL;
Stringname= intf.getName();
intn= name.lastIndexOf('.');
Stringpkg= ((n == -1) ? "" : name.substring(0, n + 1));
if (proxyPkg == null) {
proxyPkg = pkg;
} elseif (!pkg.equals(proxyPkg)) {
thrownewIllegalArgumentException(
"non-public interfaces from different packages");
}
}
}
if (proxyPkg == null) {
// if no non-public proxy interfaces, use com.sun.proxy package
proxyPkg = ReflectUtil.PROXY_PACKAGE + ".";
}
/*
* Choose a name for the proxy class to generate.
*/longnum= nextUniqueNumber.getAndIncrement();
StringproxyName= proxyPkg + proxyClassNamePrefix + num;
/*
* Generate the specified proxy class.
*/byte[] proxyClassFile = ProxyGenerator.generateProxyClass(
proxyName, interfaces, accessFlags);
try {
return defineClass0(loader, proxyName,
proxyClassFile, 0, proxyClassFile.length);
} catch (ClassFormatError e) {
/*
* A ClassFormatError here means that (barring bugs in the
* proxy class generation code) there was some other
* invalid aspect of the arguments supplied to the proxy
* class creation (such as virtual machine limitations
* exceeded).
*/thrownewIllegalArgumentException(e.toString());
}
}
/**
* Loads the class with the specified <a href="#name">binary name</a>. The
* default implementation of this method searches for classes in the
* following order:
*
* <ol>
*
* <li><p> Invoke {@link #findLoadedClass(String)} to check if the class
* has already been loaded. </p></li>
*
* <li><p> Invoke the {@link #loadClass(String) <tt>loadClass</tt>} method
* on the parent class loader. If the parent is <tt>null</tt> the class
* loader built-in to the virtual machine is used, instead. </p></li>
*
* <li><p> Invoke the {@link #findClass(String)} method to find the
* class. </p></li>
*
* </ol>
*
* <p> If the class was found using the above steps, and the
* <tt>resolve</tt> flag is true, this method will then invoke the {@link
* #resolveClass(Class)} method on the resulting <tt>Class</tt> object.
*
* <p> Subclasses of <tt>ClassLoader</tt> are encouraged to override {@link
* #findClass(String)}, rather than this method. </p>
*
* <p> Unless overridden, this method synchronizes on the result of
* {@link #getClassLoadingLock <tt>getClassLoadingLock</tt>} method
* during the entire class loading process.
*
* @param name
* The <a href="#name">binary name</a> of the class
*
* @param resolve
* If <tt>true</tt> then resolve the class
*
* @return The resulting <tt>Class</tt> object
*
* @throws ClassNotFoundException
* If the class could not be found
*/protected Class<?> loadClass(String name, boolean resolve)
throws ClassNotFoundException
{
synchronized (getClassLoadingLock(name)) {
// First, check if the class has already been loaded
Class<?> c = findLoadedClass(name);
if (c == null) {
longt0= System.nanoTime();
try {
if (parent != null) {
c = parent.loadClass(name, false);
} else {
c = findBootstrapClassOrNull(name);
}
} catch (ClassNotFoundException e) {
// ClassNotFoundException thrown if class not found// from the non-null parent class loader
}
if (c == null) {
// If still not found, then invoke findClass in order// to find the class.longt1= System.nanoTime();
c = findClass(name);
// this is the defining class loader; record the stats
sun.misc.PerfCounter.getParentDelegationTime().addTime(t1 - t0);
sun.misc.PerfCounter.getFindClassTime().addElapsedTimeFrom(t1);
sun.misc.PerfCounter.getFindClasses().increment();
}
}
if (resolve) {
resolveClass(c);
}
return c;
}
}
/**
* Returns the lock object for class loading operations.
* For backward compatibility, the default implementation of this method
* behaves as follows. If this ClassLoader object is registered as
* parallel capable, the method returns a dedicated object associated
* with the specified class name. Otherwise, the method returns this
* ClassLoader object.
*
* @param className
* The name of the to-be-loaded class
*
* @return the lock for class loading operations
*
* @throws NullPointerException
* If registered as parallel capable and <tt>className</tt> is null
*
* @see #loadClass(String, boolean)
*
* @since 1.7
*/protected Object getClassLoadingLock(String className) {
Objectlock=this;
if (parallelLockMap != null) {
ObjectnewLock=newObject();
lock = parallelLockMap.putIfAbsent(className, newLock);
if (lock == null) {
lock = newLock;
}
}
return lock;
}
/**
* Creates a thread pool that reuses a fixed number of threads
* operating off a shared unbounded queue. At any point, at most
* {@code nThreads} threads will be active processing tasks.
* If additional tasks are submitted when all threads are active,
* they will wait in the queue until a thread is available.
* If any thread terminates due to a failure during execution
* prior to shutdown, a new one will take its place if needed to
* execute subsequent tasks. The threads in the pool will exist
* until it is explicitly {@link ExecutorService#shutdown shutdown}.
*
* @param nThreads the number of threads in the pool
* @return the newly created thread pool
* @throws IllegalArgumentException if {@code nThreads <= 0}
*/publicstatic ExecutorService newFixedThreadPool(int nThreads) {
returnnewThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
newLinkedBlockingQueue<Runnable>());
}
/**
* Creates a thread pool that creates new threads as needed, but
* will reuse previously constructed threads when they are
* available. These pools will typically improve the performance
* of programs that execute many short-lived asynchronous tasks.
* Calls to {@code execute} will reuse previously constructed
* threads if available. If no existing thread is available, a new
* thread will be created and added to the pool. Threads that have
* not been used for sixty seconds are terminated and removed from
* the cache. Thus, a pool that remains idle for long enough will
* not consume any resources. Note that pools with similar
* properties but different details (for example, timeout parameters)
* may be created using {@link ThreadPoolExecutor} constructors.
*
* @return the newly created thread pool
*/publicstatic ExecutorService newCachedThreadPool() {
returnnewThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
newSynchronousQueue<Runnable>());
}