不得不知道的golang之sync.Mutex互斥鎖源碼分析

針對Golang 1.9的sync.Mutex進行分析，與Golang 1.10基本一樣除了將panic改為了throw之外其他的都一樣。
源代碼位置：sync\mutex.go。
可以看到注釋如下:

Mutex can be in 2 modes of operations: normal and starvation.
 In normal mode waiters are queued in FIFO order, but a woken up waiter does not own the mutex and competes with new arriving goroutines over the ownership. New arriving goroutines have an advantage -- they are already running on CPU and there can be lots of them, so a woken up waiter has good chances of losing. In such case it is queued at front of the wait queue. If a waiter fails to acquire the mutex for more than 1ms, it switches mutex to the starvation mode.

In starvation mode ownership of the mutex is directly handed off from the unlocking goroutine to the waiter at the front of the queue. New arriving goroutines don't try to acquire the mutex even if it appears  to be unlocked, and don't try to spin. Instead they queue themselves at  the tail of the wait queue.

If a waiter receives ownership of the mutex and sees that either (1) it is the last waiter in the queue, or (2) it waited for less than 1 ms, it switches mutex back to normal operation mode.

 Normal mode has considerably better performance as a goroutine can acquire a mutex several times in a row even if there are blocked waiters.
Starvation mode is important to prevent pathological cases of tail latency.

博主英文很爛，就粗略翻譯一下，僅供參考：

互斥量可分為兩種操作模式:正常和饑餓。
在正常模式下，等待的goroutines按照FIFO（先進先出）順序排隊，但是goroutine被喚醒之后并不能立即得到mutex鎖，它需要與新到達的goroutine爭奪mutex鎖。
因為新到達的goroutine已經在CPU上運行了，所以被喚醒的goroutine很大概率是爭奪mutex鎖是失敗的。出現這樣的情況時候，被喚醒的goroutine需要排隊在隊列的前面。
如果被喚醒的goroutine有超過1ms沒有獲取到mutex鎖，那么它就會變為饑餓模式。
在饑餓模式中，mutex鎖直接從解鎖的goroutine交給隊列前面的goroutine。新達到的goroutine也不會去爭奪mutex鎖（即使沒有鎖，也不能去自旋），而是到等待隊列尾部排隊。
在饑餓模式下，有一個goroutine獲取到mutex鎖了，如果它滿足下條件中的任意一個，mutex將會切換回去正常模式：
1. 是等待隊列中的最后一個goroutine
2. 它的等待時間不超過1ms。
正常模式有更好的性能，因為goroutine可以連續多次獲得mutex鎖；
饑餓模式對于預防隊列尾部goroutine一致無法獲取mutex鎖的問題。

看了這段解釋，那么基本的業務邏輯也就了解了，可以整理一下衣裝，準備看代碼。

打開mutex.go看到如下代碼:

type Mutex struct {
    state int32    // 將一個32位整數拆分為 當前阻塞的goroutine數(29位)|饑餓狀態(1位)|喚醒狀態(1位)|鎖狀態(1位) 的形式，來簡化字段設計
    sema  uint32   // 信號量
}

const (
    mutexLocked = 1 << iota      // 1 0001 含義：用最后一位表示當前對象鎖的狀態，0-未鎖住 1-已鎖住
    mutexWoken                   // 2 0010 含義：用倒數第二位表示當前對象是否被喚醒 0-喚醒 1-未喚醒
    mutexStarving                // 4 0100 含義：用倒數第三位表示當前對象是否為饑餓模式，0為正常模式，1為饑餓模式。
    mutexWaiterShift = iota      // 3，從倒數第四位往前的bit位表示在排隊等待的goroutine數
    starvationThresholdNs = 1e6  // 1ms
)

可以看到Mutex中含有：

• 一個非負數信號量sema；
• state表示Mutex的狀態。

常量：

• mutexLocked表示鎖是否可用（0可用，1被別的goroutine占用）
• mutexWoken=2表示mutex是否被喚醒
• mutexWaiterShift=4表示統計阻塞在該mutex上的goroutine數目需要移位的數值。

將3個常量映射到state上就是

state:   |32|31|...| |3|2|1|
         \__________/ | | |
              |       | | |
              |       | |  mutex的占用狀態（1被占用，0可用）
              |       | |
              |       |  mutex的當前goroutine是否被喚醒
              |       |
              |       饑餓位，0正常，1饑餓
              |
               等待喚醒以嘗試鎖定的goroutine的計數，0表示沒有等待者

如果同學們熟悉Java的鎖，就會發現與AQS的設計是類似，只是沒有AQS設計的那么精致，不得不感嘆，JAVA的牛逼。
有同學是否會有疑問為什么使用的是int32而不是int64呢，因為32位原子性操作更好,當然也滿足的需求。

Mutex在1.9版本中就兩個函數Lock()和Unlock()。
下面我們先來分析最難的Lock()函數：

func (m *Mutex) Lock() {
    // 如果m.state=0，說明當前的對象還沒有被鎖住，進行原子性賦值操作設置為mutexLocked狀態，CompareAnSwapInt32返回true
    // 否則說明對象已被其他goroutine鎖住，不會進行原子賦值操作設置，CopareAndSwapInt32返回false
    if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) 
        if race.Enabled {
            race.Acquire(unsafe.Pointer(m))
        }
        return
    }

    // 開始等待時間戳
    var waitStartTime int64
    // 饑餓模式標識
    starving := false
    // 喚醒標識
    awoke := false
    // 自旋次數
    iter := 0
    // 保存當前對象鎖狀態
    old := m.state
    // 看到這個for {}說明使用了cas算法
    for {
        // 相當于xxxx...x0xx & 0101 = 01，當前對象鎖被使用
        if old&(mutexLocked|mutexStarving) == mutexLocked && 
            // 判斷當前goroutine是否可以進入自旋鎖
            runtime_canSpin(iter) {

            // 主動旋轉是有意義的。試著設置mutexwake標志，告知解鎖，不要喚醒其他阻塞的goroutines。
            if !awoke &&
            // 再次確定是否被喚醒： xxxx...xx0x & 0010 = 0
            old&mutexWoken == 0 &&
            // 查看是否有goroution在排隊
            old>>mutexWaiterShift != 0 &&
                // 將對象鎖改為喚醒狀態：xxxx...xx0x | 0010 = xxxx...xx1x 
                atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
                awoke = true
            }//END_IF_Lock

            // 進入自旋鎖后當前goroutine并不掛起，仍然在占用cpu資源，所以重試一定次數后，不會再進入自旋鎖邏輯
            runtime_doSpin()
            // 自加，表示自旋次數
            iter++
            // 保存mutex對象即將被設置成的狀態
            old = m.state
            continue
        }// END_IF_spin

        // 以下代碼是不使用**自旋**的情況
        new := old

        // 不要試圖獲得饑餓的互斥，新來的goroutines必須排隊。
        // 對象鎖饑餓位被改變，說明處于饑餓模式
        // xxxx...x0xx & 0100 = 0xxxx...x0xx
        if old&mutexStarving == 0 {
            // xxxx...x0xx | 0001 = xxxx...x0x1，標識對象鎖被鎖住
            new |= mutexLocked
        }
        // xxxx...x1x1 & (0001 | 0100) => xxxx...x1x1 & 0101 != 0;當前mutex處于饑餓模式并且鎖已被占用，新加入進來的goroutine放到隊列后面
        if old&(mutexLocked|mutexStarving) != 0 {
            // 更新阻塞goroutine的數量,表示mutex的等待goroutine數目加1
            new += 1 << mutexWaiterShift
        }

        // 當前的goroutine將互斥鎖轉換為饑餓模式。但是，如果互斥鎖當前沒有解鎖，就不要打開開關,設置mutex狀態為饑餓模式。Unlock預期有饑餓的goroutine
        if starving && 
            // xxxx...xxx1 & 0001 != 0；鎖已經被占用
            old&mutexLocked != 0 {
            // xxxx...xxx | 0101 =>   xxxx...x1x1，標識對象鎖被鎖住
            new |= mutexStarving
        }

        // goroutine已經被喚醒，因此需要在兩種情況下重設標志
        if awoke {
            // xxxx...xx1x & 0010 = 0,如果喚醒標志為與awoke不相協調就panic
            if new&mutexWoken == 0 {
                panic("sync: inconsistent mutex state")
            }
            // new & (^mutexWoken) => xxxx...xxxx & (^0010) => xxxx...xxxx & 1101 = xxxx...xx0x  ：設置喚醒狀態位0,被喚醒
            new &^= mutexWoken
        }
        // 獲取鎖成功
        if atomic.CompareAndSwapInt32(&m.state, old, new) {
            // xxxx...x0x0 & 0101 = 0，已經獲取對象鎖
            if old&(mutexLocked|mutexStarving) == 0 {
                // 結束cas
                break
            }
            // 以下的操作都是為了判斷是否從饑餓模式中恢復為正常模式
            // 判斷處于FIFO還是LIFO模式
            queueLifo := waitStartTime != 0
            if waitStartTime == 0 {
                waitStartTime = runtime_nanotime()
            }
            runtime_SemacquireMutex(&m.sema, queueLifo)
            starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
            old = m.state
            // xxxx...x1xx & 0100 != 0
            if old&mutexStarving != 0 {
                // xxxx...xx11 & 0011 != 0
                if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
                    panic("sync: inconsistent mutex state")
                }
                delta := int32(mutexLocked - 1<<mutexWaiterShift)
                if !starving || old>>mutexWaiterShift == 1 {
                    delta -= mutexStarving
                }
                atomic.AddInt32(&m.state, delta)
                break
            }
            awoke = true
            iter = 0
        } else {
            // 保存mutex對象狀態
            old = m.state
        }
    }// cas結束

    if race.Enabled {
        race.Acquire(unsafe.Pointer(m))
    }
}

看了Lock()函數之后是不是覺得一片懵逼狀態，告訴大家一個方法，看Lock()函數時候需要想著如何Unlock。下面就開始看看Unlock()函數。

func (m *Mutex) Unlock() {
    if race.Enabled {
        _ = m.state
        race.Release(unsafe.Pointer(m))
    }

    // state-1標識解鎖
    new := atomic.AddInt32(&m.state, -mutexLocked)
    // 驗證鎖狀態是否符合
    if (new+mutexLocked)&mutexLocked == 0 {
        panic("sync: unlock of unlocked mutex")
    }
    // xxxx...x0xx & 0100 = 0 ;判斷是否處于正常模式
    if new&mutexStarving == 0 {
        old := new
        for {
            // 如果沒有等待的goroutine或goroutine已經解鎖完成
            if old>>mutexWaiterShift == 0 || 
            // xxxx...x0xx & (0001 | 0010 | 0100) => xxxx...x0xx & 0111 != 0
            old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
                return
            }
            // Grab the right to wake someone.
            new = (old - 1<<mutexWaiterShift) | mutexWoken
            if atomic.CompareAndSwapInt32(&m.state, old, new) {
                runtime_Semrelease(&m.sema, false)
                return
            }
            old = m.state
        }
    } else {
        // 饑餓模式:將mutex所有權移交給下一個等待的goroutine
        // 注意:mutexlock沒有設置，goroutine會在喚醒后設置。
        // 但是互斥鎖仍然被認為是鎖定的，如果互斥對象被設置，所以新來的goroutines不會得到它
        runtime_Semrelease(&m.sema, true)
    }
}

在網上還會有一些基于go1.6的分析，但是與go 1.9的差距有點大。
上面的分析，因個人水平有限，難免存在錯誤，請各位老師同學多多指點，不喜勿噴。

附錄

https://github.com/golang/go/blob/dev.boringcrypto.go1.9/src/sync/mutex.go
https://segmentfault.com/a/1190000000506960