«

如何掌握go的sync.RWMutex锁

时间:2024-8-5 09:34     作者:韩俊     分类: Go语言


本文小编为大家详细介绍“如何掌握go的sync.RWMutex锁”,内容详细,步骤清晰,细节处理妥当,希望这篇“如何掌握go的sync.RWMutex锁”文章能帮助大家解决疑惑,下面跟着小编的思路慢慢深入,一起来学习新知识吧。

在简略的说之前,首先要对RW锁的结构有一个大致的了解

type RWMutex struct {
    w           Mutex  // 写锁互斥锁,只锁写锁,和读锁无关
    writerSem   uint32 // sema锁--用于“写协程”排队等待
    readerSem   uint32 // sema锁--用于“读协程”排队等待
    readerCount int32  // 读锁的计数器
    readerWait  int32  // 等待读锁释放的数量
}

这里要额外说一句,writerSem和readerSem底层都是semaRoot,这个结构体有兴趣可以了解下,他的用法有点类似于一个简版的channel,很多地方把他的初始值设置为0,使得所有想获取该sema锁的协程都排队等待,也就是说初始值为0的sema锁,他本身起到的作用是成为一个协程等待队列,就像没有缓冲区的channel一样。

前提:

readerCount这个参数非常重要

    为负数时:说明此锁已经被写协程占据,所有渴望加读锁的协程被阻塞在readerSem

    为正数时:正数的数值为当前持有该锁的所有读协程的数量总和,所有渴望加写锁的协程被阻塞在writerSem

读写锁互斥性

    读锁是并发的,可以多个协程持有一把读锁。

    写锁是唯一的,互斥的,同一时刻只能有一个写协程拥有写锁

    读锁和写锁是互斥的,写锁生效时,是不能有读锁被获取到,同样,必须所有的读锁都被释放,或者压根没有读协程获取读锁,写锁方可被获取。

一个很重要的参数:const rwmutexMaxReaders = 1 << 30 ,rwmutexMaxReaders 非常大,意思是最多能有rwmutexMaxReaders(1 << 30  十进制为  4294967296)个协程同时持有读锁。

写锁上锁场景:

首先分析写锁,因为读锁的很多操作是根据写锁来的,如果一上来就说读锁,很多东西没法串起来

 func (rw *RWMutex) Lock() {
    // race.Enabled是官方的一些测试,性能检测的东西,无需关心,这个只在编译阶段才能启用
    if race.Enabled {
        _ = rw.w.state
        race.Disable()
    }
    // 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)
    }
    if race.Enabled {
        race.Enable()
        race.Acquire(unsafe.Pointer(&rw.readerSem))
        race.Acquire(unsafe.Pointer(&rw.writerSem))
    }
}

1.获取写锁--没有读锁等待

    rw.w.Lock进行加锁,阻塞后续的其他写协程的锁请求。

    atomic.AddInt32进行原子操作,减去rwmutexMaxReaders,减成功才说明没有并发问题,可以继续下面的操作。然后再加上rwmutexMaxReaders,得到真正的readerCount的数值。

    此时还需要再次进行一个原子操作,把当前readerCount的值搬运到readerWait里面,意思是当前要获取写锁的协程需要等待的读锁的数量。

    此时readerCount只有两种情况,一种是0,一种是正数,因为只有写锁上的时候才为负数,而上面的操作已经还原了加写锁之前的值,而w.Lock保证了不会有2个及以上的写协程去同时操作

    readerCount 如果是 0,加锁成功。

    如果不为0则说明有读锁等待,详见场景2

2.获取写锁--有读锁等待

    接上面的判断,如果readrCount不为0,说明前面有读锁正在运行,写锁需要等待所有读锁释放才能获取写锁,当前协程执行 runtime_SemacquireMutex 进入 waiterSem 的休眠队列等待被唤醒

3.获取写锁--前面已经有写锁了

后面的写协程也调用 rw.w.Lock() 进行加锁,因为前面有写锁已经获取了w,所以后续的写协程会因为获取不到w,而进入到w的sema队列里面,w是一个mutex的锁,mutex锁里是一个sema锁,sema锁因为没有设置初始值,所以退化为一个队列,而获取不到w锁的就会直接被阻塞在w的sema队列里,从而无法进行接下来的操作

写锁释放锁场景:

func (rw *RWMutex) Unlock() {
    if race.Enabled {
        _ = rw.w.state
        race.Release(unsafe.Pointer(&rw.readerSem))
        race.Disable()
    }
 
    // Announce to readers there is no active writer.
    r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    if r >= rwmutexMaxReaders {
        race.Enable()
        throw("sync: Unlock of unlocked RWMutex")
    }
    // 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()
    if race.Enabled {
        race.Enable()
    }
}

1.释放写锁--后面【没有】读锁等待

    执行atomic.AddInt32进行原子操作,把已经为负值的readerCount还原为正数,此时已经算释放了写锁

    (此步骤不重要,就是个判错)如果还原后的readerCount比rwmutexMaxReaders还大,这就是说明出错了,直接throw弹出错误,throw这个方法是内部方法,对go来说就是panic了

    此场景因为没有读锁等待,此时的readerCount为0,不会进入for循环,直接rw.w.Unlock释放w锁,允许其他写协程加锁,此时其他的写协程会被从w里的sema队列里唤醒

2.释放写锁--后面【有】读锁等待

    接场景1,原子操作readerCount释放写锁后,如果r是大于0,说明有读锁等待,for循环readerSem里面所有的等待的读协程,因为读锁是共享锁,所以所有的读协程都会获取锁并被唤醒

    rw.w.Unlock释放w锁,允许其他写协程加锁,其他的写协程会被从w里的sema队列里唤醒

3.释放写锁--后面有【写锁】等待

    上接场景2,当rw.w.Unlock释放w锁,其他的写协程会被从w里的sema队列里唤醒

    写锁释放的时候,是先唤醒所有等待的读锁,再解除rw.w锁,所以,并不会造成读锁的饥饿

    后面读锁再次对rw.w进行上锁,重复上面所述写锁获取锁的场景

读锁上锁场景:

func (rw *RWMutex) RLock() {
    // race.Enabled都是测试用的代码,在阅读源码的时候可以跳过
    if race.Enabled {
        _ = rw.w.state
        race.Disable()
    }
    
    if atomic.AddInt32(&rw.readerCount, 1) < 0 {
        // A writer is pending, wait for it.
        runtime_SemacquireMutex(&rw.readerSem, false, 0)
    }
    if race.Enabled {
        race.Enable()
        race.Acquire(unsafe.Pointer(&rw.readerSem))
    }
}

1.获取读锁--此时没有写锁.

最简单的场景,协程对rw.readerCount进行原子操作加一,如果得到的结果为正数,说明获取读锁成功。

2.获取读锁--前方已经有写锁抢占了该锁

    当协程对rw.readerCount进行原子加1操作的时候,发现加完,readerCount还是负数,说明在这个时间点以前,已经有协程获取了写锁

     runtime_SemacquireMutex 方法将当前协程加入readerSem队列,等待写锁释放后被批量唤醒(写锁释放会一次性放出所有的堆积的读协程)

3.获取读锁--前方有写锁抢已经被抢占,后方有写锁等待

    写锁在获取的时候,对RWMutex.w进行加锁,是独占锁,如果前方一个写锁已经得到了锁正在处理业务,那么后方的写锁进来就会发现加不上锁,直接在rw.w.lock阶段就阻塞了,后面的逻辑是无法继续运行的,所以进入不了writerSem,它只会进入到w这个mutex锁的sema队列里,读锁则进入休眠队列readerSem

读锁释放锁场景:

func (rw *RWMutex) RUnlock() {
    if race.Enabled {
        _ = rw.w.state
        race.ReleaseMerge(unsafe.Pointer(&rw.writerSem))
        race.Disable()
    }
    if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
        // Outlined slow-path to allow the fast-path to be inlined
        rw.rUnlockSlow(r)
    }
    if race.Enabled {
        race.Enable()
    }
}

1.释放读锁--后方没有写锁等待

    atomic.AddInt32 进行原子操作,让readerCount 减1,操作后,如果readerCount 大于0,说明后方是没有写锁等待的,释放锁后整个流程就结束了

2.释放读锁--后方有写锁等待

    原子操作eaderCount 减1后,发现eaderCount是小于0的,此时说明已经有等待写锁的协程在尝试获取写锁。执行 rw.rUnlockSlow(r) 。               

func (rw *RWMutex) rUnlockSlow(r int32) {
    if r+1 == 0 || r+1 == -rwmutexMaxReaders {
        race.Enable()
        throw("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, false, 1)
    }
}

这里是有个前提的,上面提到(详见上面的获取写锁的场景1),如果写协程进来想加写锁,需要把它需要等待的读锁数量从readerCount里赋值给readerWait。当它等待的读锁释放后,就需要用rUnlockSlow方法对readerWait进行减1,如果readWait == 0 ,说明这是最后一个需要等待的读锁也释放了,释放后就通知该写锁可以被唤醒了,锁给你了。

标签: golang

热门推荐