Jdk1.8 JUC源码增量解析(1)-atomic-Striped64

2024年1月10日 318次阅读来源: JUC

转载自 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

    原文作者：JUC
    原文地址: https://blog.csdn.net/moakun/article/details/80590893
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。