GoLangsync.Mutex实现分析

定义介绍

先从sync.Mutex定义出发&＃xff0c;看看mutex是如何实现; sync.Mutex源码

• sema: 信号量, 用于阻塞和唤醒goroutine; 注意, sema并不存储阻塞的goroutine的数量;
• state: 状态(精彩之处); 它的使用被切分2部分&＃xff0c;一部分是状态标识位&＃xff0c;另一部分存储wait goroutine的数量;

所以&＃xff0c;锁竞争可以理解为 当前goroutine是否将锁定标志位 设置成1; 正常模式是多goroutine抢占锁定标志位的过程&＃xff1b;饥饿模式是饥饿标志位被设置成1;

模式转换

结合代码分析以上3种方式&＃xff0c;实际是2种&＃xff0c;即正常模式和饥饿模式&＃xff1b;我这里将正常模式细分出简单模式和正常模式; 首先回忆一下锁的用法&＃xff0c; 即

func xxxxx() {mu.Lock()defer mu.Unlock()......
}


当多个goroutine 抢锁时&＃xff0c;成功的可以继续执行&＃xff0c;失败的则被阻塞&＃xff1b;等同于 抢锁成功的可以从Lock函数中返回&＃xff0c;失败则留在Lock函数中无法返回。

简单模式

• 加锁
• 解锁

func (m *Mutex) Lock() {if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {...return}m.lockSlow()
}func (m *Mutex) Unlock() {...new :&＃61; atomic.AddInt32(&m.state, -mutexLocked)if new !&＃61; 0 {m.unlockSlow(new)}
}


代码容易理解 当G抢锁时state&＃61;0变成state&＃61;1&＃xff0c;加锁成功; 当G解锁时state&＃61;1变成state&＃61;0 解锁成功;

正常模式

• 加锁
• 解锁

func (m *Mutex) unlockSlow(new int32) {...if new&mutexStarving &＃61;&＃61; 0 {old :&＃61; newfor {if old>>mutexWaiterShift &＃61;&＃61; 0 || old&(mutexLocked|mutexWoken|mutexStarving) !&＃61; 0 {return}new &＃61; (old - 1<<mutexWaiterShift) | mutexWokenif atomic.CompareAndSwapInt32(&m.state, old, new) {runtime_Semrelease(&m.sema, false, 1)return}old &＃61; m.state}} else {...}
}


先看正常模式的解锁操作&＃xff0c;已将饥饿模式代码隐藏; new &＃61; m.state - mutexLocked, 意味着 锁定标志被解除; 正常模式解锁 会将state 唤醒标志置1, 并唤醒G

old :&＃61; newfor {//old>>mutexWaiterShift &＃61; 0 表示无等待的G; 这种情况可以直接返回; 因为没有需要唤醒的G;//old>>mutexWaiterShift > 0 表示存在等待G;////此时: old 可能为 ..1 000 或者 ..1 010; 因为已解锁且非饥饿////if old>>mutexWaiterShift &＃61;&＃61; 0 || old&(mutexLocked|mutexWoken|mutexStarving) !&＃61; 0 {return}//// (old - 1<//// old 为 ...1 000 --> ...1 010//new &＃61; (old - 1<<mutexWaiterShift) | mutexWokenif atomic.CompareAndSwapInt32(&m.state, old, new) {runtime_Semrelease(&m.sema, false, 1)return}old &＃61; m.state}


runtime_Semrelease(&m.sema, false, 1) 队尾唤醒G
runtime_Semrelease(&m.sema, true, 1) 队首唤醒G

然后看一下正常模式的加锁操作&＃xff1b;已将饥饿模式代码隐藏; 前文所过加锁成功可以等同于Lock函数返回&＃xff0c;所以紧盯break; 接下来将分析如何退出;

func (m *Mutex) lockSlow() {var waitStartTime int64 //G等待时间纳秒starving :&＃61; false //是否饥饿awoke :&＃61; false //是否唤醒iter :&＃61; 0 //自旋计数old :&＃61; m.state //快照状态for {......new :&＃61; old//非饥饿模式加锁 确保锁定标志为1if old&mutexStarving &＃61;&＃61; 0 {new |&＃61; mutexLocked}//等待G计数增加if old&(mutexLocked|mutexStarving) !&＃61; 0 {new &＃43;&＃61; 1 << mutexWaiterShift}// 进入饥饿模式if starving && old&mutexLocked !&＃61; 0 {new |&＃61; mutexStarving}//唤醒标志重置, 因为唤醒G时都会将唤醒标志置1if awoke {if new&mutexWoken &＃61;&＃61; 0 {throw("sync: inconsistent mutex state")}new &^&＃61; mutexWoken}if atomic.CompareAndSwapInt32(&m.state, old, new) {if old&(mutexLocked|mutexStarving) &＃61;&＃61; 0 { //唤醒G抢锁成功 010 解锁时为010break // locked the mutex with CAS}//竞争失败重新等待; 新G将放入等待队列尾&＃xff0c;唤醒G将放入等待队列首;queueLifo :&＃61; waitStartTime !&＃61; 0if waitStartTime &＃61;&＃61; 0 {waitStartTime &＃61; runtime_nanotime()}runtime_SemacquireMutex(&m.sema, queueLifo, 1)//这个位置很关键; 它是唤醒G 开始执行的位置; 阻塞的底层是将G与P解耦&＃xff0c;然后将G交由sema保存;////进入饥饿条件是等待时间 > 1msstarving &＃61; starving || runtime_nanotime()-waitStartTime > starvationThresholdNs//唤醒G 需要重新快照state; 因为G已睡眠一段时间, state可能已改变;old &＃61; m.stateif old&mutexStarving !&＃61; 0 {...}//唤醒G 开启新一轮锁竞争awoke &＃61; trueiter &＃61; 0} else {old &＃61; m.state}}...
}


饥饿模式

当唤醒G等待时间大于1ms时&＃xff0c;即将进入模式&＃xff0c;那么后续锁竞争的G 将直接进入等待队列&＃xff0c;无需竞争;

func (m *Mutex) lockSlow() {var waitStartTime int64 //G等待时间纳秒starving :&＃61; false //是否饥饿awoke :&＃61; false //是否唤醒iter :&＃61; 0 //自旋计数old :&＃61; m.state //快照状态for {...new :&＃61; old...//等待G计数增加if old&(mutexLocked|mutexStarving) !&＃61; 0 {new &＃43;&＃61; 1 << mutexWaiterShift}// 进入饥饿模式if starving && old&mutexLocked !&＃61; 0 {new |&＃61; mutexStarving}...if atomic.CompareAndSwapInt32(&m.state, old, new) {....//竞争失败重新等待; 新G将放入等待队列尾&＃xff0c;唤醒G将放入等待队列首;queueLifo :&＃61; waitStartTime !&＃61; 0if waitStartTime &＃61;&＃61; 0 {waitStartTime &＃61; runtime_nanotime()}runtime_SemacquireMutex(&m.sema, queueLifo, 1)//这个位置很关键; 它是唤醒G 开始执行的位置; 阻塞的底层是将G与P解耦&＃xff0c;然后将G交由sema保存;////进入饥饿条件是等待时间 > 1msstarving &＃61; starving || runtime_nanotime()-waitStartTime > starvationThresholdNs//唤醒G 需要重新快照state; 因为G已睡眠一段时间, state可能已改变;old &＃61; m.state//饥饿模式处理逻辑&＃xff0c;有点绕;if old&mutexStarving !&＃61; 0 {////没有等待G是不会进入饥饿模式的 old>>mutexWaiterShift &＃61;&＃61; 0////此时应该 ..1 100//if old&(mutexLocked|mutexWoken) !&＃61; 0 || old>>mutexWaiterShift &＃61;&＃61; 0 {throw("sync: inconsistent mutex state")}delta :&＃61; int32(mutexLocked - 1<<mutexWaiterShift)//// delta &＃61; 1 - 8 &＃61; -7 饥饿 1.....1 111; 符号位为1&＃xff0c;后3位为111;//if !starving || old>>mutexWaiterShift &＃61;&＃61; 1 {//解除饥饿//-7-4 &＃61; -11 &＃61; 1...1 011 ////此时 0...1 100 &＃43; 1...1 011 运算结果为 ...1 011 解除饥饿模式delta -&＃61; mutexStarving}//等待G计数减1;//此时 0....1 100 &＃43; 1....0 111 运算结果后 为 ...1 111, 仍然为饥饿模式atomic.AddInt32(&m.state, delta)break}...} else {old &＃61; m.state}}...
}


阻塞与解锁

加锁自旋

自旋就是空跑阻止G进入阻塞队列; 因为G的唤醒和阻塞涉及G的调度, 调度是耗时的; 如果通过短暂的自选可以获取锁&＃xff0c;那就避免G的调度耗时; 同时可以看到自旋会浪费CPU的;

func (m *Mutex) lockSlow() {...awoke :&＃61; falseiter :&＃61; 0old :&＃61; m.statefor {if old&(mutexLocked|mutexStarving) &＃61;&＃61; mutexLocked && runtime_canSpin(iter) {if !awoke && old&mutexWoken &＃61;&＃61; 0 && old>>mutexWaiterShift !&＃61; 0 &&atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {awoke &＃61; true}runtime_doSpin()iter&＃43;&＃43;old &＃61; m.statecontinue}new :&＃61; old...}


runtime_canSpin&＃xff0c;进入自旋的条件是严苛的&＃xff0c;尽力避免浪费CPU;

• 不能超过4次
• 多核
• 多P
• 空闲待运行队列

如有不对之处&＃xff0c;欢迎留言评论区