Go 源码之互斥锁 Mutex

文章目录

一、总结
二、源码
- （一）Mutex
- （二） Lock
- （三）Unlock
三、常见问题
- - - - 有劳各位看官 `点赞、关注➕收藏 `，你们的支持是我最大的动力！！！
      - 接下来会不断更新 `golang` 的一些`底层源码及个人开发经验`（个人见解）！！！
      - 同时也欢迎大家在评论区提问、分享您的经验和见解！！！

一、总结

锁不可复制：拷贝互斥锁同时会拷贝锁的状态，容易造成死锁
不是可重入锁，并且一个协程上锁，可以由另外一个协程解锁
mutex 锁结构
- state：32位，锁状态，bitmap 设计，
  - 1 mutexLocked ：低1位锁定状态
  - 2 mutexWoken ：低2位，从正常模式被唤醒
  - 3 mutexStarving 是低3位，进入饥饿模式
  - 1. mutexWaiterShift 剩下 29 位，当前互斥锁上等待者的数量
- sema：协程等待信号量，用于控制goroutine的阻塞与唤醒
上锁
解锁
锁的两种模式

正常模式
在正常模式下等待的 g 按照先进先出的方式获取锁
新 g 会自旋，并且和刚唤醒的 g 竞争锁，新 g 会优先获得锁，会导致刚被唤起的 g 一直获取不到锁，
这种情况的出现会导致线程长时间被阻塞下去，所以Go语言在1.9中进行了优化，引入了饥饿模式
饥饿模式
为了解决等待 goroutine 队列的长尾问题（饿死）
当 g 超过 1ms 没有获取到锁，就会将当前互斥锁切换到饥饿模式
等待的 g 按照先进先出的方式获取锁
饥饿模式下，新进来的 G 不会参与抢锁也不会进入自旋状态，会直接进入等待队列的尾部。
在这种情况下，这个被唤醒的 goroutine 会优先加入到等待队列的前面，防止饿死
如果一个 goroutine 获得了互斥锁并且它在队列的末尾或者它等待的时间少于 1ms，那么当前的互斥锁就会切换回正常模式

二、源码

（一）Mutex

const (
   mutexLocked = 1 << iota 									// 1 mutex 锁定状态
   mutexWoken 													// 2 mutex 从正常模式被唤醒
   mutexStarving 												// 4 mutex进入饥饿状态
   mutexWaiterShift = iota 									// 3 当前互斥锁上等待者的数量
)

type Mutex struct { 												// Mutex 不可被复制
    state int32 // 32位，锁状态，bitmap 设计，低三位表示锁的状态，剩下 29 位表当前互斥锁上等待者的数量
    sema  uint32 // 缓冲信号量，用来控制等待goroutine的阻塞休眠和唤醒，可以理解为一个队列
}

（二） Lock

直接 CAS 进行原子操作上锁，成功则返回，失败则执行 lockSlow()
上锁失败，执行 lockSlow()，内部持续 for 循环
- 支持自旋（正常模式、cpu空闲、自旋次数<4），则进入自旋
- 不支持自旋：两种模式
  - 正常模式：加入尾部队列，按照先进先出的方式加入队列等待获取锁
  - 饥饿模式：当goroutine超过1ms没有获取到锁，就会将当前互斥锁切换到饥饿模式，如果当前goroutine 存在队列中，则移动到队头，然后按照先进先出的方式获取锁，防止饿死


func (m *Mutex) Lock() {
	if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) { 	// 直接 CAS 修改锁的状态，将 state=0 改为 1 
		return
	}
	m.lockSlow()															// CAS 无法直接上锁，则执行慢路径
}
func (m *Mutex) lockSlow() {
	var waitStartTime int64 											// 用来计算waiter的等待时间
	starving := false 													// 是否是饥饿模式
	awoke := false 														// 是否唤醒
	iter := 0 															// 自旋次数
	old := m.state 														// 旧的锁状态
	for {
    // 支持自旋：锁不是饥饿模式 && cpu 支持继续自旋（<=4次）
		if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
      // g不是唤醒状态 && 
      // 没有其他正在唤醒的goroutine && 
      // 等待队列中有正在等待的goroutine
      // && 尝试将当前锁的低2位的Woken状态位设置为1，表示已被唤醒, 这是为了通知在解锁Unlock()中不再唤醒其他waiter
			if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
				atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
				awoke = true                                     // 设置当前goroutine唤醒成功
			}
			runtime_doSpin()                                    // 进行自旋
			iter++                                              // 自旋次数++
			old = m.state                                       //更新锁状态
			continue
		}
        new := old
		
		if old&mutexStarving == 0 { 
			new |= mutexLocked                          // 非饥饿模式下进行加锁
		}
                
		if old&(mutexLocked|mutexStarving) != 0 {
			new += 1 << mutexWaiterShift                // 等待着数量+1
		}
		
	
		if starving && old&mutexLocked != 0 {
			new |= mutexStarving	                    // 加锁的情况下切换为饥饿模式
		}
               
		if awoke {                                     // goroutine 唤醒的时候进行重置标志
			if new&mutexWoken == 0 {
				throw("sync: inconsistent mutex state")
			}
			new &^= mutexWoken
		}
         if atomic.CompareAndSwapInt32(&m.state, old, new) {     //设置新的状态
			if old&(mutexLocked|mutexStarving) == 0 {
				break 
			}
			queueLifo := waitStartTime != 0
			if waitStartTime == 0 {                       // 判断是不是第一次加入队列
				waitStartTime = runtime_nanotime()         // 如果之前就在队列里面等待了，加入到对头
			}        
			runtime_SemacquireMutex(&m.sema, queueLifo, 1) // 阻塞等待
          // 检查锁是否处于饥饿状态
			starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
			old = m.state
          // 如果锁处于饥饿状态，直接抢到锁
			if old&mutexStarving != 0 {
				if old&(mutexLocked|mutexWoken) != 0 || old>>mutexWaiterShift == 0 {
					throw("sync: inconsistent mutex state")
				}
              //设置标志，进行加锁并且waiter-1
				delta := int32(mutexLocked - 1<<mutexWaiterShift)
				//如果是最后一个的话清除饥饿标志
              if !starving || old>>mutexWaiterShift == 1 {
                 //退出饥饿模式				
					delta -= mutexStarving
				}
				atomic.AddInt32(&m.state, delta)
				break
			}
			awoke = true
			iter = 0
		} else {
			old = m.state
		}
	}
}

（三）Unlock

直接 CAS 进行原子操作解锁，成功则返回，失败则执行 unlockSlow()
解锁失败，执行 unlockSlow()
- 正常模式：
  - 如果当前队列中没有waiter，只有自己本身，直接解锁返回
  - 如果当前队列中有waiter，解锁后唤醒下个等待者 runtime_Semrelease(&m.sema, false, 1)
- 饥饿模式：
  - 饥饿模式直接将锁的控制权交给队列中队头等待的waiter

func (m *Mutex) Unlock() {
	new := atomic.AddInt32(&m.state, -mutexLocked)  // 直接 CAS 修改锁的状态 
	if new != 0 {
        // 不等于0说明解锁失败，
		m.unlockSlow(new)
	}
}
func (m *Mutex) unlockSlow(new int32) {
        //解锁一个未加锁的Mutex会报错
	if (new+mutexLocked)&mutexLocked == 0 {
		throw("sync: unlock of unlocked mutex")
	}
	if new&mutexStarving == 0 {
		old := new
		for {
			// 正常模式下，没有waiter或者在处理事情的情况下直接返回
			if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
				return
			}
			//如果有等待者，设置mutexWoken标志，waiter-1，更新state
			new = (old - 1<<mutexWaiterShift) | mutexWoken
			if atomic.CompareAndSwapInt32(&m.state, old, new) {
				runtime_Semrelease(&m.sema, false, 1)
				return
			}
			old = m.state
		}
	} else {
		// 饥饿模式下会直接将mutex交给下一个等待的waiter，让出时间片，以便waiter执行
		runtime_Semrelease(&m.sema, true, 1)
	}
}

三、常见问题

1. sema 字段的含义作用
在正常模式下，一个goroutine先通过自旋方式获得锁，如果还不能获取锁，则通过信号量进行排队等
待
(所有等待者都会按照先入先出的顺序排队)但是当被唤醒后，第一个等待者并不会立即获得锁，而是需要和那些正在处于自旋阶段，尚未加入到队列中的routine竞争,如果抢不到锁的话，重新插入到队列的头部，而当这个goroutine加锁等待的时间超过了1ms之后，它会把mutex由正常模式切换到饥饿模式，这种模式下锁的所有权直接传递给头部的routine。后来者不会自旋，也不会尝试获取锁，直接加到队列尾部

2.什么是CAS，什么是原子操作
CAS（Compare and Swap）比较并交换，比较两个值，如果他们两者相等就把他们交换。这是一个由CPU硬件提供并实现的原子操作。
原子操作：操作系统提高的锁机制来保证操作的原子性和线程安全性。这种锁机制可以使执行原子操作的 CPU 独占内存总线或者缓存，并防止其他 CPU 对同一内存地址进行读写操作，从而避免了数据竞争的问题
具体来说，在执行原子操作时，CPU 会向内存总线或者缓存发送锁请求信号，然后等待锁授权。一旦锁授权成功，CPU 就可以将操作的结果写入内存，然后释放锁。其他 CPU 在锁被释放之前不能对同一内存地址进行读写操作，从而保证了操作的原子性和线程安全性。
需要注意的是，原子操作增加 CPU 的开销和内存带宽的消耗

3. 锁的正常模式和饥饿模式？

4.为什么锁不可复制
因为互斥锁没有绑定 gid，复制锁会复制锁的状态，容易出现死锁

5.什么情况下mutex会从饥饿模式变成正常模式呢?
如果当前 goroutine 等待锁的时间超过了 1ms，互斥锁就会切换到饥饿模式。
如果当前 goroutine 是互斥锁最后一个waiter，或者等待的时间小于 1ms，互斥锁切换回正常模式。

5. goroutine能进入自旋的条件

当前互斥锁处于正常模式，不处于饥饿模式
积累的自旋次数小于最大自旋次数（active_spin=4
cpu 核数大于 1
有空闲的 P
当前 goroutine 所挂载的 P 下，本地待运行队列为空

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