读写锁RWMutex

读写锁RWMutex

type RWMutex struct {
    w           Mutex  // 互斥锁解决多个writer的竞争
    writerSem   uint32 // writer信号量
    readerSem   uint32 // reader信号量
    readerCount int32  // reader的数量
    readerWait  int32  // writer等待完成的reader的数量
}

// 最大的 reader 数量。
const rwmutexMaxReaders = 1 << 30

RLock/RUnlock

当 writer 请求锁的时候，是无法改变既有的reader 持有锁的现实的，也不会强制这些 reader 释放锁，它的优先权只是限定后来的reader 不要和它抢。所以，rUnlockSlow 将持有锁的 reader 计数减少 1 的时候，会检查既有的 reader 是不是都已经释放了锁，如果都释放了锁，就会唤醒 writer，让 writer 持有锁。

func (rw *RWMutex) RLock() {
    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))
    }
}

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()
    }
}

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

Lock

一旦一个 writer 获得了内部的互斥锁，就会反转 readerCount 字段，把它从原来的正整
数 readerCount(>=0) 修改为负数（readerCount-rwmutexMaxReaders），让这个字段
保持两个含义（既保存了 reader 的数量，又表示当前有 writer）。

func (rw *RWMutex) Lock() {
    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.
    // 反转readerCount，告诉reader有writer竞争锁
    // 记录当前活跃的 reader 数量(持有读锁还没有释放的那些 reader)
    r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
    // Wait for active readers.
    // 如果当前有reader持有锁，那么需要等待
    // 如果 readerCount 不是 0，就说明当前有持有读锁的 reader，RWMutex 需要把这个当前 readerCount 赋值给 readerWait 字段保存下来， 同时，这个 writer 进入阻塞等待状态。每当一个 reader 释放读锁的时候（调用 RUnlock 方法时），readerWait 字段就减 1，直到所有的活跃的 reader 都释放了读锁，才会唤醒这个 writer
    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

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.
    // 反转 readerCount 字段
    r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    if r >= rwmutexMaxReaders {
        race.Enable()
        fatal("sync: Unlock of unlocked RWMutex")
    }
    // Unblock blocked readers, if any.
    // 唤醒阻塞的reader们
    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()
    }
}

在 Lock 方法中，是先获取内部互斥锁，才会修改的其他字段；而在 Unlock 方法中，是先修改的其他字段，才会释放内部互斥锁，这样才能保证字段的修改也受到互斥锁的保护。

TryRLock /TryLock

TryRLock 尝试锁定 rw 以进行读取并报告它是否成功

func (rw *RWMutex) TryRLock() bool {
    if race.Enabled {
        _ = rw.w.state
        race.Disable()
    }
    for {
        c := atomic.LoadInt32(&rw.readerCount)
        if c < 0 {
            if race.Enabled {
                race.Enable()
            }
            return false
        }
        if atomic.CompareAndSwapInt32(&rw.readerCount, c, c+1) {
            if race.Enabled {
                race.Enable()
                race.Acquire(unsafe.Pointer(&rw.readerSem))
            }
            return true
        }
    }
}

TryLock 尝试锁定 rw 以进行写入并报告它是否成功。

func (rw *RWMutex) TryLock() bool {
    if race.Enabled {
        _ = rw.w.state
        race.Disable()
    }
    if !rw.w.TryLock() {
        if race.Enabled {
            race.Enable()
        }
        return false
    }
    if !atomic.CompareAndSwapInt32(&rw.readerCount, 0, -rwmutexMaxReaders) {
        rw.w.Unlock()
        if race.Enabled {
            race.Enable()
        }
        return false
    }
    if race.Enabled {
        race.Enable()
        race.Acquire(unsafe.Pointer(&rw.readerSem))
        race.Acquire(unsafe.Pointer(&rw.writerSem))
    }
    return true
}

RLocker

RLocker 返回一个 Locker 接口，该接口通过调用 rw.RLock 和 rw.RUnlock 来实现 Lock 和 Unlock 方法。

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() }