go RWMutex的实现示例
Overview
go 里面的 rwlock 是 write preferred 的，可以避免写锁饥饿。
读锁和写锁按照先来后到的规则持有锁，一旦有协程持有了写锁，后面的协程只能在写锁被释放后才能得到读锁。
同样，一旦有 >= 1 个协程写到了读锁，只有等这些读锁全部释放后，后面的协程才能拿到写锁。
下面了解一下 Go 的 RWMutex 是如何实现的吧，下面的代码取自 go1.17.2/src/sync/rwmutex.go，并删减了 race 相关的代码。
PS: rwmutex 的代码挺短的，其实读源码也没那么可怕...
RWMutex 的结构
RWMutex 总体上是通过: 普通锁和条件变量来实现的
type RWMutex struct {
w Mutex // held if there are pending writers
writerSem uint32 // semaphore for writers to wait for completing readers
readerSem uint32 // semaphore for readers to wait for completing writers
readerCount int32 // number of pending readers
readerWait int32 // number of departing readers
}
Lock
func (rw *RWMutex) Lock() {
// First, resolve competition with other writers.
rw.w.Lock()
// Announce to readers there is a pending writer.
r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
// Wait for active readers.
if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
runtime_SemacquireMutex(&rw.writerSem, false, 0)
}
}
Unlock
const rwmutexMaxReaders = 1 << 30
func (rw *RWMutex) Unlock() {
// Announce to readers there is no active writer.
r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
// Unblock blocked readers, if any.
for i := 0; i < int(r); i++ {
runtime_Semrelease(&rw.readerSem, false, 0)
}
// Allow other writers to proceed.
rw.w.Unlock()
}
RLock
func (rw *RWMutex) RLock() {
if atomic.AddInt32(&rw.readerCount, 1) < 0 {
// A writer is pending, wait for it.
runtime_SemacquireMutex(&rw.readerSem, false, 0)
}
}
RUnlock
func (rw *RWMutex) RUnlock() {
if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
// Outlined slow-path to allow the fast-path to be inlined
rw.rUnlockSlow(r)
}
}
func (rw *RWMutex) rUnlockSlow(r int32) {
// A writer is pending.
if atomic.AddInt32(&rw.readerWait, -1) == 0 {
// The last reader unblocks the writer.
runtime_Semrelease(&rw.writerSem, false, 1)
}
}
Q1: 多个协程并发拿读锁，如何保证这些读锁协程都不会被阻塞？
func (rw *RWMutex) RLock() {
if atomic.AddInt32(&rw.readerCount, 1) < 0 {
// A writer is pending, wait for it.
runtime_SemacquireMutex(&rw.readerSem, false, 0)
}
}
拿读锁时，仅仅会增加 readerCount，因此读锁之间是可以正常并发的
Q2: 多个协程并发拿写锁，如何保证只会有一个协程拿到写锁？
func (rw *RWMutex) Lock() {
// First, resolve competition with other writers.
rw.w.Lock()
// Announce to readers there is a pending writer.
r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
// Wait for active readers.
if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
runtime_SemacquireMutex(&rw.writerSem, false, 0)
}
}
拿写锁时，会获取 w.Lock，自然能保证同一时间只会有一把写锁
Q3: 在读锁被拿到的情况下，新协程拿写锁，如果保证写锁现成会被阻塞？
func (rw *RWMutex) Lock() {
// First, resolve competition with other writers.
rw.w.Lock()
// Announce to readers there is a pending writer.
r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
// Wait for active readers.
if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
runtime_SemacquireMutex(&rw.writerSem, false, 0)
}
}
假设此时有 5 个协程拿到读锁，则 readerCount = 5，假设 rwmutexMaxReaders = 100。
此时有一个新的协程 w1 想要拿写锁。
在执行
r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
后， rw.readerCount = -95，r = 5。
在执行
atomic.AddInt32(&rw.readerWait, r)
后，rw.readerWait = 5。
readerWait 记录了在获取写锁的这一瞬间有多少个协程持有读锁。这一瞬间之后，就算有新的协程尝试获取读锁，也只会增加 readerCount ，而不会动到 readerWait。
之后执行 runtime_SemacquireMutex() 睡在了 writerSem 这个信号量上面。
Q4: 在读锁被拿到的情况下，新协程拿写锁被阻塞，当旧有的读锁协程全部释放，如何唤醒等待的写锁协程
func (rw *RWMutex) RUnlock() {
if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
// Outlined slow-path to allow the fast-path to be inlined
rw.rUnlockSlow(r)
}
}
func (rw *RWMutex) rUnlockSlow(r int32) {
// A writer is pending.
if atomic.AddInt32(&rw.readerWait, -1) == 0 {
// The last reader unblocks the writer.
runtime_Semrelease(&rw.writerSem, false, 1)
}
}
继续上一步的场景，每当执行 RUnlock 时，readerCount 都会减去1。当 readerCount 为负数时，意味着有协程正在持有或者正在等待持有写锁。
之前的五个读协程中的四个，每次 RUnlock() 之后，readerCount = -95 - 4 = -99，readerWait = 5 - 4 = 1。
当最后一个读协程调用 RUnlock() 之后，readerCount 变成了 -100，readerWait 变成 0，此时会唤醒在 writerSem 上沉睡的协程 w1。
Q5: 在写锁被拿到的情况下，新协程拿读锁，如何让新协程被阻塞？
func (rw *RWMutex) RLock() {
if atomic.AddInt32(&rw.readerCount, 1) < 0 {
// A writer is pending, wait for it.
runtime_SemacquireMutex(&rw.readerSem, false, 0)
}
}
继续上面的场景，readerCount = -100 + 1 = -99 < 0。
新的读协程 r1 被沉睡在 readerSem 下面。
假设此时再来一个读协程 r2，则 readerCount = -98，依旧沉睡。
Q6: 在写锁被拿到的情况下，新协程拿读锁，写锁协程释放，如何唤醒等待的读锁协程？
继续上面的场景，此时协程 w1 释放写锁
func (rw *RWMutex) Unlock() {
// Announce to readers there is no active writer.
r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
// Unblock blocked readers, if any.
for i := 0; i < int(r); i++ {
runtime_Semrelease(&rw.readerSem, false, 0)
}
// Allow other writers to proceed.
rw.w.Unlock()
}
在执行
atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
后，r = readerCount = -98 + 100 = 2，代表此时有两个读协程 r1 和 r2 在等待
ps: 如果此时有一些新的协程想要拿读锁，他会因为 readerCount = 2 + 1 = 3 > 0 而顺利执行下去，不会被阻塞
之后 for 循环执行两次，将协程 r1 和 协程 r2 都唤醒了。
Q7: 在写锁被拿到的情况下，有两个协程分别去抢读锁和写锁，当写锁被释放时，这两个协程谁会胜利？
func (rw *RWMutex) Unlock() {
// Announce to readers there is no active writer.
r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
// Unblock blocked readers, if any.
for i := 0; i < int(r); i++ {
runtime_Semrelease(&rw.readerSem, false, 0)
}
// Allow other writers to proceed.
rw.w.Unlock()
}
由于是先唤醒读锁，再调用 w.Unlock() ，因此肯定是读协程先胜利！
认为写的比较巧妙的两个点
readerCount 与 rwmutexMaxReaders 的纠缠
通过
readerCount + rwmutexMaxReaders以及readerCount - rwmutexMaxReaders这两个操作可以得知当前是否有协程等待/持有写锁以及当前等待/持有读锁的协程数量readerCount 与 readerWait 的纠缠
在 Lock() 时直接将 readerCount 的值赋给 readerWait，在 readerWait = 0 而非 readerCount = 0 是唤醒写协程，可以避免在 Lock() 后来达到的读协程先于写协程被执行。
到此这篇关于go RWMutex的实现示例的文章就介绍到这了,更多相关go RWMutex内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
这篇文章主要介绍了如何利用Go语言实现LRU Cache，LRU是Least Recently Used的缩写，是一种操作系统中常用的页面置换算法，下面我们一起进入文章了解更多内容吧,需要的朋友可以参考一下2022-03-03
这篇文章主要介绍了Go中的 panic / recover 简介与实践,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2023-04-04
这篇文章主要为大家介绍了Go gorilla/sessions库安装使用详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪2022-08-08
在现代软件开发中,日志记录是一项至关重要的任务,它不仅帮助开发人员诊断问题,还有助于监控和维护应用程序,本文主要来和大家聊聊日志的统一处理,感兴趣的小伙伴可以了解下2024-01-01
本文主要介绍了记一次go语言使用time.Duration类型踩过的坑，文中通过示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2022-01-01
golang jsoniter extension 处理动态字段的实现方法
这篇文章主要介绍了golang jsoniter extension 处理动态字段的实现方法,我们使用实例级别的 extension, 而非全局，可以针对不同业务逻辑有所区分，jsoniter 包提供了比较完善的定制能力，通过例子可以感受一下扩展性，需要的朋友可以参考下2023-04-04
golang中拿slice当queue和拿list当queue使用分析
这篇文章主要为大家介绍了golang 中拿slice当queue和拿list当queue使用分析,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪2023-08-08
本文主要介绍了Go的固定时长定时器和周期性时长定时器，文中通过示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2021-08-08
这篇文章主要介绍了GO语言实现列出目录和遍历目录的方法,涉及ioutil.ReadDir()与filepath.Walk()的应用,是非常实用的技巧,需要的朋友可以参考下2014-12-12
这篇文章主要介绍了Go和PHP 实现爬楼梯算法,本文通过动态规划和斐波那契数列两种解决思路给大家讲解的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2020-05-05
最新评论