用信号量和读写锁解决读者写者问题
Posted xybaby
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了用信号量和读写锁解决读者写者问题相关的知识，希望对你有一定的参考价值。
读者写者问题是非常经典的同步问题，本文首先用信号量来解决这个问题，并结合代码分析什么是读者优先、什么是写者优先，然后给出读写锁的解决方案，并指出在Linux下读写锁的注意事项。
读者写者问题
读者写者问题描述的是这么一种情况：对象在多个线程（或者进程）之间共享，其中一些线程只会读数据，另外一些线程只会写数据。为了保证写入和读取的正确性，我们需要保证，只要有线程在写，那么其他线程不能读，否则可能读到写了一半的数据；另外，也不能有两个线程同时写，否则导致数据错乱。当然，多个线程是可以同时读数据。
读者写者问题在计算机领域非常普遍，大家最容易想到的就是数据库，数据库的读写分离也是为了减少因为读者写者问题加锁带来的对并发的影响。
读者写者问题的解决方案一般都有两种不同的侧重：读者优先或者写者优先。简单来说，读者优化就是尽量满足并发的读操作，当已经有线程在读数据的时候，其他读线程无需等待，而写线程需要等待所有正在进行的读操作之后才能执行。写者优先就是尽量满足写操作，尽管写操作不能并发，但是可以排队，优先于等待的读线程获得执行权。
信号量解决读者写者问题
关于信号量的基础知识，可以参见上一篇文章《并发与同步、信号量与管程、生产者消费者问题》的讲解。这个章节通过信号量来解决读者写者问题，为了更好的分析，可以用下面的情况来概括所有的并发的可能。首先r代表读线程 w代表写线程，我们给出三元组（X， Y Z）， X代表当前已经获取资源访问权的线程， Y和Z代表随后并发到达的线程（注意，YZ是并发的），那么综合有以下情况：
(r rr)
(r rw)
(r wr)
(r ww)
(w rr)
(w rw)
(w wr)
(w ww)
另外，当V操作释放信号量时，如果有多个线程阻塞在当前信号量，那么该唤醒哪一个线程呢？有的教程说有队列，先来的先服务（FCFS）；也有文章说，it is undefined，所以不要做任何假设。为了方便后文分析，假设唤醒的线程是随机的。因此上面的八种情况中YZ的值为 rw 和 wr是一样的。
另外YZ为rr 和 ww的情况也很简单，因此实际上需要分析的是下面两种情况：
（r1 r2w2）： 当已经有写线程执行的时候，同时来了一个读线程（w2）和一个写线程(r2)，那么是读线程w2先执行还是写线程r2先执行
（w1 r2w2）：当已经有读线程执行的时候，同时来了一个读线程和一个写线程，那么是读线程w2先执行还是写线程r2先执行
后面的分析都会结合这两种情况。
Linux下信号量API
Linux下提供了信号量相关的API，定义在/usr/include/semaphore.h, 我们常用的是以下四个函数
int sem_init (sem_t *__sem, int __pshared, unsigned int __value);
int sem_destroy (sem_t *__sem);
int sem_wait (sem_t *__sem);
int sem_post (sem_t *__sem);
这些API都是需要配对使用，如果配合C++，可以使用RAII来保证。sem_init初始化信号量，注意第二个参数，代表了该信号量是否与其他进程共享，第三个参数是共享资源的数量。sem_wait即P操作，尝试获取资源；sem_wait即V操作，释放资源。个人感觉sem_wait这个名字不是很友好，让人以为会wait，但事实上不一定会wait，所以感觉叫acquire好一些。
读者优先：
由于读者写者问题运行并发的读操作，写操作必须是互斥的，所以写出来的程序最直观的就是读者优先。源代码（rw_semaphore_rp.cpp）
1 #include <stdio.h> 2 #include <pthread.h> 3 #include <time.h> 4 #include <stdlib.h> 5 #include <unistd.h> 6 #include <semaphore.h> 7 #include <time.h> 8 struct data{ 9 int read_counter; 10 int write_counter; 11 int value; 12 }; 13 14 data share_data; 15 16 int READ_THREAD_NUM = 5; 17 int WRITE_THREAD_NUM = 3; 18 int TOTAL_TRY = 10000; // 尝试这么多次程序就结束吧 19 20 bool stop = false; 21 22 int reader_num = 0; 23 sem_t lock; 24 sem_t lock_writer; 25 sem_t w_or_r; 26 27 28 void* func_read(void *ptr){ 29 while(!stop){ 30 sem_wait(&lock); 31 reader_num += 1; 32 share_data.read_counter += 1; 33 if(reader_num == 1) 34 sem_wait(&w_or_r); 35 sem_post(&lock); 36 37 // do sth with share_data 38 int tmp = share_data.value; 39 40 sem_wait(&lock); 41 reader_num -= 1; 42 if(reader_num == 0) 43 sem_post(&w_or_r); 44 sem_post(&lock); 45 sleep(0.1); 46 } 47 return NULL; 48 } 49 50 void* func_write(void *ptr){ 51 int idx = *(int*)ptr; 52 while(!stop){ 53 #ifdef LOCK_WRITER 54 sem_wait(&lock_writer); 55 #endif 56 sem_wait(&w_or_r); 57 58 share_data.write_counter += 1; 59 share_data.value = idx; 60 61 if(share_data.write_counter >= TOTAL_TRY) 62 stop = true; 63 64 sem_post(&w_or_r); 65 #ifdef LOCK_WRITER 66 sem_post(&lock_writer); 67 #endif 68 69 70 sleep(0.1); 71 } 72 return NULL; 73 } 74 75 int main(int argc, char * argv[]){ 76 share_data.read_counter = 0; 77 share_data.write_counter = 0; 78 79 sem_init(&lock, 0, 1); 80 #ifdef LOCK_WRITER 81 sem_init(&lock_writer, 0, 1); 82 #endif 83 sem_init(&w_or_r, 0, 1); 84 85 pthread_t readers[READ_THREAD_NUM]; 86 pthread_t writers[WRITE_THREAD_NUM]; 87 88 for(int i = 0; i < READ_THREAD_NUM; i++){ 89 pthread_create(&readers[i], NULL, func_read, NULL); 90 } 91 92 int thread_args[WRITE_THREAD_NUM]; 93 for(int i = 0; i < WRITE_THREAD_NUM; i++){ 94 thread_args[i] = i + 1; 95 pthread_create(&writers[i], NULL, func_write, (thread_args + i)); 96 } 97 98 for(int i = 0; i < READ_THREAD_NUM; i++) 99 pthread_join(readers[i],0); 100 101 //for(int i = 0; i < WRITE_THREAD_NUM; i++) 102 // pthread_join(writers[i],0); 103 104 sem_destroy(&lock); 105 #ifdef LOCK_WRITER 106 sem_destroy(&lock_writer); 107 #endif 108 sem_destroy(&w_or_r); 109 printf("Finally read count %d, write count %d\\n", share_data.read_counter, share_data.write_counter); 110 return 0; 111 }
首先忽略掉宏定义LOCK_WRITER, 那么使用的就是lock 和 w_or_r这两个信号量。w_or_r使用来保护对共享资源（读写的资源）的访问，在func_write中对share_data的写操作之前，sem_wait(w_or_r)即信号量的P操作，操作完成之后sem_post。在读线程func_read中，用reader_num记录并发的读线程数量，对这个变量的操作用lock这个信号量来互斥（在并发的读线程之间互斥）。从代码可以看到，只有当第一个读线程发起读操作时(reader_num==1）才会去sem_wait(w_or_r)，而且只有当所有的读线程都结束时(reader_num == 0），才会释放w_or_r。
我们来考虑本章开始提出的问题：
（r1 r2w2），r1（读线程）获取了管程的执行权，那么w2（写线程）一定会阻塞，而r2不用对w_or_r执行P操作，因此可以立即执行，那么后面r2w2的执行顺序一定是r2 然后是 w2
（w1 r2w2），w1获取了管程的执行权，那么r2和w2都一定会阻塞在w_or_r，前面提到多个线程被阻塞时，V操作唤醒的线程是随机的，因此r2 w2各有50%的概率先执行
用以下命令可编译并执行代码 EXE=rw_semaphore && g++ -g -lpthread $EXE.cpp -o $EXE && ./$EXE，下面是执行了5次的结果：
Finally read count 14378, write count 10001
Finally read count 3870, write count 10002
Finally read count 8040, write count 10001
Finally read count 9077, write count 10001
Finally read count 14386, write count 10002
接下来考虑宏定义LOCK_WRITER, 在这种情况下有lock_writer对func_write加锁，所以当写线程正在写数据时，后面来的读线程阻塞在lock_writer，而不是w_or_r。
我们来考虑本章开始提出的问题：
（r1 r2w2），r1（读线程）获取了管程的执行权，那么w2（写线程）一定会阻塞，而r2不用对w_or_r执行P操作，因此可以立即执行，那么后面r2w2的执行顺序一定是r2 然后是 w2
（w1 r2w2），w1获取了管程的执行权，那么r2会阻塞在w_or_r，此时w2阻塞在lock_writer，当w1释放w_or_r时，一定是r2获得w_or_r, 那么后面r2w2的执行顺序一定是r2 然后是 w2
用以下命令可编译并执行代码 EXE=rw_semaphore && g++ -DLOCK_WRITER -g -lpthread $EXE.cpp -o $EXE && ./$EXE，下面是执行了5次的结果：
Finally read count 25983, write count 10001
Finally read count 14378, write count 10001
Finally read count 54344, write count 10001
Finally read count 35147, write count 10001
Finally read count 27105, write count 10002
可以看到，read操作的数量明显高于之前的版本。
写者优先：
按照第一章节对读者写者问题的描述，所谓的写者优先即如果当前是写线程获得了执行权，那么后续并发有读请求和写请求到来的时候，我们优先考虑后续的读请求。为了达到这个目的，我们可以记录当前正在尝试写操作的请求数量，当这个数量大于1时，读操作需要被阻塞，当所有写操作都完成，再通知等到的读操作请求，这个解决思路比较简单，和读者优先的解决方法也很类似，下面是具体的代码(rw_semaphore_wnp.cpp):
1 #include <stdio.h> 2 #include <pthread.h> 3 #include <time.h> 4 #include <stdlib.h> 5 #include <unistd.h> 6 #include <semaphore.h> 7 #include <time.h> 8 struct data{ 9 int read_counter; 10 int write_counter; 11 int value; 12 int value1; 13 }; 14 15 data share_data; 16 17 int READ_THREAD_NUM = 5; 18 int WRITE_THREAD_NUM = 3; 19 int TOTAL_TRY = 10000; // 尝试这么多次程序就结束吧 20 21 bool stop = false; 22 int reader_num = 0; 23 int writer_num = 0; 24 sem_t lock, wlock; 25 sem_t w_or_r; 26 sem_t try_read; 27 28 void* func_read(void *ptr){ 29 while(!stop){ 30 sem_wait(&try_read); 31 32 sem_wait(&lock); 33 34 reader_num += 1; 35 share_data.read_counter += 1; 36 if(reader_num == 1){ 37 sem_wait(&w_or_r); 38 } 39 40 sem_post(&lock); 41 42 sem_post(&try_read); 43 44 // do sth with share_data 45 int tmp = share_data.value; 46 sleep(0); 47 if(tmp != share_data.value1){ 48 printf("Error happens %d %d \\n", tmp, share_data.value1); 49 stop = true; 50 } 51 52 sem_wait(&lock); 53 reader_num -= 1; 54 if(reader_num == 0){ 55 sem_post(&w_or_r); 56 } 57 sem_post(&lock); 58 59 sleep(0.1); 60 } 61 return NULL; 62 } 63 64 void* func_write(void *ptr){ 65 int idx = *(int*)ptr; 66 while(!stop){ 67 sem_wait(&wlock); 68 writer_num += 1; 69 if(writer_num == 1) 70 sem_wait(&try_read); 71 sem_post(&wlock); 72 73 sem_wait(&w_or_r); 74 75 share_data.write_counter += 1; 76 share_data.value = idx; 77 share_data.value1 = idx; 78 79 if(share_data.write_counter >= TOTAL_TRY) 80 stop = true; 81 sem_post(&w_or_r); 82 83 sem_wait(&wlock); 84 writer_num -= 1; 85 if(writer_num == 0) 86 sem_post(&try_read); 87 sem_post(&wlock); 88 sleep(0.1); 89 } 90 return NULL; 91 } 92 93 int main(int argc, char * argv[]){ 94 share_data.read_counter = 0; 95 share_data.write_counter = 0; 96 97 sem_init(&lock, 0, 1); 98 sem_init(&wlock, 0, 1); 99 sem_init(&w_or_r, 0, 1); 100 sem_init(&try_read, 0, 1); 101 102 pthread_t readers[READ_THREAD_NUM]; 103 pthread_t writers[WRITE_THREAD_NUM]; 104 105 for(int i = 0; i < READ_THREAD_NUM; i++){ 106 pthread_create(&readers[i], NULL, func_read, NULL); 107 } 108 109 int thread_args[WRITE_THREAD_NUM]; 110 for(int i = 0; i < WRITE_THREAD_NUM; i++){ 111 thread_args[i] = i + 1; 112 pthread_create(&writers[i], NULL, func_write, (thread_args + i)); 113 } 114 115 for(int i = 0; i < READ_THREAD_NUM; i++) 116 pthread_join(readers[i],0); 117 118 //for(int i = 0; i < WRITE_THREAD_NUM; i++) 119 // pthread_join(writers[i],0); 120 121 sem_destroy(&lock); 122 sem_destroy(&wlock); 123 sem_destroy(&w_or_r); 124 sem_destroy(&try_read); 125 printf("Finally read count %d, write count %d\\n", share_data.read_counter, share_data.write_counter); 126 return 0; 127 }
对比读者优先的代码，主要是增加了一个计数器（writer_num）和两个信号量(wlock, try_read)。writer_num用来记录排队的写线程数目，wlock用来保证writer_num的互斥修改。try_read的作用是保证读线程每次操作之前都得获得这个信号量，而在func_write中，只有当所有的读线程都结束(writer_num == 0)时才会释放try_write这个信号量，这样就达到了读操作优先的目的。
同样我们做一下理论上的分析：
（r1 r2w2），r1（读线程）获取了管程的执行权，此时W2和r2都会try_lock(分别是第70 和 第30行），当r1释放try_read时，唤醒r2还是w2是随机的，因此r2 w2各有50%的概率先执行
（w1 r2w2），w1获取了管程的执行权，那么r2会阻塞在try_read，同事w2阻塞在w_or_r。w1是先释放w_or_r，后释放try_read, 所以w2会先获得执行权， 那么后面r2w2的执行顺序一定是w2 然后是r2
用以下命令可编译并执行代码 EXE=rw_semaphore_wnp && g++ -g -lpthread $EXE.cpp -o $EXE && ./$EXE，下面是执行了5次的结果：
Finally read count 3342, write count 10001
Finally read count 2252, write count 10002
Finally read count 7795, write count 10002
Finally read count 8394, write count 10001
Finally read count 676, write count 10002
读写锁解决读者写者问题
读者写者问题是一个如此普遍的并发问题，因此很多语言为了便于开发者的使用都提供的个高级的抽象，即读写锁。
Linux下读写锁API
int pthread_rwlock_init(pthread_rwlock_t *restrict rwlock, const pthread_rwlockattr_t *restrict attr);
int pthread_rwlock_destroy(pthread_rwlock_t *rwlock);int pthread_rwlock_rdlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_wrlock(pthread_rwlock_t *rwlock);
int pthread_rwlock_unlock(pthread_rwlock_t *rwlock);int pthread_rwlockattr_setkind_np(pthread_rwlockattr_t *attr, int pref);
int pthread_rwlockattr_getkind_np(const pthread_rwlockattr_t *attr, int *pref);
前面两个函数用来初始和销毁读写锁，注意init函数可以指定属性参数（attr）。中间有三个函数用于读写锁的lock与unlock，pthread_rwlock_rdlock用来获取read lock（用于读数据），pthread_rwlock_wrlock用来获取write lock（用来写数据）。最后两个函数用来设置、判断是读者优先，还是写着优先，在pthread.h源码枚举值如下：
enum
{
PTHREAD_RWLOCK_PREFER_READER_NP,
PTHREAD_RWLOCK_PREFER_WRITER_NP,
PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP,
PTHREAD_RWLOCK_DEFAULT_NP = PTHREAD_RWLOCK_PREFER_READER_NP
};
从上面可以看到有三种模式：读者优先，写者优先，写者非递归优先。但坑爹的是，如果设置读者优先（即PTHREAD_RWLOCK_PREFER_WRITER_NP）根本不起作用。如果希望写者优先，那么要使用PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP，这个问题描述可以参见这篇文章和man page。
实现：
基于读写锁rw_lock，解决读者写者问题就比较简单了。先设置好优先选项，然后再写线程中调用pthread_rwlock_wrlock，在读线程中调用pthread_rwlock_rdlock。具体的代码如下(rw_rwlock.cpp)：
1 #include <stdio.h> 2 #include <pthread.h> 3 #include <time.h> 4 #include <stdlib.h> 5 #include <unistd.h> 6 #include <time.h> 7 8 struct data{ 9 int read_counter; 10 int write_counter; 11 int value; 12 }; 13 14 data share_data; 15 16 int READ_THREAD_NUM = 5; 17 int WRITE_THREAD_NUM = 3; 18 int TOTAL_TRY = 10000; // 尝试这么多次程序就结束吧 19 20 bool stop = false; 21 pthread_rwlock_t rwlock; 22 pthread_mutex_t lock; 23 24 void* func_read(void *ptr){ 25 while(!stop){ 26 pthread_rwlock_rdlock(&rwlock); 27 28 pthread_mutex_lock(&lock); 29 share_data.read_counter += 1; 30 pthread_mutex_unlock(&lock); 31 // do sth about value 32 int tmp = share_data.value; 33 34 pthread_rwlock_unlock(&rwlock); 35 sleep(0.1); 36 } 37 return NULL; 38 } 39 40 void* func_write(void *ptr){ 41 int idx = *(int*)ptr; 42 while(!stop){ 43 pthread_rwlock_wrlock(&rwlock); 44 45 share_data.write_counter += 1; 46 share_data.value = idx; 47 if(share_data.write_counter >= TOTAL_TRY) 48 stop = true; 49 pthread_rwlock_unlock(&rwlock); 50 sleep(0.1); 51 52 } 53 return NULL; 54 } 55 56 int main(int argc, char * argv[]){ 57 share_data.read_counter = 0; 58 share_data.write_counter = 0; 59 60 pthread_rwlockattr_t attr; 61 pthread_rwlockattr_init(&attr); 62 63 64 #ifdef WRITER_NONRECURSIVE_NP 65 pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP); 66 #endif 67 68 int np; 69 pthread_rwlockattr_getkind_np(&attr, &np); 70 printf("readnp: %d, writenp: %d, write_nonrecursivenp: %d, and we are using %d\\n", \\ 71 PTHREAD_RWLOCK_PREFER_READER_NP, PTHREAD_RWLOCK_PREFER_WRITER_NP, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP , np); 72 73 pthread_rwlock_init(&rwlock, &attr); 74 75 pthread_mutex_init(&lock, 0); 76 77 pthread_t readers[READ_THREAD_NUM]; 78 pthread_t writers[WRITE_THREAD_NUM]; 79 80 for(int i = 0; i < READ_THREAD_NUM; i++){ 81 pthread_create(&readers[i], NULL, func_read, NULL); 82 } 83 84 int thread_args[WRITE_THREAD_NUM]; 85 for(int i = 0; i < WRITE_THREAD_NUM; i++){ 86 thread_args[i] = i + 1; 87 pthread_create(&writers[i], NULL, func_write, (thread_args + i)); 88 } 89 90 for(int i = 0; i < READ_THREAD_NUM; i++) 91 pthread_join(readers[i],0); 92 93 //for(int i = 0; i < WRITE_THREAD_NUM; i++) 94 // pthread_join(writers[i],0); 95 96 pthread_rwlockattr_destroy(&attr); 97 pthread_rwlock_destroy(&rwlock); 98 pthread_mutex_destroy(&lock); 99 printf("Finally read count %d, write count %d\\n", share_data.read_counter, share_data.write_counter); 100 return 0; 101 }
上面的代码中，为了统计读请求执行的次数，使用了互斥锁(lock)，在编译的时候通过是否预定于WRITER_NONRECURSIVE_NP这个宏来指定是读者优先还是写着优先。
首先是读者优先时得指令 EXE=rw_rwlock && g++ -g -lpthread $EXE.cpp -o $EXE && ./$EXE，输出结果如下（注意，为了统计 所以在读线程中加了互斥锁lock，去掉这个互斥锁 读操作的次数会更多）：
Finally read count 9820, write count 10002
Finally read count 58371, write count 10002
Finally read count 2297, write count 10002
Finally read count 23586, write count 10001
Finally read count 4217, write count 10002
然后是写者优先时得指令 EXE=rw_rwlock && g++ -DWRITER_NONRECURSIVE_NP -g -lpthread $EXE.cpp -o $EXE && ./$EXE，输出结果如下
Finally read count 376, write count 10001
Finally read count 711, write count 10001
Finally read count 928, write count 10001
Finally read count 1468, write count 10001
Finally read count 449, write count 10001
references：
以上是关于用信号量和读写锁解决读者写者问题的主要内容，如果未能解决你的问题，请参考以下文章