This class may be applicable when many threads frequently * update a common sum that is used for purposes such as collecting * statistics. In this case, performance may be significantly faster * than using a shared {@link AtomicLong}, at the expense of using - * significantly more space. On the other hand, if it is known that - * only one thread can ever update the sum, performance may be - * significantly slower than just updating a local variable. + * much more space. On the other hand, if it is known that only one + * thread can ever update the sum, performance may be significantly + * slower than just updating a local variable. + * + *
A StripedAdder may optionally be constructed with a given + * expected contention level; i.e., the number of threads that are + * expected to concurrently update the sum. Supplying an accurate + * value may improve performance by reducing the need for dynamic + * adjustment. * * @author Doug Lea */ @@ -37,19 +43,41 @@ public class StripedAdder implements Ser private static final long serialVersionUID = 7249069246863182397L; /* - * Overview: We maintain a table of AtomicLongs (padded to reduce - * false sharing). The table is indexed by per-thread hash codes - * that are initialized as random values. The table doubles in - * size upon contention (as indicated by failed CASes when - * performing add()), but is capped at the nearest power of two >= - * #cpus: At that point, contention should be infrequent if each - * thread has a unique index; so we instead adjust hash codes to - * new random values upon contention rather than expanding. A - * single spinlock is used for resizing the table as well as + * A StripedAdder maintains a table of Atomic long variables. The + * table is indexed by per-thread hash codes that are initialized + * to random values. + * + * The table doubles in size upon contention (as indicated by + * failed CASes when performing add()), but is capped at the + * nearest power of two >= #CPUS. This reflects the idea that, + * when there are more threads than CPUs, then if each thread were + * bound to a CPU, there would exist a perfect hash function + * mapping threads to slots that eliminates collisions. When we + * reach capacity, we search for this mapping by randomly varying + * the hash codes of colliding threads. Because search is random, + * and failures only become known via CAS failures, convergence + * will be slow, and because threads are typically not bound to + * CPUS forever, may not occur at all. However, despite these + * limitations, observed contention is typically low in these + * cases. + * + * Table entries are of class Adder; a form of AtomicLong padded + * to reduce cache contention on most processors. Padding is + * overkill for most Atomics because they are most often + * irregularly scattered in memory and thus don't interfere much + * with each other. But Atomic objects residing in arrays will + * tend to be placed adjacent to each other, and so will most + * often share cache lines without this precaution. Adders are + * constructed upon first use, which further improves per-thread + * locality and helps reduce (an already large) footprint. + * + * A single spinlock is used for resizing the table as well as * populating slots with new Adders. Upon lock contention, threads - * just try other slots rather than blocking. We guarantee that at + * try other slots rather than blocking. After initialization, at * least one slot exists, so retries will eventually find a - * candidate Adder. + * candidate Adder. During these retries, there is increased + * contention and reduced locality, which is still better than + * alternatives. */ /** @@ -58,8 +86,7 @@ public class StripedAdder implements Ser static final int NCPU = Runtime.getRuntime().availableProcessors(); /** - * Version of AtomicLong padded to avoid sharing cache - * lines on most processors + * Padded version of AtomicLong */ static final class Adder extends AtomicLong { long p0, p1, p2, p3, p4, p5, p6, p7, p8, p9, pa, pb, pc, pd; @@ -67,7 +94,9 @@ public class StripedAdder implements Ser } /** - * Holder for the thread-local hash code. + * Holder for the thread-local hash code. The code starts off with + * a given random value, but may be set to a different value upon + * collisions in retryAdd. */ static final class HashCode { int code; @@ -93,7 +122,7 @@ public class StripedAdder implements Ser static final ThreadHashCode threadHashCode = new ThreadHashCode(); /** - * Table of adders. Initially of size 2; grows to be at most NCPU. + * Table of adders. Minimum size 2. Size grows to be at most NCPU. */ private transient volatile Adder[] adders; @@ -105,20 +134,28 @@ public class StripedAdder implements Ser private final AtomicInteger mutex; /** - * Marsaglia XorShift for rehashing on collisions + * Creates a new adder with zero sum. */ - private static int xorShift(int r) { - r ^= r << 13; - r ^= r >>> 17; - return r ^ (r << 5); + public StripedAdder() { + this.mutex = new AtomicInteger(); + // remaining initialization on first call to add. } /** - * Creates a new adder with initially zero sum. + * Creates a new adder with zero sum, and with stripes presized + * for the given expected contention level. + * + * @param expectedContention the expected number of threads that + * will concurrently update the sum. */ - public StripedAdder() { - Adder[] as = new Adder[2]; - as[0] = new Adder(0); // ensure at least one available adder + public StripedAdder(int expectedContention) { + int cap = (expectedContention < NCPU) ? expectedContention : NCPU; + int size = 2; + while (size < cap) + size <<= 1; + Adder[] as = new Adder[size]; + for (int i = 0; i < size; ++i) + as[i] = new Adder(0); this.adders = as; this.mutex = new AtomicInteger(); } @@ -129,43 +166,60 @@ public class StripedAdder implements Ser * @param x the value to add */ public void add(long x) { + Adder[] as; Adder a; int n; long v; // locals to hold volatile reads HashCode hc = threadHashCode.get(); - for (int h = hc.code;;) { - Adder[] as = adders; - int n = as.length; - Adder a = as[h & (n - 1)]; - if (a != null) { - long v = a.get(); - if (a.compareAndSet(v, v + x)) - break; - if (n >= NCPU) { // Collision when table at max - h = hc.code = xorShift(h); // change code - continue; + if ((as = adders) == null || (n = as.length) < 1 || + (a = as[hc.code & (n - 1)]) == null || + !a.compareAndSet(v = a.get(), v + x)) + retryAdd(x, hc); + } + + /** + * Handle cases of add involving initialization, resizing, + * creating new Adders, and/or contention. + */ + private void retryAdd(long x, HashCode hc) { + int h = hc.code; + final AtomicInteger mutex = this.mutex; + AtomicInteger lock = null; // nonnull when held + try { + for (;;) { + Adder[] as; Adder a; long v; int n, k; // locals for volatiles + boolean needLock = true; + if ((as = adders) == null || (n = as.length) < 1) { + if (lock != null) // default-initialize + adders = new Adder[2]; } - } - final AtomicInteger mutex = this.mutex; - if (mutex.get() != 0) - h = xorShift(h); // Try elsewhere - else if (mutex.compareAndSet(0, 1)) { - boolean created = false; - try { - Adder[] rs = adders; - if (a != null && rs == as) // Resize table - rs = adders = Arrays.copyOf(as, as.length << 1); - int j = h & (rs.length - 1); - if (rs[j] == null) { // Create adder - rs[j] = new Adder(x); - created = true; + else if ((a = as[k = h & (n - 1)]) == null) { + if (lock != null) { // attach new adder + as[k] = new Adder(x); + break; } - } finally { - mutex.set(0); } - if (created) { - hc.code = h; // Use this adder next time + else if (a.compareAndSet(v = a.get(), v + x)) break; + else if (n >= NCPU) // cannot expand + needLock = false; + else if (lock != null) // expand table + adders = Arrays.copyOf(as, n << 1); + + if (lock == null) { + if (needLock && mutex.get() == 0 && + mutex.compareAndSet(0, 1)) + lock = mutex; + else { // try elsewhere + h ^= h << 13; // Marsaglia XorShift + h ^= h >>> 17; + h ^= h << 5; + } } } + } finally { + if (lock != null) + lock.set(0); } + if (hc.code != h) // avoid unneeded writes + hc.code = h; } /** @@ -176,13 +230,15 @@ public class StripedAdder implements Ser * @return the estimated sum */ public long sum() { - long sum = 0; + long sum = 0L; Adder[] as = adders; - int n = as.length; - for (int i = 0; i < n; ++i) { - Adder a = as[i]; - if (a != null) - sum += a.get(); + if (as != null) { + int n = as.length; + for (int i = 0; i < n; ++i) { + Adder a = as[i]; + if (a != null) + sum += a.get(); + } } return sum; } @@ -194,11 +250,13 @@ public class StripedAdder implements Ser */ public void reset() { Adder[] as = adders; - int n = as.length; - for (int i = 0; i < n; ++i) { - Adder a = as[i]; - if (a != null) - a.set(0L); + if (as != null) { + int n = as.length; + for (int i = 0; i < n; ++i) { + Adder a = as[i]; + if (a != null) + a.set(0L); + } } } @@ -222,14 +280,16 @@ public class StripedAdder implements Ser * @return the estimated sum */ public long sumAndReset() { - long sum = 0; + long sum = 0L; Adder[] as = adders; - int n = as.length; - for (int i = 0; i < n; ++i) { - Adder a = as[i]; - if (a != null) { - sum += a.get(); - a.set(0L); + if (as != null) { + int n = as.length; + for (int i = 0; i < n; ++i) { + Adder a = as[i]; + if (a != null) { + sum += a.get(); + a.set(0L); + } } } return sum; @@ -244,13 +304,8 @@ public class StripedAdder implements Ser private void readObject(ObjectInputStream s) throws IOException, ClassNotFoundException { s.defaultReadObject(); - long c = s.readLong(); - Adder[] as = new Adder[2]; - as[0] = new Adder(c); - this.adders = as; mutex.set(0); + add(s.readLong()); } } - -