哲学家就餐问题

小咪咪 2022-05-22 08:12 303阅读 0赞

# 哲学家就餐问题 #

## 1 问题描述 ##

由Dijkstra提出并解决的哲学家就餐问题是典型的同步问题。该问题描述的是五个哲学家共用一张圆桌，分别坐在周围的五张椅子上，在圆桌上有五个碗和五只筷子，他们的生活方式是交替的进行思考和进餐。平时，一个哲学家进行思考，饥饿时便试图取用其左右最靠近他的筷子，只有在他拿到两只筷子时才能进餐。进餐完毕，放下筷子继续思考。

## 2 解题思路 ##

因为是五位哲学家，并且每位哲学家的各自做自己的事情（思考和吃饭），因此可以创建五个线程表示五位哲学家，五个线程相互独立（异步）。并对五位哲学家分别编号为`0~4`。

同时，有五根筷子，每根筷子只对其相邻的两位哲学家是共享的，因此这五根筷子可以看做是五种不同的临界资源（不是一种资源有5个，因为每根筷子只能被固定编号的哲学家使用）。并对五根筷子分别编号为`0~4`，其中第`i`号哲学家左边的筷子编号为`i`，则其右边的筷子编号就应该为`(i + 1) % 5`。

因为筷子是临界资源，因此当一个线程在使用某根筷子的时候，应该给这根筷子加锁，使其不能被其他进程使用。

根据以上分析，可以使用pthread\_create函数创建五个线程，可以使用pthread\_mutex\_t chops\[5\]表示有五根筷子，五个不同的临界资源，并用pthread\_mutex\_init(&chops\[i\], NULL);来初始化他们。

## 3 问题求解 ##

根据上面的分析，可以得到一个基本的解决方案如下：

void philosopher (void* arg) {
        while (1) {
            think();
            hungry();
            pthread_mutex_lock(&chopsticks[left]);
            pthread_mutex_lock(&chopsticks[right]);
            eat();
            pthread_mutex_unlock(&chopsticks[left]);    
            pthread_mutex_unlock(&chopsticks[right]);
        }
    }

这段伪代码的思路很明确，这个函数代表的是一个哲学家的活动，可以将其创建为五个不同的线程代表五位不同的哲学家。每位哲学家先思考，当某位哲学家饥饿的时候，就拿起他左边的筷子，然后拿起他右边的筷子，然后进餐，然后放下他左右的筷子并进行思考。因为筷子是临界资源，所以当一位哲学家拿起他左右的筷子的时候，就会对他左右的筷子进行加锁，使其他的哲学家不能使用，当该哲学家进餐完毕后，放下了筷子，才对资源解锁，从而使其他的哲学家可以使用。

这个过程看似没什么问题，但是当你仔细分析之后，你会发现这里面有一个很严重的问题，就是死锁，就是每个线程都等待其他线程释放资源从而被唤醒，从而每个线程陷入了无限等待的状态。在哲学家就餐问题中，一种出现死锁的情况就是，假设一开始每位哲学家都拿起其左边的筷子，然后每位哲学家又都尝试去拿起其右边的筷子，这个时候由于每根筷子都已经被占用，因此每位哲学家都不能拿起其右边的筷子，只能等待筷子被其他哲学家释放。由此五个线程都等待被其他进程唤醒，因此就陷入了死锁。

## 4 死锁问题解决 ##

解决死锁问题的办法有很多，下面对各种办法做一下详细的介绍：

### 4.1 Plan A ###

第一种解决死锁问题的办法就是同时只允许四位哲学家同时拿起同一边的筷子，这样就能保证一定会有一位哲学家能够拿起两根筷子完成进食并释放资源，供其他哲学家使用，从而实现永动，避免了死锁。举个最简单的栗子，假定0~3号哲学家已经拿起了他左边的筷子，然后当4号哲学家企图去拿他左边的筷子的时候，将该哲学家的线程锁住，使其拿不到其左边的筷子，然后其左边的筷子就可以被3号哲学家拿到，然后3号哲学家进餐，释放筷子，然后更多的哲学家拿到筷子并进餐。

如何才能实现当4号哲学家企图拿起其左边的筷子的时候将该哲学家的线程阻塞？这个时候就要用到该问题的提出者迪杰斯特拉（这货还提出了迪杰斯特拉最短路径算法，著名的银行家算法也是他发明的）提出的信号量机制。因为同时只允许有四位哲学家同时拿起左筷子，因此我们可以设置一个信号量r，使其初始值为4，然后每当一位哲学家企图去拿起他左边的筷子的时候，先对信号量做一次P操作，从而当第五位哲学家企图去拿做筷子的时候，对r做一次P操作，r = -1，由r < 0得第五位哲学家的线程被阻塞，从而不能拿起左筷子，因此也就避免了死锁问题。然后当哲学家放下他左边的筷子的时候，就对r做一次V操作。

根据上面的分析，代码可以修改为：

void philosopher (void* arg) {
        while (1) {
            think();
            hungry();
            P(&r);//C语言提供的P操作的函数是sem_wait
            pthread_mutex_lock(&chopsticks[left]);
            pthread_mutex_lock(&chopsticks[right]);
            eat();
            pthread_mutex_unlock(&chopsticks[left]);
            V(&r);//C语言提供的V操作的函数是sem_post
            pthread_mutex_unlock(&chopsticks[right]);
        }
    }

### 4.2 Plan B ###

第二种解决死锁问题的办法就是使用AND信号量机制，意思就是如果想给某个哲学家筷子，就将他需要的所有资源都给他，然后让他进餐，否则就一个都不给他。

根据上面的分析，可以将代码修改为：

void philosopher (void* arg) {
        while (1) {
            think();
            hungry();
            Swait(chopsticks[left], chopsticks[right]);
            eat();
            Spost(chopsticks[left], chopsticks[right]);
        }
    }

但是C语言的库里面并没有给提供AND型信号量，但是我们可以利用互斥量简单的替代一下AND信号量，就是设置一个全局互斥量mutex，用来锁住全部的临界资源，当一个哲学家企图拿筷子的时候，就将所有的资源锁住，然后让他去拿他需要的筷子，等他取到他需要的筷子之后，就解锁，然后让其他哲学家取筷子。代码如下：

void philosopher (void* arg) {
        while (1) {
            think();
            hungry();
            pthread_mutex_lock(mutex);
            pthread_mutex_lock(&chopsticks[left]);
            pthread_mutex_lock(&chopsticks[right]);
            pthread_mutex_unlock(mutex);
            eat();
            pthread_mutex_unlock(&chopsticks[left]);
            pthread_mutex_unlock(&chopsticks[right]);
        }
    }

### 4.3 Plan C ###

第三种解决的办法就是规定奇数号哲学家先拿起他左边的筷子，然后再去拿他右边的筷子，而偶数号的哲学家则相反，这样的话总能保证一个哲学家能获得两根筷子完成进餐，从而释放其所占用的资源，代码如下：

void philosopher (void* arg) {
        int i = *(int *)arg;
        int left = i;
        int right = (i + 1) % N;
        while (1) {
            printf("哲学家%d正在思考问题\n", i);
            delay(50000);
    
            printf("哲学家%d饿了\n", i);
            if (i % 2 == 0) {
       //偶数哲学家，先右后左
                pthread_mutex_lock(&chopsticks[right]);
                pthread_mutex_lock(&chopsticks[left]);
                eat();
                pthread_mutex_unlock(&chopsticks[left]);
                pthread_mutex_unlock(&chopsticks[right]);
            } else {
       //奇数哲学家，先左后又
                pthread_mutex_lock(&chopsticks[left]);
                pthread_mutex_lock(&chopsticks[right]);
                eat();
                pthread_mutex_unlock(&chopsticks[right]);
                pthread_mutex_unlock(&chopsticks[left]);
            }
        }
    }

## 5 扩展 ##

这里主要是说一下怎么自己模拟实现互斥量。注意是模拟，不是真的实现。

互斥量主要做的就是当临界区在执行的时候，就将临界区锁住，锁住的目的是使其他需要操作这个临界区中使用的资源的进程阻塞，其实互斥量只要做的就是可以阻塞和唤醒其他的进程，因此我们可以设计下面一种结构来模拟互斥量：

islocked = False
    
    void swap(int *x, int *y) {
        int temp = *x;
        *x = *y;
        *y = temp;
    }
    void philosopher (void* arg) {
        key = True;
        while (1) {
            do {
                swap(&islocked, &key);
            } while (key != False);
    
            //临界区操作
            islocked = False;
        }
    }

这里定义了一个全局变量islocked，表示临界资源是否被锁住，然后当第一次进入临界区之前，执行交换，这时候islocked就为true，表示将临界资源加锁，然后key得到islocked的状态false，表示资源未被锁定，然后就可以退出do…while循环，进而进行临界区操作。这时候当第二个进程来访问临界区资源的时候，同样先获取临界区的状态，获得islocked为true，因此将一直循环，也就是第一个线程通过改变互斥量从而使第二个线程阻塞。直到前一个进程执行完临界区操作，将islocked设置为false，然后第二个进程获得到临界资源的状态可用，就可以退出循环，进而进行临界区操作，也就是第一个线程通过改变互斥量唤醒了第二个线程。

以上就是对互斥量进行简单的模拟，但是这里会有个问题，就是swap函数不是原子操作，在多个线程同时调用swap函数的时候容易产生指令错排从而影响结果，这个没关系，intel X86的指令集中提供了一个交换两数的指令`xchg`，这是一条汇编指令，用法如下：

void swap(int *x, int *y) {
        __asm__("xchgl %0, %1" : "=r" (*x) : "m" (*y));
    }

这样就可以使用一条汇编来进行交换两个数了，就不会产生指令错排的情况了。

## 6 代码 ##

#include <stdio.h>
    #include <stdlib.h>
    #include <malloc.h>
    #include <time.h>
    #include <unistd.h>
    #include <pthread.h>
    #include <semaphore.h>
    
    #define N 5
    
    //信号量使用的参数
    sem_t chopsticks[N];
    sem_t r;
    int philosophers[N] = {
       0, 1, 2, 3, 4};
    
    //swap指令需要的参数
    int islocked[N] = {
       0};
    
    //互斥量使用的参数
    pthread_mutex_t chops[N];
    
    //延迟函数
    void delay (int len) {
        int i = rand() % len;
        int x;
        while (i > 0) {
            x = rand() % len;
            while (x > 0) {
                x--;
            }
            i--;
        }
    }
    
    //交换函数：目前为发现bug
    void xchg(int *x, int *y) {
        __asm__("xchgl %0, %1" : "=r" (*x) : "m" (*y));
    }
    
    //这个函数使用的解决办法是最多允许四个哲学家拿起左筷子
    void philosopher (void* arg) {
        int i = *(int *)arg;
        int left = i;
        int right = (i + 1) % N;
        int leftkey;
        int rightkey;
        while (1) {
            leftkey = 1;
            rightkey = 1;
    
    
            printf("哲学家%d正在思考问题\n", i);
            delay(50000);
    
            printf("哲学家%d饿了\n", i);
            sem_wait(&r);
    
            //sem_wait(&chopsticks[left]);
            //pthread_mutex_lock(&chopsticks[left]);
            do {
                xchg(&leftkey, &islocked[left]);
            }while (leftkey);
            printf("哲学家%d拿起了%d号筷子,现在只有一支筷子,不能进餐\n", i, left);
    
            //sem_wait(&chopsticks[right]);
            //pthread_mutex_lock(&chopsticks[right]);
            do {
                xchg(&rightkey, &islocked[right]);
            }while (rightkey);
            printf("哲学家%d拿起了%d号筷子, 现在有两支筷子,开始进餐\n", i, right);
            delay(50000);
    
            //sem_post(&chopsticks[left]);
            //pthread_mutex_unlock(&chopsticks[left]);
            islocked[left] = 0;
            printf("哲学家%d放下了%d号筷子\n", i, left);
    
            //sem_post(&chopsticks[right]);
            //pthread_mutex_unlock(&chopsticks[right]);
            islocked[right] = 0;
            printf("哲学家%d放下了%d号筷子\n", i, right);
    
            sem_post(&r);
        }
    }
    
    //这个函数使用的解决办法是奇数号哲学家先拿左筷子再拿右筷子，而偶数号哲学家相反。
    void philosopher2 (void* arg) {
        int i = *(int *)arg;
        int left = i;
        int right = (i + 1) % N;
        while (1) {
            printf("哲学家%d正在思考问题\n", i);
            delay(50000);
    
            printf("哲学家%d饿了\n", i);
            if (i % 2 == 0) {
       //偶数哲学家，先右后左
                sem_wait(&chopsticks[right]);
                printf("哲学家%d拿起了%d号筷子,现在只有一支筷子,不能进餐\n", i, right);
                sem_wait(&chopsticks[left]);
                printf("哲学家%d拿起了%d号筷子, 现在有两支筷子,开始进餐\n", i, left);
                delay(50000);
                sem_post(&chopsticks[left]);
                printf("哲学家%d放下了%d号筷子\n", i, left);
                sem_post(&chopsticks[right]);
                printf("哲学家%d放下了%d号筷子\n", i, right);
            } else {
       //奇数哲学家，先左后又
                sem_wait(&chopsticks[left]);
                printf("哲学家%d拿起了%d号筷子, 现在有两支筷子,开始进餐\n", i, left);
                sem_wait(&chopsticks[right]);
                printf("哲学家%d拿起了%d号筷子,现在只有一支筷子,不能进餐\n", i, right);
                delay(50000);
                sem_post(&chopsticks[right]);
                printf("哲学家%d放下了%d号筷子\n", i, right);
                sem_post(&chopsticks[left]);
                printf("哲学家%d放下了%d号筷子\n", i, left);
            }
        }
    }
    
    int main (int argc, char **argv) {
        srand(time(NULL));
        pthread_t PHD[N];
    
    
        for (int i=0; i<N; i++) {
            sem_init(&chopsticks[i], 0, 1);
        }
        sem_init(&r, 0, 4);
    
        for (int i=0; i<N; i++) {
            pthread_mutex_init(&chops[i], NULL);
        }
    
        for (int i=0; i<N; i++) {
            pthread_create(&PHD[i], NULL, (void*)philosopher2, &philosophers[i]);
        }
        for (int i=0; i<N; i++) {
            pthread_join(PHD[i], NULL);
        }
    
        for (int i=0; i<N; i++) {
            sem_destroy(&chopsticks[i]);
        }
        sem_destroy(&r);
        for (int i=0; i<N; i++) {
            pthread_mutex_destroy(&chops[i]);
        }
        return 0;
    }

运行截图：  
![这里写图片描述][70]

## 7 补充：什么是信号量机制 ##

并发编程领域的先锋人物迪杰斯特拉提出了一种经典的解决同步不同执行线程问题的方法，这种方法是基于一种叫做信号量的特殊类型变量的。信号量s是具有非负整数值的全局变量，只能由两种特殊的操作来处理，这两种操作称为P和V：

*  P(s)：如果s是非零的，那么P将s减1，并且立即返回。如果s为零，那么就挂起这个线程，直到s变为非零，而一个V操作会重启这个线程。在重启之后，P操作将s减1，并将控制返回给调用者。
 *  V(s)：V操作将s加1。如果有任何线程阻塞在P操作等待s变成非零，那么V操作会重启这些线程中的一个，然后该线程将s减1，完成它的P操作。

P中的测试和减1操作是不可分割的，也就是说，一旦预测信号量s变成非零，就会将s减1，不能有中断。V中的加1操作也是不可分割的，也就是加载、加1和存储信号量的过程中没有中断。注意，V的定义中没有定义等待线程被重启动的顺序。唯一的要求是V必须只能重启一个正在等待的线程。因此，当有多个线程在等待同一个信号量时，你不能预测V操作要重启哪一个线程。

P和V的定义确保了一个正在运行的程序绝不可能进入这样一种状态，也就是一个正确初始化了的信号量有一个负值。这个属性为信号量不变性，为控制并发程序的轨迹线提供了强有力的工具。

Posix标准定义了许多操作信号量的函数：

#include <semaphore.h>
    
    int sem_init(sem_t *sem, 0, unsigned int value);
    int sem_wait(sem_t *s);    /* P(s) */
    int sem_post(sem_t *s);    /* V(s) */

[70]: /images/20220522/09f646b922d6427e98ac8987452c742a.png