关于golang:Go精妙的互斥锁设计

Some people, when confronted with a problem, think,“I know, I’ll use threads,”and then two they hav erpoblesms.

多线程程序在多核 CPU 机器上访问共享资源时，难免会遇到问题。咱们能够来看一个例子

var Cnt int

func Add(iter int) {
    for i := 0; i < iter; i++ {Cnt++}
}

func main() {wg := &sync.WaitGroup{}
    for i := 0; i < 2; i++ {wg.Add(1)
        go func() {Add(100000)
            wg.Done()}()}
    wg.Wait()
    fmt.Println(Cnt)
}

很显著，程序的预期后果是 200000，但理论的输入却是不可确定的，可能为 100910、101364 或者其余数值，这就是典型的多线程拜访抵触问题。

利用 go tool trace 剖析工具（须要在代码中退出 runtime/trace 包获取程序运行信息，此处省略），查看该程序运行期间 goroutine 的执行状况如上图所示。其中 G20 和 G19 就是执行 Add() 函数的两个 goroutine，它们在执行期间并行地拜访了共享变量Cnt。

相似这种状况，即两个或者多个线程读写某些共享数据，而最初的后果取决于程序运行的准确时序，这就是 竞争条件（race condition）

怎么防止竞争条件？实际上凡波及共享内存、共享文件以及共享任何资源的状况都会引发上文例子中相似的谬误，要防止这种谬误，要害是要找出某种路径来阻止多线程同时读写共享的数据。换言之，咱们须要的是互斥（mutual exclusion），即以某种伎俩确保当一个线程在应用一个共享变量或文件时，其余线程不能做同样的操作。

咱们把对共享内存进行拜访的程序片段称作 临界区 （critical section），例如上例中的Cnt++ 片段。从形象的角度看，咱们心愿的多线程行为如下图所示。线程 A 在 t1 时刻进入临界区，执行一段时间后，在 t2 时刻线程 B 试图进入临界区，然而这是不能被容许的，因为同一时刻只能运行一个线程在临界区内，而此时曾经有一个线程在临界区内。咱们通过某种互斥伎俩，将 B 临时挂起直到线程 A 来到临界区，即 t3 时刻 B 进入临界区。最初，B 执行完临界区代码后，来到临界区。

如果咱们可能正当地安顿，使得两个线程不可能同时处于临界区中，就可能防止竞争条件。因而，咱们将代码稍作调整如下

var (
    Cnt int
    mu sync.Mutex
)

func Add(iter int) {mu.Lock()
    for i := 0; i < iter; i++ {Cnt++}
    mu.Unlock()}

此时，程序执行失去了预期后果 200000。

程序运行期间的执行状况如上图所示。其中 G8 和 G7 是执行 Add() 函数的两个 goroutine，通过退出 sync.Mutex 互斥锁，G8 和 G7 就不再存在竞争条件。

须要明确的是，只有在多核机器上才会产生竞争条件，只有多线程对共享资源做了写操作时才有可能产生竞态问题，只有资源没有发生变化，多个线程读取雷同的资源就是平安的。

互斥锁是实现互斥性能的常见实现，Go 中的互斥锁即sync.Mutex。本文将基于 Go 1.15.2 版本，对互斥锁的实现深入研究。

type Mutex struct {
    state int32
    sema  uint32
}

const (
    mutexLocked = 1 << iota
    mutexWoken
    mutexStarving
    mutexWaiterShift = iota   // mutexWaiterShift 值为 3，通过右移 3 位的位运算，可计算 waiter 个数
    starvationThresholdNs = 1e6 // 1ms，进入饥饿状态的等待时间
)

state字段示意以后互斥锁的状态信息，它是 int32 类型，其低三位的二进制位均有相应的状态含意。

mutexLocked是 state 中的低 1 位，用二进制示意为0001（为了不便，这里只形容后 4 位），它代表该互斥锁是否被加锁。
mutexWoken是低 2 位，用二进制示意为0010，它代表互斥锁上是否有被唤醒的 goroutine。
mutexStarving是低 3 位，用二进制示意为0100，它代表以后互斥锁是否处于饥饿模式。
state剩下的 29 位用于统计在互斥锁上的期待队列中 goroutine 数目（waiter）。

默认的 state 字段（无锁状态）如下图所示。

sema字段是信号量，用于管制 goroutine 的阻塞与唤醒，下文中会有介绍到。

Go 实现的互斥锁有两种模式，别离是 失常模式 和饥饿模式。

在失常模式下，waiter 依照先进先出（FIFO）的形式获取锁，然而一个刚被唤醒的 waiter 与新达到的 goroutine 竞争锁时，大概率是干不过的。新来的 goroutine 有一个劣势：它曾经在 CPU 上运行，并且有可能不止一个新来的，因而 waiter 极有可能失败。在这种状况下，waiter 还须要在期待队列中排队。为了防止 waiter 长时间抢不到锁，当 waiter 超过 1ms 没有获取到锁，它就会将以后互斥锁切换到饥饿模式，避免期待队列中的 waiter 被饿死。

在饥饿模式下，锁的所有权间接从解锁（unlocking）的 goroutine 转移到期待队列中的队头 waiter。新来的 goroutine 不会尝试去获取锁，也不会自旋。它们将在期待队列的队尾排队。

如果某 waiter 获取到了锁，并且满足以下两个条件之一，它就会将锁从饥饿模式切换回失常模式。

它是期待队列的最初一个 goroutine
它期待获取锁的工夫小于 1ms

饥饿模式是在 Go 1.9 版本引入的，它避免了队列尾部 waiter 始终无奈获取锁的问题。与饥饿模式相比，失常模式下的互斥锁性能更好。因为相较于将锁的所有权明确赋予给唤醒的 waiter，间接竞争锁能升高整体 goroutine 获取锁的延时开销。

既然被称作锁，那就存在加锁和解锁的操作。sync.Mutex的加锁 Lock() 代码如下

func (m *Mutex) Lock() {if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
        if race.Enabled {race.Acquire(unsafe.Pointer(m))
        }
        return
    }
    m.lockSlow()}

代码十分简洁，首先通过 CAS 判断以后锁的状态（CAS 的原理和实现能够参照小菜刀写的《同步原语的基石》一文）。如果锁是齐全闲暇的，即 m.state 为 0，则对其加锁，将 m.state 的值赋为 1，此时加锁后的 state 如下

如果，以后锁曾经被其余 goroutine 加锁，则进入 m.lockSlow() 逻辑。lockSlow函数比拟长，这里咱们分段论述。

初始化

func (m *Mutex) lockSlow() {
    var waitStartTime int64  // 用于计算 waiter 的等待时间
    starving := false        // 饥饿模式标记
    awoke := false           // 唤醒标记
    iter := 0                // 统计以后 goroutine 的自旋次数
    old := m.state           // 保留以后锁的状态
    ...
}

第一段程序是做一些初始化状态、标记的动作。

自旋

lockSlow函数余下的代码，就是一个大的 for 循环，首先看自旋局部。

for { 
    // 判断是否能进入自旋
    if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
        // !awoke 判断以后 goroutine 是不是在唤醒状态
        // old&mutexWoken == 0 示意没有其余正在唤醒的 goroutine
        // old>>mutexWaiterShift != 0 示意期待队列中有正在期待的 goroutine
        if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
            // 尝试将以后锁的低 2 位的 Woken 状态位设置为 1，示意已被唤醒
            // 这是为了告诉在解锁 Unlock()中不要再唤醒其余的 waiter 了
            atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {awoke = true}
        // 自旋
        runtime_doSpin()
        iter++
        old = m.state
        continue
    }
    ...
}

对于自旋，这里须要简略论述一下。自旋是自旋锁的行为，它通过忙期待，让线程在某段时间内始终放弃执行，从而防止线程上下文的调度开销。自旋锁对于线程只会阻塞很短时间的场景是十分适合的。很显然，单核 CPU 是不适宜应用自旋锁的，因为，在同一时间只有一个线程是处于运行状态，假如运行线程 A 发现无奈获取锁，只能期待解锁，但因为 A 本身不挂起，所以那个持有锁的线程 B 没有方法进入运行状态，只能等到操作系统分给 A 的工夫片用完，能力有机会被调度。这种状况下应用自旋锁的代价很高。

在本场景中，之所以想让以后 goroutine 进入自旋行为的根据是，咱们乐观地认为：以后正在持有锁的 goroutine 能在较短的工夫内偿还锁。

runtime_canSpin()函数的实现如下

//go:linkname sync_runtime_canSpin sync.runtime_canSpin
func sync_runtime_canSpin(i int) bool {
  // active_spin = 4
    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
}

因为自旋自身是空转 CPU 的，所以如果使用不当，反倒会升高程序运行性能。联合函数中的判断逻辑，这里总结进去 goroutine 能进入自旋的条件如下

以后互斥锁处于失常模式
以后运行的机器是多核 CPU，且 GOMAXPROCS>1
至多存在一个其余正在运行的处理器 P，并且它的本地运行队列（local runq）为空
以后 goroutine 进行自旋的次数小于 4

后面说到，自旋行为就是让以后 goroutine 并不挂起，占用 cpu 资源。咱们看一下 runtime_doSpin() 的实现。

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

runtime_doSpin调用了procyield，其实现如下（以 amd64 为例）

TEXT runtime·procyield(SB),NOSPLIT,$0-0    MOVL    cycles+0(FP), AXagain:    PAUSE    SUBL    $1, AX    JNZ    again    RET

很显著，所谓的忙期待就是执行 30 次 PAUSE 指令，通过该指令占用 CPU 并耗费 CPU 工夫。

计算冀望状态

后面说过，以后 goroutine 进入自旋是须要满足相应条件的。如果不满足自旋条件，则进入以下逻辑。

        // old 是锁以后的状态，new 是冀望的状态，以期于在前面的 CAS 操作中更改锁的状态    new := old        if old&mutexStarving == 0 {// 如果以后锁不是饥饿模式，则将 new 的低 1 位的 Locked 状态位设置为 1，示意加锁            new |= mutexLocked}        if old&(mutexLocked|mutexStarving) != 0 {// 如果以后锁已被加锁或者处于饥饿模式，则将 waiter 数加 1，示意以后 goroutine 将被作为 waiter 置于期待队列队尾            new += 1 << mutexWaiterShift}        if starving && old&mutexLocked != 0 {// 如果以后锁处于饥饿模式，并且已被加锁，则将低 3 位的 Starving 状态位设置为 1，示意饥饿            new |= mutexStarving}    // 当 awoke 为 true，则表明以后 goroutine 在自旋逻辑中，胜利批改锁的 Woken 状态位为 1        if awoke {if new&mutexWoken == 0 {                throw("sync: inconsistent mutex state")            }      // 将唤醒标记位 Woken 置回为 0      // 因为在后续的逻辑中，以后 goroutine 要么是拿到锁了，要么是被挂起。// 如果是挂起状态，那就须要期待其余开释锁的 goroutine 来唤醒。// 如果其余 goroutine 在 unlock 的时候发现 Woken 的地位不是 0，则就不会去唤醒，那该 goroutine 就无奈再醒来加锁。new &^= mutexWoken        }

这里须要重点了解一下位操作 A |= B，它的含意就是在 B 的二进制位为 1 的位，将 A 对应的二进制位设为 1，如下图所示。因而，new |= mutexLocked 的作用就是将 new 的最低一位设置为 1。

更新冀望状态

在上一步，咱们失去了锁的冀望状态，接下来通过 CAS 将锁的状态进行更新。

        // 尝试将锁的状态更新为冀望状态    if atomic.CompareAndSwapInt32(&m.state, old, new) {// 如果锁的原状态既不是被获取状态，也不是处于饥饿模式      // 那就间接返回，示意以后 goroutine 已获取到锁            if old&(mutexLocked|mutexStarving) == 0 {break // locked the mutex with CAS}      // 如果走到这里，那就证实以后 goroutine 没有获取到锁      // 这里判断 waitStartTime != 0 就证实以后 goroutine 之前曾经期待过了，则须要将其搁置在期待队列队头            queueLifo := waitStartTime != 0            if waitStartTime == 0 {// 如果之前没有期待过，就以当初的工夫来初始化设置                waitStartTime = runtime_nanotime()            }      // 阻塞期待            runtime_SemacquireMutex(&m.sema, queueLifo, 1)      // 被信号量唤醒之后查看以后 goroutine 是否应该示意为饥饿      //（这里示意为饥饿之后，会在下一轮循环中尝试将锁的状态更改为饥饿模式）// 1. 如果以后 goroutine 曾经饥饿（在上一次循环中更改了 starving 为 true）// 2. 如果以后 goroutine 曾经期待了 1ms 以上            starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs            // 再次获取锁状态      old = m.state      // 走到这里，如果此时锁依然是饥饿模式      // 因为在饥饿模式下，锁是间接交给唤醒的 goroutine      // 所以，即把锁交给以后 goroutine            if old&mutexStarving != 0 {        // 如果以后锁既不是被获取也不是被唤醒状态，或者期待队列为空        // 这代表锁状态产生了不统一的问题                if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {throw("sync: inconsistent mutex state")                }        // 因为以后 goroutine 曾经获取了锁，delta 用于将期待队列 -1                delta := int32(mutexLocked - 1<<mutexWaiterShift)        // 如果以后 goroutine 中的 starving 标记不是饥饿        // 或者以后 goroutine 曾经是期待队列中的最初一个了        // 就通过 delta -= mutexStarving 和 atomic.AddInt32 操作将锁的饥饿状态位设置为 0，示意为失常模式                if !starving || old>>mutexWaiterShift == 1 {delta -= mutexStarving}                atomic.AddInt32(&m.state, delta)        // 拿到锁退出，业务逻辑解决完之后，须要调用 Mutex.Unlock()办法开释锁                break}      // 如果锁不是饥饿状态      // 因为以后 goroutine 曾经被信号量唤醒了      // 那就将示意以后 goroutine 状态的 awoke 设置为 true      // 并且将自旋次数的计数 iter 重置为 0，如果能满足自旋条件，从新自旋期待            awoke = true            iter = 0        } else {// 如果 CAS 未胜利, 更新锁状态，从新一个大循环            old = m.state}

这里须要了解一下 runtime_SemacquireMutex(s *uint32, lifo bool, skipframes int) 函数， 它是用于同步库的 sleep 原语 ，它的实现是位于src/runtime/sema.go 中的 semacquire1 函数，与它相似的还有 runtime_Semacquire(s *uint32) 函数。两个睡眠原语须要等到 *s>0（本场景中 m.sema>0），而后原子递加 *s。SemacquireMutex 用于剖析竞争的互斥对象，如果 lifo（本场景中queueLifo）为 true，则将期待者排在期待队列的队头。skipframes 是从 SemacquireMutex 的调用方开始计数，示意在跟踪期间要疏忽的帧数。

所以，运行到 SemacquireMutex 就证实以后 goroutine 在后面的过程中获取锁失败了，就须要 sleep 原语来阻塞以后 goroutine，并通过信号量来排队获取锁：如果是新来的 goroutine，就须要放在队尾；如果是被唤醒的期待锁的 goroutine，就放在队头。

后面说过，有加锁就必然有解锁。咱们来看解锁的过程

func (m *Mutex) Unlock() {    if race.Enabled {        _ = m.state        race.Release(unsafe.Pointer(m))    }  // new 是解锁的冀望状态    new := atomic.AddInt32(&m.state, -mutexLocked)    if new != 0 {m.unlockSlow(new)    }}

通过原子操作 AddInt32 想将锁的低 1 位 Locked 状态地位为 0。而后判断新的 m.state 值，如果值为 0，则代表以后锁曾经齐全闲暇了，完结解锁，否则进入 unlockSlow() 逻辑。

这里须要留神的是，锁闲暇有两种状况，第一种是齐全闲暇，它的状态就是锁的初始状态。

第二种闲暇，是指的以后锁没被占有，然而会有期待拿锁的 goroutine，只是还未被唤醒，例如以下状态的锁也是闲暇的，它有两个期待拿锁的 goroutine（未唤醒状态）。

以下是 unlockSlow 函数实现。

func (m *Mutex) unlockSlow(new int32) {// 1. 如果 Unlock 了一个没有上锁的锁，则会产生 panic。if (new+mutexLocked)&mutexLocked == 0 {throw("sync: unlock of unlocked mutex")   }  // 2. 失常模式   if new&mutexStarving == 0 {old := new      for {        // 如果锁没有 waiter, 或者锁有其余以下已产生的状况之一，则前面的工作就不必做了，间接返回        // 1. 锁处于锁定状态，示意锁曾经被其余 goroutine 获取了        // 2. 锁处于被唤醒状态，这表明有期待 goroutine 被唤醒，不必再尝试唤醒其余 goroutine        // 3. 锁处于饥饿模式，那么锁之后会被间接交给期待队列队头 goroutine         if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {return}        // 如果能走到这，那就是下面的 if 判断没通过        // 阐明以后锁是闲暇状态，然而期待队列中有 waiter，且没有 goroutine 被唤醒        // 所以，这里咱们想要把锁的状态设置为被唤醒，期待队列 waiter 数 -1         new = (old - 1<<mutexWaiterShift) | mutexWoken        // 通过 CAS 操作尝试更改锁状态         if atomic.CompareAndSwapInt32(&m.state, old, new) {// 通过信号量唤醒 goroutine，而后退出            runtime_Semrelease(&m.sema, false, 1)            return         }        // 这里是 CAS 失败的逻辑        // 因为在 for 循环中，锁的状态有可能曾经被扭转了，所以这里须要及时更新一下状态信息        // 以便下个循环里作判断解决         old = m.state      }   // 3. 饥饿模式   } else {// 因为是饥饿模式，所以非常简单     // 间接唤醒期待队列队头 goroutine 即可      runtime_Semrelease(&m.sema, true, 1)   }}

在这里，须要了解一下 runtime_Semrelease(s *uint32, handoff bool, skipframes int) 函数。它是用于同步库的 wakeup 原语 ，Semrelease 原子减少 *s 值（本场景中 m.sema），并告诉阻塞在Semacquire 中正在期待的 goroutine。如果 handoff 为真，则跳过计数，间接唤醒队头 waiter。skipframes是从 Semrelease 的调用方开始计数，示意在跟踪期间要疏忽的帧数。

从代码量而言，go 中互斥锁的代码十分轻量简洁，通过奇妙的位运算，仅仅采纳 state 一个字段就实现了四个字段的成果，十分之精彩。

然而，代码量少并不代表逻辑简略，相同，它很简单。互斥锁的设计中蕴含了大量的位运算，并包含了两种不同锁模式、信号量、自旋以及调度等内容，读者要真正了解加解锁的过程并不容易，这里再做一个简略回顾总结。

在失常模式下，waiter 依照先进先出的形式获取锁；在饥饿模式下，锁的所有权间接从解锁的 goroutine 转移到期待队列中的队头 waiter。

模式切换

如果以后 goroutine 期待锁的工夫超过了 1ms，互斥锁就会切换到饥饿模式。

如果以后 goroutine 是互斥锁最初一个 waiter，或者期待的工夫小于 1ms，互斥锁切换回失常模式。

加锁

如果锁是齐全闲暇状态，则通过 CAS 间接加锁。
如果锁处于失常模式，则会尝试自旋，通过持有 CPU 期待锁的开释。
如果以后 goroutine 不再满足自旋条件，则会计算锁的冀望状态，并尝试更新锁状态。
在更新锁状态胜利后，会判断以后 goroutine 是否能获取到锁，能获取锁则间接退出。
以后 goroutine 不能获取到锁时，则会由 sleep 原语 SemacquireMutex 陷入睡眠，期待解锁的 goroutine 发出信号进行唤醒。
唤醒之后的 goroutine 发现锁处于饥饿模式，则能间接拿到锁，否则重置自旋迭代次数并标记唤醒位，从新进入步骤 2 中。

解锁

如果通过原子操作 AddInt32 后，锁变为齐全闲暇状态，则间接解锁。
如果解锁一个没有上锁的锁，则间接抛出异样。
如果锁处于失常模式，且没有 goroutine 期待锁开释，或者锁被其余 goroutine 设置为了锁定状态、唤醒状态、饥饿模式中的任一种（非闲暇状态），则会间接退出；否则，会通过 wakeup 原语 Semrelease 唤醒 waiter。
如果锁处于饥饿模式，会间接将锁的所有权交给期待队列队头 waiter，唤醒的 waiter 会负责设置 Locked 标记位。

另外，从 Go 的互斥锁带有自旋的设计而言，如果咱们通过 sync.Mutex 只锁定执行耗时很低的要害代码，例如锁定某个变量的赋值，性能是十分不错的（因为期待锁的 goroutine 不必被挂起，持有锁的 goroutine 会很快开释锁）。所以，咱们在应用互斥锁时，应该只锁定真正的临界区。

mu.Lock()defer mu.Unlock()

写如上的代码，是很爽。然而，你有想过这会带来没必要的性能损耗吗？

关于golang:Go精妙的互斥锁设计

竞争条件

临界区与互斥

Go 互斥锁设计

两种模式

加锁

解锁

总结