OpenJDK:JVM對CAS的設計與實現

CAS簡介

CAS即Compare-and-Swap的縮寫,即比較並交換,它是一種實現樂觀鎖的技術.在CAS中包含三個操作數:

• V: 需要讀寫的內存位置,從java角度你可以把它當成一個變量
• A: 預期值,也就是要進行比較的值
• B: 擬寫入的新值

當且僅當V的值等於A時,CAS才會通過原子方式用新值B來更新V的值,否則不會執行任何操作.無論位置V的值是否等於A,最終都會返回V原有的值.換句話說:"我認為V的值應該是A,如果是,那么就將V的值更新為B,否則不修改並告訴V的實際值是多少".

當多個線程使用CAS同時更新同一個變量時,只有其中一個線程能夠成功更新變量的值,其他線程都將失敗.和鎖機制不同,失敗的線程並不會被掛起,而是告知用戶當前失敗的情況,並由用戶決定是否要再次嘗試或者執行其他操作,其典型的流程如下:

 

image-20180910121551030

傳統鎖實現CAS語義

在明白CAS的語義后,我們用傳統的鎖機制來實現該語義.

public class SimpleCAS { private int value; public int getValue() { return value; } // 比較並交換語義,最終都返回原有值 public synchronized int compareAndSwap(int exectedValue, int newValue) { int oldValue = value; if (oldValue == exectedValue) { value = newValue; } return value; } // 比較並設置語義,返回成功與否 public synchronized boolean compareAndSet(int exectedValue, int newValue) { return exectedValue == compareAndSwap(exectedValue, newValue); } } 

在上述代碼中,compareAndSwap()用於實現"比較並交換"的語義,在此之上我們還實現了"比較並設置"的語義.

使用場景

CAS典型使用模式是:首先從V中讀取值A,並根據A計算出新值B,然后再通過CAS以原子方式將V中的值變成B(如果在此期間沒有任何線程將V的值修改為其他值).我們借助剛才的SimpleCAS實現一個計數器,借此來說明其使用場景:

// 線程安全的計數器 public class SafeCounter { private SimpleCAS cas; public SafeCounter() { this.cas = new SimpleCAS(); } public int getValue() { return cas.getValue(); } public int increment() { int value; int newValue; do { // 讀取舊值A value = cas.getValue(); // 根據A計算新值B newValue = value + 1; } while (!cas.compareAndSet(value, newValue));// 使用CAS來設置新值B return newValue; } } 

SafeCounter不會阻塞,如果其他線程同時更新計數器,那么會執行多次重試操作直至成功.到現在有關CAS的語義和使用已經說完,下面我們要說的是CAS在JAVA中的應用以及JVM中如何實現CAS.

CAS實現

通過傳統的鎖實現的CAS語義並非JVM真正對CAS的實現,這點需要記住.JVM中能夠實現CAS本質是現代CPU已經支持Compare-and-Swap指令.從Java 5.0開始,JVM中直接調用了相關指令.

JVM對CAS的支持

有關原子性變量的操作被統一定義在atomic.hpp,並以模板方法提供,其路徑為:

/OpenJDK10/hotspot/src/share/vm/runtime/atomic.hpp

template<typename T, typename D, typename U> inline D Atomic::cmpxchg(T exchange_value, D volatile* dest, U compare_value, cmpxchg_memory_order order) { return CmpxchgImpl<T, D, U>()(exchange_value, dest, compare_value, order); } template<typename T> struct Atomic::CmpxchgImpl< T, T, T, typename EnableIf<IsIntegral<T>::value || IsRegisteredEnum<T>::value>::type> VALUE_OBJ_CLASS_SPEC { T operator()(T exchange_value, T volatile* dest, T compare_value, cmpxchg_memory_order order) const { // Forward to the platform handler for the size of T. return PlatformCmpxchg<sizeof(T)>()(exchange_value, dest, compare_value, order); } }; 

不同的平台PlatformCmpxchg實現不同,比如在mac平台上,其實現在

/OpenJDK10/hotspot/src/os_cpu/bsd_x86/vm/atomic_bsd_x86.hpp

// 1字節長度 template<typename T> inline T Atomic::PlatformCmpxchg<1>::operator()(T exchange_value, T volatile* dest, T compare_value, cmpxchg_memory_order /* order */) const { STATIC_ASSERT(1 == sizeof(T)); // 內嵌匯編代碼,最終調用cmpxchgb指令實現"比較並交換" __asm__ volatile ( "lock cmpxchgb %1,(%3)" : "=a" (exchange_value) : "q" (exchange_value), "a" (compare_value), "r" (dest) : "cc", "memory"); return exchange_value; } // 4字節長度 template<typename T> inline T Atomic::PlatformCmpxchg<4>::operator()(T exchange_value, T volatile* dest, T compare_value, cmpxchg_memory_order /* order */) const { STATIC_ASSERT(4 == sizeof(T)); __asm__ volatile ( "lock cmpxchgl %1,(%3)" : "=a" (exchange_value) : "r" (exchange_value), "a" (compare_value), "r" (dest) : "cc", "memory"); return exchange_value; } // 8字節長度 template<typename T> inline T Atomic::PlatformCmpxchg<8>::operator()(T exchange_value, T volatile* dest, T compare_value, cmpxchg_memory_order /* order */) const { STATIC_ASSERT(8 == sizeof(T)); __asm__ __volatile__ ( "lock cmpxchgq %1,(%3)" : "=a" (exchange_value) : "r" (exchange_value), "a" (compare_value), "r" (dest) : "cc", "memory"); return exchange_value; } 

在window_x86平台中,其實現在/OpenJDK10/hotspot/src/os_cpu/windows_x86/vm/atomic_windows_x86.hpp

template<> template<typename T> inline T Atomic::PlatformCmpxchg<1>::operator()(T exchange_value, T volatile* dest, T compare_value, cmpxchg_memory_order order) const { STATIC_ASSERT(1 == sizeof(T)); // alternative for InterlockedCompareExchange // 內嵌匯編代碼,最終調用cmpxchg指令實現"比較並交換" __asm { mov edx, dest mov cl, exchange_value mov al, compare_value lock cmpxchg byte ptr [edx], cl } } template<> template<typename T> inline T Atomic::PlatformCmpxchg<4>::operator()(T exchange_value, T volatile* dest, T compare_value, cmpxchg_memory_order order) const { STATIC_ASSERT(4 == sizeof(T)); // alternative for InterlockedCompareExchange __asm { mov edx, dest mov ecx, exchange_value mov eax, compare_value lock cmpxchg dword ptr [edx], ecx } } template<> template<typename T> inline T Atomic::PlatformCmpxchg<8>::operator()(T exchange_value, T volatile* dest, T compare_value, cmpxchg_memory_order order) const { STATIC_ASSERT(8 == sizeof(T)); jint ex_lo = (jint)exchange_value; jint ex_hi = *( ((jint*)&exchange_value) + 1 ); jint cmp_lo = (jint)compare_value; jint cmp_hi = *( ((jint*)&compare_value) + 1 ); // 內嵌匯編代碼,最終調用cmpxchg8b指令實現"比較並交換"8字節 __asm { push ebx push edi mov eax, cmp_lo mov edx, cmp_hi mov edi, dest mov ebx, ex_lo mov ecx, ex_hi lock cmpxchg8b qword ptr [edi] pop edi pop ebx } } 

不難發現,最終都是通過內嵌匯編代碼的形式來實現對於CPU指令cmpxchg的調用,關於該指令后續單獨進行說明.到目前為止,對於JVM中的CAS操作已經了解的差不多了,但在Java層又是如何使用的呢?在開始了解Java層之前,我們先來看JVM是如何向Java層暴露這些操作的.

Java層無法直接調用CPU指令,必須借助JNI,這里對CAS的調用在Java層就體現在sun.misc.Unsafe類上,UnSafe類中定義了很多Native方法:

public final class Unsafe { private static native void registerNatives(); static { registerNatives(); } private Unsafe() {} private static final Unsafe theUnsafe = new Unsafe(); public final native boolean compareAndSetObject(Object o, long offset, Object expected, Object x); public final native boolean compareAndSetInt(Object o, long offset, int expected, int x); ....... } 

其對應的C++實現類是:

/OpenJDK10/hotspot/src/share/vm/prims/unsafe.cpp,這里的compareAndSetInt()其實就對應於unsafe.cpp中的Unsafe_CompareAndSetInt():

UNSAFE_ENTRY(jboolean, Unsafe_CompareAndSetInt(JNIEnv *env, jobject unsafe, jobject obj, jlong offset, jint e, jint x)) { oop p = JNIHandles::resolve(obj); jint* addr = (jint *)index_oop_from_field_offset_long(p, offset); // 調用Atomic.cpp中的cmpxchg() return (jint)(Atomic::cmpxchg(x, addr, e)) == e; } UNSAFE_END 

總結一下,sun.misc.Unsafe中Native方法的調用,最終都會通過JNI調用到unsafe.cpp中,而unsafe.cpp中的實現本質都是調用CPU的cmpxchg指令.關於cmpxchg指令將在后續單獨說明.

到現在為止,JVM如何實現CAS以及如何向Java層暴露CAS操作這兩個流程已經比較明了了,接下來還是要回歸到Java層,來明白Java層中對CAS的支持.

Java層對CAS的支持

在Java層面,原子變量類(java.util.concurrent.atomic中的AtomicXXX)在底層充分使用了來此JVM對CAS的支持,來實現高效的原子操作,此外,java.util.concurrent中的大多數類在實現時也是借助了這些原子變量類.以AtomicInteger為例,來了解下Atomic如何使用CAS.

public class AtomicInteger extends Number implements java.io.Serializable { private static final long serialVersionUID = 6214790243416807050L; // Unsafe類型的成員變量U private static final jdk.internal.misc.Unsafe U = jdk.internal.misc.Unsafe.getUnsafe(); private static final long VALUE = U.objectFieldOffset(AtomicInteger.class, "value"); private volatile int value; public AtomicInteger(int initialValue) { value = initialValue; } public AtomicInteger() { } public final int get() { return value; } public final boolean compareAndSet(int expectedValue, int newValue) { return U.compareAndSetInt(this, VALUE, expectedValue, newValue); } ...... } 

通過上述代碼不難看出,AtomicInteger本質就是CAS在Java層的應用.在明白CAS原理后,對AtomicInteger並不會感到難以理解.可以說正是有了CPU對Compare-and-Swap的支持,才使得Java在並發上有了突飛猛進的提升.另外在不支持Compare-and-Swap的平台上,JVM將使用自旋鎖來代替.

CAS缺陷

ABA問題是CAS中是一種常見的問題:在於並發環境下,當第一個線程執行CAS(V,A,B)操作,在已經獲取到當前變量V，但還沒將其修改為新值B前，其他線程在其期間連續修改了兩次變量V的值，使得該值又恢復為舊值.在這種情況下,我們無法判斷這個變量是否已被修改過.其流程如下:

 

image-20180910153730654

在大多數情況下,ABA問題並不會影響最終的計算結果,但如果需要避免ABA問題該怎么辦呢?從上述流程看出導致ABA的問題個根源是在時間線推進過程中沒有為每次修改記錄版本號導致.解決該問題只需要在每次修改時記錄下其當前時間戳作為版本號就可以避免.在Java中,AtomicStampedReference原子類實現原理就是如此:設置值時要求對象值以及時間戳都必須滿足期望值才能寫入成功.(可以理解為git中的commit-id,先將A修改為B,再修改成A,最終內容雖然沒變,但通過版本commit-id,我們仍然可以知道中間發生了變化)

除了ABA問題外,如果自旋CAS長時間失敗會給CPU帶來很大的開銷,在並發激烈的時候該問題尤其明顯.另外,CAS使用場景比較單一,只能用於保證一個共享變量的原子操作.

作者：涅槃1992
鏈接：https://www.jianshu.com/p/f009da2e4110
來源：簡書
著作權歸作者所有。商業轉載請聯系作者獲得授權，非商業轉載請注明出處。