Go-知识并发控制mutex

gitio: https://a18792721831.github.io/

1. 介绍

互斥锁是并发程序中对共享资源进行访问控制的主要手段，Go 语言提供了非常简单易用的 Mutex。
Mutex 是结构体类型，对外暴露了 Lock 和 Unlock 两个方法，用于加锁和解锁。

2. 数据结构

2.1 Mutex 结构体

在源码包 src/sync/mutex.go中定义了互斥锁的数据结构：

Mutex.state 表示互斥锁的状态，比如是否被锁定等等
Mutex.sema 表示信号量，协程阻塞等待该信号量，解锁的写成释放信号量，从而唤醒等待信号量的协程

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

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

协程之间的抢锁实际上是抢给 Locked 赋值的权利，能给 Locked 赋值 1，就说明抢锁成功。
抢不到就阻塞等待 Mutex.sema 信号量，一旦持有锁的协程解锁，等待的协程就会依次被唤醒。

2.2 Mutex 方法

Mutex 对外提供的方法主要是加锁和解锁：

• Lock(): 加锁
• Unlock(): 解锁
• TryLock(): 非阻塞的方式尝试加锁(Go 1.18 引入)

3. 加锁解锁过程

3.1 简单加锁

假设当前只有一个协程在加锁，没有其他协程干扰：

加锁过程中会判断 Locked 的标志位是否为0，如果是0则把 Locked 位置为1，代表大锁成功。
如上图所示，加锁成功后，只有 Locked 位置为1，其他位置都没变。

func (m *Mutex) Lock() {
	// 快速尝试，使用 cas 进行加锁，如果 state 等于 0 表示无锁
	if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
		return
	}
	// 否则就需要慢加锁了
	m.lockSlow()
}

func (m *Mutex) lockSlow() {
    // 等待开始时间
	var waitStartime int64
	// 是否饥饿
	starving := false
	// 是否有协程被唤醒
	awoke := false
	iter := 0
	// 获取 state 值
	old := m.state
	// 死循环
	for {
		// old & mutexLocked 表示取出 Locked 位
		// old & mutexStarving 表示取出 Starving 位
		// old & (mutexLocked | mutexStarving) 表示取出 Locked 或 starving 位
		// 所以 old & (mutexLocked|mutexStarving) == mutexLocked 表示 Locked 或 Starving 等于 1。
		// 要么有锁，要么饥饿
		// 并且当前 goroutine 可以进行自旋
		if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
			// 没有饥饿 而且 没有协程被唤醒 并且 Waiter 等待的协程不为空
			// 尝试使用 CAS 设置 Worken 值，表示有 goroutine 被唤醒了，在自旋等待锁
			if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
				atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
				// 是否有协程被唤醒标志置为 true 
				awoke = true
			}
			// 自旋 ，我认为可以理解为 continue ..
			runtime_doSpin()
			// 自旋次数加1
			iter++
			// 重新获取最新的 state 值
			old = m.state
			// 自旋
			continue
		}
		// 重新获取 state
		new := old
		// 没有饥饿
		if old&mutexStarving == 0 {
		    // 新值设置为加锁了
			new |= mutexLocked
		}
		// 如果存在饥饿或者有锁
		if old&(mutexLocked|mutexStarving) != 0 {
		    // Waiter 数量增加 1 
		    // 1< starvationThresholdNs
			// 重新获取 state 
			old = m.state
			// 如果存在饥饿
			if old&mutexStarving != 0 {
			    // 有锁 或 有唤醒的协程 或者
			    // 没有等待的协程
				if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
				    // 不能即处于饥饿，有存在唤醒，还没有锁
				    // 不能没有等待的协程，又处于饥饿状态
					throw("sync: inconsistent mutex state")
				}
				// 表示 1 减去 1000 = -7 
				delta := int32(mutexLocked - 1<>mutexWaiterShift == 1 {
					// -7 减去 -4 ,移除饥饿模式
					delta -= mutexStarving
				}
				// 更新 state
				// 防止自旋死锁
				atomic.AddInt32(&m.state, delta)
				break
			}
			awoke = true
			iter = 0
		} else {
			old = m.state
		}
	}
}

3.2 加锁被阻塞

假设加锁时，锁已经被其他协程占用了：

当协程B对一个已被占用的锁再次加锁时，Waiter 计数器增加1，此时协程B被阻塞，直到 Locked 值变为0后才会被唤醒。

3.3 简单解锁

假设解锁时，没有其他协程阻塞：

由于没有其他协程阻塞等待加锁，所以此时解锁时只需要把 Locked 位置为0即可，不需要释放信号量。

func (m *Mutex) Unlock() {
	// 使用 CAS 快速解锁
	new := atomic.AddInt32(&m.state, -mutexLocked)
	// 如果 CAS 解锁失败，那么尝试慢解锁
	if new != 0 {
		m.unlockSlow(new)
	}
}

func (m *Mutex) unlockSlow(new int32) {
    // 如果 无锁 new 最低位 1，因为在 Unlock 中已经减去 1 了 ，1+1 = 10 ， 0 & 1 = 0 ， 0 == 0 
    // 如果 有锁 new 最低位 0，0+1 = 1, 1 & 1 = 1, 1 == 0
    // 所以，无锁异常，有锁继续 
	if (new+mutexLocked)&mutexLocked == 0 {
		throw("sync: unlock of unlocked mutex")
	}
	// 如果没有饥饿
	if new&mutexStarving == 0 {
	    // 临时赋值
		old := new
		for {
		    // Waiter 等于 0 ，没有等待的协程 或
		    // 处于 饥饿，有锁，有唤醒 ，结束
			if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
				return
			}
			// Waiter 减 1，并且 保留唤醒状态
			new = (old - 1<