关于CAS等原子操作,说点别人没说的
Java中提供了原子操作,可以简单看一下AtomicInteger类中的一个典型的原子操作incrementAndGet(),表示对原子整数变量进行加操作,并返回新的值。实现如下:
public class AtomicInteger extends Number implements java.io.Serializable { 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 incrementAndGet() { return unsafe.getAndAddInt(this, valueOffset, 1) + 1; } }
在实现incrementAndGet()操作时,由于后续要执行CAS(compare and swap,比较并交换)操作,这个操作需要对旧值与某个地址处的值进行比较,但是在Java层无法操作地址,所以只能计算出某个字段在当前类实例中的偏移,然后在HotSpot VM中根据偏移转换为对应的地址。
调用的getAndAddInt()方法的实现如下:
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()是native方法,对应的实现如下:
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
调用的Atomic::cmpxchg()函数的实现如下:
#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; }
如上在C++函数中内联了一段汇编程序。使用精练的汇编不但可以缩小目标代码的大小,还可以使用汇编来提高某些经常被卧调用的代码的性能。
内联汇编的基本格式如下:
__asm__ [__volatile__] ( assembler template // 汇编代码模板 : [output operand list] // 输出操作数列表 : [input operand list] // 输入操作数列表 : [clobbered register list] // 修改的寄存器列表 );
内联汇编可以将C++函数中相关信息通过输入操作数列表传送到汇编指令中,也可以通过输出操作数列表接收到由汇编指令执行后的输出值。下面详细介绍所一下Atomic::cmpxchg()函数中内联汇编的具体意思。
1、汇编代码模板:当操作系统为多核时,mp为true,此时会在cmpxhgl指令之前加一个lock前缀。因为cmpxhgl指令本身并不是原子的(cmpxhg解码为多个微指令,这些微指令加载、检查是否相等,然后根据比较结果存储或不存储新值),但是加lock前缀后就会变为原子的。cmpxhg的操作数可以是reg + reg,也可以是mem + reg,前者不需要lock,因为在同一个核上,寄存器只会有一套。只有cmpxhg mem, reg才可能会需要lock,这个lock是对多核有效的。使用的cmpxhgl指令有个后缀l,表示操作数是4字节大小。
2、输出操作数列表,=表示操作数在指令中是只写的(输出操作数),a表示将变量放入eax寄存器。在64位模式下,只有%rax可用,因为在执行内联汇编相关的指令时之前会自动保存%rax的值,这样避免重要数据丢失。
3、r表示将输入变量放入通用寄存器,也就是eax,ebx,ecx,edx,esi,edi中的一个。a同样表示eax寄存器。%1就是
exchange_value,
%3
是
dest,
%4
就是
mp。
4、在修改的寄存器列表中,cc表示编译器汇编代码会导致CPU状态位的改变,也就是eflags指示了CPU状态。这里由于执行cmpxhgl,所以会更改eflags的状态;memory告诉编译器汇编代码会读取或修改内存中某个地址存放的值。
在HotSpot的atomic.hpp中声明了许多原子操作,这些操作不但为Java层原子操作提供实现,也会在HotSpot内部经常使用。主要是因为CAS相对互斥量来说更加轻量级,效率更高,但是达到同样的目的时,实现也相对复杂了一些。下面就举几个小例子,如下:
1、CAS保证在多线程竞争下,通过指针碰撞分配TLAB
在分配TLAB时会通过CAS来保证并发安全。实际上采用CAS配合上失败重试的方式保证更新操作的原子性,如下:
inline HeapWord* ContiguousSpace::par_allocate_impl( size_t size, HeapWord* const end_value ) { do { HeapWord* obj = top(); // 当前的空闲空间足够分配时尝试分配 if (pointer_delta(end_value, obj) >= size) { HeapWord* new_top = obj + size; HeapWord* result = (HeapWord*)Atomic::cmpxchg_ptr(new_top, top_addr(), obj); if (result == obj) { return obj; // 分配成功时返回,否则继续循环 } } else { return NULL; // 没有足够空间时候返回 } } while (true); }
2、保证一个或多个共享变量的原子操作
首先说一下,CAS只能保证一个共享变量的原子操作。当对一个共享变量执行操作时,我们可以使用循环CAS的方式来保证原子操作,但是对多个共享变量操作时,循环CAS就无法保证操作的原子性,这个时候就可以用锁来保证原子性。
还有一个取巧的办法,就是把多个共享变量合并成一个共享变量来操作。比如,有两个共享变量 i=2,j=a,合并一下 ij=2a,然后用 CAS 来操作 ij。从 Java 1.5 开始,JDK 提供了 AtomicReference 类来保证引用对象之间的原子性,就可以把多个变量放在一个对象里来进行 CAS 操作。在HotSpot VM实现轻量级锁时,也会有类似的操作。MarkWord将多个变量拼接为了一个64位数,如下:
在偏向锁的实现过程中,需要同时判断thread、epoch及biased_lock值来确定接下来的逻辑时,就将这几个数看成了一个64位的数进行了原子操作。
3、CAS实现自旋等待
在HotSpot VM内部锁Monitor的实现过程中,使用CAS进行自旋等待,以避免上下文切换。在Monitor::ILock()函数中,如果产生锁竞争,当前线程会调用Monitor::TrySpin ()进行自旋等待。这里等待时间的选取非常关键,因为如果自旋时间长则浪费CPU时间,旋转短了又不能有效避免上下文切换。其中的等待时间与Marsaglia的xor-shift算法产生的伪随机数有直接关系,有兴趣的可自行研究。
4、原子更新变量保护代码段线程安全
多线程竞争时,可以保护一段代码同一时刻只有一个线程在执行。在Monitor中有一个volatile变量,如下:
ParkEvent * volatile _OnDeck
这个变量被HotSpot VM作者标注为内部锁,也就是借助它可实现一段代码保护。
当执行一段代码时,可以通过_OnDeck将NULL设置为_LBIT,在退出时将_OnDeck再次设置为_LBIT,这样其它的CAS就又可以执行这段被保护的代码了。如下:
void Monitor::IUnlock (bool RelaxAssert) { ... // 获取内部锁 if (CASPTR (&_OnDeck, NULL, _LBIT) != UNS(NULL)) { return ; } // 确保同一时只有一个线程在执行这里的代码 // 释放内部锁 _OnDeck = NULL ; }
CAS操作无处不在,只要用的好、用的巧,还是能极大减少互斥量的使用的。
手写Java虚拟机HotSpot已经录制一系列视频啦!有兴趣关注
B站
。
有对虚拟机、Java性能故障诊断与调优等感兴趣的人可以入群讨论。