Longadder of the source code of JUC atomic accumulator

LongAdder It's a concurrency master @author Doug Lea （ Big brother li ） The works of , The design is very exquisite
LongAdder Class has several key fields

//  Cumulative cell array ,  Lazy initialization 
transient volatile Cell[] cells;
//  Base value ,  If there is no competition ,  Then use  cas  Add up this field 
transient volatile long base;
//  stay  cells  When creating or expanding capacity ,  Set as  1,  Means to lock 
transient volatile int cellsBusy;

cas lock

//  Don't use it in practice ！！！
public class LockCas {
    
 private AtomicInteger state = new AtomicInteger(0);
 public void lock() {
    
 while (true) {
    
 if (state.compareAndSet(0, 1)) {
    
 break;
 }
 }
 }
 public void unlock() {
    
 log.debug("unlock...");
 state.set(0);
 }
}

among Cell Accumulation unit

//  Prevent cache line pseudo sharing 
@sun.misc.Contended
static final class Cell {
    
 volatile long value;
 Cell(long x) {
     value = x; }
 
 //  The most important method ,  be used for  cas  Accumulate in a way , prev  Old value , next  Represents the new value 
 final boolean cas(long prev, long next) {
    
 return UNSAFE.compareAndSwapLong(this, valueOffset, prev, next);
 }
 //  Omit unimportant code 
}

We have to start with caching
Speed comparison between cache and memory



because Cell It's in the form of an array , It's continuously stored in memory , One Cell by 24 byte （16 Byte object header and 8 Bytes of value）, Therefore, cache lines can be saved 2 One of the Cell object . Here comes the question ：

• Core-0 To be modified Cell[0]
• Core-1 To be modified Cell[1]
  No matter who modifies it successfully , Will lead to each other Core Cache row invalidation for , such as Core-0 in Cell[0]=6000, Cell[1]=8000 To accumulate Cell[0]=6001, Cell[1]=8000 , This will allow Core-1 Cache row invalidation for

@sun.misc.Contended To solve this problem , Its principle is to add... Before and after the object or field using this annotation 128 Byte size
padding, So that CPU Different cache lines are used when pre reading objects to the cache , such , It will not invalidate the other party's cache lines

Accumulation mainly calls the following methods

public void add(long x) {
    
 // as  For the cumulative cell array 
 // b  As the base value 
 // x  Is the cumulative value 
 Cell[] as; long b, v; int m; Cell a;
 //  Get into  if  Two conditions of 
 // 1. as  Valuable ,  Indicates that competition has occurred ,  Get into  if
 // 2. cas  to  base  Accumulation failed ,  Express  base  There is competition ,  Get into  if
 if ((as = cells) != null || !casBase(b = base, b + x)) {
    
 // uncontended  Express  cell  There is no competition 
 boolean uncontended = true;
 if (
 // as  Not yet created 
 as == null || (m = as.length - 1) < 0 ||
 //  Corresponding to the current thread  cell  Not yet 
 (a = as[getProbe() & m]) == null ||
 // cas  For the current thread  cell  Accumulation failed  uncontended=false ( a  For the current thread  cell )
 !(uncontended = a.cas(v = a.value, v + x))
 ) {
    
 //  Get into  cell  Array creation 、cell  Created process 
 longAccumulate(x, null, uncontended);
 }
 }
}

final void longAccumulate(long x, LongBinaryOperator fn,
 boolean wasUncontended) {
    
 int h;
 //  The current thread has no corresponding  cell,  Need to generate a random  h  Used to bind the thread to the current value  cell
 if ((h = getProbe()) == 0) {
    
 //  initialization  probe
 ThreadLocalRandom.current();
 // h  Corresponding to the new  probe  value ,  To correspond to  cell
 h = getProbe();
 wasUncontended = true;
 }
 // collide  by  true  Indicates that capacity expansion is required 
 boolean collide = false; 
 for (;;) {
    
 Cell[] as; Cell a; int n; long v;
 //  Already there.  cells
 if ((as = cells) != null && (n = as.length) > 0) {
    
 //  Not yet  cell
 if ((a = as[(n - 1) & h]) == null) {
    
 //  by  cellsBusy  Lock ,  establish  cell, cell  The initial cumulative value of is  x
 //  Success principle  break,  Otherwise continue  continue  loop 
 }
 //  There is competition ,  Change the corresponding thread  cell  Try again  cas
 else if (!wasUncontended)
 wasUncontended = true;
 // cas  Try to accumulate , fn  coordination  LongAccumulator  Not for  null,  coordination  LongAdder  by  null
 else if (a.cas(v = a.value, ((fn == null) ? v + x : fn.applyAsLong(v, x))))
 break;
 //  If  cells  The length has exceeded the maximum length ,  Or it has been expanded ,  Change the corresponding thread  cell  Try again  cas
 else if (n >= NCPU || cells != as)
 collide = false;
 //  Make sure  collide  by  false  Enter this branch ,  You won't go down there  else if  The capacity has been expanded 
 else if (!collide)
 collide = true;
 //  Lock 
 else if (cellsBusy == 0 && casCellsBusy()) {
    
 //  Locking success ,  Capacity expansion 
 continue;
 }
 //  Change the corresponding thread  cell
 h = advanceProbe(h);
 }
 //  Not yet  cells,  Try to give  cellsBusy  Lock 
 else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
    
 //  Locking success ,  initialization  cells,  The initial length is  2,  And fill in a  cell
 //  Success principle  break;
 }
 //  The last two cases failed ,  Try to give  base  Add up 
 else if (casBase(v = base, ((fn == null) ? v + x : fn.applyAsLong(v, x))))
 break;
 }
}

Get the final result through sum Method

public long sum() {
    
 Cell[] as = cells; Cell a;
 long sum = base;
 if (as != null) {
    
 for (int i = 0; i < as.length; ++i) {
    
 if ((a = as[i]) != null)
 sum += a.value;
 }
 }
 return sum; }