Go 语言中的读写锁实现：RWMutex

上一篇文章说了 Go 语言中的互斥锁（Mutex）的实现，今天来看看读写锁（RWMutex）的实现。读写锁相比于互斥锁性能损耗更低。

1. 读写锁是什么

下面是来自维基百科的解释: 

In computer science, a readers–writer (RW) or shared-exclusive lock (also known as a multiple readers/single-writer lock[1] or multi-reader lock[2]) is a synchronization primitive that solves one of the readers–writers problems. 

An RW lock allows concurrent access for read-only operations, while write operations require exclusive access. This means that multiple threads can read the data in parallel but an exclusive lock is needed for writing or modifying data. When a writer is writing the data, all other writers or readers will be blocked until the writer is finished writing.

简明地说就是：读写锁是一种同步机制，允许多个读操作同时读取数据，但是只允许一个写操作写数据。读写锁要根据进程进入临界区的具体行为（读，写）来决定锁的占用情况。这样锁的状态就有三种了：读模式加锁、写模式加锁、无锁。

• 无锁。读/写进程都可以进入。

• 读模式锁。读进程可以进入。写进程不可以进入。

• 写模式锁。读/写进程都不可以进入。

2. 为什么需要读写锁

互斥锁只容许一个进程/线程处于代码临界区，而且由于锁竞争会极大地拖慢进程的执行效率。如果程序中对于共享数据的读取操作特别多，这样就很不明智了。所以对于共享数据的读取操作比较多的情况下采用读写锁。读写锁对于读操作不会导致锁竞争和进程/线程切换。

3. 读写锁的实现思路

其实简单想一下就知道读写锁的核心是记录reader个数。有很多种实现，下面介绍两种简单的实现。

3.1 two mutex

使用两个互斥锁和一个整数记录 reader 个数，一个互斥锁 r 用来保证 reader个数 r_cnt 的更新，一个 w 用来保证 writer 互斥。伪代码如下：

//读锁
RLock()
    r.lock()
    r_cnt++
    if r_cnt==1:
        w.lock()
    r.unlock()
RUnlock()
    r.lock()
    r_cnt--
    if r_cnt==0:
        w.unlock()
    r.unlock()
//写锁
Lock()
    w.lock()
Unlock()
    w.unlock()

3.2 mutex and condition

另外一种常见的实现方式是使用 mutex 和 condition variable，思路很简单：RLock 的时候如果发现有 writer，则阻塞到condition variable；Lock的时候如果发现有writer，则阻塞。

RLock()
    m.lock()
    while w:
        c.wait(m)
    r++
    m.unlock()
RUnlock()
    m.lock()
    r--
    if r==0:
        c.signal_all()
    m.unlock()
Lock()
    m.lock()
    while w || r > 0 :
        c.wait(m)
    w = true
    m.unlock()
Unlock()
    m.lock()
    w = false
    c.signal_all()
    m.unlock()

4. golang读写锁代码实现

读写锁的定义在sync/rwmutex.go文件中。

// An RWMutex is a reader/writer mutual exclusion lock.
// The lock can be held by an arbitrary number of readers
// or a single writer.
// RWMutexes can be created as part of other
// structures; the zero value for a RWMutex is
// an unlocked mutex.
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
}

从定义中可以看出 RWMutex 内部使用了 Mutex，Mutex 其实是用来对写操作进行互斥的。同时使用两个信号量 writerSem 和 readerSem 来处理读写进程之间的关系。最后使用 readerCount 记录 reader 的个数，使用 readerWait 记录 writer 前面阻塞的 reader 的个数。感觉说的有点不太清楚，还是直接看代码吧。

RWMutex struct定义了四个方法如下。

// RLock locks rw for reading.
func (rw *RWMutex) RLock()
// RUnlock undoes a single RLock call;
func (rw *RWMutex) RUnlock()
// Lock locks rw for writing
func (rw *RWMutex) Lock()
// Unlock unlocks rw for writing.
func (rw *RWMutex) Unlock()

去掉一些golang做race detect代码，所有的代码实现如下：

func (rw *RWMutex) RLock() {
    if atomic.AddInt32(&rw.readerCount, 1) < 0 {
        // A writer is pending, wait for it.
        runtime_Semacquire(&rw.readerSem)
    }
}
func (rw *RWMutex) RUnlock() {
    if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
        if r+1 == 0 || r+1 == -rwmutexMaxReaders {
            raceEnable()
            panic("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)
        }
    }
}
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_Semacquire(&rw.writerSem)
    }
}
func (rw *RWMutex) Unlock() {
    // Announce to readers there is no active writer.
    r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
    if r >= rwmutexMaxReaders {
        raceEnable()
        panic("sync: Unlock of unlocked RWMutex")
    }
    // Unblock blocked readers, if any.
    for i := 0; i < int(r); i++ {
        runtime_Semrelease(&rw.readerSem)
    }
    // Allow other writers to proceed.
    rw.w.Unlock()
}

首先看一下这两个函数，分别对应我这篇文章《当我们谈论锁，我们谈什么》中的 wait 和 signal 操作。

//Semacquire waits until *s > 0 and then atomically decrements it.
func runtime_Semacquire(s *uint32)
// Semrelease atomically increments *s and notifies a waiting goroutine
func runtime_Semrelease(s *uint32)

首先看一下 RLock() 函数，它对 readerCount 进行加一操作（原子操作），如果 readerCount < 0，则 wait(readerSem)。等等，为什么 readerCount 会小于 0 呢？往下看发现 writer 的 Lock() 会对 readerCount 做减法操作（原子操作），用来表示现在有 writer。这个时候表示reader前面有 writer，reader 阻塞到信号量 readerSem 上，然后我们发现 Unlock() 中做了 signal(readerSem)，也就是唤醒阻塞在 readerSem上 的进程。同时 Lock() 将当前的readerCount 值加到 readerWait 上，如果 readerWait 不为0，则表示这个 writer 前有 reader，选择 wait(writerSem)。类似的 RUnlock() 里面做了 signal(writerSem)。稍微整理一下信号量的wait和signal。

//reader到来，前面有writer
RLock()
    if writer exist:
        wait(readerSem)
Unlock()
    signal(readerSem)
//writer到来，前面有reader
Lock()
    if reader exist:
        wait(writerSem)
RUnlock()
    signal(writerSem)

不得不说golang的读写锁实现还是挺巧妙的。

参考：

* [wikipedia](https://en.wikipedia.org/wiki/Readers–writer_lock#cite_note-pthreads-7)

* [stackoverflow](http://stackoverflow.com/questions/12033188/how-would-a-readers-writer-lock-be-implemented-in-c11)