锁的实现原理分析
文章目录
什么是锁
- 锁的本质，就是一种资源，是由操作系统维护的一种专门用于同步的资源
- 比如说互斥锁，说白了就是一种互斥的资源。只能有一个进程（线程）占有。当一个进程（线程）通过竞争获得锁的时候，其他进程（或线程）将得不到这把锁。这是内核代码决定的
- 如果我们希望某种资源在多个进程(线程/协程)之间共享，但是某一时刻最多有一个进程占有，这不就是互斥锁的概念吗，也就是说，我们希望自己的资源也变成一种锁
- 最简单的办法就是将自己的资源和操作系统定义好的锁绑定到一起。也就是说，进程要获取我的资源之前，必须要获得操作系统的锁。进一步说，得锁得资源，失锁失资源。这样的话，我们的资源也变成了一把锁
为什么使用锁
并发编程中保证数据一致性和安全性的
Golang中的锁
Golang的提供的同步机制有sync模块下的Mutex、WaitGroup以及语言自身提供的chan等。这些同步的方法都是以runtime中实现的底层同步机制（cas、atomic、spinlock、sem）为基础的
cas、atomic
cas(Compare And Swap)和原子运算是其他同步机制的基础
- 原子操作：指那些不能够被打断的操作被称为原子操作，当有一个CPU在访问这块内容addr时，其他CPU就不能访问
- CAS：比较及交换，其实也属于原子操作，但它是非阻塞的，所以在被操作值被频繁变更的情况下,CAS操作并不那么容易成功，不得不利用for循环以进行多次尝试
自旋锁(spinlock)
自旋锁是指当一个线程在获取锁的时候，如果锁已经被其他线程获取，那么该线程将循环等待，然后不断地判断是否能够被成功获取，知直到获取到锁才会退出循环。获取锁的线程一直处于活跃状态Golang中的自旋锁用来实现其他类型的锁,与互斥锁类似，不同点在于，它不是通过休眠来使进程阻塞，而是在获得锁之前一直处于活跃状态(自旋)
信号量
实现休眠和唤醒协程的一种方式
信号量有两个操作P和V
P(S):分配一个资源
- 资源数减1：S=S-1
- 进行以下判断
如果S<0，进入阻塞队列等待被释放 如果S>=0,直接返回
V(S):释放一个资源
- 资源数加1：S=S+1
- 进行如下判断
如果S>0，直接返回
如果S<=0,表示还有进程在请求资源，释放阻塞队列中的第一个等待进程
golang中信号量操作：runtime/sema.go
P操作：runtime_Semacquire
V操作：runtime_Semrelease
其实就是1表示有资源可用，0和负数表示资源被占用且有多少个进程在等待资源
mutex
mutex的使用
下面我们通过一个简单的示例的看一下互斥锁的使用
package main
import (
"fmt"
"sync"
)
var num int
var mtx sync.Mutex
var wg sync.WaitGroup
func add() {
mtx.Lock() //由于零值，mutex无需实例化，声明即可使用
defer mtx.Unlock()
defer wg.Done()
num += 1
}
func main() {
//一百个协程每个协程对同一个数进行+1，使用锁进行独占，如果不加锁，最后的num不一定是100
for i := 0; i < 100; i++ {
wg.Add(1)
go add()
}
wg.Wait()
fmt.Println("num:", num)
}mutex的必要性
为什么不能把互斥锁替换成原子(atomic)操作
锁在高度竞争时会不断挂起恢复线程从而让出cpu资源，原子变量在高度竞争时会一直占用cpu;而且原子操作是线程级别的，不支持协程，所以对于互斥锁和原子操作的选择需要根据实际情况进行决定
mutex演进
最新版的mutex比较复杂，初看很难理清它的逻辑，但是mutex最初的时候只是简单的拿不到锁就休眠然后等待唤醒，中间经历了什么过程才演变成现在的逻辑，通过一步一步的探究：为什么这么做，我们可以更加深入的了解mutex的设计思想
互斥锁
type Mutex struct {
state int32
sema uint32
}
const (
mutexLocked = 1 << iota
mutexWoken
mutexWaiterShift = iota //根据 mutex.state >> mutexWaiterShift 得到当前等待的 goroutine 数目
)state表示当前锁的状态，是一个共用变量
Lock方法申请对 mutex 加锁的时候分两种情况
无冲突 通过 CAS 操作把当前状态设置为加锁状态
有冲突 通过调用 semacquire 函数来让当前 goroutine 进入休眠状态，等待其他协程释放锁的时候唤醒
//如果已经加锁，那么当前协程进入休眠阻塞，等待唤醒
func (m *Mutex) Lock() {// 快速加锁：CAS更新state为locked if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) { return } awoke := false //当前goroutine是否被唤醒 for { old := m.state // 保存当前state的状态 new := old | mutexLocked // 新值locked位设置为1 // 如果当前处于加锁状态，新到来的goroutine进入等待队列 if old&mutexLocked != 0 { new = old + 1<<mutexWaiterShift } if awoke { //如果被唤醒，新值需要重置woken位为 0 new &^= mutexWoken }
} if raceenabled { raceAcquire(unsafe.Pointer(m)) }// 两种情况会走到这里：1.休眠中被唤醒 2.加锁失败进入等待队列 // CAS 更新，如果更新失败，说明有别的协程抢先一步，那么重新发起竞争。 // 由于CAS是非阻塞的操作，所以如果失败的时候需要通过for循环进行多次重试 if atomic.CompareAndSwapInt32(&m.state, old, new) { // 如果更新成功，有两种情况 // 1.如果为 1，说明当前 CAS 是为了更新 waiter 计数 // 2.如果为 0，说明是抢锁成功，那么直接 break 退出。 if old&mutexLocked == 0 { break } runtime_Semacquire(&m.sema) // 此时如果 sema <= 0 那么阻塞在这里等待唤醒，也就是卡住。走到这里都是要休眠了。 awoke = true // 有人释放了锁，然后当前 goroutine 被 runtime 唤醒了，设置 awoke true }}
UnLock 解锁分两步
解锁，通过CAS操作把当前状态设置为解锁状态
唤醒休眠协程，CAS操作把当前状态的waiter数减1,然后唤醒休眠goroutine
//锁没有和某个特定的协程关联，可以由一个协程lock，另一个协程unlock
func (m *Mutex) Unlock() {
if raceenabled {
_ = m.state
raceRelease(unsafe.Pointer(m))
}
// CAS更新state的状态为locked 注意：解锁的瞬间可能会有新的协程到来并抢到锁
new := atomic.AddInt32(&m.state, -mutexLocked)
// 释放了一个没上锁的锁会panic：原先的lock位为0
if (new+mutexLocked)&mutexLocked == 0 {
panic("sync: unlock of unlocked mutex")
}//判断是否需要释放资源 old := new for { /** * 不需要唤醒的情况 * 1.等待队列为0 * 2.已经有协程抢到锁（上面的瞬间抢锁） * 3.已经有协程被唤醒 */ if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken) != 0 { return } //将waiter计数位减一，并设置state为woken(唤醒) //问：会同时有多个被唤醒的协程存在吗 new = (old - 1<<mutexWaiterShift) | mutexWoken if atomic.CompareAndSwapInt32(&m.state, old, new) { runtime_Semrelease(&m.sema) // cas成功后，再做sema release操作，唤醒休眠的 goroutine return } old = m.state }}
知识点
- 使用&来判断位值，使用|来设置位值，使用&^来清空位置(内存对齐)
第一代互斥锁的问题
- 处于休眠中的goroutine优先级低于当前活跃的，unlock解锁的瞬间最新的goroutine会抢到锁，如果一直有活跃的进程在抢锁，那么处于休眠中的进程可能一直都拿不到锁
- 大多数果抢锁的时间很短，但是所有的goroutine都要休眠，可能刚睡下就被叫起来了，这样很增加runtime调度开销
自旋锁
Lock 方法申请对 mutex 加锁的时候分三种情况
无冲突 通过 CAS 操作把当前状态设置为加锁状态
有冲突 开始自旋，并等待锁释放，如果其他 goroutine 在这段时间内释放了该锁，直接获得该锁；如果没有释放，进入3
有冲突 通过调用 semacquire 函数来让当前 goroutine 进入等待状态，等待其他协程释放锁的时候唤醒
func (m *Mutex) Lock() {
//快速加锁，逻辑不变
if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
if race.Enabled {
race.Acquire(unsafe.Pointer(m))
}
return
}awoke := false iter := 0 for { old := m.state new := old | mutexLocked if old&mutexLocked != 0 { // 如果当前己经上锁，那么判断是否可以自旋 //短暂的自旋过后如果无果，就只能通过信号量让当前goroutine进入休眠等待了 if runtime_canSpin(iter) { // Active spinning makes sense. /** * 自旋的操作：设置state为woken,这样在unlock的时候就不会唤醒其他协程. * 自旋的条件： * 1.当前协程未被唤醒 !awoke * 2.其他协程未被唤醒 old&mutexWoken == 0 * 3.等待队列大于0 */ if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 && atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) { awoke = true } //进行自旋操作 runtime_doSpin() iter++ continue } new = old + 1<<mutexWaiterShit } if awoke { if new&mutexWoken == 0 { panic("sync: inconsistent mutex state") } new &^= mutexWoken } if atomic.CompareAndSwapInt32(&m.state, old, new) { if old&mutexLocked == 0 { break } runtime_Semacquire(&m.sema) awoke = true iter = 0 } } if race.Enabled { race.Acquire(unsafe.Pointer(m)) }}
path: runtime/proc.go
const (
mutex_unlocked = 0
mutex_locked = 1
mutex_sleeping = 2
active_spin = 4
active_spin_cnt = 30
passive_spin = 1
)
/**
* 有四种情况会返回false
* 1.已经执行了很多次 iter >= active_spin 默认为4。避免长时间自旋浪费CPU
* 2.是单核CPU ncpu <= 1 || GOMAXPROCS < 1 保证除了当前运行的Goroutine之外，还有其他的Goroutine在运行
* 3.没有其他正在运行的p
* 4 当前P的G队列为空 避免自旋锁等待的条件是由当前p的其他G来触发，这样会导致再自旋变得没有意义，因为条件永远无法触发
*/
func sync_runtime_canSpin(i int) bool {
// sync.Mutex is cooperative, so we are conservative with spinning.
// Spin only few times and only if running on a multicore machine and
// GOMAXPROCS>1 and there is at least one other running P and local runq is empty.
// As opposed to runtime mutex we don't do passive spinning here,
// because there can be work on global runq or on other Ps.
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
}
// 自旋逻辑
// procyeld函数内部循环调用PAUSE指令，PAUSE指令什么都不做，但是会消耗CPU时间
// 在这里会执行30次PAUSE指令消耗CPU时间等待锁的释放；
func sync_runtime_doSpin() {
procyield(active_spin_cnt)
}
TEXT runtime·procyield(SB),NOSPLIT,$0-0
MOVL cycles+0(FP), AX
again:
PAUSE
SUBL $1, AX
JNZ again
RET问题：
- 自旋锁解决了所有进程都要休眠的问题，每个进程在第一次拿不到锁的时候都会首先自旋等待一小段时间，如果还没拿到锁才会进入休眠，但是对于休眠进程优先级低的问题还是没有解决
公平锁
基本逻辑
- Mutex 两种工作模式，normal 正常模式，starvation 饥饿模式。normal 情况下锁的逻辑与老版相似，休眠的 goroutine 以 FIFO 链表形式保存在 sudog 中，被唤醒的 goroutine 与新到来活跃的 goroutine 竞解，但是很可能会失败。如果一个 goroutine 等待超过 1ms，那么 Mutex 进入饥饿模式
- 饥饿模式下，解锁后，锁直接交给 waiter FIFO 链表的第一个，新来的活跃 goroutine 不参与竞争，并放到 FIFO 队尾
- 如果当前获得锁的 goroutine 是 FIFO 队尾，或是等待时长小于 1ms，那么退出饥饿模式
- normal 模式下性能是比较好的，但是 starvation 模式能减小长尾 latency
LOCK流程：
无冲突 通过 CAS 操作把当前状态设置为加锁状态
有冲突 开始自旋 如果是饥饿模式禁止自旋，开始自旋，并等待锁释放，如果其他 goroutine 在这段时间内释放了该锁，直接获得该锁；如果没有释放，进入3
有冲突,且已经过了自旋阶段 通过调用 semacquire 函数来让当前 goroutine 进入等待状态，等待其他协程释放锁的时候唤醒，休眠前：如果是饥饿模式，把当前协程放到队列最前面；唤醒后：如果是饥饿模式唤醒的，直接获得锁
type Mutex struct {
state int32
sema uint32
}
// A Locker represents an object that can be locked and unlocked.
type Locker interface {
Lock()
Unlock()
}//为什么使用位掩码表达式
//第3位到第32位表示等待在mutex上协程数量
const (
mutexLocked = 1 << iota // mutex is locked
mutexWoken
mutexStarving //新增饥饿状态
mutexWaiterShift = iota
starvationThresholdNs = 1e6 //饥饿状态的阈值：等待时间超过1ms就会进入饥饿状态
)func (m *Mutex) Lock() {
//快速加锁：逻辑不变
if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
if race.Enabled {
race.Acquire(unsafe.Pointer(m))
}
return
}var waitStartTime int64 //等待时间
starving := false //饥饿标记
awoke := false //唤醒标记
iter := 0 //循环计数器
old := m.state //保存当前锁状态
for {
// 自旋的时候增加了一个判断：如果处于饥饿状态就不进入自旋，因为饥饿模式下，释放的锁会直接给等待队列的第一个，当前协程直接进入等待队列
if old&(mutexLocked|mutexStarving) == mutexLocked && 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
}
new := old
// 当mutex不处于饥饿状态的时候，将new值设置为locked，也就是说如果是饥饿状态，新到来的goroutine直接排队
if old&mutexStarving == 0 {
new |= mutexLocked
}
// 当mutex处于加锁锁或者饥饿状态时，新到来的goroutine进入等待队列
if old&(mutexLocked|mutexStarving) != 0 {
new += 1 << mutexWaiterShift } // 当等待时间超过阈值，当前goroutine切换mutex为饥饿模式，如果未加锁，就不需要切换 if starving && old&mutexLocked != 0 { new |= mutexStarving } if awoke { if new&mutexWoken == 0 { throw("sync: inconsistent mutex state") } new &^= mutexWoken } if atomic.CompareAndSwapInt32(&m.state, old, new) { // mutex 处于未加锁，正常模式下，当前 goroutine 获得锁 if old&(mutexLocked|mutexStarving) == 0 { break // locked the mutex with CAS } // 如果已经在排队了，就排到队伍的最前面 queueLifo := waitStartTime != 0 if waitStartTime == 0 { waitStartTime = runtime_nanotime() } // queueLifo 为真的时候，当前goroutine会被放到队头， // 也就是说被唤醒却没抢到锁的goroutine放到最前面 runtime_SemacquireMutex(&m.sema, queueLifo) // 当前goroutine等待时间超过阈值，切换为饥饿模式，starving设置为true 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")
}
// 如果切换为饥饿模式，等待队列计数减1
delta := int32(mutexLocked - 1<
delta -= mutexStarving
}
atomic.AddInt32(&m.state, delta)
break
}
awoke = true
iter = 0
} else {
old = m.state
}
}if race.Enabled {
race.Acquire(unsafe.Pointer(m))
}
}
UnLock 解锁分两步
解锁，通过CAS操作把当前状态设置为解锁状态
唤醒休眠协程，CAS操作把当前状态的waiter数减1,然后唤醒休眠goroutine，如果是饥饿模式的话，唤醒等待队列的第一个
func (m *Mutex) Unlock() {
if race.Enabled {
_ = m.state
race.Release(unsafe.Pointer(m))
}new := atomic.AddInt32(&m.state, -mutexLocked)
if (new+mutexLocked)&mutexLocked == 0 {
throw("sync: unlock of unlocked mutex")
}if new&mutexStarving == 0 {
// 正常模式
old := new
for {
/**
* 不需要唤醒的情况
* 1.等待队列为0
* 2.已经有协程抢到锁（上面的瞬间抢锁）
* 3.已经有协程被唤醒
* 4.处于饥饿模式 在饥饿模式获取到锁的协程仍然处于饥饿状态，新的goroutine无法获取到锁
*/
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)
return
}
old = m.state
}
} else {
// 饥饿模式
runtime_Semrelease(&m.sema, true)
}
}
公平锁解决了休眠进程有可能一直拿不到锁的问题，如果有进程等待了超过1ms还没有拿到锁，那么mutex会进入饥饿模式，释放的资源会优先给等待队列的第一个进程
参考文献
【1】《go sync.Mutex 设计思想与演化过程 （一）博客园-暮夏》
【2】《GO: sync.Mutex 的实现与演进 简书-董泽润》
【3】《golang之sync.Mutex互斥锁源码分析 简书-freelang》
【4】《Golang同步机制的实现 go语言中文网-无心之祸》
【5】《Golang 并发编程与同步原语 segmentfault-draveness》
【6】《锁的本质 csdn-DIY-GEEKER》
手机扫一扫
移动阅读更方便
你可能感兴趣的文章