Jdk1.6 JUC源码解析(2)-atomic-AtomicXXXArray

Jdk1.6 JUC源码解析(2)-atomic-AtomicXXXArray

作者：大飞

 

功能简介：

• 数组原子量。

 

源码分析：

• 和原子量一样，数组原子量内部有一个Unsafe的静态引用。

 

    private static final Unsafe unsafe = Unsafe.getUnsafe();

 

 

 

 

• 首先先看下AtomicIntegerArray。

       AtomicIntegerArray的构造方法如下：

    /**
     * Creates a new AtomicIntegerArray of given length.
     *
     * @param length the length of the array
     */
    public AtomicIntegerArray(int length) {
        array = new int[length];
        // must perform at least one volatile write to conform to JMM
        if (length > 0)
            unsafe.putIntVolatile(array, rawIndex(0), 0);
    }
    /**
     * Creates a new AtomicIntegerArray with the same length as, and
     * all elements copied from, the given array.
     *
     * @param array the array to copy elements from
     * @throws NullPointerException if array is null
     */
    public AtomicIntegerArray(int[] array) {
        if (array == null)
            throw new NullPointerException();
        int length = array.length;
        this.array = new int[length];
        if (length > 0) {
            int last = length-1;
            for (int i = 0; i < last; ++i)
                this.array[i] = array[i];
            // Do the last write as volatile
            unsafe.putIntVolatile(this.array, rawIndex(last), array[last]);
        }
    }

       注：当前源码来之jdk1.6，在里面会看到第一个构造方法最后就会添加一个volatile write，但在jdk1.8(从jdk1.7某个小版本开始)里面就看不到这个volatile write了，与群友讨论得知这应该是一个遗留代码，因为final完全可以保证这个语义(其他线程可以看到完全构造的内部array)；第二个构造方法首先对内部final修饰的array进行赋值，然后进行数组元素copy，为了保证其他线程可以看到完全构造(内部元素copy完成)的array，所以要在copy最后加一个volatile write。

 

       相对于AtomicInteger来说，AtomicIntegerArray里面的方法都带有下标：

    /**
     * Atomically increments by one the element at index {@code i}.
     *
     * @param i the index
     * @return the updated value
     */
    public final int incrementAndGet(int i) {
        while (true) {
            int current = get(i);
            int next = current + 1;
            if (compareAndSet(i, current, next))
                return next;
        }
    }

        接下来看一下AtomicIntegerArray根据下标取值的get方法：

    /**
     * Gets the current value at position {@code i}.
     *
     * @param i the index
     * @return the current value
     */
    public final int get(int i) {
        return unsafe.getIntVolatile(array, rawIndex(i));
    }

       方法里调用了unsafe的getIntVolatile方法。在hotspot/src/share/vm/classfile/vmSymbols.hpp中找到：

  do_intrinsic(_getIntVolatile,           sun_misc_Unsafe,        getIntVolatile_name, getInt_signature,         F_RN)  \

        接着在hotspot/src/share/vm/opto/library_call.cpp中找到实现：

  case vmIntrinsics::_getIntVolatile:
    return inline_unsafe_access(!is_native_ptr, !is_store, T_INT, true);
  ...
bool LibraryCallKit::inline_unsafe_access(bool is_native_ptr, bool is_store, BasicType type, bool is_volatile) {
  if (callee()->is_static())  return false;  // caller must have the capability!
...//省略不重要部分
  if (is_volatile) {
    if (!is_store)
      insert_mem_bar(Op_MemBarAcquire);
    else
      insert_mem_bar(Op_MemBarVolatile);
  }
  if (need_mem_bar) insert_mem_bar(Op_MemBarCPUOrder);
  return true;
}

       在方法的最后可以看到，如果是volatile且不是写操作的话，会加入一个Op_MemBarAcquire的内存屏障，再看下hotspot/src/cpu/x86/vm/x86_64.ad：

instruct membar_acquire()
%{
  match(MemBarAcquire);
  ins_cost(0);
  size(0);
  format %{ "MEMBAR-acquire ! (empty encoding)" %}
  ins_encode();
  ins_pipe(empty);
%}

       可见在x86_64下也相当于是对一个普通域的读取。

 

       最后看一下AtomicIntegerArray根据下标设置值的set方法：

    /**
     * Sets the element at position {@code i} to the given value.
     *
     * @param i the index
     * @param newValue the new value
     */
    public final void set(int i, int newValue) {
        unsafe.putIntVolatile(array, rawIndex(i), newValue);
    }

       方法里调用了unsafe的putIntVolatile方法。在hotspot/src/share/vm/classfile/vmSymbols.hpp中找到：

  do_intrinsic(_putIntVolatile,           sun_misc_Unsafe,        putIntVolatile_name, putInt_signature,         F_RN)  \

       接着在hotspot/src/share/vm/opto/library_call.cpp中找到实现：

  case vmIntrinsics::_putIntVolatile:
    return inline_unsafe_access(!is_native_ptr, is_store, T_INT, true);

       和上面getIntVolatile的实现一样：

  if (is_volatile) {
    if (!is_store)
      insert_mem_bar(Op_MemBarAcquire);
    else
      insert_mem_bar(Op_MemBarVolatile);
  }

       可见，如果是写操作做的话，加入Op_MemBarVolatile内存屏障，继续看hotspot/src/cpu/x86/vm/x86_64.ad：

instruct membar_volatile(rFlagsReg cr) %{
  match(MemBarVolatile);
  effect(KILL cr);
  ins_cost(400);
  format %{
    $$template
    if (os::is_MP()) {
      $$emit$$"lock addl [rsp + #0], 0\t! membar_volatile"
    } else {
      $$emit$$"MEMBAR-volatile ! (empty encoding)"
    }
  %}
  ins_encode %{
    __ membar(Assembler::StoreLoad);
  %}
  ins_pipe(pipe_slow);
%}

       如果是多核CPU，就会加入lock addl... 这个指令。其实就相当于是对一个volatile修饰的域的写操作喽。

 

 

• 再看下AtomicLongArray。

       AtomicLongArray的内部结构和AtomicIntegerArray类似，这里不做分析，只看一下get和set方法中内部调用的unsafe的getLongVolatile和putLongVolatile方法。

       首先看下get方法中的getLongVolatile方法：

    /**
     * Gets the current value at position {@code i}.
     *
     * @param i the index
     * @return the current value
     */
    public final long get(int i) {
        return unsafe.getLongVolatile(array, rawIndex(i));
    }

       看下内联实现，vmSymbols.hpp，里面有如下代码：

  do_intrinsic(_getLongVolatile,          sun_misc_Unsafe,        getLongVolatile_name, getLong_signature,       F_RN)  \

       然后找到library_call.cpp中找到相应实现：

  case vmIntrinsics::_getLongVolatile:
    return inline_unsafe_access(!is_native_ptr, !is_store, T_LONG, true);
  ...
bool LibraryCallKit::inline_unsafe_access(bool is_native_ptr, bool is_store, BasicType type, bool is_volatile) {
  if (callee()->is_static())  return false;  // caller must have the capability!
...
  if (!is_store) {
    Node* p = make_load(control(), adr, value_type, type, adr_type, is_volatile);
...
  if (is_volatile) {
    if (!is_store)
      insert_mem_bar(Op_MemBarAcquire);
    else
      insert_mem_bar(Op_MemBarVolatile);
  }
  if (need_mem_bar) insert_mem_bar(Op_MemBarCPUOrder);
  return true;
}

       首先，后面那个MemBarAcquire之前看过，在x86-64下没什么用，看下前面的make_load，找到hotspot/src/share/vm/opto/graphKit.cpp：

Node* GraphKit::make_load(Node* ctl, Node* adr, const Type* t, BasicType bt,
                          int adr_idx,
                          bool require_atomic_access) {
  assert(adr_idx != Compile::AliasIdxTop, "use other make_load factory" );
  const TypePtr* adr_type = NULL; // debug-mode-only argument
  debug_only(adr_type = C->get_adr_type(adr_idx));
  Node* mem = memory(adr_idx);
  Node* ld;
  if (require_atomic_access && bt == T_LONG) {
    ld = LoadLNode::make_atomic(C, ctl, mem, adr, adr_type, t);
  } else {
    ld = LoadNode::make(_gvn, ctl, mem, adr, adr_type, t, bt);
  }
  return _gvn.transform(ld);
}

       可见，至少对这个long值得加载是原子的(这里的原子操作应该指的是将long的高4字节和低4字节的操作合并成一个原子操作，比如某些平台不支持非volatile的long/double域的原子操作)。

 

       然后看下set方法中的setLongVolatile方法：

    /**
     * Sets the element at position {@code i} to the given value.
     *
     * @param i the index
     * @param newValue the new value
     */
    public final void set(int i, long newValue) {
        unsafe.putLongVolatile(array, rawIndex(i), newValue);
    }

       直接到library_call.cpp中看实现：

  case vmIntrinsics::_putLongVolatile:
    return inline_unsafe_access(!is_native_ptr, is_store, T_LONG, true);
  ...
bool LibraryCallKit::inline_unsafe_access(bool is_native_ptr, bool is_store, BasicType type, bool is_volatile) {
  if (callee()->is_static())  return false;  // caller must have the capability!
...
  if (is_volatile) {
    if (!is_store)
      insert_mem_bar(Op_MemBarAcquire);
    else
      insert_mem_bar(Op_MemBarVolatile);
  }
  if (need_mem_bar) insert_mem_bar(Op_MemBarCPUOrder);
  return true;
}

       之前说过，这里加入了一个lock addl ... 的内存屏障。

 

 

 

• 最后看下AtomicReferenceArray。

      AtomicReferenceArray的内部结构和AtomicIntegerArray类似，有一点点细节上的区别：

public class AtomicReferenceArray<E> implements java.io.Serializable {
    private static final long serialVersionUID = -6209656149925076980L;
    private static final Unsafe unsafe;
    private static final int base;
    private static final int shift;
    private static final long arrayFieldOffset;
    private final Object[] array; // must have exact type Object[]
    static {
        int scale;
        try {
            unsafe = Unsafe.getUnsafe();
            arrayFieldOffset = unsafe.objectFieldOffset
                (AtomicReferenceArray.class.getDeclaredField("array"));
            base = unsafe.arrayBaseOffset(Object[].class);
            scale = unsafe.arrayIndexScale(Object[].class);
        } catch (Exception e) {
            throw new Error(e);
        }
        if ((scale & (scale - 1)) != 0)
            throw new Error("data type scale not a power of two");
        shift = 31 - Integer.numberOfLeadingZeros(scale);
    }
    private long checkedByteOffset(int i) {
        if (i < 0 || i >= array.length)
            throw new IndexOutOfBoundsException("index " + i);
        return byteOffset(i);
    }
    private static long byteOffset(int i) {
        return ((long) i << shift) + base;
    }

       首先，AtomicReferenceArray里面多了一个arrayFieldOffset，这个域用来支持反序列化的。其次，与AtomicIntegerArray不同，AtomicReferenceArray并没有scale域，取而代之的是shift域。阅读代码可知，其实目的都是计算rawIndex = base + index * scale，只不过AtomicReferenceArray里面把一部分运算转换为等价的位操作(当然前提是scala为2的幂)。(了解更多位操作技巧，可参考Hacker's Delight)。

       其他涉及到unsafe的操作前面都分析过，这里不做分析了。

 

 

       好了，源码就分析到这里！