golang 互斥鎖

go語言提供了一種開箱即用的共享資源的方式，互斥鎖(sync.Mutex), sync.Mutex的零值表示一個沒有被鎖的，可以直接使用的，一個goroutine獲得互斥鎖后其他的goroutine只能等到這個gorutine釋放該互斥鎖,在Mutex結構中只公開了兩個函數，分別是Lock和Unlock，在使用互斥鎖的時候非常簡單，本文並不闡述使用。

在使用sync.Mutex的時候千萬不要做值拷貝，因為這樣可能會導致鎖失效。當我們打開我們的IDE時候跳到我們的sync.Mutex 代碼中會發現它有如下的結構:

type Mutex struct {
 state int32   //互斥鎖上鎖狀態枚舉值如下所示
 sema uint32  //信號量，向處於Gwaitting的G發送信號
}
 
const (
 mutexLocked = 1 << iota // 1 互斥鎖是鎖定的
 mutexWoken       // 2 喚醒鎖
 mutexWaiterShift = iota // 2 統計阻塞在這個互斥鎖上的goroutine數目需要移位的數值
)

上面的state值分別為 0(可用) 1(被鎖) 2~31等待隊列計數

下面是互斥鎖的源碼，這里會有四個比較重要的方法需要提前解釋，分別是runtime_canSpin，runtime_doSpin，runtime_SemacquireMutex，runtime_Semrelease，

1、runtime_canSpin：比較保守的自旋，golang中自旋鎖並不會一直自旋下去，在runtime包中runtime_canSpin方法做了一些限制, 傳遞過來的iter大等於4或者cpu核數小等於1，最大邏輯處理器大於1，至少有個本地的P隊列，並且本地的P隊列可運行G隊列為空。

//go:linkname sync_runtime_canSpin sync.runtime_canSpin
func sync_runtime_canSpin(i int) bool {
 if i >= active_spin || ncpu <= 1 || gomaxprocs <= int32(sched.npidle+sched.nmspinning)+1 {
 return false
 }
 if p := getg().m.p.ptr(); !runqempty(p) {
 return false
 }
 return true
}

2、 runtime_doSpin：會調用procyield函數(讓出線程)，該函數也是匯編語言實現。函數內部循環調用PAUSE指令。PAUSE指令什么都不做，但是會消耗CPU時間，在執行PAUSE指令時，CPU不會對它做不必要的優化。

//go:linkname sync_runtime_doSpin sync.runtime_doSpin
func sync_runtime_doSpin() {
 procyield(active_spin_cnt)
}

3、runtime_SemacquireMutex:

//go:linkname sync_runtime_SemacquireMutex sync.runtime_SemacquireMutex
func sync_runtime_SemacquireMutex(addr *uint32) {
 semacquire(addr, semaBlockProfile|semaMutexProfile)
}

4、runtime_Semrelease:

//go:linkname sync_runtime_Semrelease sync.runtime_Semrelease
func sync_runtime_Semrelease(addr *uint32) {
	semrelease(addr)
}

Lock函數定義：

func (m *Mutex) Lock() {
    //先使用CAS嘗試獲取鎖
    if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
    //這里是-race不需要管它
        if race.Enabled {
            race.Acquire(unsafe.Pointer(m))
        }
    //成功獲取返回
    return
    }

    awoke := false  //循環標記
    iter := 0       //循環計數器
    for {
        old := m.state //獲取當前鎖狀態
        new := old | mutexLocked  //將當前狀態最后一位指定1
        if old&mutexLocked != 0 { //如果鎖被占用
            if runtime_canSpin(iter) { //檢查是否可以進入自旋鎖
                if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
                    atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
                    awoke = true   //awoke標記為true
                }
                //進入自旋狀態
                runtime_doSpin()
                iter++
                continue
            }
            //沒有獲取到鎖，當前G進入Gwaitting狀態,,waiting 序列加1
            new = old + 1<<mutexWaiterShift
        }
        if awoke {
            if new&mutexWoken == 0 {
                throw("sync: inconsistent mutex state")
            }
            //清除標記
            new &^= mutexWoken
        }
        //更新狀態
        if atomic.CompareAndSwapInt32(&m.state, old, new) {
            if old&mutexLocked == 0 {
                break
            }

            // 鎖請求失敗,進入休眠狀態,等待信號喚醒后重新開始循環
            runtime_SemacquireMutex(&m.sema)
            awoke = true
            iter = 0
        }
    }

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

Unlock函數定義：

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

    // 移除標記
    new := atomic.AddInt32(&m.state, -mutexLocked)
    if (new+mutexLocked)&mutexLocked == 0 {
        throw("sync: unlock of unlocked mutex")
    }

    old := new
    for {
    //當休眠隊列內的等待計數為0或者自旋狀態計數器為0，退出
        if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken) != 0 {
            return
        }
        // 減少等待次數，添加清除標記
        new = (old - 1<<mutexWaiterShift) | mutexWoken
        if atomic.CompareAndSwapInt32(&m.state, old, new) {
            // 釋放鎖,發送釋放信號
            runtime_Semrelease(&m.sema)
            return
        }
        old = m.state
    }
}

互斥鎖無沖突是最簡單的情況了，有沖突時，首先進行自旋，因為大多數的Mutex保護的代碼段都很短，經過短暫的自旋就可以獲得；如果自旋等待無果，就只好通過信號量來讓當前Goroutine進入Gwaitting狀態。

轉載自 https://zhuanlan.zhihu.com/p/27608263