C/C++ 讀寫鎖Readers-Write Lock
讀寫鎖基本概念
讀寫鎖（readers-writer lock），又稱為多讀單寫鎖（multi-reader single-writer lock，或者MRSW lock），共享互斥鎖（shared-exclusive lock），以下簡稱RW lock。
讀寫鎖用來解決讀寫操作併發的問題。多個執行緒可以並行讀取資料，但只能獨佔式地寫或修改資料。
write-mode和read-mode
RW lock有兩種模式：write-mode，read-mode。
- write-mode
在write-mode下，一個writer取得RW lock。當writer寫資料時，其他所有writer或reader將阻塞，直到該writer完成寫操作； - read-mode
在read-mode下，至少一個reader取得RW lock。當reader讀資料時，其他reader也能同時讀取資料，但writer將阻塞，直到所有reader完成讀操作;
RW lock升級與降級
當writer取得RW lock，進入write-mode，對資料進行寫操作時，進入read-mode進行讀操作。我們把這個稱為鎖降級（downgraded RW lock）。
當reader取得RW lock，進入read-mode，對資料進行讀操作時，進入write-mode進行寫操作。我們把這個稱為鎖升級（upgradable RW lock）。
鎖降級是安全的；而鎖升級是不安全的，容易造成死鎖，應當避免。
讀寫鎖與互斥鎖的關係
相同點在於對寫操作是互斥的。
主要區別在於鎖的粒度，針對讀操作，reader可以共享資料；而針對寫操作，與其他任意reader或writer都是互斥的。
讀寫鎖與互斥鎖的詳細區別，可以參見這篇文章：Linux 自旋鎖，互斥量（互斥鎖），讀寫鎖
優先順序策略
針對reader與writer訪問，RW lock能設計成不同的優先順序策略：read-preferring（讀優先），write-preferring（寫優先），unspecified priority（不確定優先順序）。
- read-preferring，允許最大併發量，但如果爭用較多時，將導致寫飢餓：writer執行緒將長期不能完成寫操作。因為只要有一個reader執行緒持有lock，writer就無法取得RW lock。而連續不斷新來的reader，將導致writer長期無法取得RW lock。
- write-preferring，能有效避免寫飢餓問題，但相對地，會帶來讀飢餓問題。
- unspecified priority，不保證優先讀訪問，或寫訪問。
介面
通常，RW lock需要對外提供以下介面：
1）初始化Initialize
2）銷燬Destroy
3）取得讀鎖，進入read-mode
4）釋放讀鎖，退出read-mode
5）取得寫鎖，進入write-mode
6）釋放寫鎖，退出write-mode
linux的pthread執行緒庫中的pthread_rwlock是RW lock的一個實現，其介面為：
#include <pthread.h>
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock); /* 銷燬RW lock */
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock,
const pthread_rwlockattr_t *restrict attr); /* 初始化RW lock */
pthread_rwlock_t rwlock = PTHREAD_RWLOCK_INITIALIZER; /* 直接賦值方式初始化RW lock */
int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock); /* 取得讀鎖，進入read-mode */
int pthread_rwlock_tryrdlock(pthread_rwlock_t *rwlock); /* 嘗試取得讀鎖，失敗立即返回 */
int pthread_rwlock_trywrlock(pthread_rwlock_t *rwlock); /* 取得寫鎖，進入write-mode */
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock); /* 嘗試取得寫鎖，失敗立即返回 */
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock); /* 釋放讀/寫鎖 */
實現
如果是在linux中，我們可以直接使用pthread執行緒庫的pthread_rwlock。而如果是其他平臺，如Win32，就需要自行實現讀寫鎖。
注：C++17中，std::shared_lock支援RW lock。
RW lock實現有多種方式，其中代表性的有兩種：
使用2個mutex
要求：2個mutex，1個int計數器。
其中，計數器b記錄阻塞等待的reader數量。1個mutex r，用來保護b只被reader使用；另外1個mutex（全域性的）確保writers互斥。用英文來解釋：
b: a counter, tracking for number of readers waiting RW lock;
r: a mutex, protect b and only used by readers;
g: a global mutex, ensure mutual exclusion of writers. It can be acquired by one thread, but released by another.
虛擬碼
- 初始化Initialize
Set b to 0; /* clear counter b */
r is unlocked; /* init mutex r */
g is unlocked; /* init mutex g */
- 取得讀鎖Begin Read
Lock r; // 注意這裡的r，只是用來鎖住RW lock內部資源
b++;
if b = 1, lock g; // g的lock執行緒和unlock執行緒可能並非同一個
Unlock r;
- 釋放讀鎖End Read
Lock r;
b--;
if b = 0, unlock g;
Unlock r;
- 取得寫鎖Begin Write
Lock g; // 只有處於write-mode時，對g進行unlock和lock的才要求是同一個執行緒
- 釋放寫鎖End Write
Unlock g;
這種方式一個具體的實現，可參見：41 C++ 讀寫鎖的實現及使用樣例 | 知乎
使用1個condition variable + 1個mutex
要求：1個condition variable（條件變數）cond，1個普通mutex g，若干個計數器、標誌，用於表示執行緒當前處於啟用或阻塞狀態。
1）num_readers_active，取得lock的readers數量；
2）num_writers_waiting，阻塞等待lock的writers數量；
3）writer_active，表示一個writer是否已經取得lock；
虛擬碼
- 取得讀鎖Begin Read
採用寫優先方式（write-preferring），會影響到加鎖方式。
Lock g;
while num_writers_waiting > 0 or writer_active: /* 等待所有writer */
wait cond, g; /* 等待條件變數cond, 釋放互斥鎖g */
num_readers_active++;
Unlock g;
- 釋放讀鎖End Read
Lock g;
num_readers_active--;
if num_readers_active == 0:
Notify cond(broadcast) /* why not signal? */
Unlock g;
思考：為什麼這裡條件變數喚醒用的是broadcast（廣播，喚醒所有），而不是signal（喚醒單個）？
答：個人認為，broadcast和signal效果是一樣的。首先，能執行這段程式碼，說明已經之前已經取得了read-lock，處於read-mode，現在是準備釋放read-lock。也就是說，已經等待條件變數cond上的執行緒，只可能是writer（因為之前的reader會立即取得read lock）。使用signal是隨機喚醒一個write執行緒，接著直接取得寫鎖；而使用broadcast會喚醒所有write執行緒，再通過下面的取得寫鎖來爭用。
- 取得寫鎖Begin Write
Lock g;
num_writers_waiting++;
while num_readers_active > 0 or writer_active is true: /* 等待所有readers或其他writer */
wait cond, g;
num_writers_waiting--;
Set writer_active to true;
Unlock g;
- 釋放寫鎖End Write
Lock g;
Set writer_active to false;
Notify cond(broadcast);
Unlock g;
使用1個mutex + 2個條件變數
問題：能否將釋放讀鎖和釋放寫鎖寫在同一個函式中？
就像POSIX的pthread_rwlock_unlock一樣，不論持有的是讀鎖，還是寫鎖，解鎖操作都是一個，我們也可以把兩者設計到同一個介面中。參照UNP卷2，我們寫出讀寫鎖的C++版本實現：1個mutex + 2個條件變數。
實現RW lock完整程式碼：
class RWLock {
public:
RWLock() : rw_nwaitreaders(0), rw_nwaitwriters(0), rw_refcount(0) { }
~RWLock() = default;
RWLock(const RWLock&) = delete;
RWLock& operator=(const RWLock&) = delete;
public:
void rdlock(); /* wait for reader lock */
bool tryrdlock(); /* try to get reader lock */
void wrlock(); /* wait for writer lock */
bool trywrlock(); /* try to get writer lock */
void unlock(); /* release reader or writer lock */
private:
std::mutex rw_mutex;
std::condition_variable_any rw_condreaders;
std::condition_variable_any rw_condwriters;
int rw_nwaitreaders; /* the number of waiting readers */
int rw_nwaitwriters; /* the number of waiting writers */
int rw_refcount; /* 0: not locked; -1: locked by one writer; > 0: locked by rw_refcount readers */
};
// 阻塞獲取讀鎖
void RWLock::rdlock()
{
rw_mutex.lock();
{
/* give preference to waiting writers */
while (rw_refcount < 0 || rw_nwaitwriters > 0) { // 寫優先
rw_nwaitreaders++;
rw_condreaders.wait(rw_mutex);
rw_nwaitreaders--;
}
rw_refcount++; /* another reader has a read lock */
}
rw_mutex.unlock();
}
// 嘗試獲取讀鎖，失敗立即返回
bool RWLock::tryrdlock()
{
bool res = true;
rw_mutex.lock();
{
if (rw_refcount < 0 || rw_nwaitwriters > 0) { // 寫優先
res = false; /* held by a writer or waiting writers */
}
else {
rw_refcount++; /* increment count of reader locks */
}
}
rw_mutex.unlock();
return res;
}
// 阻塞獲取寫鎖
void RWLock::wrlock()
{
rw_mutex.lock();
{
while (rw_refcount != 0) { /* wait other readers release the rd or wr lock */
rw_nwaitwriters++;
rw_condwriters.wait(rw_mutex);
rw_nwaitwriters--;
}
rw_refcount = -1; /* acquire the wr lock */
}
rw_mutex.unlock();
}
// 嘗試獲取寫鎖，失敗立即返回
bool RWLock::trywrlock()
{
bool res = true;
rw_mutex.lock();
{
if (rw_refcount != 0) /* the lock is busy */
res = false;
else
rw_refcount = -1; /* acquire the wr lock */
}
rw_mutex.unlock();
return res;
}
// 釋放寫鎖或讀鎖
void RWLock::unlock()
{
rw_mutex.lock();
{
if (rw_refcount > 0)
rw_refcount--;
else if (rw_refcount == -1)
rw_refcount = 0;
else
// unexpected error
fprintf(stderr, "RWLock::unlock unexpected error. rw_refcount = %d\n", rw_refcount);
/* give preference to waiting writers over waiting readers */
if (rw_nwaitwriters > 0) {
if (rw_refcount == 0) {
rw_condwriters.notify_one();
}
}
else if (rw_nwaitreaders > 0) {
rw_condreaders.notify_one();
}
}
rw_mutex.unlock();
}
測試程式
#include <thread>
#include <mutex>
#include <iostream>
#include <vector>
using namespace std;
volatile int v = 0;
RWLock rwlock;
void WriteFunc()
{
this_thread::sleep_for(chrono::milliseconds(10)); // 為了演示效果，先讓write執行緒休眠10ms
rwlock.wrlock();
{
v++;
cout << "Write:" << v << endl;
}
rwlock.unlock();
}
void ReadFunc()
{
rwlock.rdlock();
{
cout << "Read:" << v << endl;
}
rwlock.unlock();
}
void test_rwlock()
{
vector<thread> writers;
vector<thread> readers;
for (int i = 0; i < 20; ++i) {
writers.push_back(thread(WriteFunc));
}
for (int i = 0; i < 200; ++i) {
readers.push_back(thread(ReadFunc));
}
for (auto & e : writers) {
e.join();
}
for (auto & e : readers) {
e.join();
}
getchar();
}
參考
[1]Stevens, W.Richard. UNIX network programming. Volume 2, Interprocess communications. UNIX網路程式設計. 卷2, 程序間通訊 / 3r[M]. 人民郵電出版社, 2009.
[2]https://en.wikipedia.org/wiki/Readers–writer_lock