Java 中自旋鎖的實現

Java中初始是使用mutex互斥鎖，因為互斥鎖是會線程等待掛起，而對獲取鎖後的操作時間比較短暫的應用場景來說，這樣的鎖會讓競爭鎖的線程不停的park,unpark 的操作，這樣的系統的調用性能是非常糟糕的，為了提高鎖的性能，java 在6 默認使用了自旋鎖。

在Linux中本身就已經提供了自旋鎖的系統調用，在glibc-2.9中就有它的比較簡單的實現方法

1. int pthread_spin_lock (lock) pthread_spinlock_t *lock;
2. {
3. asm ("\n"
4. "1:\t" LOCK_PREFIX "decl %0\n\t"
5. "jne 2f\n\t"
6. ".subsection 1\n\t"
7. ".align 16\n"
8. "2:\trep; nop\n\t"
9. "cmpl $0, %0\n\t"
10. "jg 1b\n\t"
11. "jmp 2b\n\t"
12. ".previous"
13. : "=m" (*lock)
14. : "m" (*lock));
15. return 0;
16. }

Java 並沒有使用系統自帶的自旋鎖，自己重寫了自旋鎖的邏輯，並且增加了自旋的次數的控制。詳細見-XX:+UseSpinning 和 -XX:PreBlockSpin=xx

讓我們具體來看是如何實現的，注意這是mutex鎖中所實現的lock，而並不是synchinized 的鎖的spin lock的實現（這個你可以參考synchronizer.cpp裡的方法TrySpin_VaryDuration）

1. int Monitor::TrySpin (Thread * const Self) {
2. if (TryLock()) return 1 ;
3. if (!os::is_MP()) return 0 ;
5. int Probes = 0 ;
6. int Delay = 0 ;
7. int Steps = 0 ;
8. int SpinMax = NativeMonitorSpinLimit ;
9. int flgs = NativeMonitorFlags ;
10. for (;;) {
11. intptr_t v = _LockWord.FullWord;
12. if ((v & _LBIT) == 0) {
13. if (CASPTR (&_LockWord, v, v|_LBIT) == v) {
14. return 1 ;
15. }
16. continue ;
17. }
19. if ((flgs & 8) == 0) {
20. SpinPause () ;
21. }
23. // Periodically increase Delay -- variable Delay form
24. // conceptually: delay *= 1 + 1/Exponent
25. ++ Probes;
26. if (Probes > SpinMax) return 0 ;
28. if ((Probes & 0x7) == 0) {
29. Delay = ((Delay << 1)|1) & 0x7FF ;
30. // CONSIDER: Delay += 1 + (Delay/4); Delay &= 0x7FF ;
31. }
33. if (flgs & 2) continue ;
35. // Consider checking _owner's schedctl state, if OFFPROC abort spin.
36. // If the owner is OFFPROC then it's unlike that the lock will be dropped
37. // in a timely fashion, which suggests that spinning would not be fruitful
38. // or profitable.
40. // Stall for "Delay" time units - iterations in the current implementation.
41. // Avoid generating coherency traffic while stalled.
42. // Possible ways to delay:
43. // PAUSE, SLEEP, MEMBAR #sync, MEMBAR #halt,
44. // wr %g0,%asi, gethrtime, rdstick, rdtick, rdtsc, etc. ...
45. // Note that on Niagara-class systems we want to minimize STs in the
46. // spin loop. N1 and brethren write-around the L1$ over the xbar into the L2$.
47. // Furthermore, they don't have a W$ like traditional SPARC processors.
48. // We currently use a Marsaglia Shift-Xor RNG loop.
49. Steps += Delay ;
50. if (Self != NULL) {
51. jint rv = Self->rng[0] ;
52. for (int k = Delay ; --k >= 0; ) {
53. rv = MarsagliaXORV (rv) ;
54. if ((flgs & 4) == 0 && SafepointSynchronize::do_call_back()) return 0 ;
55. }
56. Self->rng[0] = rv ;
57. } else {
58. Stall (Delay) ;
59. }
60. }
61. }

a. os::is_MP() 判斷系統是否是多核的系統，在單核下，自旋鎖是沒有意義的。

b. CASPTR 使用了 Atomic::cmpxchg_ptr 原子語義 cmpxchg 比較替換，如果比較的值相等就替換成需要的值並且返回去比較的值，如果不相同返回被比較的值的內容。

在這裡的語義是比較_LockWord.FullWord 和 _Lockword 的值是否相同，如果相同就把_Lockword 的值置換成v|_LBIT（_LBIT的值是1）。

自旋鎖的邏輯：判斷_LockWord.FullWord bit 0 是否是0，如果是0代表沒有占有鎖，那就嘗試去占有鎖，通過原子替換置bit0 為1，如果置換成功那麼代表擁有鎖，沒有則進入自旋。

SpinPause () 函數
在linux_x86 64位機器上 定義了
.globl SpinPause
.align 16
.type SpinPause,@function
SpinPause:
rep
nop
movq $1, %rax
ret

主要在rep, nop 的指令經過編譯器後的指令是pause,是用於提高cpu性能的，在官方上描述pase指令是為了避免memory order violation ，有種說法就是cpu是流水線的處理指令的，當原子指令store的時候，而如果有線程同時也在load他的值，那麼load 必須等到store 執行成功,這樣cpu就無法進行流水線作業了。但我更覺的這是個加強版的nop 也就是多增加幾個空的機器周期，一來省電，二來本身spin lock就需要cpu空運行，並且不需要訪問內存。

c. SafepointSynchronize::do_call_back()這是一個安全點，提供一個停止自旋鎖的切入點，比如vm thread,在做線程dump, 內存 dump的時候，是需要讓 自旋鎖提前停止的。

d. if (Probes > SpinMax) return 0 ; 當大於自旋的次數的時候，自旋自動退出,也就是前面所說的參數 -XX:PreBlockSpin

最後這裡還有個比較有意思的方法MarsagliaXORV (rv) ; 是算隨機數的，不清楚為什麼java讓cpu自旋的過程中計算隨機數的意義何在，為了不讓cpu空轉？感覺用spinpase 更合理一點。