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

Mutex数据结构

• sync.Mutex 数据结构

  type Mutex struct {
    state int32   // 表示了当前互斥锁处于的状态
    sema  int32   // 表示信号量，协程阻塞等待该信号量，解锁的协程释放信号量从而唤醒等待信号量的协程。
  }
  

• Mutex.state 数据结构

  • Mutex.state是32位的整型变量，内部实现时把该变量分成四份，用于记录Mutex的四种状态。

  

  1. Waiter: 表示阻塞等待锁的协程个数，协程解锁时根据此值来判断是否需要释放信号量。
  2. Starving：表示该Mutex是否处于饥饿状态，0：没有饥饿 1：饥饿状态，说明有协程阻塞了超过1ms。
  3. Woken: 表示是否有协程已被唤醒，0：没有协程唤醒 1：已有协程唤醒，正在加锁过程中。
  4. Locked: 表示该Mutex是否已被锁定，0：没有锁定 1：已被锁定。

互斥锁的状态

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

Mutex模式

• 每个Mutex都有两个模式，称为Normal和Starving(饥饿模式)

1. normal模式

• 默认情况下，Mutex的模式为normal。

• 该模式下，协程如果加锁不成功不会立即转入阻塞排队，而是判断是否满足自旋的条件，如果满足则会启动自旋过程，尝试抢锁。

2 starvation模式

• 自旋过程中能抢到锁，一定意味着同一时刻有协程释放了锁，我们知道释放锁时如果发现有阻塞等待的协程，还会释放一个信号量来唤醒一个等待协程，被唤醒的协程得到CPU后开始运行，此时发现锁已被抢占了，自己只好再次阻塞，不过阻塞前会判断自上次阻塞到本次阻塞经过了多长时间，如果超过1ms的话，会将Mutex标记为”饥饿”模式，然后再阻塞。

• 处于饥饿模式下，不会启动自旋过程，也即一旦有协程释放了锁，那么一定会唤醒协程，被唤醒的协程将会成功获取锁，同时也会把等待计数减1。

加锁逻辑

1. 加锁

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

2. lockSlow

• code

  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循环尝试获取锁资源

• code

  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进入自旋的条件非常苛刻，通常需要满足以下三个条件

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

自旋

• 自旋对应于CPU的”PAUSE”指令，CPU对该指令什么都不做，相当于CPU空转，对程序而言相当于sleep了一小段时间，时间非常短，当前实现是30个时钟周期。
• 自旋过程中会持续探测Locked是否变为0，连续两次探测间隔就是执行这些PAUSE指令，它不同于sleep，不需要将协程转为睡眠状态。
• 会在CPU上执行若干次PAUSE指令，也就是什么都不做，sleep一小会儿，但是会占用CPU资源。

1. 自旋条件

• 加锁时程序会自动判断是否可以自旋，无限制的自旋将会给CPU带来巨大压力，所以判断是否可以自旋就很重要了。

• 进行自旋的条件:

  1. 互斥锁只有在普通模式才能进行自旋,也就是协程调度机制中的可运行队列必须为空，否则会延迟协程调度
  2. **sync_runtime_canSpin(iter)**返回true
  3. 自旋次数（iter）小于4
  4. ncpu > 1 也就是CPU核数大于1

2. 自旋优缺点

• 优点
  1. 自旋的优势是更充分的利用CPU，尽量避免协程切换。因为当前申请加锁的协程拥有CPU，如果经过短时间的自旋可以获得锁，当前协程可以继续运行，不必进入阻塞状态。
• 缺点
  1. 如果自旋过程中获得锁，那么之前被阻塞的协程将无法获得锁，如果加锁的协程特别多，每次都通过自旋获得锁，那么之前被阻塞的进程将很难获得锁，从而进入饥饿状态。
  2. 为了避免协程长时间无法获取锁，自1.8版本以来增加了一个状态，即Mutex的Starving状态。这个状态下不会自旋，一旦有协程释放锁，那么一定会唤醒一个协程并成功加锁。

解锁逻辑

• code

  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)
  	}
  }
  

• 如果直接解锁一个没有被锁定的锁，抛出异常

• 判断锁是否为饥饿状态

  • 如果锁不为饥饿状态，且锁不为（锁住、唤醒、饥饿）状态的任一，直接解锁
  • 如果为上面三种情况的一种，需要唤醒在等待队列中的goroutine
  • 如果锁处于饥饿状态，直接唤醒等待队列中的goroutine.