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

竞争条件

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

var Cnt intfunc 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互斥锁设计

互斥锁是实现互斥性能的常见实现，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函数比拟长，这里咱们分段论述。

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

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

  1. 自旋

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_canSpinfunc 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_doSpinfunc 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 工夫。

  1. 计算冀望状态

后面说过，以后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。

  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，互斥锁切换回失常模式。

加锁

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

解锁

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

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

mu.Lock()defer mu.Unlock()

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