转载自 Jdk1.8 JUC源码增量解析(1)-atomic-Striped64
功能简介:
- Striped64是jdk1.8提供的用于支持如Long累加器,Double累加器这样机制的基础类。
- Striped64的设计核心思路就是通过内部的分散计算来避免竞争(比如多线程CAS操作时的竞争)。
- Striped64内部包含一个基础值和一个单元哈希表。没有竞争的情况下,要累加的数会累加到这个基础值上;如果有竞争的话,会将要累加的数累加到单元哈希表中的某个单元里面。所以整个Striped64的值包括基础值和单元哈希表中所有单元的值的总和。
源码分析:
- 先看一下内部结构:
- /**
- * 存放Cell的hash表,大小为2的幂。
- */
- transient volatile Cell[] cells;
- /**
- * 基础值,没有竞争时会使用(更新)这个值,同时做为初始化竞争失败的回退方案。
- * 原子更新。
- */
- transient volatile long base;
- /**
- * 自旋锁,通过CAS操作加锁,用于保护创建或者扩展Cell表。
- */
- transient volatile int cellsBusy;
看下Cell的内部结构:
- @sun.misc.Contended
- static final class Cell {
- volatile long value;
- Cell(long x) { value = x; }
- final boolean cas(long cmp, long val) {
- return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
- }
- // Unsafe mechanics
- private static final sun.misc.Unsafe UNSAFE;
- private static final long valueOffset;
- static {
- try {
- UNSAFE = sun.misc.Unsafe.getUnsafe();
- Class<?> ak = Cell.class;
- valueOffset = UNSAFE.objectFieldOffset
- (ak.getDeclaredField(“value”));
- } catch (Exception e) {
- throw new Error(e);
- }
- }
- }
Cell内部保存了一个volatile修饰的long型域,同时提供了原子操作,看起来像一个原子量。
注意到Cell类被一个Contended注解修饰,Contended的作用是对Cell做缓存行填充,避免伪共享。
- Striped64主要提供了longAccumulate和doubleAccumulate方法来支持子类,先看下longAccumulate:
- static final int NCPU = Runtime.getRuntime().availableProcessors();
- final void longAccumulate(long x, LongBinaryOperator fn,
- boolean wasUncontended) {
- int h;
- //获取当前线程的probe值作为hash值。
- if ((h = getProbe()) == 0) {
- //如果probe值为0,强制初始化当前线程的probe值,这次初始化的probe值不会为0。
- ThreadLocalRandom.current();
- //再次获取probe值作为hash值。
- h = getProbe();
- //这次相当于再次计算了hash,所以设置未竞争标记为true。
- wasUncontended = true;
- }
- boolean collide = false;
- for (;;) {
- Cell[] as; Cell a; int n; long v;
- if ((as = cells) != null && (n = as.length) > 0) {
- //通过h从cell表中选定一个cell位置。
- if ((a = as[(n – 1) & h]) == null) {
- //如果当前位置没有cell,尝试新建一个。
- if (cellsBusy == 0) {
- //创建一个Cell。
- Cell r = new Cell(x);
- //尝试或者cellsBusy锁。
- if (cellsBusy == 0 && casCellsBusy()) {
- boolean created = false;
- try {
- Cell[] rs; int m, j;
- //在获取锁的情况下再次检测一下。
- if ((rs = cells) != null &&
- (m = rs.length) > 0 &&
- rs[j = (m – 1) & h] == null) {
- //设置新建的cell到指定位置。
- rs[j] = r;
- //创建标记设置为true。
- created = true;
- }
- } finally {
- //释放cellsBusy锁。
- cellsBusy = 0;
- }
- if (created)
- //如果创建成功,直接跳出循环,退出方法。
- break;
- //说明上面指定的cell的位置上有cell了,继续尝试。
- continue;
- }
- }
- //走到这里说明获取cellsBusy锁失败。
- collide = false;
- }
- //以下条件说明上面通过h选定的cell表的位置上有Cell,就是a。
- else if (!wasUncontended) // CAS already known to fail
- //如果之前的CAS失败,说明已经发生竞争,
- //这里会设置未竞争标志位true,然后再次算一个probe值,然后重试。
- wasUncontended = true; // Continue after rehash
- //这里尝试将x值加到a的value上。
- else if (a.cas(v = a.value, ((fn == null) ? v + x :
- fn.applyAsLong(v, x))))
- //如果尝试成功,跳出循环,方法退出。
- break;
- else if (n >= NCPU || cells != as)
- //如果cell表的size已经最大,或者cell表已经发生变化(as是一个过时的)。
- collide = false;
- else if (!collide)
- //设置冲突标志,表示发生了冲突,重试。
- collide = true;
- //尝试获取cellsBusy锁。
- else if (cellsBusy == 0 && casCellsBusy()) {
- try {
- //检测as是否过时。
- if (cells == as) {
- //给cell表扩容。
- Cell[] rs = new Cell[n << 1];
- for (int i = 0; i < n; ++i)
- rs[i] = as[i];
- cells = rs;
- }
- } finally {
- //释放cellsBusy锁。
- cellsBusy = 0;
- }
- collide = false;
- //扩容cell表后,再次重试。
- continue;
- }
- //算出下一个hash值。
- h = advanceProbe(h);
- }
- //如果cell表还未创建,先尝试获取cellsBusy锁。
- else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
- boolean init = false;
- try {
- if (cells == as) {
- //初始化cell表,初始容量为2。
- Cell[] rs = new Cell[2];
- rs[h & 1] = new Cell(x);
- cells = rs;
- init = true;
- }
- } finally {
- //释放cellsBusy锁。
- cellsBusy = 0;
- }
- if (init)
- //初始化cell表成功后,退出方法。
- break;
- }
- //如果创建cell表由于竞争导致失败,尝试将x累加到base上。
- else if (casBase(v = base, ((fn == null) ? v + x :
- fn.applyAsLong(v, x))))
- break;
- }
- }
说明一下这个方法,方法的作用是将给定的值x累加到当前值(Striped64本身)上,x值为正就是加、为负就是减。
方法流程细节:
首先,方法内部首先会算一个hash值,用来确定cell数组的下标。hash值初始源于当前Thread中的threadLocalRandomProbe域,如果hash值初始后为0,会初始化一下当前线程的threadLocalRandomProbe值,然后再次赋给hash值。注意方法传入第三个参数wasUncontended表示调用方法之前是否未发生竞争,加入前面走了初始化threadLocalRandomProbe的过程,就会将wasUncontended设置为true。
接下来,方法进入主循环。
1.先判断Cell表是否创建。
1.1.如果Cell表未创建,尝试获取cellsBusy锁。
1.1.1.如果获取cellsBusy锁成功,会创建一个size为2的Cell表作为初始cell表,然后新建一个保存给定x的Cell实例,然后根据hash值设置到Cell表对应的位置上;
1.1.2.如果获取cellsBusy锁失败,会尝试将x累加到base上,失败重试。
1.2.如果Cell表已经创建,通过hash值算出一个Cell表中的位置,然后获取这个位置上的Cell,称为a。
1.2.1.如果a为null,尝试获取cellsBusy锁。
1.2.1.1.如果获取cellsBusy成功,创建一个新的Cell,然后赋值给a,方法退出。(过程中需要多次检测冲突)
1.2.1.2.如果获取cellsBusy失败,会将collide设置为false(实际上是表示发生了冲突),然后重试。
1.2.2.如果a不为null。
1.2.2.1.如果wasUncontended为false,说明之前发生过CAS竞争失败,设置wasUncontended为true,重新计算hash值,重试;如果wasUncontended为true,继续尝试下面过程。
1.2.2.2.尝试通过CAS方式将x累加到a的value上,如果尝试成功,方法退出;如果尝试失败,继续尝试下面过程。
1.2.2.3.如果当前Cell表的大小以及达到最大值(当前处理器核数),或者Cell表发生了变化(竞争导致过时),那么会设置collide为false,重新计算hash值,然后重试;否则,继续尝试下面过程。
1.2.2.4.如果collide为false,说明之前发生过冲突,将collide设置为true,重新计算hash值,然后重试;否则,继续尝试下面过程。
1.2.2.5.尝试获取cellsBusy,如果成功,扩展Cell表,并将collide设置为false,然后重试;否则,重新计算hash值,然后重试;
看下longAccumulate中使用到的一些方法:
- final boolean casBase(long cmp, long val) {
- return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);
- }
- final boolean casCellsBusy() {
- return UNSAFE.compareAndSwapInt(this, CELLSBUSY, 0, 1);
- }
- static final int getProbe() {
- return UNSAFE.getInt(Thread.currentThread(), PROBE);
- }
- //计算下一个随机值作为hash值,使用xorshift算法。
- static final int advanceProbe(int probe) {
- probe ^= probe << 13;
- probe ^= probe >>> 17;
- probe ^= probe << 5;
- //设置到当前线程的threadLocalRandomProbe域。
- UNSAFE.putInt(Thread.currentThread(), PROBE, probe);
- return probe;
- }
- 再看下doubleAccumulate:
- final void doubleAccumulate(double x, DoubleBinaryOperator fn,
- boolean wasUncontended) {
- int h;
- if ((h = getProbe()) == 0) {
- ThreadLocalRandom.current(); // force initialization
- h = getProbe();
- wasUncontended = true;
- }
- boolean collide = false; // True if last slot nonempty
- for (;;) {
- Cell[] as; Cell a; int n; long v;
- if ((as = cells) != null && (n = as.length) > 0) {
- if ((a = as[(n – 1) & h]) == null) {
- if (cellsBusy == 0) { // Try to attach new Cell
- Cell r = new Cell(Double.doubleToRawLongBits(x));
- if (cellsBusy == 0 && casCellsBusy()) {
- boolean created = false;
- try { // Recheck under lock
- Cell[] rs; int m, j;
- if ((rs = cells) != null &&
- (m = rs.length) > 0 &&
- rs[j = (m – 1) & h] == null) {
- rs[j] = r;
- created = true;
- }
- } finally {
- cellsBusy = 0;
- }
- if (created)
- break;
- continue; // Slot is now non-empty
- }
- }
- collide = false;
- }
- else if (!wasUncontended) // CAS already known to fail
- wasUncontended = true; // Continue after rehash
- else if (a.cas(v = a.value,
- ((fn == null) ?
- Double.doubleToRawLongBits
- (Double.longBitsToDouble(v) + x) :
- Double.doubleToRawLongBits
- (fn.applyAsDouble
- (Double.longBitsToDouble(v), x)))))
- break;
- else if (n >= NCPU || cells != as)
- collide = false; // At max size or stale
- else if (!collide)
- collide = true;
- else if (cellsBusy == 0 && casCellsBusy()) {
- try {
- if (cells == as) { // Expand table unless stale
- Cell[] rs = new Cell[n << 1];
- for (int i = 0; i < n; ++i)
- rs[i] = as[i];
- cells = rs;
- }
- } finally {
- cellsBusy = 0;
- }
- collide = false;
- continue; // Retry with expanded table
- }
- h = advanceProbe(h);
- }
- else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
- boolean init = false;
- try { // Initialize table
- if (cells == as) {
- Cell[] rs = new Cell[2];
- rs[h & 1] = new Cell(Double.doubleToRawLongBits(x));
- cells = rs;
- init = true;
- }
- } finally {
- cellsBusy = 0;
- }
- if (init)
- break;
- }
- else if (casBase(v = base,
- ((fn == null) ?
- Double.doubleToRawLongBits
- (Double.longBitsToDouble(v) + x) :
- Double.doubleToRawLongBits
- (fn.applyAsDouble
- (Double.longBitsToDouble(v), x)))))
- break; // Fall back on using base
- }
- }
doubleAccumulate方法是针对double值做累加的,逻辑和longAccumulate一致。但由于Cell内部用long保存数据,所以在累加的时候会利用Double的doubleToRawLongBits和longBitsToDouble方法做double和longBits形式的double之间的转换。 Striped64的代码解析完毕! 参考资料:
http://mail.openjdk.java.net/pipermail/hotspot-dev/2012-November/007309.html