Java并发包同步工具之Exchanger

/**
* 单槽位交换函数，
* 正常情况下返回的是另一个线程交换的对象，在产生竞争或者在完成之前被中断返回null，超时返回TIMED_OUT对象。
*/
private final Object slotExchange(Object item, boolean timed, long ns) {
Node p = participant.get(); //得到一个初始Node
Thread t = Thread.currentThread();
if (t.isInterrupted()) // 如果发生了中断，返回null，isInterrupted不会复位中断中断，所以在exchange方法中可以继续使用Thread.interrupted()来检测其中断状态
return null; for (Node q;;) {
//该slot槽位不为空，表示已经被先来一步的等待交换数据的线程占据
if ((q = slot) != null) {
//交换数据之前，尝试将被占据的槽位重置为空
if (U.compareAndSwapObject(this, SLOT, q, null)) {
Object v = q.item; //拿到别人的数据
q.match = item; //将自己的数据给别人
Thread w = q.parked;
if (w != null)
U.unpark(w); //唤醒别人
return v; //拿到交换的数据了，返回
}
// create arena on contention, but continue until slot null
if (NCPU > 1 && bound == 0 &&
U.compareAndSwapInt(this, BOUND, 0, SEQ))
arena = new Node[(FULL + 2) << ASHIFT];
}
else if (arena != null) //单槽位为空，但是多槽位不为空表示发生了竞争，需要执行多槽位交换逻辑
return null; // caller must reroute to arenaExchange
else {
//单槽位还没被占据，那么带着自己想交换的数据尝试占据该槽位
p.item = item;
if (U.compareAndSwapObject(this, SLOT, null, p))
break; //成功占据该槽位，退出自旋。
p.item = null; //占据失败，清除数据继续自旋。
}
} //执行到这里，表示最先来交换的线程已经成功占据了单槽位，所以需要将自己阻塞（阻塞之前可以适当自旋），直到另一个交换线程来释放它。
// await release
int h = p.hash;
long end = timed ? System.nanoTime() + ns : 0L;
int spins = (NCPU > 1) ? SPINS : 1; //自旋次数，如果是单核CPU将不会自旋
Object v;
while ((v = p.match) == null) { //表示除非等到另一个交换线程将要交换的数据设置到match才跳出循环
if (spins > 0) {
h ^= h << 1; h ^= h >>> 3; h ^= h << 10;
if (h == 0)
h = SPINS | (int)t.getId();
else if (h < 0 && (--spins & ((SPINS >>> 1) - 1)) == 0)
Thread.yield(); //主动让出cpu
}
else if (slot != p) //其他线程来交换数据了（上面for循环里面交换数据之前会将槽位重置为空），但是match还没有被设置
spins = SPINS; //设置自旋等待的次数
else if (!t.isInterrupted() && arena == null &&
(!timed || (ns = end - System.nanoTime()) > 0L)) {
//自旋等待太久，迫不得已，在没被中断，没有竞争，且超时未到达时将自己阻塞起来吧。
U.putObject(t, BLOCKER, this);
p.parked = t;
if (slot == p)
U.park(false, ns); //阻塞
p.parked = null;
U.putObject(t, BLOCKER, null);
}
//执行到这里表示被中断或者超时了或者产生了竞争，所以要退出循环了，但是先要重置槽位slot
else if (U.compareAndSwapObject(this, SLOT, p, null)) {
v = timed && ns <= 0L && !t.isInterrupted() ? TIMED_OUT : null;
break;
}
}
//一些清理工作
U.putOrderedObject(p, MATCH, null);
p.item = null;
p.hash = h;
return v; //返回交换得到的数据（失败则为null）
}

private volatile Node[] arena; //表示多槽位的数组private volatile int bound; //表示当前槽位数组的界限//两个有效槽位之间的字节地址长度，1 << 7 至少是缓存行的大小
private static final int ASHIFT = 7;
//最大支持的数组索引，即255，二进制低八位8个1
private static final int MMASK = 0xff;
//bound的递增单元，每次递增都会同时递增低8位和高8位
private static final int SEQ = MMASK + 1;
//CPU个数，用于控制自旋和多槽位数组的大小
private static final int NCPU = Runtime.getRuntime().availableProcessors();
//实际支持的最大有效索引
static final int FULL = (NCPU >= (MMASK << 1)) ? MMASK : NCPU >>> 1;
//自旋次数，实际上单核CPU的时候没启用自旋
private static final int SPINS = 1 << 10;

if (NCPU > 1 && bound == 0 && U.compareAndSwapInt(this, BOUND, 0, SEQ))
arena = new Node[(FULL + 2) << ASHIFT];

U.compareAndSwapInt(this, BOUND, b, b + SEQ + 1) //扩张1
U.compareAndSwapInt(this, BOUND, b, b + SEQ - 1) //缩减1

Node q = (Node)U.getObjectVolatile(a, j = (i << ASHIFT) + ABASE); //i 从0 开始最多递增到 m，m = bound & MMASK，不就是128， 256， 384， 512吗

static {
try {
......//省略
s = U.arrayIndexScale(ak); //arean数组元素的增量地址长度
// ABASE absorbs padding in front of element 0。数组首元素偏移地址
ABASE = U.arrayBaseOffset(ak) + (1 << ASHIFT);
} catch (Exception e) {
throw new Error(e);
}
if ((s & (s-1)) != 0 || s > (1 << ASHIFT)) //不支持避免伪共享，抛异常
throw new Error("Unsupported array scale");

private final Object arenaExchange(Object item, boolean timed, long ns) {
Node[] a = arena;
Node p = participant.get(); //获取当前线程对于的Node对象
for (int i = p.index;;) { // 根据线程对于的index访问数组的有效槽位
int b, m, c; long j; // j 是包括填充在内的偏移
//获取偏移地址为(i << ASHIFT) + ABASE的内存值，就是实际有效的槽位
Node q = (Node)U.getObjectVolatile(a, j = (i << ASHIFT) + ABASE);
if (q != null && U.compareAndSwapObject(a, j, q, null)) {
//这里说明已经有线程在等着交换了，当前线程先重置该槽位，然后再和它交换
Object v = q.item; // 取出等待线程携带的数据
q.match = item; // 将当前线程携带的数据交给等待线程
Thread w = q.parked;
if (w != null)
U.unpark(w); // 唤醒等待线程
return v; // 交换成功，返回得到的数据
}
//否则，当前位置再实际有效索引之内，并且槽位还没被占据
//m = bound & MMASK, 就是bound低8位表示的当前最大有效索引
else if (i <= (m = (b = bound) & MMASK) && q == null) {
p.item = item; // 把准备交换的数据设置好，一旦占据槽位成功就可以等着被交换了
if (U.compareAndSwapObject(a, j, null, p)) { // 槽位占据成功
long end = (timed && m == 0) ? System.nanoTime() + ns : 0L; //计算超时时间
Thread t = Thread.currentThread(); // wait
for (int h = p.hash, spins = SPINS;;) {
Object v = p.match; //match就是来交换的线程携带的数据
if (v != null) { //已经有线程来交换了
U.putOrderedObject(p, MATCH, null);
p.item = null; // 清空，下次接着使用
p.hash = h;
return v; // 交换成功，返回得到的数据
}
else if (spins > 0) { //自旋 + yield
h ^= h << 1; h ^= h >>> 3; h ^= h << 10; // xorshift
if (h == 0) // initialize hash
h = SPINS | (int)t.getId();
else if (h < 0 && // approx 50% true
(--spins & ((SPINS >>> 1) - 1)) == 0)
Thread.yield(); // two yields per wait
}
else if (U.getObjectVolatile(a, j) != p)
spins = SPINS; //交换线程已经来了，正在准备数据，继续自旋，马上就可以交换成功
else if (!t.isInterrupted() && m == 0 &&
(!timed ||
(ns = end - System.nanoTime()) > 0L)) {
//等太久了，没有打断，没有超时，而且已经再0这个最左侧的有效索引位了
U.putObject(t, BLOCKER, this); // emulate LockSupport
p.parked = t; // minimize window
if (U.getObjectVolatile(a, j) == p)
U.park(false, ns); //只有阻塞等待唤醒了
p.parked = null;
U.putObject(t, BLOCKER, null);
}
else if (U.getObjectVolatile(a, j) == p &&
U.compareAndSwapObject(a, j, p, null)) {
//超时或打断或不是在最左侧0这个有效索引位
if (m != 0) // 不是在最左侧的0索引位，那么有效索引位减1，bound版本号会加1
U.compareAndSwapInt(this, BOUND, b, b + SEQ - 1);
p.item = null;
p.hash = h;
i = p.index >>>= 1; // 索引也减半，这样可以快速找到交换点
if (Thread.interrupted())
return null; //中断返回
if (timed && m == 0 && ns <= 0L)
return TIMED_OUT; //超时返回
break; // 跳出等待内循环，重新开始
}
}
}
else
p.item = null; //槽位占据失败，清理之后重试
}
else {
//到这里还是失败，那只有重新从右往左遍历，以期望能够发生交换了
if (p.bound != b) { //bound已经发生变化了，
p.bound = b; //更新自身的bound
p.collides = 0; //重置CAS失败次数
i = (i != m || m == 0) ? m : m - 1;
//这里的i，其实就是当i已经在最右边的最大有效索引了，就减一继续往左查找，否则重新从最右侧的最大索引开始往左查找
}
//CAS失败次数已经达到最大了，也就是在每一个有效索引位都失败了，并且当前最大有效索引位还没有到达FULL，就扩大有效索引位
else if ((c = p.collides) < m || m == FULL ||
!U.compareAndSwapInt(this, BOUND, b, b + SEQ + 1)) {
p.collides = c + 1; //本次失败了，CAS失败次数加1
i = (i == 0) ? m : i - 1; // 还是在左移i，i已经到最左侧了，重新从最右侧开始往左遍历
}
else
i = m + 1; //bound加1成功了，实际最大有效索引位扩张了，那么可以去占据新出炉的空的有效索引位
p.index = i;
}
}
}