I.读者-写者问题

1.读者-写者问题和分析

• 读者: 读数据
• 写者: 写数据
• 访问规则:多个读者可以同时读数据，任何时刻只能有一个写者写数据，一个读者与一个写者不能同时在相应的临界区中
• 实质: 一个写者不能与其它的读者或写者同时访问相应的临界资源。

  这算是一个非常经典的问题，实际上这也是数据库系统中遇到的最重要的一个问题，因为对于同一个文件，可以多人读，但是不能多人写，如何才能更好地安排资源的分配呢？
  对于这个问题首先肯定应该给出一个基本解法，那就是保证读者可以读，写者在一定情况下也能写，因此比较常用的是这样一种解法：首先有一把排他锁(写锁)，只有一个线程可以获得这把排他锁，并且有若干把共享锁(读锁)，共享锁可以被任意读线程获取，但是这些线程都不能写，不过如果用信号量实现的话，我们总是需要一个可能无限大的变量进行锁的分配，因此可以改变一下思路：用一个变量记录读者数量，当读者为0时允许获取写锁，而当读者不为0时记录数量，并且不允许获取写锁，因为这个变量本身属于临界区，因此只需要对这个变量加一把互斥锁即可。

2.读者—写者问题基本解法

  在1中，我们已经分析了这个问题的一种基本解法，因此可以用C语言实现如下的代码：

#include <stdio.h>
#include <semaphore.h>
#include <pthread.h>
#define P sem_wait
#define V sem_post
#define RNUMS 10
#define WNUMS 3
sem_t mutex, x;
int readers_cnt;
int rids[RNUMS], wids[WNUMS];void init()
{sem_init(&mutex, 0, 1);sem_init(&x, 0, 1);readers_cnt = 0;for (int i = 0; i < RNUMS; i++) {rids[i] = i + 1;} for (int i = 0; i < WNUMS; i++) {wids[i] = i + 1;} 
}void* Treader(void* ID)
{while (1) {P(&mutex);readers_cnt++;if (readers_cnt == 1) P(&x);V(&mutex);printf("Reader %d: read\n", *(int*)ID);P(&mutex);readers_cnt--;if (readers_cnt == 0) V(&x);V(&mutex);}pthread_exit(NULL);
}void* Twriter(void* ID)
{while (1) {P(&x);printf("Writer %d: write\n", *(int*)ID);V(&x);}pthread_exit(NULL);
}int main()
{init();pthread_t ws[WNUMS], rs[RNUMS];for (int i = 0; i < WNUMS; i++) {pthread_create(&ws[i], NULL, Twriter, &wids[i]);}for (int i = 0; i < RNUMS; i++) {pthread_create(&rs[i], NULL, Treader, &rids[i]);}for (int i = 0; i < WNUMS; i++) {pthread_join(ws[i], NULL);}for (int i = 0; i < RNUMS; i++) {pthread_join(rs[i], NULL);}return 0;
}

  非常简单的实现，对于当前设定的10个读者和3个写者的情况下，观察一段时间输出会发现：几乎根本就不存在写者写入的情况，这是由于读者的锁很明显比排他锁更好获得，对于更易达成的条件，从概率和期望的角度上来说，实现的次数也应该会更多

  这就有问题了：写者想要获取排他锁，看样子非常困难，甚至说，我们可以构造一个读顺序，让写者几乎不可能获取到排他锁：有五个线程，四个读线程和一个写线程，两个读线程为了保障读取到的数据永远是最新的，总是会每隔一分钟读取一次，但是非常巧妙的是，这四个读线程始终有一小段时间是重合的，在这种情况下，因为引用计数一直不能清零，所以排他锁一直不能被获取，此时就造成了写者的饥饿问题

3.饥饿现象和解决方案

  上面已经相对比较详细地描述了饥饿现象的产生，那么一个解决方案是这样的：我既然饥饿现象起源于读者和写者获取锁的难度不公平，那我们就让二者再次公平，在写者全程加上wait_mutex锁，在读者试图增加读者数量的时候也加上wait_mutex锁，因此我们可以写出下面的代码：

#include <stdio.h>
#include <semaphore.h>
#include <pthread.h>
#define P sem_wait
#define V sem_post
#define RNUMS 10
#define WNUMS 1
sem_t mutex, x, wait_mutex;
int readers_cnt;
int rids[RNUMS], wids[WNUMS];void init()
{sem_init(&mutex, 0, 1);sem_init(&x, 0, 1);sem_init(&wait_mutex, 0, 1);readers_cnt = 0;for (int i = 0; i < RNUMS; i++) {rids[i] = i + 1;} for (int i = 0; i < WNUMS; i++) {wids[i] = i + 1;} 
}void* Treader(void* ID)
{while (1) {P(&wait_mutex);P(&mutex);readers_cnt++;if (readers_cnt == 1) P(&x);V(&mutex);V(&wait_mutex);printf("Reader %d: read\n", *(int*)ID);P(&mutex);readers_cnt--;if (readers_cnt == 0) V(&x);V(&mutex);}pthread_exit(NULL);
}void* Twriter(void* ID)
{while (1) {P(&wait_mutex);P(&x);printf("Writer %d: write\n", *(int*)ID);V(&x);V(&wait_mutex);}pthread_exit(NULL);
}int main()
{init();pthread_t ws[WNUMS], rs[RNUMS];for (int i = 0; i < WNUMS; i++) {pthread_create(&ws[i], NULL, Twriter, &wids[i]);}for (int i = 0; i < RNUMS; i++) {pthread_create(&rs[i], NULL, Treader, &rids[i]);}for (int i = 0; i < WNUMS; i++) {pthread_join(ws[i], NULL);}for (int i = 0; i < RNUMS; i++) {pthread_join(rs[i], NULL);}return 0;
}

  在原先代码的基础上简单加上了一个新的wait_mutex作为写者的特权，当写时就会申请wait_mutex作为特权，因此读者和写者在初始状态需要竞争这把互斥锁，在写者竞争到后，读者就无法继续操作，无法增加读者数量，直到写线程的写结束，由此一来发现，由于写者和读者的竞争再次公平，因此写者的写入次数明显提升，对比原先代码的写读比1:10的情况，后者的结果中写入次数明显增加：

总结

  读者—写者问题是一类非常经典的问题，实际上代表了一系列的文件共享的互斥问题，在数据库系统当中，经常性地查表和写表也属于读者—写者问题的实例，数据库还会采取更多的措施来增加数据库的并发效率，因为目前我们的解决方案上的锁实际上是整个数据库的锁，为了同时让更多读写操作能够进行，数据库采取了表级锁和行级锁这些更加细粒度的锁，例如同处在同一个文件的一张表中有两行数据，但是一个事务读取行1，而另一个事务写入行2，这两个操作实际上不会冲突，而采取简单的锁直接锁住会明显降低这个操作的效率

II.Peterson算法实现

1.Peterson算法问题与分析

(1).如何无锁访问临界区呢？

  当两个进程/线程希望访问同一个临界区的时候，应该怎么让这两个线程能够在不发生冲突的情况下获得临界区的数据呢？在Peterson算法之前出现的大部分解决方案实际上都不能完美地解决问题，虽然Peterson算法本身也只能用于解决两个进程之间的互斥问题，但它的确完成了任务。

(2).Peterson算法的基本逻辑

  所以Peterson算法本身究竟做了什么呢？Peterson算法融合了意向和严格轮换的想法，对于希望访问临界区的变量，首先它需要将自己的访问意向设置为true，在这之后，有两种方案（本质一样，只是实际意义不同）：将自己的进程代号放进turn变量，将对方的进程代号放进turn变量，在这两个步骤之后，每个线程就可以去检测对方是否有意向访问 && 目前轮到了对方访问，如果这个条件满足，则需要循环等待。

(3).写对方/自己进程号的区别是？

  所以把对方的和自己的进程号写入turn变量的区别在哪呢？其实比较简单，因为对于同一个变量的写入有一个先来后到的顺序，如果两个进程均写入对方的进程号，则手快的进程会优先把对方进程号写在turn中，手慢的会在自己进程号已经被写入turn之后，把对方的进程号再写入turn当中，这种情况下，进程对于临界区的访问是抢占式的，也就是谁的速度更快，谁就能抢到临界区进行访问。

  而两个进程均写入自己的进程号则是遵循了让步的原则，因为对于上述的情况，手快的进程会优先把自己的进程号写在turn中，手慢的则会在对方进程号已经被写入turn之后，把自己的进程号再写入turn当中，这种情况下，相当于手快的进程把临界区的访问权让给了手慢的进程。

  所以这两种方案其实区别不是很大，只要我们没有一个进程放自己进程号，一个进程放对方进程号，就不会出现很大的问题。

2.只包含意向的解决方案

  所以我们知道只包含意向的解决方案是会出问题的，我们可以用下面的代码进行尝试：

#include <stdio.h>
#include <pthread.h>
#include <stdbool.h>
bool flags[2] = {false, false};void* T1()
{while (true) {flags[0] = true;while (flags[1]);printf("T1 access!\n");flags[0] = false; }pthread_exit(NULL);
}void* T2()
{while (true) {flags[1] = true;while (flags[0]);printf("T2 access!\n");flags[1] = false; }pthread_exit(NULL);
}int main()
{pthread_t p1, p2;pthread_create(&p1, NULL, T1, NULL);pthread_create(&p2, NULL, T2, NULL);pthread_join(p1, NULL);pthread_join(p2, NULL);return 0;
}

  非常好代码，T1访问了一次就顺利地进入了死锁的阶段：

  我用死锁这个词可能都不太好，因为这个解决方案连锁都没有用，开个玩笑。那到底为什么会出这个问题呢？其实很简单，因为我们在等待的时候用了一个完全没法完成互斥的操作—我们只关注对方是不是true，如果true就不访问，那如果两个线程都有意向访问，相当于谁都进不去，换言之，这个方法完全没有解决互斥问题，只是检测是否别人可能在访问罢了。

3.严格轮换法

  这个方法看起来好像要正确一点点，它的代码如下：

#include <stdio.h>
#include <pthread.h>
#include <stdbool.h>
int turn;void* T1()
{while (true) {while (turn != 0);printf("T1 access!\n");turn = 1; }pthread_exit(NULL);
}void* T2()
{while (true) {while (turn != 1);printf("T2 access!\n");turn = 0;}pthread_exit(NULL);
}int main()
{turn = 0;pthread_t p1, p2;pthread_create(&p1, NULL, T1, NULL);pthread_create(&p2, NULL, T2, NULL);pthread_join(p1, NULL);pthread_join(p2, NULL);return 0;
}

  从结果上看：它好像还真没出问题，严格轮换法的确可以让不同的进程轮流访问临界区，但问题在于，这种方法会在轮不到某个进程的时候让进程持续进入轮询阶段，这会造成CPU的忙等待，浪费了CPU的资源，这种策略之下，快的进程总是不能优先地完成任务，从而造成一定的浪费和调度问题。

4.完整的Peterson算法

  来吧，让竞争更激烈一点。Peterson算法实际上完成了上述两种方法的融合，它的代码实现如下：

#include <stdio.h>
#include <pthread.h>
#include <stdbool.h>
#include <unistd.h>
bool flags[2] = {false, false};
int turn;
int cnt = 0;void* T1()
{while (true) {flags[0] = true;turn = 1;while (flags[1] && turn == 1);sleep(1);printf("T1 access! cnt = %d\n", cnt++);flags[0] = false;}pthread_exit(NULL);
}void* T2()
{while (true) {flags[1] = true;turn = 0;while (flags[0] && turn == 0);printf("T2 access!\n");flags[1] = false;}pthread_exit(NULL);
}int main()
{pthread_t p1, p2;pthread_create(&p1, NULL, T1, NULL);pthread_create(&p2, NULL, T2, NULL);pthread_join(p1, NULL);pthread_join(p2, NULL);return 0;
}

  结果看起来已经很正确了，它基本完成了Peterson算法的思想，好像问题到这儿就解决了，对吗？这台机器运行在x86-64架构的处理器下，实际上x86-64架构的CPU保证了至少对于int类型的load和store操作是原子的，如果它们不是原子的，会发生什么？具体的实例我没有查到相关的资料，我也没有成功复现出来，之后我可能还会继续研究一下。

总结

  Peterson算法的确是真正通过软件的方式完成了临界区互斥访问的问题，不过编译器并不一定能够让我们的指令依照顺序执行，编译器可能会对我们写的代码进行乱序，而这样的乱序可能导致load和store指令顺序调整而导致Peterson算法失效，我们可能需要使用：

__sync_synchronize();
// 或者
asm("mfence");

  利用内存屏障指令从而避免对指令顺序进行优化，从而避免出现关键指令乱序执行的问题，所以Peterson算法的实现可能还是需要一些特别的软硬件结合以避免出现乱序的问题。

参考资料

• 1.并发控制：基础 (Peterson 算法、模型检验、原子操作)
• 2.博客园—内存栅栏（memory barrier）：解救peterson算法的应用陷阱
  
• 3.知乎—对int变量赋值的操作是原子的吗？