使用场景

在实际编码中，我们经常遇到并发问题，产生并发问题的原因主要有以下两点：

1. 存在共享资源
2. 存在并发操作共享资源的场景，并且这些操作是非原子性的

解决并发问题，常见的方式之一就是加锁。sync.Mutex就是golang提供的对锁的支持

如何使用

golang中的锁接口：

type Locker interface {
   Lock()
   Unlock()
}

• Locker表示可以锁定和解锁的对象

Mutex实现了Locker接口中的方法，Mutex结构如下：

type Mutex struct {
   state int32
   sema  uint32
}

// Mutex提供了以下三个方法
// 加锁
func (m *Mutex) Lock() {}
// 尝试加锁
func (m *Mutex) TryLock() bool {}
// 解锁
func (m *Mutex) Unlock() {}

• Mutex是一个互斥锁
• Mutex的零值是未锁定的互斥锁，不用做额外的初始化
• Mutex在首次使用后禁止被复制
• state表示锁状态
• sema表示信号量，用来阻塞和唤醒goroutine

使用Mutex进行加锁和解锁代码：

var mu sync.Mutex

func LockFunc() {
   // 加锁
   mu.Lock()
   // 函数执行结束，自动释放锁
   defer mu.Unlock()
   // todo 需要加锁的代码
}

底层原理

前提知识

CAS

CAS是cpu硬件同步原语，全称为compare and swap（比较和交换）

CAS 操作包含三个操作数 —— 内存位置（V）、预期原值（A）和新值(B)。如果内存位置的值与预期原值相匹配，那么处理器会自动将该位置值更新为新值，并且保证这是个原子操作。否则，处理器不做任何操作

对应的伪代码为：

if V == A {
    V = B
    return true
} else {
    return false
}

正常模式 vs 饥饿模式

一个好的互斥锁，需要考虑以下两个因素：

• 性能
• 公平

Mutex有两种模式：正常模式和饥饿模式

在正常模式下，一个尝试加锁的goroutine会先自旋几次，在锁被释放或者不能自旋后，会开始尝试获取锁，此时若排队等待的队列数不为空，则队列头部的goroutine会被唤醒，并和这些自旋goroutine一起竞争锁。因为这些自旋的goroutine不需要进行唤醒操作，且正在cpu上运行，同时自旋的goroutine可以有很多，而被唤醒的goroutine只有一个，所以这些自旋的goroutine获取锁的概率更大，减少了阻塞唤醒的时间，提高了获取锁的效率。此时更加注重性能，实现为非公平锁

当锁竞争十分激烈时，有的调用方可能等待了很长的时间，但一直得不到锁。此时更加注重公平，实现为公平锁，对应mutex的饥饿模式

在饥饿模式下，Mutex的所有权从执行Unlock的goroutine，直接传递给等待队列头部的goroutine，后来的goroutine不会自旋，也不会尝试获得锁，即使Mutex处于未加锁状态，这些新来的goroutine会直接去队列的尾部进行排队，严格的先来后到

正常模式与饥饿模式的切换

正常模式切换为饥饿模式：

• 当有goroutine获取锁的等待时间大于等待阈值1ms时，mutex切换为饥饿模式

饥饿模式切换为正常模式：

• 当前waiter的等待时间小于等于等待阈值1ms
• 当前goroutine已经是排队等待队列的最后一个waiter

state字段含义

Mutex中的state字段，占32位，被分成了四个部分：

• 最后一位表示是否已加锁，0表示未加锁，1表示已加锁
• 倒数第二位表示是否有goroutine已被唤醒，0表示没有被唤醒，1表示已被唤醒
• 倒数第三位表示是否为饥饿模式，0表示正常模式，1表示饥饿模式
• 其余29位用来表示排队等待的goroutine数量

state表示含义

源码中的常量：

const (
   // 锁标识掩码常量 = 1
   mutexLocked = 1 << iota // mutex is locked
   // 唤醒标识掩码常量 = 2
   mutexWoken
   // 饥饿模式掩码常量 = 4
   mutexStarving
   // 表示排队数量需要排除的位数 = 3
   mutexWaiterShift = iota
   // 切换为饥饿模式等待的阈值 = 1ms
   starvationThresholdNs = 1e6
)

Lock

Fast path

func (m *Mutex) Lock() {
   // Fast path: grab unlocked mutex.
   // CAS快速加锁
   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)
   // 方便编译器对fast path进行内联优化
   // fast path加锁失败了，则执行slow path   
   m.lockSlow()
}

先执行fast path，通过cas进行快速加锁

• 若cas加锁成功，则直接返回
• 若cas加锁失败，则执行slow path

Slow path

func (m *Mutex) lockSlow() {
   var waitStartTime int64
   starving := false
   awoke := false
   iter := 0
   old := m.state
   // 循环尝试加锁
   for {
      // slow path加锁核心逻辑
      // ...
   }
}

接下来对slow path中的加锁核心逻辑进行详细分析，主要可以分成以下三个步骤：

1. 自旋
2. 判断状态
3. 设置状态

自旋

// old&(mutexLocked|mutexStarving) == mutexLocked 判断当前锁状态是否为非饥饿模式的已加锁状态
// runtime_canSpin(iter) 判断是否可以进行自旋
if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
   // !awoke 判断当前goroutine是否已被唤醒
   // old&mutexWoken == 0 判断当前锁状态是否为未被唤醒
   // old>>mutexWaiterShift != 0 判断排队等待队列是否不为空
   // atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) 尝试把锁设置为已唤醒状态
   if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
      atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
      // 把当前goroutine标记为已唤醒
      awoke = true
   }
   // 进行自旋
   runtime_doSpin()
   // 自旋次数+1
   iter++
   old = m.state
   continue
}

先判断是否需要进行自旋，需要同时满足以下三个条件：

1. 当前模式为正常模式。若当前模式为饥饿模式，不要进行自旋操作了，直接去等待队列尾部进行排队，严格的先来后到
2. 当前锁状态为已加锁。若当前锁状态为未加锁，不要进行自旋操作了，赶快去尝试获取锁
3. 可以进行自旋操作

const (
   active_spin = 4
)

// 判断是否可以进行自旋
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
}

可以进行自旋的判断逻辑如下：

1. 自旋次数小于4
2. cpu大于1，也就需要是多核cpu
3. GOMAXPROCS > 1并且至少有一个其他正在运行的P并且本地runq为空

执行的自旋操作：

1. 若当前的goroutine是被唤醒的，且当前锁状态为未被唤醒状态，且排队等待队列不为空，且成功把锁状态设置为已唤醒状态。则把当前goroutine标记为已唤醒
2. 进行自旋
3. 自旋次数+1
4. 重新获取state
5. 继续下一次循环

为什么需要自旋操作

如果不进行自旋操作，直接就去判断状态和设置状态，那么当前goroutine大概率是进行排队。进行合理的自旋操作后，可以避免当前goroutine被阻塞和唤醒的次数，提高获取锁的效率

判断状态

// new表示需要重新设置的状态
new := old
// old&mutexStarving == 0 判断是否是正常模式
// 如果是正常模式，则设置lock位，尝试加锁
// 如果是饥饿模式，不要尝试加锁，新到达的goroutine直接去排队   
if old&mutexStarving == 0 {
   new |= mutexLocked
}
// old&(mutexLocked|mutexStarving) != 0 判断是否是已加锁或者是饥饿模式
// 若是的话，排队数+1
if old&(mutexLocked|mutexStarving) != 0 {
   new += 1 << mutexWaiterShift
}
// 切换饥饿模式
// 当前goroutine等待超过1ms 且 当前锁没有被释放
// 为什么要求锁没有被释放？？？
// 锁被释放了，可以直接去抢锁，此时设置成饥饿模式，那么就只能去排队了
if starving && old&mutexLocked != 0 {
   new |= mutexStarving
}
// 当前唤醒标识为true，则释放唤醒标识
if awoke {
   if new&mutexWoken == 0 {
      throw("sync: inconsistent mutex state")
   }
   new &^= mutexWoken
}

判断当前锁状态是否为正常模式

• 如果是正常模式，则设置lock位，尝试加锁。不需要关心原来的锁状态为已加锁还是未加锁
• 如果是饥饿模式，不要尝试加锁，新到达的goroutine直接去等待队列尾部排队

判断是否是已加锁或者是饥饿模式

• 如果是的话，排队等待数+1

判断当前goroutine是否已经饥饿（等待加锁时间超过了1ms），且当前锁状态为已加锁

• 如果是的话，则切换为饥饿模式
• 这里还需要当前锁状态为已加锁的原因是，如果当前锁为未加锁，那么我们直接去抢锁就好了，直接设置成饥饿模式，我们就只能去排队等待了

判断当前goroutine是否已被唤醒

• 如果已经被唤醒，则释放锁的唤醒标识
• 释放唤醒标识的原因，只要下一步的设置状态成功，那么当前goroutine要么是加锁成功退出，要么是进行排队阻塞，都不是已唤醒的状态了

设置状态

// 尝试设置锁的状态
if atomic.CompareAndSwapInt32(&m.state, old, new) {
   // 设置状态成功
   // 两种可能：1加锁成功 2排队成功
   // 加锁成功，则直接返回
   if old&(mutexLocked|mutexStarving) == 0 {
      break // locked the mutex with CAS
   }
   // 排队成功
   // 需要判断当前goroutine是否是第一次进行排队
   // 如果是第一次进行排队，则排到队列尾部
   // 如果已经排过队，则排到队列头部
   queueLifo := waitStartTime != 0
   if waitStartTime == 0 {
      waitStartTime = runtime_nanotime()
   }
   // 进入排队队列，当前goroutine阻塞
   // queueLifo为true，则排到队列头部
   // queueLifo为false，则排到队列尾部
   runtime_SemacquireMutex(&m.sema, queueLifo, 1)
   // 被唤醒后继续执行
   // 判断等待时间是否超过1ms
   starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
   // 重新获取state
   old = m.state
   // 判断是否为饥饿模式
   if old&mutexStarving != 0 {
      if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
         throw("sync: inconsistent mutex state")
      }
      // 直接加锁，同时设置等待队列数-1
      delta := int32(mutexLocked - 1<>mutexWaiterShift == 1 {
         // 退出饥饿模式
         delta -= mutexStarving
      }
      atomic.AddInt32(&m.state, delta)
      break
   }
   // 正常模式，当前goroutine被唤醒，继续进行抢锁操作
   awoke = true
   // 自旋次数清0
   iter = 0
} else {
   // 设置失败，重新进行循环
   old = m.state
}

尝试设置锁的状态

• 设置失败，重新进行循环
• 设置成功，那么有两种可能：加锁成功或者排队成功

正常模式，加锁成功，则直接返回

饥饿模式，排队成功

判断当前goroutine是否是第一次进行排队

• 如果是第一次进行排队，则排到队列尾部，严格的先来后到
• 如果已经排过队，则排到队列头部

当前goroutine阻塞

当前goroutine被唤醒后继续执行

判断当前goroutine等待时间是否超过1ms

• 如果超过1ms，则标记当前goroutine已经饥饿

重新获取state，判断当前锁是否为饥饿模式

• 如果是正常模式，则标记当前goroutine已被唤醒，清空自旋次数，继续开始循环尝试获取锁
• 如果为饥饿模式，直接进行加锁，且对排队数进行-1

为什么这里判断为饥饿模式，可以直接进行加锁？

因为在饥饿模式下，被唤醒就是等待队列头部的goroutine，饥饿模式下也不允许其他goroutine进行自旋和尝试加锁操作，此时当前goroutine直接进行加锁操作即可

何时退出饥饿模式？

以下两种情况下，会把锁模式从饥饿模式切换为正常模式：

• 当前goroutine等待的时间小于等于1ms
• 当前goroutine是等待队列的最后一个waiter

Unlock

Fast path

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

   // Fast path: drop lock bit.
   // 快速解锁
   new := atomic.AddInt32(&m.state, -mutexLocked)
   if new != 0 {
      // Outlined slow path to allow inlining the fast path.
      // To hide unlockSlow during tracing we skip one extra frame when tracing GoUnblock.
      // 方便编译器对fast path进行内联优化
      // fast path解锁失败了，则执行slow path 
      m.unlockSlow(new)
   }
}

通过原子操作修改锁的状态

• 如果修改后的锁状态为0，则解锁成功，直接退出
• 如果修改后的锁状态不为0，则执行slow path

Slow path

func (m *Mutex) unlockSlow(new int32) {
   if (new+mutexLocked)&mutexLocked == 0 {
      throw("sync: unlock of unlocked mutex")
   }
   if new&mutexStarving == 0 {
      // 正常模式
      old := new
      for {
         // 等待队列为空 或者 已经有其他goroutine已经获得了锁，已经被唤醒，锁进入饥饿模式
         if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
            // 不需要唤醒某个goroutine，直接返回即可
            return
         }
         // 等待队列数-1 加 唤醒goroutine
         new = (old - 1<

正常模式

• 如果等待队列为空 或者 已经有其他goroutine已经获得了锁，或者已经被唤醒，或者锁进入饥饿模式，则直接返回即可
• 如果不满足，则等待队列数-1，唤醒等待队列的第一个goroutine。对应加锁方法Slow path标记当前goroutine已被唤醒，清空自旋次数，继续开始循环尝试获取锁
• 如果设置状态不成功，则继续循环，进行下一次尝试

饥饿模式

• Mutex的所有权从执行Unlock的goroutine，直接传递给等待队列头部的goroutine

TryLock

func (m *Mutex) TryLock() bool {
   old := m.state
   // 已加锁或为饥饿模式，直接返回
   if old&(mutexLocked|mutexStarving) != 0 {
      return false
   }

   // cas尝试加锁
   if !atomic.CompareAndSwapInt32(&m.state, old, old|mutexLocked) {
      return false
   }

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

已加锁或为饥饿模式，直接返回false

cas尝试加锁，加锁成功，返回true，加锁失败，返回false

支付宝打赏

微信打赏

golang

发表评论

留言与评论（共有 0 条评论） “”

验证码：