go学习笔记 sync/RWMutex源码

RWMutex是一个读写锁，该锁可以加多个读锁或者一个写锁，其经常用于读次数远远多于写次数的场景．
func (rw *RWMutex) Lock()　写锁，如果在添加写锁之前已经有其他的读锁和写锁，则lock就会阻塞直到该锁可用，为确保该锁最终可用，已阻塞的 Lock 调用会从获得的锁中排除新的读取器，即写锁权限高于读锁，有写锁时优先进行写锁定
func (rw *RWMutex) Unlock()　写锁解锁，如果没有进行写锁定，则就会引起一个运行时错误．
func (rw *RWMutex) RLock() 读锁，当有写锁时，无法加载读锁，当只有读锁或者没有锁时，可以加载读锁，读锁可以加载多个，所以适用于＂读多写少＂的场景
func (rw *RWMutex) RUnlock()　读锁解锁，RUnlock 撤销单次 RLock 调用，它对于其它同时存在的读取器则没有效果。若 rw 并没有为读取而锁定，调用 RUnlock 就会引发一个运行时错误

 
package sync
 
import (
    "internal/race"
    "sync/atomic"
    "unsafe"
)
 
// There is a modified copy of this file in runtime/rwmutex.go.
// If you make any changes here, see if you should make them there.
 
// A RWMutex is a reader/writer mutual exclusion lock.
// The lock can be held by an arbitrary number of readers or a single writer.
// The zero value for a RWMutex is an unlocked mutex.
//
// A RWMutex must not be copied after first use.
//
// If a goroutine holds a RWMutex for reading and another goroutine might
// call Lock, no goroutine should expect to be able to acquire a read lock
// until the initial read lock is released. In particular, this prohibits
// recursive read locking. This is to ensure that the lock eventually becomes
// available; a blocked Lock call excludes new readers from acquiring the
// lock.
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 获取写锁时需要等待的读锁释放数量
}
 
const rwmutexMaxReaders = 1 << 30 // 支持最多2^30个读锁
 
// RLock locks rw for reading.
//
// It should not be used for recursive read locking; a blocked Lock
// call excludes new readers from acquiring the lock. See the
// documentation on the RWMutex type.
// 它不应该用于递归读锁定;
func (rw *RWMutex) RLock() {
    if race.Enabled {
        _ = rw.w.state
        race.Disable()
    }
    // 每次goroutine获取读锁时，readerCount+1
    // 如果写锁已经被获取，那么readerCount在 - rwmutexMaxReaders与 0 之间，这时挂起获取读锁的goroutine，
    // 如果写锁没有被获取，那么readerCount>=0，获取读锁,不阻塞
    // 通过readerCount的正负判断读锁与写锁互斥,如果有写锁存在就挂起读锁的goroutine,多个读锁可以并行
    if atomic.AddInt32(&rw.readerCount, 1) < 0 {
        // A writer is pending, wait for it.
        // 将goroutine排到G队列的后面,挂起goroutine, 监听readerSem信号量
        runtime_SemacquireMutex(&rw.readerSem, false, 0)
    }
    if race.Enabled {
        race.Enable()
        race.Acquire(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.
// 读锁不会影响其他读操作
// 如果在进入RUnlock时没有锁没有被施加读锁的话，则会出现运行时错误。
func (rw *RWMutex) RUnlock() {
    if race.Enabled {
        _ = rw.w.state
        race.ReleaseMerge(unsafe.Pointer(&rw.writerSem))
        race.Disable()
    }
    // 读锁计数器 -1
    // 有四种情况，其中后面三种都会进这个 if
    // 【一】有读锁，但没有写锁被挂起
    // 【二】有读锁，且也有写锁被挂起
    // 【三】没有读锁且没有写锁被挂起的时候， r+1 == 0
    // 【四】没有读锁但是有写锁被挂起，则 r+1 == -(1 << 30)
    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()
    }
}
 
func (rw *RWMutex) rUnlockSlow(r int32) {
        // 读锁早就被没有了，那么在此 -1 是需要抛异常的
        // 这里只有当读锁没有的时候才会出现的两种极端情况
        // 【一】没有读锁且没有写锁被挂起的时候， r+1 == 0
        // 【二】没有读锁但是有写锁被挂起，则 r+1 == -(1 << 30)
    if r+1 == 0 || r+1 == -rwmutexMaxReaders {
        race.Enable()
        throw("sync: RUnlock of unlocked RWMutex")
    }
    // A writer is pending.
        // 如果获取写锁时的goroutine被阻塞，这时需要获取读锁的goroutine全部都释放，才会被唤醒
        // 更新需要释放的 写锁的等待读锁释放数目
        // 最后一个读锁解除时，写锁的阻塞才会被解除.
    if atomic.AddInt32(&rw.readerWait, -1) == 0 {
        // The last reader unblocks the writer.
        // 更新信号量，通知被挂起的写锁去获取锁
        runtime_Semrelease(&rw.writerSem, false, 1)
    }
}
 
// Lock locks rw for writing.
// If the lock is already locked for reading or writing,
// Lock blocks until the lock is available.
// 对一个已经lock的rw上锁会被阻塞
// 如果锁已经锁定以进行读取或写入，则锁定将被阻塞，直到锁定可用。
func (rw *RWMutex) Lock() {
    if race.Enabled {
        _ = rw.w.state
        race.Disable()
    }
    // First, resolve competition with other writers.
    // 首先，获取互斥锁，与其他来获取写锁的goroutine 互斥
    rw.w.Lock()
    // Announce to readers there is a pending writer.
    // 告诉其他来获取读锁操作的goroutine，现在有人获取了写锁
    // 减去最大的读锁数量，用0 -负数 来表示写锁已经被获取
    r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
    // Wait for active readers.
    // 设置需要等待释放的读锁数量，如果有，则挂起获取 竞争写锁 goroutine
    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))
    }
}
 
// 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) a RWMutex and then
// arrange for another goroutine to RUnlock (Unlock) it.
// 如果在写锁时，rw没有被解锁，则会出现运行时错误。
// 与互斥锁一样，锁定的RWMutex与特定的goroutine无关。
// 一个goroutine可以RLock（锁定）RWMutex然后安排另一个goroutine到RUnlock（解锁）它。
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.
    // 向读锁的goroutine发出通知，现在已经没有写锁了
    // 还原加锁时减去的那一部分readerCount
    r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    if r >= rwmutexMaxReaders {
        // 读锁数目超过了 最大允许数
        race.Enable()
        throw("sync: Unlock of unlocked RWMutex")
    }
    // Unblock blocked readers, if any.
    // 唤醒获取读锁期间所有被阻塞的goroutine
    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()
    }
}
 
// RLocker returns a Locker interface that implements
// the Lock and Unlock methods by calling rw.RLock and rw.RUnlock.
// RLocker返回一个Locker接口的实现
// 通过调用rw.RLock和rw.RUnlock来锁定和解锁方法。
func (rw *RWMutex) RLocker() Locker {
    return (*rlocker)(rw)
}
 
type rlocker RWMutex
 
func (r *rlocker) Lock()   { (*RWMutex)(r).RLock() }
func (r *rlocker) Unlock() { (*RWMutex)(r).RUnlock() }

从上面的代码中我们可以看到，读写锁首先是内置了一个互斥锁，然后再加上维护各种计数器来实现的读写锁，紧接着提供了四个函数支撑着读写锁操作，由 Lock 和Unlock 分别支持写锁的锁定和释放，由RLock 和RUnlock 来支持读锁的的锁定和释放。其中，读锁不涉及内置mutex的使用，写锁用了mutex来排斥其他写锁。

读写互斥锁的实现比较有技巧性一些，需要几点

1. 读锁不能阻塞读锁，引入readerCount实现

2. 读锁需要阻塞写锁，直到所以读锁都释放，引入readerSem实现

3. 写锁需要阻塞读锁，直到所以写锁都释放，引入wirterSem实现

go学习笔记 sync/RWMutex源码

【读锁的】Rlock:

【读锁的】RUnlock:

【写锁的】Lock:

【写锁的】Unlock: