Go互斥锁源码解析

Go开发大全

【导读】本文介绍了CAS理论和Mutex包中锁的实现。

我们在进行并发编程的时候，往往离不开多个并发操作修改同一份资源，当然，离不开锁这个服务。

锁是在执行并发操作时用于强行限制资源访问的同步机制，即用于在并发控制中保证对互斥要求的满足。

Go语言在Mutex包中提供了互斥锁，sync.Mutex

Go语言的sync.Mutex由两个字段组成

type Mutex struct {
  state int32
  sema  int32
}

其中Mutex.state表示了当前互斥锁处于的状态

waiterNum 表示目前互斥锁等待队列中有多少goroutine在等待

straving 表示目前互斥锁是否处于饥饿状态

woken 表示目前互斥锁是否为唤醒状态

locked 表示目前互斥锁资源是否被goroutine持有

而Mutex.sema主要用于等待队列

互斥锁的状态

互斥锁通常保持两种状态 正常模式 与 饥饿模式
引入饥饿模式的原因是，为了保持互斥锁的公平性。
在正常模式下，锁资源一般会交给刚被唤醒的goroutine，而为了怕部分goroutine被“饿死”，所以引入了饥饿模式，在饥饿模式下，goroutine在释放锁资源的时候会将锁资源交给等待队列中的下一个goroutine。

加锁逻辑

当我们在执行Mutex.Lock()，会执行以下逻辑

func (m *Mutex) Lock() {
   // Fast path: grab unlocked mutex.
   if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
      if race.Enabled {
         race.Acquire(unsafe.Pointer(m))
      }
      return
   }
   // Slow path (outlined so that the fast path can be inlined)
   m.lockSlow()
}

首先使用CAS方法判断是否可以直接获取锁资源

CAS指的是CompareAndSwap，也就是如果可以获得锁资源，则修改Mutex.state中的locked位，并成功获取，如果获取不到，则执行lowSlow()方法

比较并交换(compare and swap, CAS)，是原子操作的一种，可用于在多线程编程中实现不被打断的数据交换操作，从而避免多线程同时改写某一数据时由于执行顺序不确定性以及中断的不可预知性产生的数据不一致问题。该操作通过将内存中的值与指定数据进行比较，当数值一样时将内存中的数据替换为新的值。

lockSlow方法

func (m *MyLock)lockSlow() {
   if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked){
      // 通过CAS方式，如果锁还没被获取，则直接加上锁即可
      return
   }
   var waitStartTime int3
   starving := false
   awoke := false
   iter := 0
   old := m.state
   for {
       // 尝试获取锁资源，暂时先注释掉
   }
}

首先lowSlow会尝试使用CAS获取锁资源，如果获取不到，初始化当前goroutine需要的变量，执行for循环尝试获取锁资源

func (m *MyLock)Lock() {
   /*
   通过CAS获取锁资源，获取不到则初始化当前goroutine所需要的变量
   */
   for {
       if old&(mutexLocked|mutexStarving) == mutexLocked && sync_runtime_canSpin(iter) {
           if !awoke && old&mutexWoken == 0 && old >> mutexWaiterShift != 0 && atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
              awoke = true
           }
           runtime_doSpin()
           iter++
           old = m.state
           continue
        }
        /*
        处理自旋逻辑结束后的逻辑
        */
   }
}

for循环的第一段代码主要是，尝试通过自旋方式获取锁资源，自旋可以避免goroutine切换，但是消耗的资源更多，当goroutine进行自旋的时候，实际上是调用 sync_runtime_doSpin方法，该方法会在CPU上执行若干次PAUSE指令，也就是什么都不做，sleep一小会儿，但是会占用CPU资源。

所以goroutine进入自旋的条件非常苛刻，通常需要满足以下三个条件

互斥锁只有在普通模式才能进行自旋
sync_runtime_canSpin(iter)返回true
自旋次数（iter）小于4
ncpu > 1 也就是CPU核数大于1
当前机器上有一个运行的P队列，且GOMAXPROS（可以用的处理器）大于1

更新持有锁状态副本的值：

func (m *MyLock)Lock() {
   /*
   通过CAS获取锁资源，获取不到则初始化当前goroutine所需要的变量
   */
   for {
        /*
        处理自旋逻辑
        */
        new := old
        if old & mutexStarving == 0 {
           // 如果不是饥饿状态，则尝试更新锁的状态到new上
           new = new | mutexLocked
        }
        if old&(mutexLocked|mutexStarving) != 0 {
           // 如果锁处于饥饿状态或被其他goroutine持有
           // 等待队列+1
           new += 1 << mutexWaiterShift
        }
        if starving && old&mutexLocked != 0 {
           // 如果当前锁为饥饿状态，并且锁资源被其他goroutine持有
           // 更新当前锁状态副本的饥饿状态的值
           new = new | mutexStarving
        }
        if awoke {
           if new & mutexWoken == 0 {
              panic("sync: inconsistent mutex state")
           }
           // 如果本goroutine为唤醒状态，清除掉副本内的环星表金
           // 因为在这种情况下，本goroutine要么获得了锁，要么进入休眠
           new &^= mutexWoken
        }

        /*
        执行接下来的获取锁资源
        */
   }
}

当我们更新了互斥锁状态后，执行以下逻辑

func (m *MyLock)Lock() {
   /*
   通过CAS获取锁资源，获取不到则初始化当前goroutine所需要的变量
   */
   for {
        /*
        处理自旋逻辑
        */
        /*
        更新互斥锁状态
        */
        // 通过CAS设置new的值
      if atomic.CompareAndSwapInt32(&m.state, old, new) {
        // 当 锁 没有处在饥饿以及状态下，可以视作成功获取了锁
          if old&(mutexStarving|mutexLocked) == 0 {
             break
          }
          // 判断是否处于等待状态
          queueLifo := waitStartTime != 0
          // 获取等待开始的时间
          if waitStartTime == 0 {
             // runtime_nanotime 实际上是一个系统调用，获取当前时间
             waitStartTime = runtime_nanotime()
          }
          /*
          如果我们没有通过 CAS 获得锁，
          会调用 sync.runtime_SemacquireMutex 
          使用信号量保证资源不会被两个 Goroutine 获取。
          sync.runtime_SemacquireMutex 
          会在方法中不断调用尝试获取锁并休眠当前 Goroutine 等待信号量的释放，
          一旦当前 Goroutine 可以获取信号量，它就会立刻返回，
          */
          runtime_SemaacquireMutex(&m.sema, queueLifo)

          // 当等待时间超过1ms时候，变为饥饿状态
          starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
          old = m.state
          if old&mutexStarving != 0 {
             // 如果为饥饿状态的话
             if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
                panic("sync: inconsistent mutex state")
             }
             // delta的实际值为  等待的位数
             delta := int32(mutexLocked - 1 << mutexWaiterShift)
             if !starving || old >> mutexWaiterShift == 1 {
                delta -= mutexStarving
             }
             // 对m.state中的等待计进行ADD的原子操作
             atomic.AddInt32(&m.state, delta)
             break
          }
          // 唤醒次数为true, 自旋次数重置
          awoke = true
          iter = 0
       } else {
          // 如果CAS不成功，则获取新的state状态
          old = m.state
       }
       break
}
   }
}

我们首先尝试尝试使用CAS设置目前new的值。

如果没有成功设置则代表有新的goroutine更新了当前的锁资源，我们需要更新当前锁状态，重新进行for循环尝试获取锁。

如果当前锁不处于饥饿状态以及没有被别的goroutine获取，则视为拿到锁资源

判断等待实际以及更新等待时间，调用runtime_SemaacquireMutex使用信号量使资源不会被两个goroutine同时获取，而当有别的goroutine释放了锁资源，则第一时间会将信号量返回给该goroutine，立即获得锁资源。

当等待时间超过1ms得时候，更新饥饿状态。

如果锁处于饥饿状态，且当前goroutine不属于饥饿状态，或锁位未处于饥饿状态，则退出饥饿模式

如果锁不处于饥饿状态，唤醒该goroutine然后将自旋次数重置。

整个加锁过程结束。

解锁逻辑

解锁逻辑相较于加锁逻辑就没有那么复杂了。

func (m *Mutex) Unlock() {
    if race.Enabled {
        _ = m.state
        race.Release(unsafe.Pointer(m))
    }
    new := atomic.AddInt32(&m.state, -mutexLocked)
    if new != 0 {
        m.unlockSlow(new)
    }
}

解锁逻辑实际上就是判断

如果锁不处于饥饿状态，且不属于唤醒状态，则直接释放锁资源

否则执行 unlockSlow函数

func (m *Mutex) unlockSlow(new int32) {
    if (new+mutexLocked)&mutexLocked == 0 {
        throw("sync: unlock of unlocked mutex")
    }
    if new&mutexStarving == 0 {
        old := new
        for {
            if old>>mutexWaiterShift == 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, 1)
                return
            }
            old = m.state
        }
    } else {
        runtime_Semrelease(&m.sema, true, 1)
    }
}

实际上相较于加锁逻辑的lockslow,该unlockSlow也很简单

如果直接解锁一个没有被锁定的锁，抛出一场
判断锁是否为饥饿状态
如果锁不为饥饿状态，且锁不为（锁住、唤醒、饥饿）状态的任一，直接解锁
如果为上面三种情况的一种，需要唤醒在等待队列中的goroutine
如果锁处于饥饿状态，直接唤醒等待队列中的goroutine.

总结

其实整个加锁逻辑总结下来只要明白两点

一、当前锁处于什么状态

二、当前进行到哪一步了

即可很轻松的理解并掌握整个互斥锁的逻辑，解锁更为简单

转自：
zhuanlan.zhihu.com/p/152214855

- EOF -