Golang Mutex互斥鎖原始碼分析

// 互斥鎖
type Mutex struct {
 state int32  // 狀態
 sema  uint32  // 號誌
}


const (
 mutexLocked = 1 << iota // 1
 mutexWoken // 2
 mutexStarving // 4
 mutexWaiterShift = iota // 3

 starvationThresholdNs = 1e6 // 判斷是否要進入飢餓狀態的閾值
)

// 上鎖
func (m *Mutex) Lock() {
 // fastpath：期望當前鎖沒有被佔用，可以快速獲取到鎖， CAS 修改 state 最後一位的值為1（標記鎖是否被佔用）
 if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
  return
 }
 // Slow path : 單獨抽出來放到一個函數裡，方便 fast path 被內聯
 m.lockSlow()
}

func (m *Mutex) lockSlow() {
 var waitStartTime int64 // // 記錄等待時間
 starving := false // 當前的 goroutine 是否已經飢餓了（如果已經飢餓，就會將 state 的飢餓狀態置為 1）
 awoke := false  // 當前的 goroutine 是否被喚醒的
 iter := 0 // 自旋次數
 old := m.state // 儲存當前的 state 狀態
 for {
    
    /*
    自旋：如果滿足如下條件，就會進入 if 語句，然後 continue，不斷自旋:
    1. 鎖被佔用，且不處於飢餓模式（飢餓狀態直接去排隊，不允許嘗試獲取鎖）
    2. 基於當前自旋的次數，再次自旋有意義 runtime_canSpin(iter)
    
    那麼退出自旋的條件也就是：
    1. 鎖被釋放了，當前處於沒被佔用狀態（說明等到了，該goroutine就會立即去獲取鎖）
  2. mutex進入了飢餓模式，不自旋了，沒意義（飢餓狀態會直接把鎖交給等待佇列隊首的goroutine）
  3. 不符合自旋狀態（自旋次數太多了，自旋失去了意義）
    
    如下程式碼是位元運算：
    mutexLocked|mutexStarving = 00000...101
    mutexLocked = 00000...001
    如果要滿足 old & 00000...101 = 00000...001，需要 old = ...0*1,即狀態為：鎖被佔用，且不處於飢餓狀態 
    
    runtime_canSpin(iter) 會根據自旋次數，判斷是否可以繼續自旋
    */
  if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
   
      /*
      如果 
       1. 當前goroutine不是被喚醒的 (awoke=false) 
       2. 鎖狀態喚醒標誌位為0（old&mutexWoken == 0） 
       3. 等待者數量不為0 （old>>mutexWaiterShift != 0  右移三位得到的就是等待者數量）
      
       那麼利用CAS，將 state 的喚醒標記置為1，標記自己是被喚醒的 （將state的喚醒標記置為1，說明外面有喚醒著的goroutine，那麼在釋放鎖的時候，就不去等待佇列叫號了，畢竟已經有喚醒的了）
       
       如果有其他 goroutine 已經設定了 state 的喚醒標記位，那麼本次就會失敗
      */

   if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
    atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
    awoke = true
   }
   runtime_doSpin()
      
      // 迭代次數加一
   iter++
      
      // 獲取最新的狀態
   old = m.state
      
      // 想再次自旋，看看鎖釋放了沒
   continue
  }
    
    // 到這裡，說明退出了自旋，當前鎖沒被佔用 或者  系統處於飢餓模式 或者 自旋次數太多導致不符合自旋條件
  
    // new 代表當前goroutine 基於當前狀態要設定的新狀態
  new := old
    
    // 只要不是飢餓狀態，就需要獲取鎖（飢餓狀態直接去排隊，不能搶鎖）
  if old&mutexStarving == 0 {
   new |= mutexLocked
  }
    
    // 鎖被佔用 或者 處於飢餓模式下，新增一個等待者
  if old&(mutexLocked|mutexStarving) != 0 {
   new += 1 << mutexWaiterShift
  }
    
    // 當前 goroutine 已經進入飢餓了，且鎖還沒有釋放，需要把 Mutex 的狀態改為飢餓狀態
  if starving && old&mutexLocked != 0 {
   new |= mutexStarving
  }
    
    // 如果是被喚醒的，把喚醒標誌位置0，表示外面沒有被喚醒的goroutine了（搶到就獲得鎖、搶不到就睡眠，把喚醒標誌置0）
  if awoke {
   
      // 由於是被喚醒的，new 裡面的 喚醒標記位一定是 1
   if new&mutexWoken == 0 {
    throw("sync: inconsistent mutex state")
   }
      
      // a &^ b 的意思就是 清零a中，ab都為1的位，即清除喚醒標記
   new &^= mutexWoken
  }
    
    /*
      利用CAS,將狀態設定為新的
      1. 如果是飢餓狀態，只增加一個等待者數量
      2. 正常狀態，加鎖標記置為 1，如果鎖已被佔用增加一個等待者數量
      3. 如果當前 goroutine 已經飢餓了，將 飢餓標記 置為 1
      4. 如果是被喚醒的，清除喚醒標記
    */
    
  if atomic.CompareAndSwapInt32(&m.state, old, new) {
      
      // 如果改狀態之前，鎖未被佔用 且 處於正常模式，那麼就相當於獲取到鎖了
   if old&(mutexLocked|mutexStarving) == 0 {
    break 
   }
      
      // 到這裡說明：1. 之前鎖被佔用  或者 2.之前是處於飢餓狀態 
      
      // 判斷之前是否等待過（是否從佇列裡喚醒的），之前等待過，再次排隊放在隊首
   queueLifo := waitStartTime != 0
      
      // 如果之前沒等過（新來的），設定等待起始時間當前時間
   if waitStartTime == 0 {
    waitStartTime = runtime_nanotime()
   }
      
     // 之前排過隊的老人，放到等待佇列隊首；新人放到隊尾，然後等待獲取號誌
   runtime_SemacquireMutex(&m.sema, queueLifo, 1)
      
      // 鎖被釋放，goroutine 被喚醒
      
      // 設定當前 goroutine 飢餓狀態，如果之前已經飢餓，或者距離等待開始時間超過了 1ms，也變飢餓
   starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
      
      // 獲取最新的狀態
   old = m.state
      
      // 如果 state 飢餓標記為1，說明當前在飢餓模式，飢餓模式下被喚醒，已經獲取到鎖了；
      // 飢餓狀態下，釋放鎖沒有更新等待者數量和飢餓標記，需要獲得鎖的goroutine去更新狀態
   if old&mutexStarving != 0 {

        // 正確性校驗：
        // 1. 鎖還是鎖住的狀態（鎖已經釋放給當前goroutine了，不應該被鎖住）
    // 2. 或者有被喚醒的goroutine（飢餓模式下不應該有醒著的goroutine，都應該去乖乖等著）
    // 3. 或者當前goroutine 的等待者數量為0（當前goroutine就是等待者）
    // 這三種情況不應該出現，與預期狀態不符
        
    if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
     throw("sync: inconsistent mutex state")
    }
        
        // 加鎖，減去一個等待者
    delta := int32(mutexLocked - 1<<mutexWaiterShift)
        
        // 如果當前的 goroutine 非飢餓，或者等待者只有一個（也就是隻有當前goroutine，等待佇列空了），可以取消飢餓狀態，進入正常狀態
    if !starving || old>>mutexWaiterShift == 1 {
     delta -= mutexStarving
    }
        
        // 修改狀態：
        // 加鎖，減去一個等待者：m.state + mutexLocked - 1<<mutexWaiterShift : 
        // 滿足非飢餓條件，加鎖，減去一個等待者，取消飢餓狀態：
        // m.state + mutexLocked - 1<<mutexWaiterShift - mutexStarving: 
    atomic.AddInt32(&m.state, delta)
        
        // 飢餓模式下被喚醒，相當於獲得鎖了，可以結束
    break
   }
      
      // 之前是處於鎖被佔用且非飢餓狀態，被喚醒，去繼續搶鎖
   awoke = true
      
      // 新喚醒的，自旋數量置0
   iter = 0
  } else {
      
      // 修改新狀態失敗，狀態有更新，需要重試
   old = m.state
  }
 }
}

func (m *Mutex) Unlock() {

 // Fast path: 把鎖標記清除
 new := atomic.AddInt32(&m.state, -mutexLocked)
 if new != 0 {
  // 清除完鎖標記，發現還有其他狀態，比如等待佇列不為空，需要喚醒其他 goroutine
  m.unlockSlow(new)
 }
}

func (m *Mutex) unlockSlow(new int32) {
  
  /* 狀態正確性校驗：
    1. 如果解鎖一個上鎖狀態的鎖，最後一位則為1，fast path 中 new 已經減去了1, 此時 new 最後一位應當為0
   2. 如果解鎖一個未上鎖狀態的鎖，最後一位則為0，fast path 中 new 已經減去了1, 此時 new 最後一位應當為1
    如果 (new+mutexLocked)&mutexLocked == 0，說明 new 當前最後一位是1，那麼就是解鎖了一個沒有上鎖的鎖，狀態有誤
 */
 if (new+mutexLocked)&mutexLocked == 0 {
  throw("sync: unlock of unlocked mutex")
 }
  
  // 正常模式，非飢餓，可能需要喚醒佇列中的 goroutine，飢餓狀態直接移交鎖
 if new&mutexStarving == 0 {
  old := new
  for {

      /* 系統運轉正常，鎖可以正確交接，可以直接返回了：
        1. 沒有等待者了 (沒有等鎖的了，去喚醒誰？)
        2. 有喚醒狀態的 goroutine  (自旋狀態的 goroutine，將喚醒狀態置為1)
       3. 有 goroutine 已經獲取了鎖 (Unlock方法已經將鎖標記置為了0，可能自旋的此時已經搶到了鎖)
      */
   if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
    return
   }
      
      // 沒有喚醒狀態的 goroutine，喚醒一個去搶鎖
   // 減去一個等待者，並且將 喚醒標記 置為 1
   new = (old - 1<<mutexWaiterShift) | mutexWoken
   if atomic.CompareAndSwapInt32(&m.state, old, new) {
        // 第二個引數為false, 喚醒隊首的 goroutine 去搶鎖(不一定能搶到)
    runtime_Semrelease(&m.sema, false, 1)
    return
   }
      
      // 上面 CAS 失敗，可能由於新增了一個等待者，for 迴圈重試
   old = m.state
  }
 } else {
  
    /*
     1. 第二個引數為 true，直接將鎖的所有權，交給等待佇列的第一個等待者
   2. 注意，此時沒有設定 mutexLocked =1 ，被喚醒的 goroutine 會設定
   3. 雖然沒有設定 mutexLocked ，但是飢餓模式下, Mutex 始終被認為是鎖住的，都會直接排隊等待移交鎖
    */
  runtime_Semrelease(&m.sema, true, 1)
 }
}