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

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

功能简介:

  • Striped64是jdk1.8提供的用于支持如Long累加器,Double累加器这样机制的基础类。
  • Striped64的设计核心思路就是通过内部的分散计算来避免竞争(比如多线程CAS操作时的竞争)。
  • Striped64内部包含一个基础值和一个单元哈希表。没有竞争的情况下,要累加的数会累加到这个基础值上;如果有竞争的话,会将要累加的数累加到单元哈希表中的某个单元里面。所以整个Striped64的值包括基础值和单元哈希表中所有单元的值的总和。

源码分析:

  • 先看一下内部结构:
  1. /** 
  2.  * 存放Cell的hash表,大小为2的幂。 
  3.  */  
  4. transient volatile Cell[] cells;  
  5. /** 
  6.  * 基础值,没有竞争时会使用(更新)这个值,同时做为初始化竞争失败的回退方案。 
  7.  * 原子更新。 
  8.  */  
  9. transient volatile long base;  
  10. /** 
  11.  * 自旋锁,通过CAS操作加锁,用于保护创建或者扩展Cell表。 
  12.  */  
  13. transient volatile int cellsBusy;  

 

       看下Cell的内部结构:

  1. @sun.misc.Contended   
  2. static final class Cell {  
  3.     volatile long value;  
  4.     Cell(long x) { value = x; }  
  5.     final boolean cas(long cmp, long val) {  
  6.         return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);  
  7.     }  
  8.     // Unsafe mechanics  
  9.     private static final sun.misc.Unsafe UNSAFE;  
  10.     private static final long valueOffset;  
  11.     static {  
  12.         try {  
  13.             UNSAFE = sun.misc.Unsafe.getUnsafe();  
  14.             Class<?> ak = Cell.class;  
  15.             valueOffset = UNSAFE.objectFieldOffset  
  16.                 (ak.getDeclaredField(“value”));  
  17.         } catch (Exception e) {  
  18.             throw new Error(e);  
  19.         }  
  20.     }  
  21. }  

  
     Cell内部保存了一个volatile修饰的long型域,同时提供了原子操作,看起来像一个原子量。
       注意到Cell类被一个Contended注解修饰,Contended的作用是对Cell做缓存行填充,避免伪共享。  

  • Striped64主要提供了longAccumulate和doubleAccumulate方法来支持子类,先看下longAccumulate:
  1. static final int NCPU = Runtime.getRuntime().availableProcessors();  
  2. final void longAccumulate(long x, LongBinaryOperator fn,  
  3.                           boolean wasUncontended) {  
  4.     int h;  
  5.     //获取当前线程的probe值作为hash值。  
  6.     if ((h = getProbe()) == 0) {  
  7.         //如果probe值为0,强制初始化当前线程的probe值,这次初始化的probe值不会为0。  
  8.         ThreadLocalRandom.current();   
  9.         //再次获取probe值作为hash值。  
  10.         h = getProbe();  
  11.         //这次相当于再次计算了hash,所以设置未竞争标记为true。  
  12.         wasUncontended = true;  
  13.     }  
  14.     boolean collide = false;  
  15.     for (;;) {  
  16.         Cell[] as; Cell a; int n; long v;  
  17.         if ((as = cells) != null && (n = as.length) > 0) {  
  18.             //通过h从cell表中选定一个cell位置。  
  19.             if ((a = as[(n – 1) & h]) == null) {  
  20.                 //如果当前位置没有cell,尝试新建一个。  
  21.                 if (cellsBusy == 0) {  
  22.                     //创建一个Cell。         
  23.                     Cell r = new Cell(x);   
  24.                     //尝试或者cellsBusy锁。  
  25.                     if (cellsBusy == 0 && casCellsBusy()) {  
  26.                         boolean created = false;  
  27.                         try {                 
  28.                             Cell[] rs; int m, j;  
  29.                             //在获取锁的情况下再次检测一下。  
  30.                             if ((rs = cells) != null &&  
  31.                                 (m = rs.length) > 0 &&  
  32.                                 rs[j = (m – 1) & h] == null) {  
  33.                                 //设置新建的cell到指定位置。  
  34.                                 rs[j] = r;  
  35.                                 //创建标记设置为true。  
  36.                                 created = true;  
  37.                             }  
  38.                         } finally {  
  39.                             //释放cellsBusy锁。  
  40.                             cellsBusy = 0;  
  41.                         }  
  42.                         if (created)  
  43.                             //如果创建成功,直接跳出循环,退出方法。  
  44.                             break;  
  45.                         //说明上面指定的cell的位置上有cell了,继续尝试。  
  46.                         continue;    
  47.                     }  
  48.                 }  
  49.                 //走到这里说明获取cellsBusy锁失败。  
  50.                 collide = false;  
  51.             }  
  52.             //以下条件说明上面通过h选定的cell表的位置上有Cell,就是a。  
  53.             else if (!wasUncontended)       // CAS already known to fail  
  54.                 //如果之前的CAS失败,说明已经发生竞争,  
  55.                 //这里会设置未竞争标志位true,然后再次算一个probe值,然后重试。  
  56.                 wasUncontended = true;      // Continue after rehash  
  57.             //这里尝试将x值加到a的value上。  
  58.             else if (a.cas(v = a.value, ((fn == null) ? v + x :  
  59.                                          fn.applyAsLong(v, x))))  
  60.                 //如果尝试成功,跳出循环,方法退出。  
  61.                 break;  
  62.             else if (n >= NCPU || cells != as)  
  63.                 //如果cell表的size已经最大,或者cell表已经发生变化(as是一个过时的)。  
  64.                 collide = false;             
  65.             else if (!collide)  
  66.                 //设置冲突标志,表示发生了冲突,重试。  
  67.                 collide = true;  
  68.             //尝试获取cellsBusy锁。  
  69.             else if (cellsBusy == 0 && casCellsBusy()) {  
  70.                 try {  
  71.                     //检测as是否过时。  
  72.                     if (cells == as) {        
  73.                         //给cell表扩容。  
  74.                         Cell[] rs = new Cell[n << 1];  
  75.                         for (int i = 0; i < n; ++i)  
  76.                             rs[i] = as[i];  
  77.                         cells = rs;  
  78.                     }  
  79.                 } finally {  
  80.                     //释放cellsBusy锁。  
  81.                     cellsBusy = 0;  
  82.                 }  
  83.                 collide = false;  
  84.                 //扩容cell表后,再次重试。  
  85.                 continue;                    
  86.             }  
  87.             //算出下一个hash值。  
  88.             h = advanceProbe(h);  
  89.         }  
  90.         //如果cell表还未创建,先尝试获取cellsBusy锁。  
  91.         else if (cellsBusy == 0 && cells == as && casCellsBusy()) {  
  92.             boolean init = false;  
  93.             try {                            
  94.                 if (cells == as) {  
  95.                     //初始化cell表,初始容量为2。  
  96.                     Cell[] rs = new Cell[2];  
  97.                     rs[h & 1] = new Cell(x);  
  98.                     cells = rs;  
  99.                     init = true;  
  100.                 }  
  101.             } finally {  
  102.                 //释放cellsBusy锁。  
  103.                 cellsBusy = 0;  
  104.             }  
  105.             if (init)  
  106.                 //初始化cell表成功后,退出方法。  
  107.                 break;  
  108.         }  
  109.         //如果创建cell表由于竞争导致失败,尝试将x累加到base上。  
  110.         else if (casBase(v = base, ((fn == null) ? v + x :  
  111.                                     fn.applyAsLong(v, x))))  
  112.             break;                            
  113.     }  
  114. }  

       说明一下这个方法,方法的作用是将给定的值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中使用到的一些方法:

  1. final boolean casBase(long cmp, long val) {  
  2.     return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);  
  3. }  
  4.   
  5. final boolean casCellsBusy() {  
  6.     return UNSAFE.compareAndSwapInt(this, CELLSBUSY, 01);  
  7. }  
  8.   
  9. static final int getProbe() {  
  10.     return UNSAFE.getInt(Thread.currentThread(), PROBE);  
  11. }  
  12.   
  13. //计算下一个随机值作为hash值,使用xorshift算法。  
  14. static final int advanceProbe(int probe) {  
  15.     probe ^= probe << 13;  
  16.     probe ^= probe >>> 17;  
  17.     probe ^= probe << 5;  
  18.     //设置到当前线程的threadLocalRandomProbe域。  
  19.     UNSAFE.putInt(Thread.currentThread(), PROBE, probe);  
  20.     return probe;  
  21. }  

  

  • 再看下doubleAccumulate:
  1. final void doubleAccumulate(double x, DoubleBinaryOperator fn,  
  2.                             boolean wasUncontended) {  
  3.     int h;  
  4.     if ((h = getProbe()) == 0) {  
  5.         ThreadLocalRandom.current(); // force initialization  
  6.         h = getProbe();  
  7.         wasUncontended = true;  
  8.     }  
  9.     boolean collide = false;                // True if last slot nonempty  
  10.     for (;;) {  
  11.         Cell[] as; Cell a; int n; long v;  
  12.         if ((as = cells) != null && (n = as.length) > 0) {  
  13.             if ((a = as[(n – 1) & h]) == null) {  
  14.                 if (cellsBusy == 0) {       // Try to attach new Cell  
  15.                     Cell r = new Cell(Double.doubleToRawLongBits(x));  
  16.                     if (cellsBusy == 0 && casCellsBusy()) {  
  17.                         boolean created = false;  
  18.                         try {               // Recheck under lock  
  19.                             Cell[] rs; int m, j;  
  20.                             if ((rs = cells) != null &&  
  21.                                 (m = rs.length) > 0 &&  
  22.                                 rs[j = (m – 1) & h] == null) {  
  23.                                 rs[j] = r;  
  24.                                 created = true;  
  25.                             }  
  26.                         } finally {  
  27.                             cellsBusy = 0;  
  28.                         }  
  29.                         if (created)  
  30.                             break;  
  31.                         continue;           // Slot is now non-empty  
  32.                     }  
  33.                 }  
  34.                 collide = false;  
  35.             }  
  36.             else if (!wasUncontended)       // CAS already known to fail  
  37.                 wasUncontended = true;      // Continue after rehash  
  38.             else if (a.cas(v = a.value,  
  39.                            ((fn == null) ?  
  40.                             Double.doubleToRawLongBits  
  41.                             (Double.longBitsToDouble(v) + x) :  
  42.                             Double.doubleToRawLongBits  
  43.                             (fn.applyAsDouble  
  44.                              (Double.longBitsToDouble(v), x)))))  
  45.                 break;  
  46.             else if (n >= NCPU || cells != as)  
  47.                 collide = false;            // At max size or stale  
  48.             else if (!collide)  
  49.                 collide = true;  
  50.             else if (cellsBusy == 0 && casCellsBusy()) {  
  51.                 try {  
  52.                     if (cells == as) {      // Expand table unless stale  
  53.                         Cell[] rs = new Cell[n << 1];  
  54.                         for (int i = 0; i < n; ++i)  
  55.                             rs[i] = as[i];  
  56.                         cells = rs;  
  57.                     }  
  58.                 } finally {  
  59.                     cellsBusy = 0;  
  60.                 }  
  61.                 collide = false;  
  62.                 continue;                   // Retry with expanded table  
  63.             }  
  64.             h = advanceProbe(h);  
  65.         }  
  66.         else if (cellsBusy == 0 && cells == as && casCellsBusy()) {  
  67.             boolean init = false;  
  68.             try {                           // Initialize table  
  69.                 if (cells == as) {  
  70.                     Cell[] rs = new Cell[2];  
  71.                     rs[h & 1] = new Cell(Double.doubleToRawLongBits(x));  
  72.                     cells = rs;  
  73.                     init = true;  
  74.                 }  
  75.             } finally {  
  76.                 cellsBusy = 0;  
  77.             }  
  78.             if (init)  
  79.                 break;  
  80.         }  
  81.         else if (casBase(v = base,  
  82.                          ((fn == null) ?  
  83.                           Double.doubleToRawLongBits  
  84.                           (Double.longBitsToDouble(v) + x) :  
  85.                           Double.doubleToRawLongBits  
  86.                           (fn.applyAsDouble  
  87.                            (Double.longBitsToDouble(v), x)))))  
  88.             break;                          // Fall back on using base  
  89.     }  
  90. }  

       doubleAccumulate方法是针对double值做累加的,逻辑和longAccumulate一致。但由于Cell内部用long保存数据,所以在累加的时候会利用Double的doubleToRawLongBits和longBitsToDouble方法做double和longBits形式的double之间的转换。          Striped64的代码解析完毕!               参考资料:
http://mail.openjdk.java.net/pipermail/hotspot-dev/2012-November/007309.html

    原文作者:JUC
    原文地址: https://blog.csdn.net/moakun/article/details/80590893
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞