Java原子类中CAS的底层实现

时间 2019-12-04

标签 java 子类 cas 底层实现栏目 Java 繁體版

原文原文链接

Java原子类中CAS的底层实现

从Java到c++到汇编, 深刻讲解cas的底层原理.html

介绍原理前, 先来一个Demo

以AtomicBoolean类为例.先来一个调用cas的demo.java

主线程在for语句里cas忙循环, 直到cas操做成功返回true为止.linux

而新开的一个县城new Thread 会在4秒后,将flag设置为true, 为了让主线程可以设置成功.(由于cas的预期值是true, 而flag被初始化为了false)ios

现象就是主线程一直在跑for循环. 4秒后, 主线程将会设置成功, 而后输出时间差, 而后终止for循环.c++

public class TestAtomicBoolean {
    public static void main(String[] args) {
        AtomicBoolean flag = new AtomicBoolean(false);

        long start = System.currentTimeMillis();

        new Thread(()->{
            try {
                Thread.sleep(4000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            flag.set(true);
        }).start();

        for(;;){
            if(flag.compareAndSet(true,false)){
                System.out.println(System.currentTimeMillis() - start);
                System.out.println("inner loop OK!");
                break;
            }
        }
    }
}

这里只是举了一个例子, 也许这个例子也不太恰当, 本文只是列出了这个api的调用方法而已, 重点在于介绍compareAndSet()方法的底层原理.windows

Java级源码AtomicBoolean.java

发现AtomicBoolean的compareAndSet()调用的是unsafe里的compareAndSwapInt()方法.api

Java级源码Unsafe.java

有的同窗可能好奇, 其中的unsafe是怎么来的.缓存

在AtomicBoolean类中的静态成员变量:多线程

若是还要细究Unsafe.getUnsafe()是怎么实现的话....那么我再贴一份Unsafe类里的getUnsafe的代码:函数

首先, 在Unsafe类里, 本身就有了一个本身的实例.(并且是单例的)

而后Unsafe类里的getUnsafe()方法会进行检查, 最终会return这个单例 theUnsafe.

刚刚跑去取介绍了getUnsafe()方法...接下来继续讲解cas...

刚才说到了AtomicBoolean类里的compareAndSet()方法内部其实调用了Unsafe类里的compareAndSwapInt()方法.

Unsafe类里的compareAndSwapInt源码以下:

(OpenJDK8的源码里路径: openjdk/jdk/src/share/classes/sun/misc/Unsafe.java)

发现这里是一段native方法.说明继续看源码的话, 从这里就开始脱离Java语言了....

c++级源码Unsafe.cpp

本源码在OpenJDK8里的路径为: openjdk/hotspot/src/share/vm/prims/unsafe.cpp

(这里临时跑题一下: 若是说要细究 UNSAFE_ENTRY 是什么的话...UNSAFE_ENTRY 就是 JVM_ENTRY, 而 JVM_ENTRY 在interfaceSupport.hpp里面定义了, jni相关.若是想看的话, 源码路径在OpenJDK8中的路径是这个:

openjdk/hotspot/src/share/vm/runtime/interfaceSupport.hpp)

回到本文的主题cas....上面截图的这段代码, 看后面那一句return, 发现其中的使用到的是Atomic下的cmpxchg()方法.

c++级源码atomic.cpp

本段源码对应OpenJDK8的路径是这个: openjdk/hotspot/src/share/vm/runtime/atomic.cpp

其中的cmpxchg为核心内容. 可是这句代码根据操做系统和处理器的不一样, 使用不一样的底层代码.

而atomic.inline.hpp里声明以下:

可见 ...不一样不一样操做系统, 不一样的处理器, 都要走不一样的cmpxchg()方法的实现.

我们接下来以其中的linux操做系统 x86处理器为例 , atomic_linux_x86.inline.hpp

汇编级源码atomic_linux_x86.inline.hpp

OpenJDK中路径以下: openjdk/hotspot/src/os_cpu/linux_x86/vm/atomic_linux_x86.inline.hpp

看到了__asm__, 说明c++要开始内联汇编了,说明继续看代码的话, 将会是汇编语言.

这是一段内联汇编:

其中 __asm__ volatile 指示了编译器不要改动优化后面的汇编语句, 若是进行了优化(优化是为了减小访问内存, 直接经过缓存, 加快取读速度), 那么就在这段函数的周期内, 某几个变量就至关于常亮了, 其值可能会与内存中真实的值有差别.

2018.6.7更新: 10天没更新了, 因为实习结束, 这几天手头没电脑, 并且租房火车回学校等各类问题繁杂...

了解汇编指令cmpxchg

环境

怼代码以前...先把环境介绍一下...汇编的指令细节方便会有所不一样, 好比__asm__ 和asm不必定同样, 又好比 asm(...) 和 asm{...} 的括号问题, 又好比汇编指令的首操做数和第2操做数的含义是颠倒的, 也就是`mov 操做数1, 操做数2` , 要改写成`mov 操做数2, 操做数1`...反正很乱..我也不专门搞汇编, 我只能保证我这里使用g++进行编译能正常运行.(linux下或者mac下用g++编译是没问题的, windows就不知道了...)

内联汇编格式

有些同窗可能会对上面突如其来的汇编语句感到迷茫...

先来一下内联汇编的语法格式:

asm　volatile("Instruction List"

: Output

: Input

: Clobber/Modify);

其中的Output和Input你们很好理解, 可是Clobber/Modify是什么意思呢:

(下面关于Clobber/Modify的内容引用自:https://blog.csdn.net/dlh0313/article/details/52172833)
有时候,当你想通知GCC当前内联汇编语句可能会对某些寄存器或内存进行修改,但愿GCC在编译时可以将这一点考虑进去;那么你就能够在Clobber/Modify部分声明这些寄存器或内存

你们能够看看上面图片中openJDK中的汇编代码, 里面的Clobber/Modify部分是"cc"和"memory", memory好理解, 就是内存. 那么"cc"是什么呢?

当一个内联汇编中包含影响标志寄存器eflags的条件,那么也须要在Clobber/Modify部分中使用"cc"来向GCC声明这一点

内联汇编的简单例子

接下来展现一个简单的c++内联汇编的程序: a是1000, 经过汇编来给b变量也赋值为1000

#include<iostream>

using namespace std;

int main() {
    int a = 1000, b = 0;

    asm("movl %1,%%eax\n"
        "movl %%eax,%0\n"
    : "=r"(b)
    : "r" (a)
    : "%eax");
    cout << "a := " << a << endl;
    cout << "b := " << b << endl;
    return 0;
}

首先在c++里定义了a=1000, b=0;

而后看asm里的第一行冒号, 这里表示Output ` : "=r"(b) `, b是第0个参数, 等号(=)表示当前输出表达式的属性为只写, r表示寄存器

而后看asm里的第一行冒号, 这里表示Input `: "r" (a)`, a是第1个参数, r表示寄存器.

而后看asm里的第一行指令 `movl %1,%%eax\n` , 将第0个参数的值(变量a)传入到寄存器eax中

而后看asm里的第二行指令 `movl %%eax,%0\n` , 将寄存器eax的值传给第一个参数(变量b)

接下来使用c++的cout进行输出, 查看是否赋值成功, 结果以下:

再来一个简单汇编例子

用汇编给a变量赋值为100

#include<iostream>

using namespace std;

int main() {
    int a = 0;

    asm("movl $100,%%eax\n"
        "movl %%eax,%0\n"
    : "=r"(a)
    : /*no input*/
    : "%eax", "memory");
    cout << "a := " << a << endl;
    return 0;
}

(固然若是用的熟练, 就不须要使用两句mov来实现这个功能. 能够直接 movl $100, %0 ,就能够实现把100赋值给a变量了. 这里只是为了演示使用寄存器, 因此先给寄存器eax存上100, 再把寄存器的值赋值给a, 用寄存器eax作了一个中间的临时存储)

程序运行结果以下:

内联汇编cmpxchg(cmpxchg比对成功)

(有同窗不会编译运行g++下的cpp, 我顺便也在这里作一个示范吧.....)

#include<iostream>

using namespace std;

int main() {
    int cpp_eax = 0;
    int cpp_ebx = 0;
    int expect = 2222;
    int target = 8888;
    asm("movl    %2,    %%eax \n"  /*将2222存入到eax寄存器中*/
        "movl    %3,    %%ebx \n"  /*将8888存入到ebx寄存器中*/
        "cmpxchg %%ebx, %2    \n"  /*若是变量expect的值与寄存器eax的值相等(成功), 那么ebx的值就赋给expect*/
        "movl    %%eax, %0    \n"  /*将寄存器eax的值赋值给变量cpp_eax*/
        "movl    %%ebx, %1    \n"  /*将寄存器ebx的值赋值给变量cpp_ebx*/
    :"=r"(cpp_eax), "=r"(cpp_ebx), "+r"(expect)   /*等于号表示可写, 加号表示可写可读*/
    :"r"(target)
    /*cpp_eax是%0, cpp_ebx是%1, expect是%2, target是%3*/
    :"%eax", "%ebx", "memory", "cc");
    cout << "eax := " << cpp_eax << endl;
    cout << "ebx := " << cpp_ebx << endl;
    cout << "expect := " << expect << endl;
    cout << "target := " << target << endl;
}

运行方式和结果以下:

内联汇编cmpxchg(cmpxchg比对失败)

若是cmpxchg指令中的第二个操做数与寄存器eax进行比对, 发现值不同的时候, 就会比对失败, 那么expect的值就会赋值给eax, 而expect的值保持不变.

#include<iostream>

using namespace std;

int main() {
    int cpp_eax = 0;
    int cpp_ebx = 0;
    int expect = 2222;
    int target = 8888;
    asm("movl    $77,   %%eax \n"  /*将字面量77存入到eax寄存器中*/
        "movl    %3,    %%ebx \n"  /*将8888存入到ebx寄存器中*/
        "cmpxchg %%ebx, %2    \n"  /*若是变量expect的值与寄存器eax的值不相等(失败), 那么expect的值就赋给eax*/
        "movl    %%eax, %0    \n"  /*将寄存器eax的值赋值给变量cpp_eax*/
        "movl    %%ebx, %1    \n"  /*将寄存器ebx的值赋值给变量cpp_ebx*/
    :"=r"(cpp_eax), "=r"(cpp_ebx), "+r"(expect)   /*等于号表示可写, 加号表示可写可读*/
    :"r"(target)
    /*cpp_eax是%0, cpp_ebx是%1, expect是%2, target是%3*/
    :"%eax", "%ebx", "memory", "cc");
    cout << "eax := " << cpp_eax << endl;
    cout << "ebx := " << cpp_ebx << endl;
    cout << "expect := " << expect << endl;
    cout << "target := " << target << endl;
}

cmpxchg指令结论

1.指令格式: CMPXCHG 操做数1 (8位/16位/32位寄存器), 操做数2(能够是任意寄存器或者内存memory)

2.指令做用: 将累加器AL/AX/EAX(也就是%eax)中的值与第2操做数（目的操做数）比较，若是相等，第首操做数（源操做数）的值装载到第2操做数，zf置1。若是不等，第2操做数的值装载到AL/AX/EAX并将zf清0

3.该指令只能用于486及其后继机型。

lock前缀

本段的文字理论介绍引用自(http://www.weixianmanbu.com/article/736.html):

在单处理器系统中是不须要加lock的，由于可以在单条指令中完成的操做均可以认为是原子操做，中断只能发生在指令与指令之间。

在多处理器系统中,因为系统中有多个处理器在独立的运行，即便在能单条指令中完成的操做也可能受到干扰。

在全部的 X86 CPU 上都具备锁定一个特定内存地址的能力，当这个特定内存地址被锁定后，它就能够阻止其余的系统总线读取或修改这个内存地址。这种能力是经过 LOCK 指令前缀再加上下面的汇编指令来实现的。当使用 LOCK 指令前缀时，它会使 CPU 宣告一个 LOCK# 信号，这样就能确保在多处理器系统或多线程竞争的环境下互斥地使用这个内存地址。当指令执行完毕，这个锁定动做也就会消失。

#include<iostream>

using namespace std;

int main() {
    int cpp_eax = 0;
    int cpp_ebx = 0;
    int expect = 2222;
    int target = 8888;

    __asm__ __volatile__("movl    $2222, %%eax \n"  /*将字面量2222存入到eax寄存器中*/
                         "movl    %3   , %%ebx \n"  /*将8888存入到ebx寄存器中*/
                 "lock;" "cmpxchg %%ebx, %2    \n"  /*若是变量expect的值与寄存器eax的值不相等(失败), 那么expect的值就赋给eax*/
                         "movl    %%eax, %0    \n"  /*将寄存器eax的值赋值给变量cpp_eax*/
                         "movl    %%ebx, %1    \n"  /*将寄存器ebx的值赋值给变量cpp_ebx*/
    :"=r"(cpp_eax), "=r"(cpp_ebx), "=m"(expect)     /*等于号表示可写, 加号表示可写可读*/
    :"r"(target)
    /*cpp_eax是%0, cpp_ebx是%1, expect是%2, target是%3*/
    :"%eax", "%ebx", "memory", "cc");
    cout << "eax := " << cpp_eax << endl;
    cout << "ebx := " << cpp_ebx << endl;
    cout << "expect := " << expect << endl;
    cout << "target := " << target << endl;
}

结果以下:

注意, 若是加lock前缀的话, 指令的第2操做数必须是存储在内存中的, 语句格式是这样` lock; cmpxchg 操做数1(寄存器), 操做数2(内存) ` .因此, expect做为output的时候是这样声明的: "=m"(expect)　

若是不是操做的内存, 那么会报异常. 操做数2必须是内存...在这里卡了好久