rwmutex.go

2019-04-05 本文已影响0人 killtl

概述

rwmutex.go即读写锁，内部基于atomic和sync.mutex实现，提供四个方法

RLock 读锁，阻塞Lock，不阻塞其他的Rlock
RUnlock 解读锁，与Rlock匹配
Lock 写锁，阻塞Rlock和其他Lock，不阻塞RUnlock
Unlock 解写锁，与Lock匹配

Struct

type RWMutex struct {
    w           Mutex  // 排他锁，用于Lock阻塞其他Lock
    writerSem   uint32 // 写信号
    readerSem   uint32 // 读信号
    readerCount int32  // Rlock锁的数量
    readerWait  int32  // Lock阻塞等待的Rlock锁的数量
}

const rwmutexMaxReaders = 1 << 30  // 注意这里不仅是一个限制读锁最大数量的标记，也是Lock阻塞RLock的工具

RLock

/注/ 后续代码均抹掉race竞争检测部分

func (rw *RWMutex) RLock() {
    if atomic.AddInt32(&rw.readerCount, 1) < 0 {
        // readerCount值加1，注意这里不是CAS操作
        // readerCount < 0 必然是有写Lock住了，等待
        runtime_Semacquire(&rw.readerSem)
    }
}

这里的readerCount为啥是int32，因为它可为负值，这里设计是很巧妙的，readerCount为正值的时候表示没有写锁，当readerCount为负值的时候表示有写锁，同时readerCount无论为正值还是负值，都可以对当前Rlock的数量进行记录

RUnlock

func (rw *RWMutex) RUnlock() {
    // readerCount减一
    if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
        // 如果readerCount是正数，减1后小于0，重复RUnlock了
        // 如果readerCount是负数，说明此时有写锁占用着，减1后小于-rwmutexMaxReaders，重复RUnlock了
        if r+1 == 0 || r+1 == -rwmutexMaxReaders {
            race.Enable()
            throw("sync: RUnlock of unlocked RWMutex")
        }
        // 没有重复RUnlock，r < 0只可能此时被写锁占用了
        // 因为只有写锁才会把readerCount置为负数进行标识
        // 将阻塞写锁的读锁数readerWait减1，结果如果为0即所有读锁都解锁了，可以唤醒阻塞中的写锁了
        if atomic.AddInt32(&rw.readerWait, -1) == 0 {
            runtime_Semrelease(&rw.writerSem, false)
        }
    }
}

这里用到了readerWait来标识阻塞写锁的读锁的数量，这里可能会有人觉得困惑，为啥不直接用readerCount = -rwmutexMaxReaders来标识读锁全部解锁完了呢，这里需要注意的是就算写锁阻塞了读锁，但是读锁是没有用到互斥锁mutex的，所以就算阻塞了也已经执行了atomic.AddInt32(&rw.readerCount, 1)，只不过阻塞于待唤醒状态，所以readerCount其实是已经获得读锁和正在阻塞获得读锁的和，没法用来标记写锁要等待的读锁数，想一下这种情况，顺序执行如下操作

先获得读锁50个此时readerCount = 50
再尝试获得写锁此时写锁阻塞，读锁readerCount = 50 - rwmutexMaxReaders，写锁需要等待的读锁数readerWait = 50
再获得读锁20个全部阻塞，但是此时readerCount = 50 - rwmutexMaxReaders + 20，readerWait = 50
解一个读锁此时readerCount = 50 - rwmutexMaxReaders + 20 - 1，readerWait = 50 -1，这里尽管在写锁占用的同时尝试获得读锁引发了readerCount增加，但是不影响readerWait，同时也就不影响RUnlock唤醒写锁

Lock

func (rw *RWMutex) Lock() {
    // 互斥锁，锁住水分，哦不，锁住其他写
    rw.w.Lock()
    // 这里看仔细了，先让readerCount = readerCount - rwmutexMaxReaders变成负值，用来挡住其他读锁
    // 再算出原来的readerCount，这个值是已经获取成功的读锁数
    r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
    // 修改阻塞写锁的读锁数，如果不为0，则睡眠等待
    if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
        runtime_Semacquire(&rw.writerSem)
    }
}

Unlock

func (rw *RWMutex) Unlock() {
    // readerCount = readerCount + rwmutexMaxReaders
    // 把readerCount变成正值，不再阻塞写锁的获取
    // 注意此时的readerCount表示所有阻塞中的读锁，因为只有所有已经获取成功的写锁全部解锁了才会走到写锁的Unlock里
    r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    // 如果r > rwmutexMaxReaders 只可能是执行了两次unlock
    if r >= rwmutexMaxReaders {
        race.Enable()
        throw("sync: Unlock of unlocked RWMutex")
    }
    // 唤醒所有阻塞中的读锁
    for i := 0; i < int(r); i++ {
        runtime_Semrelease(&rw.readerSem, false)
    }
    // 此次写锁完毕，解锁
    rw.w.Unlock()
}

这里继续上面的操作

解49个读锁此时readerCount = 50 - rwmutexMaxReaders + 20 - 1 - 49，readerWait = 50 - 1 - 49，此时readerWait = 0触发唤醒写锁
执行解写锁此时readerCount = 50 - rwmutexMaxReaders + 20 - 1 - 49 + rwmutexMaxReaders = 20，刚好是写锁占用期间尝试获取读锁的数量
这里要注意，如果把解49个读锁换成解49+20个读锁，RUnlock并不会报错，而之前那20个正在等待中的锁可能不会按照预期的时间被唤醒，举个栗子

func main() {
    var m sync.RWMutex
    m.RLock()
    m.RLock()
    fmt.Println("double rlock succ")

    go func() {
        fmt.Println("lock begin...")
        m.Lock()
        fmt.Println("lock end")
        m.Unlock()
        fmt.Println("unlock succ")
    }()

    time.Sleep(time.Second)

    go func() {
        fmt.Println("try rlock...")
        m.RLock()   // 这把读锁不能被m.Unlock唤醒了
        fmt.Println("rlock end")
    }()

    time.Sleep(time.Second)

    m.RUnlock()
    m.RUnlock()
    m.RUnlock()  // 这个解读锁把阻塞中还未获取成功的读锁给解掉了

    time.Sleep(time.Second)
}

// output 
double rlock succ
lock begin...
try rlock...
lock end
unlock succ

再看看这个，Lock后还能RLock

const rwmutexMaxReaders = 1 << 30 - 1

func main() {
    var m sync.RWMutex
    for i := rwmutexMaxReaders; i > 0; i-- {
        m.RLock()
    }

    fmt.Printf("%d rlock/n", rwmutexMaxReaders)

    go func() {
        m.Lock()
    }()

    time.Sleep(time.Second)

    fmt.Println("lock")

    m.RLock()

    fmt.Println("rlock again")
}

// output
1073741823 rlock
lock
rlock again

这是因为RLock方法中的判断if atomic.AddInt32(&rw.readerCount, 1) < 0，所以正如开头所说的rwmutexMaxReaders只是一个限制标记，不代表读锁数的最大值，这里面读锁数的最大值应该是1 << 30 - 2

总结

佩服