Java CAS机制的底层实现探究

上次搞JVM源码挑了块硬骨头（见《通过HotSpot源码详解Java堆空间创建过程》），结果憋出内伤。当时放话说一个月不碰JVM，掐指一算，今天正好一个月了。不过这次得悠着点儿，选个容易的话题说说。

所谓CAS，即“比较与交换”（Compare-and-swap），是最常见的乐观锁实现方式，看官应该对这个概念很熟悉。一次CAS过程是原子的，包含3个操作数：

• 需要访问的内存地址V；
• 该内存地址中存储的预期值A；
• 希望向该地址写入的新值B。

当且仅当V中的值与A相同时，其值才会更新成B，否则就不执行任何动作。

在JDK中，CAS机制主要应用于原子类型包java.util.concurrent.atomic的各个类，如AtomicInteger类。下面只示出部分方法。

public class AtomicInteger extends Number implements java.io.Serializable {
    private static final long serialVersionUID = 6214790243416807050L;
    private static final Unsafe unsafe = Unsafe.getUnsafe();
    private static final long valueOffset;

    static {
        try {
            valueOffset = unsafe.objectFieldOffset
                (AtomicInteger.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }

    private volatile int value;

    // ......
    public final int getAndSet(int newValue) {
        return unsafe.getAndSetInt(this, valueOffset, newValue);
    }

    public final boolean compareAndSet(int expect, int update) {
        return unsafe.compareAndSwapInt(this, valueOffset, expect, update);

    public final int getAndIncrement() {
        return unsafe.getAndAddInt(this, valueOffset, 1);
    }

    public final int getAndDecrement() {
        return unsafe.getAndAddInt(this, valueOffset, -1);
    }

    public final int getAndAdd(int delta) {
        return unsafe.getAndAddInt(this, valueOffset, delta);
    }
    // ......
}

这些方法最终都调用了sun.misc.Unsafe类的方法（之前的简介见这里），它们的具体实现如下。

    public final native boolean compareAndSwapInt(Object o, long offset, int expected, int x);

    public final int getAndSetInt(Object o, long offset, int newValue) {
        int v;
        do {
            v = getIntVolatile(o, offset);
        } while (!compareAndSwapInt(o, offset, v, newValue));
        return v;
    }

    public final int getAndAddInt(Object o, long offset, int delta) {
        int v;
        do {
            v = getIntVolatile(o, offset);
        } while (!compareAndSwapInt(o, offset, v, v + delta));
        return v;
    }

可见，最基础的方法是compareAndSwapInt()，其余的getAndSetInt()、getAndAddInt()方法则是基于它的自旋操作。它是个native方法，所以接下来还是进入HotSpot的地盘。来到src/share/vm/prims/unsafe.cpp，有如下代码。

UNSAFE_ENTRY(jboolean, Unsafe_CompareAndSwapInt(JNIEnv *env, jobject unsafe, jobject obj, jlong offset, jint e, jint x))
  UnsafeWrapper("Unsafe_CompareAndSwapInt");
  oop p = JNIHandles::resolve(obj);
  jint* addr = (jint *) index_oop_from_field_offset_long(p, offset);
  return (jint)(Atomic::cmpxchg(x, addr, e)) == e;
UNSAFE_END

该方法的执行流程如下：

1. 调用JNIHandles::resolve()方法，获取AtomicInteger对象的OOP（普通对象指针），也就是该对象的基地址；
2. 调用index_oop_from_field_offset_long()方法，经由AtomicInteger对象的OOP及其value字段的偏移量，计算出存有值的实际内存地址。value字段的偏移量是从哪里来的呢？看一下本文开头AtomicInteger代码中的static代码块就能明白，是通过反射和Unsafe提供的方法取得的。
3. 调用Atomic::cmpxchg()方法，完成CAS操作。

下面来看Atomic::cmpxchg()方法。

jbyte Atomic::cmpxchg(jbyte exchange_value, volatile jbyte* dest, jbyte compare_value) {
  assert(sizeof(jbyte) == 1, "assumption.");
  uintptr_t dest_addr = (uintptr_t)dest;
  uintptr_t offset = dest_addr % sizeof(jint);
  volatile jint* dest_int = (volatile jint*)(dest_addr - offset);
  jint cur = *dest_int;
  jbyte* cur_as_bytes = (jbyte*)(&cur);
  jint new_val = cur;
  jbyte* new_val_as_bytes = (jbyte*)(&new_val);
  new_val_as_bytes[offset] = exchange_value;
  while (cur_as_bytes[offset] == compare_value) {
    jint res = cmpxchg(new_val, dest_int, cur);
    if (res == cur) break;
    cur = res;
    new_val = cur;
    new_val_as_bytes[offset] = exchange_value;
  }
  return cur_as_bytes[offset];
}

在做了一大堆地址转换操作之后，其最终会调用cmpxchg()方法。它根据系统平台和架构的不同，会有不同的实现，如下图所示。

来看看Linux x86的实现，其中用上了内嵌汇编。大学时期曾经很热衷于它，现在已经忘得有点多了，先复习一下内嵌汇编代码块的基本格式吧。

__asm__ __volatile__(assembly template
        : output operand list 
        : input operand list
        : clobber list);

具体的代码如下。

#define LOCK_IF_MP(mp) "cmp $0, " #mp "; je 1f; lock; 1: "

inline jint     Atomic::cmpxchg    (jint     exchange_value, volatile jint*     dest, jint     compare_value) {
  int mp = os::is_MP();
  __asm__ volatile (LOCK_IF_MP(%4) "cmpxchgl %1,(%3)"
                    : "=a" (exchange_value)
                    : "r" (exchange_value), "a" (compare_value), "r" (dest), "r" (mp)
                    : "cc", "memory");
  return exchange_value;
}

其中调用的os::is_MP()方法返回当前系统的处理器核心数，额外定义的LOCK_IF_MP宏则用来检测是否为多核环境，如果是，就先用lock指令为前端总线上锁，让其他核心暂时不能访存，这是CAS机制能够保证原子性的关键所在。

然后，调用cmpxchgl指令执行比较与交换操作。通过查阅Intel指令集文档，它接受两个操作数%1和(%3)（根据输入操作数列表，%1~%4分别代表exchange_value、compare_value、dest和mp），并执行如下伪码所示的操作：

TEMP ← DEST
IF accumulator = TEMP
    THEN
        ZF ← 1;
        DEST ← SRC;
    ELSE
        ZF ← 0;
        accumulator ← TEMP;
        DEST ← TEMP;
FI;

accumulator表示AL、AX、EAX、RAX这几个寄存器（即8/16/32/64位累加器）其中之一。也就是说，它先用累加器中的值与地址dest中的值比较，如果相等，就将exchange_value的值加载到dest指向的地址，并将零标记ZF置1。否则，就将exchange_value的值加载到累加器，并将ZF置0。注意到操作数%2即compare_value没有出现在汇编模板部分，但实际上已经在输出操作数列表中把它存到了累加器，可以被cmpxcghl指令隐式访问到。

至此，整个CAS过程完成。从根源上讲，Java的CAS机制完全依赖于CPU对原子性的“比较-交换”操作的底层支持，因此效率非常高。关于CAS的主要缺点，如自旋造成的CPU开销以及ABA问题，都很容易理解，并且有专门的资料讲解，本文就不废话了。

晚安。