（转）Race condition解决

在很多门课上都接触到race condition， 其中也举了很多方法解决这个问题。于是想来总结一下这些方法。

Race condition
它旨在描述一个系统或者进程的输出依赖于不受控制的事件出现顺序或者出现时机。此词源自于两个信号试着彼此竞争，来影响谁先输出。 
举例来说，如果计算机中的两个进程同时试图修改一个共享内存的内容，在没有并发控制的情况下，最后的结果依赖于两个进程的执行顺序与时机。而且如果发生了并发访问冲突，则最后的结果是不正确的。 
竞争冒险常见于不良设计的电子系统，尤其是逻辑电路。但它们在软件中也比较常见，尤其是有采用多线程技术的软件。 – 维基百科

从维基百科的定义来看，race condition不仅仅是出现在程序中。以下讨论的race conditon全是计算机中多个进程同时访问一个共享内存，共享变量的例子。

Critical Section
要阻止出现race condition情况的关键就是不能让多个进程同时访问那块共享内存。访问共享内存的那段代码就是critical section。所有的解决方法都是围绕这个critical section来设计的。

想要成功的解决race condition问题，并且程序还可以正确运行，从理论上应该满足以下四个条件： 
1、不会有两个及以上进程同时出现在他们的critical section。 
2、不要做任何关于CPU速度和数量的假设。 
3、任何进程在运行到critical section之外时都不能阻塞其他进程。 
4、不会有进程永远等在critical section之前。

互斥
以下将介绍几种实现互斥的方法。(担心翻译不到位，就不翻译这些名字了)

Disabling interrupts(几乎没用)
Lock variables（错误）
Strict alternation（有问题）
Perterson’s solution（有效）
Test and set lock（有效）
Sleep/ Wakeup（有缺陷）
Semaphores(有效)
Mutexes（有效） 
虽然这些方法中很多是有缺陷的，但是考试会有拿这些有问题的版本来问的。
Disabling Interrupts
首先这个方法几乎没有用。不敢兴趣可以直接忽略的。

Time-slicing(时间片)依赖于timer interrupt，如果这个中断被禁用了，那么调度器（scheduler）就不能切换到另一个进程了。 （这个具体就和调度有关了，也许我会写一篇博客来总结一下调度吧）

但是很明显，这个方法是有问题的 
1、这样随意的禁用中断会导致整个系统运行不流畅。比如，如果我们的电脑是这样实现的，那么我在使用编辑器时打字就会很卡，因为当系统里某个进程在critical section时，我打的字产生的interruption就会被忽略，处理这个打字的进程就拿不到我打的字，也就没法在屏幕上显示。 
2、这仅仅在只有一个处理器的电脑上试用，因为禁用中断的本质是将一个CPU中的状态寄存器中关于中断是否可用的那一个bit 置为0。所以当在这个CPU上将此bit置为0了，其他CPU可能就不是0，或者不能保证其他CPU和当前CPU被同时置为0。

Lock variables
这个方法完全不能解决问题，但是还是介绍一下他的思路。

1、有一个全局变量“lock“初始被置为1。 
2、进程1读这个变量，将其置为0然后进入critical section 
3、进程2读lock变量，发现它是0，等待它变成1。 
4、进程1从critical section出来，将lock置为1。 
5、进程2发现lock为1，进入critical section。

这里致命的问题就是会引入对于lock 变量的race condition。

Strict Alternation
这个方法违背了之前提出的四条规则中的第三条，存在一些问题。

假设有两个进程，进程0和进程1

//进程0
while(True){
    while(turn!=0);
    critical_section();
    turn = 1;
    noncritical_section();
}

//进程1
while(True){
    while(turn!=1);
    critical_section();
    turn = 0;
    noncritical_section();
}

这个方法和之前的lock variables是不同的，请仔细体会。首先又一个int型的变量turn，被初始化为0，它用来表示当前应该是由哪个进程来进入critical section。首先进程0检查turn，发现它是0，进入critical section，进程1也来检查，发现turn为0，然后就一直循环检查turn直到它为1，才进入critical section。 
这个方法是可以避免所有的竞争（race）但是违背了规则3。

当进程0离开critical section时，将turn置为1，然后进程1进入critical section（此时进程0在noncritical section中），假设进程1很快的完成critical section，将turn置为0，然后进入noncritical section。此时进程0已经很快的完成了第一个循环，在第二个循环中，检测到turn为0，快速的执行完critical section，将turn置为1，此时两个进程都在noncritical section中，进程0很快的执行完了第二个循环，此时进程0仍然在第一个循环的noncritical section中，这时候问题来了，进程0在第三个循环中，发现turn为1，于是进程0被阻塞在了 循环中。这个违背了第三条规则，任何进程在non critical section中不可以阻塞你其他进程，这个例子中，进程1阻塞了进程0。所以如果进程1的noncritical section执行时间非常长，进程0就得一直等着。

发生这个问题的原因就是，这个算法要求，这两个进程必须交替的进入critical section。所以这个算法叫做 strict alternation。

Perterson’s Solution
有用的方法，并且是纯软件方法实现的。

#define FALSE 0
#define TRUE 1 
#define N 2 /* number of processes */
int turn; /* whose turn is it? */
int interested[N];/* all values initially 0 (FALSE)*/
void enter_region(int process) /* process is 0 or 1 */
{
    int other; /* number of the other process */
    other = 1 - process;  /* the opposite of process */
    interested[process] = TRUE; /* show that you are interested */
    turn = process;/* set flag */
    while (turn == process && interested[other] == TRUE) /* null statement */;
}

void leave_region(int process) /* process: who is leaving */ 
{
    interested[process] = FALSE; /* indicate departure from critical region */ 
}

在访问共享内存前，每个进程都必须调用enter_region，将自己的进程号码作为参数，在这个例子中只有两个进程，进程0和进程1。这个enter_region可能会造成阻塞，当每个进程完成对共享内存的操作之后必须调用leave_region 来表示自己已经结束，并允许其他进程来操作。

来看看这个方法具体怎么工作的： 
1、开始时，没有任何进程在critical section 
2、进程0调用enter_region, 它将interested数组中的自己对应的那项（0）设为TRUE。 
3、如果进程1现在调用enter_region，那么他就必须等待interested[0]变为FALSE。这个只有在进程0调用leave_region才会发生。

来考虑一下，两个进程同时调用enter_region. 
他们都会将turn设为自己进程号(此处不是pid)，但turn只有一个值，后设置完成的值将是turn的值，第一个设置不会成功，假设进程1后设置成功，turn=1。然后它们同时执行while，进程0执行0次，进入critical section，进程1等待。

这个算法只对两个进程有效，如果进程数大于2，我们需要修改这个算法。

//以下代码摘自维基百科

// initialization
level[N] = { -1 };     // current level of processes 0...N-1
waiting[N-1] = { -1 }; // the waiting process of each level 0...N-2

// code for process #i
for(l = 0; l < N-1; ++l) {
    level[i] = l;
    waiting[l] = i;
    while(waiting[l] == i &&
          (there exists k ≠ i, such that level[k] ≥ l)) {
        // busy wait
    }
}

// critical section

level[i] = -1; // exit section

数组level表示每个线程的等待级别，最小为0，最高为N-1，-1表示未设置。数组waiting模拟了一个阻塞（忙等待）的线程队列，从位置0为入队列，位置越大则入队列的时间越长。每个线程为了进入临界区，需要在队列的每个位置都经过一次，如果没有更高优先级的线程（考察数组level），cd 或者被后入队列的线程推着走（上述程序waiting[l] ≠ i），则当前线程在队列中向前走过一个位置。可见该算法满足互斥性。 
由filter算法去反思Peterson算法，可见其中的flags数组表示两个进程的等待级别，而turn变量则是阻塞（忙等待）的线程队列，这个队列只需要容纳一个元素。

以上两段文字也摘自维基百科（突然感觉写博客好累啊。。。）

Test and set lock
现在很多处理器都会支持这样一种指令 
TSL RX, LOCK 
它是这样工作的：他将内存中LOCK（地址）上的内容读到RX寄存器中，然后将一个非0的值存到LOCK地址上。这个读和写的操作是不可分割的，也就是说，其他处理器必须等到这个指令结束了才能访问这个LOCK内存。简而言之，这是通过利用硬件将这个读和写强行变成原子的。

那么现在可以利用这个TSL指令来实现互斥了。以下是enter_region 和 leave_region的汇编的版本。

enter_region:
        TSL REGISTER,LOCK
        CMP REGISTER,#0
        JNE ENTER_REGION
        RET

leave_region:
    MOVE LOCK,#0
    RET

enter_region的第一条指令将LOCK中的值copy到寄存器中，并将LOCK置为1，然后比较这个copy来的值是否为0，如果不是0，就继续回到enter_region，继续测试；如果是0，那么就从enter_region中返回，当进程离开的时候很简单只要将LOCK置为0就可以了。

Sleep/WakeUp
sleep 和 wakeup 是两个system call，也就是一种特殊的函数。它们是这样工作的。

当一个进程发现锁被锁上的时候，调用sleep，当当前进程的状态置为blocked。当另一个进程离开critical section的时候，释放锁，调用wake函数，将因为这个锁被block的进程放到ready queue中。（这又涉及到进程调度的问题了，ready queue中的进程都处于ready状态，调度器根据调度算法选择一个执行）

这个虽然听起来是个不错的方法，但是有可能导致著名的生产者消费者问题。

void producer(void) {
    int item;
    while (TRUE){
          item = produce_item();
          if (count == N) sleep();
          insert_item(item);
          count = count + 1;
          if (count == 1) wakeup(consumer);
    }
}

void consumer(void) {
    int item;
    while (TRUE){
        if (count == 0) sleep();
        item = remove_item();
        count = count - 1;
        if (count ==N - 1) wakeup(producer);
        consume_item(item);
    }
}

假设此时count = 0， consumer正在执行，当consumer执行到if (count == 0)时，调度器选择producer执行，producer生产了一个iterm，insert，然后发现count=1，调用wakeup，但是此时consumer并不是sleep状态，所以这个wakeup函数没起作用，然后调度器选择consumer执行，由于之前consumer停在if(count==0)然后consumer 进入sleep，然后producer一直生产，等生产了N个之后producer也进入sleep，这样两个进程就都sleep了，并且没有人会唤醒它们。

Semaphores
这个方法的提出者是Edsger Wybe Dijkstra，就是那个牛逼的一匹的Dijkstra，最短路算法的那个Dijkstra算法就是这个Dijkstra的。

Semaphores(信号量)是一个特殊的锁变量用来记录wakeup的数量。 
对信号量有两个原子性的操作，DOWN 和UP（注意原子性）： 
DOWN：如果信号量的值大于0，DOWN将信号量的值减1，并返回。如果信号量=0，DOWN操作被阻塞。 
UP：如果有任何进程被阻塞在DOWN操作，选择一个唤醒，如果没有进程被阻塞在DOWN，那么就将信号量加1.

再来看看使用信号量是如何解决生产者消费者问题的。

#define N 100
typedef int semaphore;
semaphore mutex = 1;
semaphore empty = N;
semaphore full = 0;
void producer(void) {
     int item;
     while (TRUE){
          item = produce_item();
          down(&empty);
          down(&mutex);
          insert_item(item);
          up(&mutex);
          up(&full);
} }
void consumer(void) {
     int item;
     while (TRUE){
          down(&full);
          down(&mutex);
          item = remove_item();
          up(&mutex);
          up(&empty);
          consume_item(item);
} }

这里有三个信号量，full, empty, mutex. 
full: 未被消费掉的物品的数量，也就是buffer中物品量，初始为0 
empty：还可以生产的物品的数量，也就是buffer中空闲的数量，初始为N 
mutex：用来控制生产者消费者不会同时访问buffer。

用信号量解决生产者消费者问题网上有很多资料，我就不细说了。

Mutex
Mutex实际上是Semaphore的简化版本，当不需要计数功能时就是mutex，比如上面的mutex信号量就是这个的用法。

总结
以上只是简单的介绍一下各个方法是如何工作的，接下来会给一些总结。

Busy waiting
Perterson‘s Solution, TSL，都有一个busy waiting的问题。给busy waiting一个定义就是：连续的测试一个变量直到它变成某一个值的现象。

不管是C版本的还是汇编版本的，我们可以看到当一个进程无法进入到critical section时，他就一直在while循环。浪费CPU时间来执行这样没有意义的事情是需要我们避免的。

IPC
那么后面的信号量 sleep/wakeup为什么没有busy waiting呢？因为它们都用到了进程间通信(InterProcess Communication,IPC)

纯软件实现
perterson’s solution是纯软件实现的哦，TSL用到了硬件，信号量那个说明了DOWN 和UP是原子性的，但是在写这篇博客的时候我还不知道它是如何实现原子性的。

信号量的问题
不要以为信号量就那么的简单，如果使用不当会造成死锁的，具体怎么造成死锁的网上应该有资料，不想写了。

生产者消费者问题
刚刚那个版本的信号量对于生产者消费者问题来说，即便有多个生产者消费者都可以正确运行，但是我记得有些算法是只能针对一个生产者一个消费者的，具体的我想不起来了。
--------------------- 
作者：Alex267 
来源：CSDN 
原文：https://blog.csdn.net/AlexZhang67/article/details/56287299 
版权声明：本文为博主原创文章，转载请附上博文链接！