我终于搞懂了"/>
Go锁,我终于搞懂了
自从前两天写了Go互斥锁实现原理后,我感觉我错了,我不应该看源码,看了也没啥用,会了也不敢用。按照源码中的写法,没人想做代码review,也不好做单元测试。
我就是想知道锁是怎么实现的,结果整一堆优化逻辑,感觉懂了又感觉啥都不懂。所以我痛定思痛,我看早期版本行吧。来让我们看2014年的版本
互斥锁:.go
读写互斥锁:.go
互斥锁
这个版本的互斥锁没有饥饿模式、没有自旋、没有各种小的性能优化点。但这段代码把最核心的逻辑展示的十分清楚。
说明
type Mutex struct {state int32sema uint32
}
-
state表示互斥锁的状态,比如是否被锁定等。
-
sema表示信号量,协程阻塞等待该信号量,解锁的协程释放信号量从而唤醒等待信号量的协程。
state的组成如下图所示:
Locked: 表示该Mutex是否已被锁定,0:没有锁定 1:已被锁定。
Woken: 表示是否有协程已被唤醒,0:没有协程唤醒 1:已有协程唤醒,正在加锁过程中。释放锁时,如果正常模式下,不会再唤醒其它协程。
Waiter: 表示阻塞等待锁的协程个数,协程解锁时根据此值来判断是否需要释放信号量
因为在Go互斥锁实现原理已经写过很多基础知识了,这次只把核心逻辑、核心点写一下,如果对其中部分知识不太理解,可以看一下上一篇文章。
源码
// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.// Package sync provides basic synchronization primitives such as mutual
// exclusion locks. Other than the Once and WaitGroup types, most are intended
// for use by low-level library routines. Higher-level synchronization is
// better done via channels and communication.
//
// Values containing the types defined in this package should not be copied.
package syncimport ("sync/atomic""unsafe"
)// A Mutex is a mutual exclusion lock.
// Mutexes can be created as part of other structures;
// the zero value for a Mutex is an unlocked mutex.
type Mutex struct {state int32sema uint32
}// A Locker represents an object that can be locked and unlocked.
type Locker interface {Lock()Unlock()
}const (//status只分为三部分,mutexLocked表示锁是否已经被其它协程占用//mutexWoken表示是否唤起协程,让协程开始抢占锁//mutexWaiterShift表示阻塞等待锁的协程个数,占int32的高30位mutexLocked = 1 << iota // mutex is lockedmutexWokenmutexWaiterShift = iota
)// Lock locks m.
// If the lock is already in use, the calling goroutine
// blocks until the mutex is available.
//加锁过程仍然以将status的Locked位置为1位加锁成功
func (m *Mutex) Lock() {// Fast path: grab unlocked mutex.//如果锁没有被占用、没有唤醒的协程、没有等待加锁的协程,直接加锁,成功便返回if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {if raceenabled {//为false,不用管raceAcquire(unsafe.Pointer(m))}return}//协程主动发起的加锁请求,肯定不是被唤醒的,所以awoke为falseawoke := falsefor {old := m.state//无论如何先尝试加锁。因为如果没被占用,肯定是要加锁的。如果被占用,因为使用CAS,所以Locked位也需要为1new := old | mutexLocked//如果锁是被占用的,则将等待协程值加1if old&mutexLocked != 0 {new = old + 1<<mutexWaiterShift}//如果是被唤醒的,CAS操作之后自己就到终止状态(获得锁或者阻塞),所以需要将Woken位置0if awoke {// The goroutine has been woken from sleep,// so we need to reset the flag in either case.new &^= mutexWoken}//CAS操作// old new//(0,1)不是唤醒的,被占用了 (+1,0,1)//(1,1)是唤醒的,被占用了 (+1,0,1)//(0,0)不是唤醒的,未被占用 (+0,0,1)//(1,0) 是唤醒的,未被占用 (+0,0,1)if atomic.CompareAndSwapInt32(&m.state, old, new) {if old&mutexLocked == 0 { //如果CAS操作时,锁是未被占用,则加锁成功break}runtime_Semacquire(&m.sema) //否则,将当前协程阻塞awoke = true //当该协程被唤醒是,用awoke进行标记}}if raceenabled {raceAcquire(unsafe.Pointer(m))}
}// Unlock unlocks m.
// It is a run-time error if m is not locked on entry to Unlock.
//
// A locked Mutex is not associated with a particular goroutine.
// It is allowed for one goroutine to lock a Mutex and then
// arrange for another goroutine to unlock it.
// 解锁逻辑也比go1.13版本简单很多
func (m *Mutex) Unlock() {if raceenabled {//默认为false,不用管_ = m.stateraceRelease(unsafe.Pointer(m))}// Fast path: drop lock bit.//将值减一,其实就是将locked位置为0new := atomic.AddInt32(&m.state, -mutexLocked)//小技巧,检查解锁的锁是否未被加锁,如果是这种情况,就panicif (new+mutexLocked)&mutexLocked == 0 {panic("sync: unlock of unlocked mutex")}//因为atomic.AddInt32操作m.state使用的是指针,如果没有其它协程操作m.state的话,new=old=m.stateold := newfor {// If there are no waiters or a goroutine has already// been woken or grabbed the lock, no need to wake anyone.//如果没有等待加锁的协程或者当前锁已经唤醒了其它协程,直接返回if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken) != 0 {return}// Grab the right to wake someone.//要唤醒等待协程了,所以需要将woken位置为1,同时将等待数量减一new = (old - 1<<mutexWaiterShift) | mutexWokenif atomic.CompareAndSwapInt32(&m.state, old, new) {runtime_Semrelease(&m.sema)//如果解锁成功,则唤醒一个等待的协程return}old = m.state//否则继续尝试解锁}
}
关注点
-
Unlock中atomic.AddInt32(&m.state, -mutexLocked)是对m.state的指针真行操作,所以其实对m.state进行了真正的减一,这也是CAS操作时atomic.CompareAndSwapInt32(&m.state, old, new),m.state和old值可能一样的原因
-
大家课本上学到的信号量伪代码一般是这样的
wait(S):while S<=0 do no-op;S:=S-1;signal(S):S:=S+1;
感觉和Mutex中的sema不一样。有这种疑问很正常,因为准确的说,Mutex才是真正意义上的信号量,Mutex中的sema虽然翻译为信号量,但其实只是为了将协程阻塞到以sema为地址的队列上,细节可查看semacquire1、semrelease1。
type Mutex struct {state int32sema uint32
}
课本上的伪代码:可以加锁,值减一;不可以加锁,死循环。
Mutex具体实现:可以加锁,Locked位置为1,不可以加锁,协程阻塞。
- 存在等待协程饿死的情况
读写锁
上一篇文章没有写读写锁,主要还是因为go1.13的代码阅读起来太过困难。这次看老版本的代码,简单的多,所以顺便写一下读写锁。
说明
读写锁结构体为:
type RWMutex struct {w Mutex // held if there are pending writerswriterSem uint32 // semaphore for writers to wait for completing readersreaderSem uint32 // semaphore for readers to wait for completing writersreaderCount int32 // number of pending readersreaderWait int32 // number of departing readers
}
-
Mutex:复用写锁,应对读写锁变为写锁的情况。获得写锁首先要获取该锁,如果有一个写锁在进行,那么再到来的写锁将会阻塞于此
-
writerSem:写阻塞等待的信号量,最后一个读者释放锁时会释放信号量
-
readerSem:读阻塞的协程等待的信号量,持有写锁的协程释放锁后会释放信号量
-
readerCount:记录读者个数
-
readerWait:记录写阻塞时读者个数。表示加写锁的时候有几个读锁,当对应数量的读锁解锁完毕后,写锁开始被唤起。主要目的为防止写锁饿死。
源码
// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.package syncimport ("sync/atomic""unsafe"
)// An RWMutex is a reader/writer mutual exclusion lock.
// The lock can be held by an arbitrary number of readers
// or a single writer.
// RWMutexes can be created as part of other
// structures; the zero value for a RWMutex is
// an unlocked mutex.
type RWMutex struct {w Mutex // held if there are pending writerswriterSem uint32 // semaphore for writers to wait for completing readersreaderSem uint32 // semaphore for readers to wait for completing writersreaderCount int32 // number of pending readersreaderWait int32 // number of departing readers
}
//读协程最大个数。
//这个值主要和readerCount相比较。如果加写锁,则将readerCount减去rwmutexMaxReaders
//这是一个小技巧,既能看出是否加了写锁,也能使用这个值方便的还原出读协程的数量
const rwmutexMaxReaders = 1 << 30// RLock locks rw for reading.
// 加读锁
func (rw *RWMutex) RLock() {if raceenabled {_ = rw.w.stateraceDisable()}//加读锁,直接将readerCount值加1//什么情况下readerCount会小于0呢,有协程请求写锁了。这种情况下,加读锁的协程需要阻塞//因为加读锁的协程到达时间比加写锁的晚,所以得等写锁用完了才能执行,否则加写锁的协程会饿死,不讲武德if atomic.AddInt32(&rw.readerCount, 1) < 0 {// A writer is pending, wait for it.runtime_Semacquire(&rw.readerSem)}if raceenabled {//为false,不用管raceEnable()raceAcquire(unsafe.Pointer(&rw.readerSem))}
}// RUnlock undoes a single RLock call;
// it does not affect other simultaneous readers.
// It is a run-time error if rw is not locked for reading
// on entry to RUnlock.
//解读锁
func (rw *RWMutex) RUnlock() {if raceenabled {//为false,不用管_ = rw.w.stateraceReleaseMerge(unsafe.Pointer(&rw.writerSem))raceDisable()}//将加读锁的协程数量值减一//如果没有尝试加写锁的协程,值r肯定大于0,则解锁结束//如果有协程加写锁,就需要看一下自己是否是请求写锁前的最后一个读锁协程了,是的话就要唤醒加写锁协程if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {//说明解锁了未加锁mutex,会panicif r+1 == 0 || r+1 == -rwmutexMaxReaders {raceEnable()panic("sync: RUnlock of unlocked RWMutex")}// A writer is pending.//自己是否是请求写锁前的最后一个读锁协程了,是的话就要唤醒加写锁协程if atomic.AddInt32(&rw.readerWait, -1) == 0 {// The last reader unblocks the writer.runtime_Semrelease(&rw.writerSem)}}if raceenabled {raceEnable()}
}// Lock locks rw for writing.
// If the lock is already locked for reading or writing,
// Lock blocks until the lock is available.
// To ensure that the lock eventually becomes available,
// a blocked Lock call excludes new readers from acquiring
// the lock.
func (rw *RWMutex) Lock() {if raceenabled {_ = rw.w.stateraceDisable()}// First, resolve competition with other writers.//调用mutex加写锁。如果是第一个加写锁的协程,肯定能成功。如果不是第一个,就阻塞了rw.w.Lock()// Announce to readers there is a pending writer.//这个操作实现两个功能,一是将readerCount变为负值,另一个是计算出加写锁时读锁个数r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders// Wait for active readers.//说明加写锁的时候还有读协程在,则加写锁的协程阻塞if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {runtime_Semacquire(&rw.writerSem)}//当加写锁前的所有读锁协程都释放了,加写锁协程会被唤起if raceenabled {raceEnable()raceAcquire(unsafe.Pointer(&rw.readerSem))raceAcquire(unsafe.Pointer(&rw.writerSem))}
}// Unlock unlocks rw for writing. It is a run-time error if rw is
// not locked for writing on entry to Unlock.
//
// As with Mutexes, a locked RWMutex is not associated with a particular
// goroutine. One goroutine may RLock (Lock) an RWMutex and then
// arrange for another goroutine to RUnlock (Unlock) it.
func (rw *RWMutex) Unlock() {if raceenabled {_ = rw.w.stateraceRelease(unsafe.Pointer(&rw.readerSem))raceRelease(unsafe.Pointer(&rw.writerSem))raceDisable()}// Announce to readers there is no active writer.//写锁释放的时候,将readerCount变为正值r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)//如果释放未加锁的mutex,panicif r >= rwmutexMaxReaders {raceEnable()panic("sync: Unlock of unlocked RWMutex")}// Unblock blocked readers, if any.//顺序唤起所有等待的读协程for i := 0; i < int(r); i++ {runtime_Semrelease(&rw.readerSem)}// Allow other writers to proceed.rw.w.Unlock()if raceenabled {raceEnable()}
}// RLocker returns a Locker interface that implements
// the Lock and Unlock methods by calling rw.RLock and rw.RUnlock.
func (rw *RWMutex) RLocker() Locker {return (*rlocker)(rw)
}type rlocker RWMutexfunc (r *rlocker) Lock() { (*RWMutex)(r).RLock() }
func (r *rlocker) Unlock() { (*RWMutex)(r).RUnlock() }
整个流程如下图
关注点
-
对于写锁,有两个阻塞。一是Mutex自身阻塞,另一个是RWMutex的writerSem阻塞。两者的作用不一样。Mutex用于处理写锁与写锁之间的关系,writerSem用于处理写锁与读锁之间的关系
-
源码使用readerWait记录请求写锁时读锁协程个数,当这些协程都释放锁,写锁加锁成功,防止写锁饿死
总结
通过阅读Go锁源码,明白真正的信号量实现逻辑。虽然整体思路上,和操作系统课本讲述是一致的,但是仍然有很多细节不一样。像Go没有选择空等,进行阻塞优化性能;像读写锁的实现要比整型信号量复杂的多。
对于源码的复杂性,我觉得分两个方面看。对于Go这种源码,一点点的性能优化,对于全球那么多使用者来说,收益是巨大的,所以可读性的重要性没那么高。如果是想学习的话,还是看一下早期的、可读性高一些的源码,在理解的基础上再看最新版本的源码,不但入门难度降低,而且能够思考别人是怎么进行思考、进行优化的。
对于具体业务而言,代码一定要有较高的可读性,不然code review的同学或者今后维护的同学,太痛苦!!!
资料
-
go中semaphore(信号量)源码解读
-
.go
最后
大家如果喜欢我的文章,可以关注我的公众号(程序员麻辣烫)
我的个人博客为:/
往期文章回顾:
-
设计模式
-
招聘
-
思考
-
存储
-
算法系列
-
读书笔记
-
小工具
-
架构
-
网络
-
Go语言
更多推荐
Go锁,我终于搞懂了
发布评论