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

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